Page MenuHomec4science
Files F60510319

trained_model_reduced_basis.py
No OneTemporary
Actions

File Metadata

Created
Tue, Apr 30, 18:17

trained_model_reduced_basis.py
View Options

# Copyright (C) 2018 by the RROMPy authors
#
# This file is part of RROMPy.
#
# RROMPy is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# RROMPy is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with RROMPy. If not, see <http://www.gnu.org/licenses/>.
#
import numpy as np
from collections.abc import Iterable
from rrompy.reduction_methods.base.trained_model.trained_model import (
TrainedModel)
from rrompy.reduction_methods.standard.reduced_basis_utils import (
projectAffineDecomposition)
from rrompy.utilities.base.types import (Np1D, ListAny, paramVal, paramList,
sampList)
from rrompy.utilities.base import verbosityManager as vbMng, freepar as fp
from rrompy.utilities.numerical.compress_matrix import compressMatrix
from rrompy.utilities.numerical.marginalize_poly_list import (
marginalizePolyList)
from rrompy.utilities.numerical.nonlinear_eigenproblem import (
eigvalsNonlinearDense)
from rrompy.utilities.expression import expressionEvaluator
from rrompy.utilities.exception_manager import RROMPyException, RROMPyWarning
from rrompy.parameter import checkParameter
from rrompy.sampling import sampleList
from rrompy.utilities.parallel import (poolRank, masterCore, listScatter,
matrixGatherv, isend, recv)
__all__ = ['TrainedModelReducedBasis']
class TrainedModelReducedBasis(TrainedModel):
"""
ROM approximant evaluation for RB approximant.
Attributes:
Data: dictionary with all that can be pickled.
"""
def reset(self):
super().reset()
if hasattr(self, "data") and hasattr(self.data, "lastSetupMu"):
self.data.lastSetupMu = None
def compress(self, collapse : bool = False, tol : float = 0., *args,
**kwargs):
if collapse:
raise RROMPyException("Cannot collapse implicit surrogates.")
if tol <= 0.: return
if hasattr(self.data, "_compressTol"):
RROMPyWarning(("Recompressing already compressed model is "
"ineffective. Aborting."))
return
self.data.projMat, RMat, _ = compressMatrix(self.data.projMat, tol,
*args, **kwargs)
self.data.ARBs, self.data.bRBs = projectAffineDecomposition(
self.data.ARBs, self.data.bRBs, RMat)
super().compress(collapse, tol)
def assembleReducedModel(self, mu:paramVal):
mu = checkParameter(mu, self.data.npar)
if not (hasattr(self.data, "lastSetupMu")
and self.data.lastSetupMu == mu):
vbMng(self, "INIT", "Assembling reduced model for mu = {}."\
.format(mu), 17)
muEff = self.mapParameterList(mu)
self.data.ARBmu, self.data.bRBmu = 0., 0.
for thA, ARB in zip(self.data.thAs, self.data.ARBs):
self.data.ARBmu = (expressionEvaluator(thA[0], muEff) * ARB
+ self.data.ARBmu)
for thb, bRB in zip(self.data.thbs, self.data.bRBs):
self.data.bRBmu = (expressionEvaluator(thb[0], muEff) * bRB
+ self.data.bRBmu)
vbMng(self, "DEL", "Done assembling reduced model.", 17)
self.data.lastSetupMu = mu
def getApproxReduced(self, mu : paramList = []) -> sampList:
"""
Evaluate reduced representation of approximant at arbitrary parameter.
Args:
mu: Target parameter.
"""
mu = self.checkParameterList(mu)
if (not hasattr(self, "lastSolvedApproxReduced")
or self.lastSolvedApproxReduced != mu):
vbMng(self, "INIT",
"Computing RB solution at mu = {}.".format(mu), 12)
mu, _, sizes = listScatter(mu, return_sizes = True)
mu = self.checkParameterList(mu)
req, emptyCores = [], np.where(np.logical_not(sizes))[0]
if len(mu) == 0:
vbMng(self, "MAIN", "Idling.", 37)
uL, uT = recv(source = 0, tag = poolRank())
uApproxR = np.empty((uL, 0), dtype = uT)
else:
for j, mj in enumerate(mu):
self.assembleReducedModel(mj)
uAppR = np.linalg.solve(self.data.ARBmu, self.data.bRBmu)
if j == 0:
uApproxR = np.empty((len(uAppR), len(mu)),
dtype = uAppR.dtype)
if masterCore():
for dest in emptyCores:
req += [isend((len(uAppR), uAppR.dtype),
dest = dest, tag = dest)]
uApproxR[:, j] = uAppR
for r in req: r.wait()
uApproxR = matrixGatherv(uApproxR, sizes)
self.uApproxReduced = sampleList(uApproxR)
vbMng(self, "DEL", "Done computing RB solution.", 12)
self.lastSolvedApproxReduced = mu
return self.uApproxReduced
def getPoles(self, marginalVals : ListAny = [fp], jSupp : int = 1,
**kwargs) -> Np1D:
"""
Obtain approximant poles.
Returns:
Numpy complex vector of poles.
"""
if not self.data.affinePoly:
RROMPyWarning(("Unable to compute approximate poles due "
"to parametric dependence (detected non-"
"polynomial). Change HFEngine.affinePoly to True "
"if necessary."))
return
if not isinstance(marginalVals, Iterable):
marginalVals = [marginalVals]
mVals = list(marginalVals)
rDim = mVals.index(fp)
if rDim < len(mVals) - 1 and fp in mVals[rDim + 1 :]:
raise RROMPyException(("Exactly 1 'freepar' entry in "
"marginalVals must be provided."))
ARBs = self.data.ARBs
if self.data.npar > 1:
mVals[rDim] = self.data.mu0(rDim)
mVals = checkParameter(mVals, return_data = True).flatten()
mVals[rDim] = fp
ARBs = marginalizePolyList(ARBs, mVals, "auto")
ev = eigvalsNonlinearDense(ARBs, jSupp = jSupp, **kwargs)
return self.mapParameterList(ev, "B", [rDim])(0)

Event Timeline

c4science · Help