diff --git a/examples/4_funnel_output/funnel_output.py b/examples/4_funnel_output/funnel_output.py
index 323d845..57a1eff 100644
--- a/examples/4_funnel_output/funnel_output.py
+++ b/examples/4_funnel_output/funnel_output.py
@@ -1,57 +1,61 @@
import numpy as np
from funnel_output_engine import FunnelOutputEngine as engine
from rrompy.reduction_methods import (NearestNeighbor as NN,
RationalInterpolant as RI,
RationalInterpolantGreedy as RIG)
from rrompy.parameter.parameter_sampling import (QuadratureSampler as QS,
EmptySampler as ES)
ks = [5., 10.]
k0, n = np.mean(ks), 50
solver = engine(k0, n)
k = 6.5
for method in ["RI", "RI_GREEDY"]:
print("Testing {} method".format(method))
if "GREEDY" not in method:
params = {'S':20, 'POD':True, 'polybasis':"CHEBYSHEV",
'sampler':QS(ks, "CHEBYSHEV")}
algo = RI
if "GREEDY" in method:
params = {'S':2, 'POD':True, 'polybasis':"LEGENDRE", 'greedyTol':1e-1,
'maxIter':25, 'sampler':QS(ks, "UNIFORM"),
'errorEstimatorKind':"LOOK_AHEAD_OUTPUT"}
algo = RIG
approx = algo(solver, mu0 = k0, approxParameters = params, verbosity = 5)
if "GREEDY" not in method:
approx.setupApprox()
else:
approx.setupApprox("LAST")
print("--- Approximant ---")
approx.plotApprox(k, name = 'u_app')
approx.plotHF(k, name = 'u_HF')
approx.plotErr(k, name = 'err_app')
- normErr = approx.normErr(k)[0]
- normSol = approx.normHF(k)[0]
+ err = approx.getErr(k)[0]
+ sol = approx.getHF(k)[0]
+ normErr = np.abs(solver.L2NormMatrix.dot(err).dot(err.conj())) ** .5
+ normSol = np.abs(solver.L2NormMatrix.dot(sol).dot(sol.conj())) ** .5
print("SolNorm:\t{:.5e}\nErr_app: \t{:.5e}\nErrRel_app:\t{:.5e}".format(
normSol, normErr, normErr / normSol))
print("--- Closest snapshot ---")
approxNN = NN(solver, mu0 = k0, verbosity = 0,
approxParameters = {'S':approx.samplingEngine.nsamples,
'POD':True, 'sampler':ES()})
approxNN.setSamples(approx.samplingEngine)
approxNN.plotApprox(k, name = 'u_close')
approxNN.plotHF(k, name = 'u_HF')
approxNN.plotErr(k, name = 'err_close')
- normErr = approxNN.normErr(k)[0]
- normSol = approxNN.normHF(k)[0]
+ err = approxNN.getErr(k)[0]
+ sol = approxNN.getHF(k)[0]
+ normErr = np.abs(solver.L2NormMatrix.dot(err).dot(err.conj())) ** .5
+ normSol = np.abs(solver.L2NormMatrix.dot(sol).dot(sol.conj())) ** .5
print("SolNorm:\t{:.5e}\nErr_close:\t{:.5e}\nErrRel_close:\t{:.5e}".format(
normSol, normErr, normErr / normSol))
print("Poles:\n{}".format(approx.getPoles()))
print("\n")
diff --git a/examples/4_funnel_output/funnel_output_engine.py b/examples/4_funnel_output/funnel_output_engine.py
index e32bc08..6a34633 100644
--- a/examples/4_funnel_output/funnel_output_engine.py
+++ b/examples/4_funnel_output/funnel_output_engine.py
@@ -1,36 +1,36 @@
import numpy as np
import fenics as fen
import mshr
from rrompy.hfengines.fenics_engines import ScatteringProblemEngine
from rrompy.solver.fenics import fenics2Sparse
class FunnelOutputEngine(ScatteringProblemEngine):
def __init__(self, k0:float, n:int):
super().__init__(mu0 = [k0])
mesh = mshr.generate_mesh(mshr.Circle(fen.Point(0., 0.), 5.)
- mshr.Rectangle(fen.Point(-5., -5.),
fen.Point(-1., 5.))
- mshr.Rectangle(fen.Point(-1., -5.),
fen.Point(0., -1.))
- mshr.Rectangle(fen.Point(-1., 1.),
fen.Point(0., 5.)), n)
self.V = fen.FunctionSpace(mesh, "P", 1)
eps = 1e-4
self.DirichletBoundary = (lambda x, on_boundary:
on_boundary and x[0] < -1. + eps)
self.NeumannBoundary = (lambda x, on_boundary:
on_boundary and x[0] >= -1. + eps
and x[0] < eps)
self.RobinBoundary = "REST"
y = fen.SpatialCoordinate(self.V.mesh())[1]
self.DirichletDatum = 1. + .25 * fen.sin(.5 * np.pi * y)
self.autoSetDS()
l2R0 = fen.inner(self.u, self.v) * self.ds(1)
L2R0 = fenics2Sparse(l2R0, {}, None, -1)
bcR = np.where(np.abs(L2R0.dot(np.ones(L2R0.shape[1]))) > 1e-10)[0]
bcR = bcR[np.argsort(self.V.tabulate_dof_coordinates()[bcR, 1])]
self._C = np.zeros((len(bcR), L2R0.shape[1]))
self._C[np.arange(len(bcR)), bcR] = 1.
- self.outputNormMatrix = L2R0[bcR][:, bcR]
+ self.L2NormMatrix = L2R0[bcR][:, bcR]
self.cutOffPolesIMax = 0.
diff --git a/examples/9_active_remeshing/active_remeshing.py b/examples/9_active_remeshing/active_remeshing.py
index f247787..7af5b1d 100755
--- a/examples/9_active_remeshing/active_remeshing.py
+++ b/examples/9_active_remeshing/active_remeshing.py
@@ -1,118 +1,129 @@
import numpy as np
from pickle import load
from matplotlib import pyplot as plt
from active_remeshing_engine import ActiveRemeshingEngine
from rrompy.reduction_methods import (
- RationalInterpolantGreedyPivotedNoMatch as RIGPNM,
- RationalInterpolantGreedyPivotedGreedyPoleMatch as RIGPG)
+ RationalInterpolantGreedyPivotedNoMatch as RIGPNM,
+ RationalInterpolantGreedyPivotedPoleMatch as RIGPG)
from rrompy.parameter.parameter_sampling import (QuadratureSampler as QS,
SparseGridSampler as SGS)
zs, ts = [0., 100.], [0., .5]
z0, t0, n = np.mean(zs), np.mean(ts), 150
solver = ActiveRemeshingEngine(z0, t0, n)
+solver.cutOffPolesRMinRel, solver.cutOffPolesRMaxRel = -2.5, 2.5
+solver.cutOffPolesIMin, solver.cutOffPolesIMax = -.01, .01
mus = [[z0, ts[0]], [z0, ts[1]]]
for mu in mus:
u = solver.solve(mu, return_state = True)[0]
- Y = solver.applyC(u, mu)[0][0]
+ Y = solver.applyC(u, mu)[0]
_ = solver.plot(u, what = "REAL", name = "u(z={}, t={})".format(*mu),
is_state = True, figsize = (12, 4))
- print("Y(z={}, t={}) = {} (solution norm squared)".format(*mu, Y))
+ print("Y(z={}, t={}) = {} (solution average)".format(*mu, np.real(Y)))
fighandles = []
with open("./active_remeshing_hf_samples.pkl", "rb") as f:
zspace, tspace, Yex = load(f)
-# zspace = np.linspace(zs[0], zs[-1], 198)
+# zspace = np.linspace(zs[0], zs[-1], 200)
# tspace = np.linspace(ts[0], ts[-1], 50)
-# Yex = np.log10([[solver.solve([z, t]) for t in tspace] for z in zspace])
-# (from a ~2.5h simulation on one node of the EPFL Helvetios cluster)
+# Yex = [[solver.solve([z, t]) for t in tspace] for z in zspace]
+# (from a ~2h45m simulation on one node of the EPFL Helvetios cluster)
+Yex = np.clip(Yex, np.quantile(Yex, .05), np.quantile(Yex, .95))
+Ymin, Ymax = np.min(Yex), np.max(Yex)
-for match in [0, 1]:
+approx = []
+for match in range(3):
params = {"POD": True, "S": 5, "greedyTol": 1e-4, "nTestPoints": 500,
"polybasis": "LEGENDRE", "trainSetGenerator": QS(zs, "UNIFORM"),
- "samplerPivot":QS(zs, "CHEBYSHEV"), "samplerMarginal":SGS(ts),
+ "samplerPivot": QS(zs, "CHEBYSHEV"), "SMarginal": 5,
+ "samplerMarginal": SGS(ts),
"errorEstimatorKind": "LOOK_AHEAD_OUTPUT"}
if match:
- print("\nTesting output-based matching with weight 1.")
- params["SMarginal"] = 3
- params["maxIterMarginal"] = 25
- params["greedyTolMarginal"] = 1e-2
+ if match == 1:
+ print("\nTesting output-based matching.")
+ else: #if match == 2:
+ print("\nTesting output-based matching in chordal metric.")
+ params["matchingChordalRadius"] = [1., "AUTO"]
params["matchingWeight"] = 1.
params["matchingShared"] = .75
params["polybasisMarginal"] = "PIECEWISE_LINEAR_UNIFORM"
- params["errorEstimatorKindMarginal"] = "LOOK_AHEAD_RECOVER"
algo = RIGPG
else:
print("\nTesting matching-free approach.")
- params["SMarginal"] = 5
algo = RIGPNM
- approx = algo([0], solver, mu0 = [z0, t0], approxParameters = params,
- verbosity = 15)
- approx.setupApprox("ALL" * match)
+ approx += [algo([0], solver, mu0 = [z0, t0], approxParameters = params,
+ verbosity = 5)]
+ if match:
+ approx[match].setTrainedModel(approx[0])
+ else:
+ approx[match].setupApprox()
- verb, verbTM = approx.verbosity, approx.trainedModel.verbosity
- approx.verbosity, approx.trainedModel.verbosity = 0, 0
+ verb = approx[match].verbosity
+ verbTM = approx[match].trainedModel.verbosity
+ approx[match].verbosity, approx[match].trainedModel.verbosity = 0, 0
for j, t in enumerate(tspace):
- out = approx.getApprox(np.pad(zspace.reshape(-1, 1), [(0, 0), (0, 1)],
- "constant", constant_values = t))
- pls = approx.getPoles([None, t])
+ out = approx[match].getApprox(np.pad(zspace.reshape(-1, 1),
+ [(0, 0), (0, 1)], "constant",
+ constant_values = t))
+ pls = approx[match].getPoles([None, t])
pls[np.abs(np.imag(pls)) > 1e-5] = np.nan
if j == 0:
Ys = np.empty((len(zspace), len(tspace)))
poles = np.empty((len(tspace), len(pls)))
- Ys[:, j] = np.log10(out.data)
+ Ys[:, j] = out.re.data
if len(pls) > poles.shape[1]:
poles = np.pad(poles, [(0, 0), (0, len(pls) - poles.shape[1])],
"constant", constant_values = np.nan)
poles[j, : len(pls)] = np.real(pls)
- approx.verbosity, approx.trainedModel.verbosity = verb, verbTM
- Ymin, Ymax = min(np.min(Ys), np.min(Yex)), max(np.max(Ys), np.max(Yex))
+ approx[match].verbosity = verb
+ approx[match].trainedModel.verbosity = verbTM
+ Ys = np.clip(Ys, np.quantile(Ys, .05), np.quantile(Ys, .95))
fighandles += [plt.figure(figsize = (15, 5))]
ax1 = fighandles[-1].add_subplot(1, 2, 1)
ax2 = fighandles[-1].add_subplot(1, 2, 2)
if match:
ax1.plot(poles, tspace)
else:
ax1.plot(poles, tspace, "k.")
ax1.set_ylim(ts)
ax1.set_xlabel("z")
ax1.set_ylabel("t")
ax1.grid()
if match:
ax2.plot(poles, tspace)
else:
ax2.plot(poles, tspace, "k.")
- for mm in approx.musMarginal:
+ for mm in approx[match].musMarginal:
ax2.plot(zs, [mm[0, 0]] * 2, "k--", linewidth = 1)
ax2.set_xlim(zs)
ax2.set_ylim(ts)
ax2.set_xlabel("z")
ax2.set_ylabel("t")
ax2.grid()
plt.show()
print("Approximate poles")
fighandles += [plt.figure(figsize = (15, 5))]
ax1 = fighandles[-1].add_subplot(1, 2, 1)
ax2 = fighandles[-1].add_subplot(1, 2, 2)
p = ax1.contourf(np.repeat(zspace.reshape(-1, 1), len(tspace), axis = 1),
np.repeat(tspace.reshape(1, -1), len(zspace), axis = 0),
- Ys, vmin = Ymin, vmax = Ymax, cmap = "gray_r",
+ Ys, vmin = Ymin, vmax = Ymax,
levels = np.linspace(Ymin, Ymax, 50))
plt.colorbar(p, ax = ax1)
ax1.set_xlabel("z")
ax1.set_ylabel("t")
ax1.grid()
p = ax2.contourf(np.repeat(zspace.reshape(-1, 1), len(tspace), axis = 1),
np.repeat(tspace.reshape(1, -1), len(zspace), axis = 0),
- Yex, vmin = Ymin, vmax = Ymax, cmap = "gray_r",
+ Yex, vmin = Ymin, vmax = Ymax,
levels = np.linspace(Ymin, Ymax, 50))
ax2.set_xlabel("z")
ax2.set_ylabel("t")
ax2.grid()
plt.colorbar(p, ax = ax2)
plt.show()
print("Approximate and exact output\n")
diff --git a/examples/9_active_remeshing/active_remeshing_engine.py b/examples/9_active_remeshing/active_remeshing_engine.py
index 41c09e0..31d2faf 100644
--- a/examples/9_active_remeshing/active_remeshing_engine.py
+++ b/examples/9_active_remeshing/active_remeshing_engine.py
@@ -1,90 +1,88 @@
import numpy as np
import ufl
import fenics as fen
import mshr
from rrompy.utilities.base.decorators import (pivot_affine_construct,
mupivot_independent)
from rrompy.hfengines.fenics_engines import HelmholtzProblemEngine
from rrompy.utilities.numerical.hash_derivative import (
hashDerivativeToIdx as hashD)
from rrompy.solver.fenics import fenZERO, fenONE, fenics2Vector
from rrompy.parameter import parameterMap as pMap
from rrompy.utilities.exception_manager import RROMPyException
class ActiveRemeshingEngine(HelmholtzProblemEngine):
def __init__(self, z0:float, t0:float, n:int):
super().__init__(mu0 = [z0, t0])
self._affinePoly = False
self._nMesh = n
self.meshGen(t0)
self.parameterMap = pMap(1., 2)
self.DirichletBoundary = lambda x, on_boundary: (on_boundary
and np.abs(x[0]) >= .75 - 1e-5
or np.abs(x[1]) >= .5 - 1e-5)
self.NeumannBoundary = "REST"
- self.setCQuadratic(2)
self.cutOffPolesIMin, self.cutOffPolesIMax = -1e-2, 1e-2
def meshGen(self, t:float):
t = np.real(t)
if (not hasattr(self, "_tMesh") or not np.isclose(self._tMesh, t)):
e, self._tMesh = .01, t
tipx, tipy = .5 * np.sin(t), .5 - .5 * np.cos(t)
tiplx, tiply = tipx + .5 * e * np.cos(t), tipy + .5 * e * np.sin(t)
tiprx, tipry = tipx - .5 * e * np.cos(t), tipy - .5 * e * np.sin(t)
basx, basy = - e * np.sin(t), .5 + e * np.cos(t)
baslx, basly = basx + .5 * e * np.cos(t), basy + .5 * e * np.sin(t)
basrx, basry = basx - .5 * e * np.cos(t), basy - .5 * e * np.sin(t)
mesh = mshr.generate_mesh(
mshr.Rectangle(fen.Point(-.75, -.5), fen.Point(.75, .5))
- mshr.Polygon([fen.Point(tiprx, tipry),
fen.Point(tiplx, tiply),
fen.Point(baslx, basly),
fen.Point(basrx, basry)])
- mshr.Circle(fen.Point(tipx, tipy), .5 * e), self._nMesh)
self.V = fen.FunctionSpace(mesh, "P", 1)
self.As, self._C = [None] * 2, None
self.autoSetDS()
if hasattr(self, "energyNormMatrix"):
del self.energyNormMatrix
if hasattr(self, "energyNormDualMatrix"):
del self.energyNormDualMatrix
def getForcingTerm(self, mu = []):
mu = self.checkParameter(mu)
self.meshGen(mu(0, 1))
x, y = fen.SpatialCoordinate(self.V.mesh())[:]
rightZone = .1875**-2 * ufl.conditional(ufl.And(
ufl.And(ufl.ge(x, -.5625), ufl.le(x, -.375)),
ufl.And(ufl.ge(y, .125), ufl.le(y, .3125))),
fenONE, fenZERO)
return rightZone, fenZERO
@pivot_affine_construct
def A(self, mu = [], der = 0):
mu = self.checkParameter(mu)
self.meshGen(mu(0, 1))
return HelmholtzProblemEngine.A(self, mu, der)
@pivot_affine_construct
def b(self, mu = [], der = 0):
derI = hashD(der) if hasattr(der, "__len__") else der
if derI < 0: return self.baselineb()
if derI > 0: raise Exception("Derivatives not implemented.")
if self.thbs[0] is None: self.thbs = self.getMonomialWeights(self.nbs)
fen0 = self.getForcingTerm(mu)[0] * self.v * fen.dx
DBC = fen.DirichletBC(self.V, fenZERO, self.DirichletBoundary)
self.bs = [fenics2Vector(fen0, {}, DBC, 1)]
return HelmholtzProblemEngine.b(self, mu, der)
@mupivot_independent
def C(self, mu):
mu = self.checkParameterList(mu)
if not np.all(np.isclose(mu(1), mu(0, 1))):
raise RROMPyException(("Simultaneous evaluation of C on multiple "
"meshes not supported."))
self.meshGen(mu(0, 1))
if self._C is None:
- self.buildEnergyNormForm()
- self._C = self.energyNormMatrix
+ self._C = fenics2Vector(self.v * fen.dx, {}).reshape(1, -1) / 1.5
return self._C
diff --git a/examples/9_active_remeshing/active_remeshing_hf_samples.pkl b/examples/9_active_remeshing/active_remeshing_hf_samples.pkl
index 86fc869..92b87c3 100644
Binary files a/examples/9_active_remeshing/active_remeshing_hf_samples.pkl and b/examples/9_active_remeshing/active_remeshing_hf_samples.pkl differ
diff --git a/rrompy/hfengines/base/hfengine_base.py b/rrompy/hfengines/base/hfengine_base.py
index 1250741..dde9851 100644
--- a/rrompy/hfengines/base/hfengine_base.py
+++ b/rrompy/hfengines/base/hfengine_base.py
@@ -1,427 +1,351 @@
# Copyright (C) 2018-2020 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 .
#
from abc import abstractmethod
import numpy as np
import scipy.sparse as scsp
from numbers import Number
from collections.abc import Iterable
from copy import copy as softcopy
from rrompy.utilities.base.decorators import (nonaffine_construct,
mu_independent)
from rrompy.utilities.base.types import (Np1D, Np2D, List, DictAny, paramVal,
paramList, sampList)
from rrompy.utilities.numerical import solve as tsolve, dot
from rrompy.utilities.expression import expressionEvaluator
-from rrompy.utilities.exception_manager import RROMPyException, RROMPyAssert
+from rrompy.utilities.exception_manager import RROMPyAssert
from rrompy.sampling.sample_list import sampleList
from rrompy.parameter import (checkParameter, checkParameterList,
parameterList, parameterMap as pMap)
from rrompy.solver.linear_solver import setupSolver
from rrompy.utilities.parallel import (poolRank, masterCore, listScatter,
matrixGatherv, isend, recv)
__all__ = ['HFEngineBase']
class HFEngineBase:
"""Generic solver for parametric problems."""
def __init__(self, verbosity : int = 10, timestamp : bool = True):
self.verbosity = verbosity
self.timestamp = timestamp
self.setSolver("SPSOLVE", {"use_umfpack" : False})
self.npar = 0
self._C = None
- self.outputNormMatrix = 1.
def name(self) -> str:
return self.__class__.__name__
def __str__(self) -> str:
return self.name()
def __repr__(self) -> str:
return self.__str__() + " at " + hex(id(self))
def __dir_base__(self):
return [x for x in self.__dir__() if x[:2] != "__"]
def __deepcopy__(self, memo):
return softcopy(self)
@property
def npar(self):
"""Value of npar."""
return self._npar
@npar.setter
def npar(self, npar):
nparOld = self._npar if hasattr(self, "_npar") else -1
if npar != nparOld:
self.parameterMap = pMap(1., npar)
self._npar = npar
@property
def spacedim(self):
return 1
def checkParameter(self, mu:paramVal) -> paramVal:
muP = checkParameter(mu, self.npar)
if self.npar == 0: muP.reset((1, 0), muP.dtype)
return muP
def checkParameterList(self, mu:paramList,
check_if_single : bool = False) -> paramList:
muL = checkParameterList(mu, self.npar, check_if_single)
return muL
def mapParameterList(self, mu:paramList, direct : str = "F",
idx : List[int] = None) -> paramList:
if idx is None: idx = np.arange(self.npar)
muMapped = checkParameterList(mu, len(idx))
for j, d in enumerate(idx):
muMapped.data[:, j] = expressionEvaluator(
self.parameterMap[direct][d],
muMapped(j)).flatten()
return muMapped
def buildEnergyNormForm(self):
"""
Build sparse matrix (in CSR format) representative of scalar product.
"""
self.energyNormMatrix = 1.
def buildEnergyNormDualForm(self):
"""
Build sparse matrix (in CSR format) representative of dual scalar
product without duality.
"""
self.energyNormDualMatrix = 1.
def innerProduct(self, u:Np2D, v:Np2D, onlyDiag : bool = False,
dual : bool = False, is_state : bool = True) -> Np2D:
"""Scalar product."""
if is_state or self.isCEye:
if dual:
if not hasattr(self, "energyNormDualMatrix"):
self.buildEnergyNormDualForm()
energyMat = self.energyNormDualMatrix
else:
if not hasattr(self, "energyNormMatrix"):
self.buildEnergyNormForm()
energyMat = self.energyNormMatrix
else:
- energyMat = self.outputNormMatrix
+ energyMat = 1.
if isinstance(u, (parameterList, sampleList)): u = u.data
if isinstance(v, (parameterList, sampleList)): v = v.data
if onlyDiag:
return np.sum(dot(energyMat, u) * v.conj(), axis = 0)
return dot(dot(energyMat, u).T, v.conj()).T
def norm(self, u:Np2D, dual : bool = False,
is_state : bool = True) -> Np1D:
return np.abs(self.innerProduct(u, u, onlyDiag = True, dual = dual,
is_state = is_state)) ** .5
def baselineA(self):
"""Return 0 of shape consistent with operator of linear system."""
if (hasattr(self, "As") and isinstance(self.As, Iterable)
and self.As[0] is not None):
d = self.As[0].shape[0]
else:
d = self.spacedim
return scsp.csr_matrix((np.zeros(0), np.zeros(0), np.zeros(d + 1)),
shape = (d, d), dtype = np.complex)
def baselineb(self):
"""Return 0 of shape consistent with RHS of linear system."""
return np.zeros(self.spacedim, dtype = np.complex)
@nonaffine_construct
@abstractmethod
def A(self, mu : paramVal = [], der : List[int] = 0) -> Np2D:
"""
Assemble terms of operator of linear system and return it (or its
derivative) at a given parameter.
"""
return
@nonaffine_construct
@abstractmethod
def b(self, mu : paramVal = [], der : List[int] = 0) -> Np1D:
"""
Assemble terms of RHS of linear system and return it (or its
derivative) at a given parameter.
"""
return
@mu_independent
def C(self, mu:paramVal):
"""
Value of C. Should be overridden (with something like
return self._C(mu)
) if a mu-dependent C is needed.
"""
if self._C is None: self._C = 1.
return self._C
- def setCQuadratic(self, quad : bool = 1):
- if isinstance(quad, (str,)):
- quad = quad.upper().strip().replace(" ","")
- if quad == "SELFADJOINT":
- quad = 2
- elif quad in ["STANDARD", "GENERAL"]:
- quad = 1
- else:
- raise RROMPyException(("String keyword for output symmetry "
- "not recognized."))
- if quad:
- self._is_C_quadratic = int(quad)
- elif hasattr(self, "_is_C_quadratic"):
- del self._is_C_quadratic
-
@property
def isCEye(self):
"""
Whether the action of C can be seen as a scalar multiplication. Should
be overridden (with
return True
) if a mu-dependent scalar C is used.
"""
return isinstance(self._C, Number)
def applyC(self, u:sampList, mu:paramVal):
"""Apply LHS of linear system."""
- if not (hasattr(self, "_is_C_quadratic") and self._is_C_quadratic):
- return dot(self.C(mu), u)
- return self._applyCQuadratic(u, mu)
-
- def _applyCQuadratic(self, u:sampList, mu:paramVal, v : sampList = None,
- onlyDiag : bool = False):
- """Apply quadratic LHS of linear system."""
- if not (hasattr(self, "_is_C_quadratic") and self._is_C_quadratic):
- raise RROMPyException(("Cannot call quadratic output routine if "
- "output is not quadratic."))
- C = self.C(mu)
- is_C_list = isinstance(C, list)
- if ((is_C_list and np.any([Ci.ndim != 2 for Ci in C]))
- or not (is_C_list or C.ndim in [2, 3])):
- raise RROMPyException(("C array for quadratic output must have 2 "
- "or 3 dimensions or be list of "
- "2-dimensional arrays."))
- symmetry = v is None # computing u^H * C * u
- selfadjoint = self._is_C_quadratic == 2 and symmetry
- if isinstance(u, sampleList): u = u.data
- if isinstance(v, sampleList): v = v.data
- while u.ndim < 2: u = np.expand_dims(u, -1)
- if v is None: v = u
- while v.ndim < 2: v = np.expand_dims(v, -1)
- N, M = u.shape[1], v.shape[1]
- if onlyDiag:
- if N != M:
- raise RROMPyException(("Cannot extract diagonal of "
- "rectangular output."))
- Ncol = (N,)
- elif symmetry:
- Ncol = (N * (N + 1) // 2,) if selfadjoint else (N ** 2,)
- else:
- Ncol = (N, M)
- for j in range(N):
- Rv = v[:, j] if onlyDiag else v[:, j :] if selfadjoint else v
- if is_C_list:
- cj = np.array([[dot(dot(Ci, u[:, j]), Rv.conj())] for Ci in C])
- else:
- cj = dot(dot(C, u[:, j]), Rv.conj())
- if C.ndim == 2: cj = np.expand_dims(cj, 0)
- if selfadjoint and (onlyDiag or N == 1): cj = np.real(cj)
- if j == 0: res = np.empty((len(cj),) + Ncol, dtype = cj[0].dtype)
- if not onlyDiag and symmetry: # map 2D idx to hierarchical 1D idx
- if self._is_C_quadratic == 2:
- res[:, j * (j + 1) // 2] = cj[:, 0]
- iD = np.arange(j + 1, N) * np.arange(j + 4, N + 3) // 2 - j
- res[:, iD] = 2 * cj[:, 1 :] # exploit symmetry
- else:
- res[:, j ** 2 : j * (j + 1)] = cj[:, : j]
- iD = np.arange(j, N) * np.arange(j + 2, N + 2) - j
- res[:, iD] = cj[:, j :]
- else:
- res[:, j] = cj
- return res
+ return dot(self.C(mu), u)
def setSolver(self, solverType:str, solverArgs : DictAny = {}):
"""Choose solver type and parameters."""
self._solver, self._solverArgs = setupSolver(solverType, solverArgs)
def solve(self, mu : paramList = [], RHS : sampList = None,
return_state : bool = False) -> sampList:
"""
Find solution of linear system.
Args:
mu: parameter value.
RHS: RHS of linear system. If None, defaults to that of parametric
system. Defaults to None.
return_state: whether to return state before multiplication by c.
Defaults to False.
"""
from rrompy.sampling import sampleList, emptySampleList
if mu == []: mu = self.mu0
mu = self.checkParameterList(mu)
if len(mu) == 0: return emptySampleList()
mu = self.checkParameterList(mu)
mu_loc, idx, sizes = listScatter(mu, return_sizes = True)
mu_loc = self.checkParameterList(mu_loc)
req, emptyCores = [], np.where(np.logical_not(sizes))[0]
if len(mu_loc) == 0:
uL, uT = recv(source = 0, tag = poolRank())
sol = np.empty((uL, 0), dtype = uT)
else:
- applyCglob = (hasattr(self, "_is_C_quadratic")
- and self._is_C_quadratic)
if RHS is None: # build RHSs
RHS = sampleList([self.b(m) for m in mu_loc])
else:
RHS = sampleList(RHS)
if len(RHS) > 1: RHS = sampleList([RHS[i] for i in idx])
mult = 0 if len(RHS) == 1 else 1
RROMPyAssert(mult * (len(mu_loc) - 1) + 1, len(RHS), "Sample size")
for j, mj in enumerate(mu_loc):
u = tsolve(self.A(mj), RHS[mult * j], self._solver,
self._solverArgs)
- if not (return_state or applyCglob): u = self.applyC(u, mj)
+ if not return_state: u = self.applyC(u, mj)
if j == 0:
sol = np.empty((len(u), len(mu_loc)), dtype = u.dtype)
if masterCore():
for dest in emptyCores:
req += [isend((len(u), u.dtype), dest = dest,
tag = dest)]
sol[:, j] = u
for r in req: r.wait()
sol = matrixGatherv(sol, sizes)
- if not return_state and applyCglob: sol = self.applyC(sol, mu)
return sampleList(sol)
def residual(self, mu : paramList = [], u : sampList = None,
post_c : bool = True) -> sampList:
"""
Find residual of linear system for given approximate solution.
Args:
mu: parameter value.
u: numpy complex array with function dofs. If None, set to 0.
post_c: whether to post-process using c. Defaults to True.
"""
from rrompy.sampling import sampleList, emptySampleList
if mu == []: mu = self.mu0
mu = self.checkParameterList(mu)
if len(mu) == 0: return emptySampleList()
mu_loc, idx, sizes = listScatter(mu, return_sizes = True)
mu_loc = self.checkParameterList(mu_loc)
req, emptyCores = [], np.where(np.logical_not(sizes))[0]
if len(mu_loc) == 0:
uL, uT = recv(source = 0, tag = poolRank())
res = np.empty((uL, 0), dtype = uT)
else:
- applyCglob = (hasattr(self, "_is_C_quadratic")
- and self._is_C_quadratic)
v = sampleList(np.zeros((self.spacedim, len(mu_loc))))
if u is not None:
u = sampleList(u)
v = v + sampleList([u[i] for i in idx])
for j, (mj, vj) in enumerate(zip(mu_loc, v)):
r = self.b(mj) - dot(self.A(mj), vj)
- if post_c and not applyCglob: r = self.applyC(r, mj)
+ if post_c: r = self.applyC(r, mj)
if j == 0:
res = np.empty((len(r), len(mu_loc)), dtype = r.dtype)
if masterCore():
for dest in emptyCores:
req += [isend((len(r), r.dtype), dest = dest,
tag = dest)]
res[:, j] = r
for r in req: r.wait()
res = matrixGatherv(res, sizes)
- if post_c and applyCglob: res = self.applyC(res, mu)
return sampleList(res)
cutOffPolesRMax,cutOffPolesRMin = np.inf, - np.inf
cutOffPolesRMaxRel, cutOffPolesRMinRel = np.inf, - np.inf
cutOffPolesIMax, cutOffPolesIMin = np.inf, - np.inf
cutOffPolesIMaxRel, cutOffPolesIMinRel = np.inf, - np.inf
cutOffResNormMin = -1
cutOffResAngleMin, cutOffResAngleMax = -1, np.pi + 1
@property
def _ignoreResidues(self):
return (self.cutOffResNormMin <= 0. and self.cutOffResAngleMin <= 0.
and self.cutOffResAngleMax >= np.pi)
def flagBadPolesResidues(self, poles:Np1D, residues : Np1D = None,
relative : bool = False,
projMat : Np2D = None) -> Np1D:
"""
Flag (numerical) poles/residues which are impossible.
Args:
poles: poles to be judged.
residues: residues norms to be judged.
relative: whether relative values should be used for poles.
projMat: matrix for projection of residues.
"""
poles = np.array(poles).flatten()
flag = np.zeros(len(poles), dtype = bool)
if relative:
RMax, RMin = self.cutOffPolesRMaxRel, self.cutOffPolesRMinRel
IMax, IMin = self.cutOffPolesIMaxRel, self.cutOffPolesIMinRel
else:
RMax, RMin = self.cutOffPolesRMax, self.cutOffPolesRMin
IMax, IMin = self.cutOffPolesIMax, self.cutOffPolesIMin
if not np.isinf(RMax):
flag = np.logical_or(flag, np.real(poles) > RMax)
if not np.isinf(RMin):
flag = np.logical_or(flag, np.real(poles) < RMin)
if not np.isinf(IMax):
flag = np.logical_or(flag, np.imag(poles) > IMax)
if not np.isinf(IMin):
flag = np.logical_or(flag, np.imag(poles) < IMin)
if residues is not None and not self._ignoreResidues:
residues = np.array(residues).reshape(-1, len(poles))
if projMat is None:
resNorm = np.linalg.norm(residues, axis = 0)
else:
residues = projMat.dot(residues)
resNorm = self.norm(residues)
if self.cutOffResNormMin > 0.:
resNormEff = resNorm / np.max(resNorm)
flag = np.logical_or(flag, resNormEff < self.cutOffResNormMin)
if self.cutOffResAngleMin > 0. or self.cutOffResAngleMax < np.pi:
if projMat is None:
angles = np.real(residues.T.conj().dot(residues))
else:
angles = np.real(self.innerProduct(residues, residues))
resNormEff = resNorm
resNormEff[np.isclose(resNormEff, 0., atol = 1e-15)] = 1.
angles = np.clip((angles / resNormEff).T / resNormEff, -1., 1.)
angles = np.arccos(angles)
badangles = np.logical_or(angles < self.cutOffResAngleMin,
angles > self.cutOffResAngleMax)
badangles[np.arange(len(angles)), np.arange(len(angles))] = 0
idx = np.zeros(len(angles), dtype = bool)
while np.sum(badangles) > 0:
idxn = np.argmax(np.sum(badangles, axis = 1))
badangles[idxn], badangles[:, idxn] = 0, 0
idx[idxn] = 1
flag = np.logical_or(flag, idx)
return flag
diff --git a/rrompy/reduction_methods/base/generic_approximant.py b/rrompy/reduction_methods/base/generic_approximant.py
index 7eafcc1..5b3a81d 100644
--- a/rrompy/reduction_methods/base/generic_approximant.py
+++ b/rrompy/reduction_methods/base/generic_approximant.py
@@ -1,889 +1,883 @@
# Copyright (C) 2018-2020 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 .
#
from abc import abstractmethod
import numpy as np
from collections.abc import Iterable
from itertools import product as iterprod
from copy import deepcopy as copy
from os import remove as osrm
from rrompy.sampling import (SamplingEngine, SamplingEngineNormalize,
SamplingEnginePOD)
from rrompy.utilities.base.types import (Np1D, DictAny, HFEng, List, Tuple,
ListAny, strLst, paramVal, paramList,
sampList)
from rrompy.utilities.base.data_structures import purgeDict, getNewFilename
from rrompy.utilities.base import verbosityManager as vbMng
from rrompy.utilities.exception_manager import (RROMPyException, RROMPyAssert,
RROMPy_READY, RROMPy_FRAGILE)
from rrompy.utilities.base.pickle_utilities import pickleDump, pickleLoad
from rrompy.parameter import (emptyParameterList, checkParameter,
checkParameterList)
from rrompy.sampling import sampleList, emptySampleList
from rrompy.utilities.parallel import (bcast, masterCore, listGather,
listScatter)
__all__ = ['GenericApproximant']
def addNormFieldToClass(self, fieldName):
def objFunc(self, mu:paramList, *args, **kwargs) -> Np1D:
uV = getattr(self.__class__, "get" + fieldName)(self, mu)
kwargs["is_state"] = False
val = self.HFEngine.norm(uV, *args, **kwargs)
return val
setattr(self.__class__, "norm" + fieldName, objFunc)
def addNormDualFieldToClass(self, fieldName):
def objFunc(self, mu:paramList, *args, **kwargs) -> Np1D:
uV = getattr(self.__class__, "get" + fieldName)(self, mu)
kwargs["is_state"] = True
if "dual" not in kwargs.keys(): kwargs["dual"] = True
val = self.HFEngine.norm(uV, *args, **kwargs)
return val
setattr(self.__class__, "norm" + fieldName, objFunc)
def addPlotFieldToClass(self, fieldName):
def objFunc(self, mu:paramList, *args, **kwargs):
uV = getattr(self.__class__, "get" + fieldName)(self, mu)
uV = listScatter(uV)[0].T
filesOut = []
if len(uV) > 0:
if "name" in kwargs.keys(): nameBase = copy(kwargs["name"])
for j, u in enumerate(uV):
if "name" in kwargs.keys(): kwargs["name"] = nameBase + str(j)
filesOut += [self.HFEngine.plot(u, *args, **kwargs)]
if "name" in kwargs.keys(): kwargs["name"] = nameBase
filesOut = listGather(filesOut)
if filesOut[0] is None: return None
return filesOut
setattr(self.__class__, "plot" + fieldName, objFunc)
def addPlotDualFieldToClass(self, fieldName):
def objFunc(self, mu:paramList, *args, **kwargs):
uV = getattr(self.__class__, "get" + fieldName)(self, mu)
uV = listScatter(uV)[0].T
filesOut = []
if len(uV) > 0:
if "name" in kwargs.keys(): nameBase = copy(kwargs["name"])
for j, u in enumerate(uV):
if "name" in kwargs.keys(): kwargs["name"] = nameBase + str(j)
filesOut += [self.HFEngine.plot(u, *args, **kwargs)]
if "name" in kwargs.keys(): kwargs["name"] = nameBase
filesOut = listGather(filesOut)
if filesOut[0] is None: return None
return filesOut
setattr(self.__class__, "plot" + fieldName, objFunc)
def addOutParaviewFieldToClass(self, fieldName):
def objFunc(self, mu:paramVal, *args, **kwargs):
if not hasattr(self.HFEngine, "outParaview"):
raise RROMPyException(("High fidelity engine cannot output to "
"Paraview."))
uV = getattr(self.__class__, "get" + fieldName)(self, mu)
uV = listScatter(uV)[0].T
filesOut = []
if len(uV) > 0:
if "name" in kwargs.keys(): nameBase = copy(kwargs["name"])
for j, u in enumerate(uV):
if "name" in kwargs.keys(): kwargs["name"] = nameBase + str(j)
filesOut += [self.HFEngine.outParaview(u, *args, **kwargs)]
if "name" in kwargs.keys(): kwargs["name"] = nameBase
filesOut = listGather(filesOut)
if filesOut[0] is None: return None
return filesOut
setattr(self.__class__, "outParaview" + fieldName, objFunc)
def addOutParaviewTimeDomainFieldToClass(self, fieldName):
def objFunc(self, mu:paramVal, *args, **kwargs):
if not hasattr(self.HFEngine, "outParaviewTimeDomain"):
raise RROMPyException(("High fidelity engine cannot output to "
"Paraview."))
uV = getattr(self.__class__, "get" + fieldName)(self, mu)
uV = listScatter(uV)[0].T
filesOut = []
if len(uV) > 0:
omega = args.pop(0) if len(args) > 0 else np.real(mu)
if "name" in kwargs.keys(): nameBase = copy(kwargs["name"])
filesOut = []
for j, u in enumerate(uV):
if "name" in kwargs.keys(): kwargs["name"] = nameBase + str(j)
filesOut += [self.HFEngine.outParaviewTimeDomain(u, omega,
*args,
**kwargs)]
if "name" in kwargs.keys(): kwargs["name"] = nameBase
filesOut = listGather(filesOut)
if filesOut[0] is None: return None
return filesOut
setattr(self.__class__, "outParaviewTimeDomain" + fieldName, objFunc)
class GenericApproximant:
"""
ABSTRACT
ROM approximant computation for parametric problems.
Args:
HFEngine: HF problem solver.
mu0(optional): Default parameter. Defaults to 0.
approxParameters(optional): Dictionary containing values for main
parameters of approximant. Recognized keys are:
- 'POD': kind of snapshots orthogonalization; allowed values
include 0, 1/2, and 1; defaults to 1, i.e. full POD;
- 'scaleFactorDer': scaling factors for derivative computation;
defaults to 'AUTO';
- 'S': total number of samples current approximant relies upon.
Defaults to empty dict.
verbosity(optional): Verbosity level. Defaults to 10.
Attributes:
HFEngine: HF problem solver.
trainedModel: Trained model evaluator.
mu0: Default parameter.
approxParameters: Dictionary containing values for main parameters of
approximant. Recognized keys are in parameterList{Soft,Critical}.
parameterListSoft: Recognized keys of soft approximant parameters:
- 'POD': kind of snapshots orthogonalization;
- 'scaleFactorDer': scaling factors for derivative computation.
parameterListCritical: Recognized keys of critical approximant
parameters:
- 'S': total number of samples current approximant relies upon.
verbosity: Verbosity level.
POD: Kind of snapshots orthogonalization.
scaleFactorDer: Scaling factors for derivative computation.
S: Number of solution snapshots over which current approximant is
based upon.
samplingEngine: Sampling engine.
uHF: High fidelity solution(s) with parameter(s) lastSolvedHF as
sampleList.
lastSolvedHF: Parameter(s) corresponding to last computed high fidelity
solution(s) as parameterList.
uApproxReduced: Reduced approximate solution(s) with parameter(s)
lastSolvedApprox as sampleList.
lastSolvedApproxReduced: Parameter(s) corresponding to last computed
reduced approximate solution(s) as parameterList.
uApprox: Approximate solution(s) with parameter(s) lastSolvedApprox as
sampleList.
lastSolvedApprox: Parameter(s) corresponding to last computed
approximate solution(s) as parameterList.
"""
__all__ += [ftype + dtype for ftype, dtype in iterprod(
["norm", "plot", "outParaview", "outParaviewTimeDomain"],
["HF", "RHS", "Approx", "Res", "Err"])]
def __init__(self, HFEngine:HFEng, mu0 : paramVal = None,
approxParameters : DictAny = {}, verbosity : int = 10,
timestamp : bool = True):
self._preInit()
self._mode = RROMPy_READY
self.verbosity = verbosity
self.timestamp = timestamp
if not hasattr(self, "_output_lvl"): self._output_lvl = []
self._output_lvl += [1]
vbMng(self, "INIT",
"Initializing engine of type {}.".format(self.name()), 10)
self._HFEngine = HFEngine
self.trainedModel = None
self.lastSolvedHF = emptyParameterList()
self.uHF = emptySampleList()
self._addParametersToList(["POD", "scaleFactorDer"], [1, "AUTO"],
["S"], [1.])
if mu0 is None:
if hasattr(self.HFEngine, "mu0"):
self.mu0 = checkParameter(self.HFEngine.mu0)
else:
raise RROMPyException(("Center of approximation cannot be "
"inferred from HF engine. Parameter "
"required"))
else:
self.mu0 = checkParameter(mu0, self.HFEngine.npar)
self.resetSamples()
self.approxParameters = approxParameters
self._postInit()
### add norm{HF,Err} methods
"""
Compute norm of * at arbitrary parameter.
Args:
mu: Target parameter.
Returns:
Target norm of *.
"""
for objName in ["HF", "Err"]:
addNormFieldToClass(self, objName)
### add norm{RHS,Res} methods
"""
Compute norm of * at arbitrary parameter.
Args:
mu: Target parameter.
Returns:
Target norm of *.
"""
for objName in ["RHS", "Res"]:
addNormDualFieldToClass(self, objName)
### add plot{HF,Approx,Err} methods
"""
Do some nice plots of * at arbitrary parameter.
Args:
mu: Target parameter.
name(optional): Name to be shown as title of the plots. Defaults to
'u'.
what(optional): Which plots to do. If list, can contain 'ABS',
'PHASE', 'REAL', 'IMAG'. If str, same plus wildcard 'ALL'.
Defaults to 'ALL'.
save(optional): Where to save plot(s). Defaults to None, i.e. no
saving.
saveFormat(optional): Format for saved plot(s). Defaults to "eps".
saveDPI(optional): DPI for saved plot(s). Defaults to 100.
show(optional): Whether to show figure. Defaults to True.
figspecs(optional key args): Optional arguments for matplotlib
figure creation.
"""
for objName in ["HF", "Approx", "Err"]:
addPlotFieldToClass(self, objName)
### add plot{RHS,Res} methods
"""
Do some nice plots of * at arbitrary parameter.
Args:
mu: Target parameter.
name(optional): Name to be shown as title of the plots. Defaults to
'u'.
what(optional): Which plots to do. If list, can contain 'ABS',
'PHASE', 'REAL', 'IMAG'. If str, same plus wildcard 'ALL'.
Defaults to 'ALL'.
save(optional): Where to save plot(s). Defaults to None, i.e. no
saving.
saveFormat(optional): Format for saved plot(s). Defaults to "eps".
saveDPI(optional): DPI for saved plot(s). Defaults to 100.
show(optional): Whether to show figure. Defaults to True.
figspecs(optional key args): Optional arguments for matplotlib
figure creation.
"""
for objName in ["RHS", "Res"]:
addPlotDualFieldToClass(self, objName)
### add outParaview{HF,RHS,Approx,Res,Err} methods
"""
Output * to ParaView file.
Args:
mu: Target parameter.
name(optional): Base name to be used for data output.
filename(optional): Name of output file.
time(optional): Timestamp.
what(optional): Which plots to do. If list, can contain 'MESH',
'ABS', 'PHASE', 'REAL', 'IMAG'. If str, same plus wildcard
'ALL'. Defaults to 'ALL'.
forceNewFile(optional): Whether to create new output file.
filePW(optional): Fenics File entity (for time series).
"""
for objName in ["HF", "RHS", "Approx", "Res", "Err"]:
addOutParaviewFieldToClass(self, objName)
### add outParaviewTimeDomain{HF,RHS,Approx,Res,Err} methods
"""
Output * to ParaView file, converted to time domain.
Args:
mu: Target parameter.
omega(optional): frequency.
timeFinal(optional): final time of simulation.
periodResolution(optional): number of time steps per period.
name(optional): Base name to be used for data output.
filename(optional): Name of output file.
forceNewFile(optional): Whether to create new output file.
"""
for objName in ["HF", "RHS", "Approx", "Res", "Err"]:
addOutParaviewTimeDomainFieldToClass(self, objName)
def _preInit(self):
if not hasattr(self, "depth"): self.depth = 0
else: self.depth += 1
@property
def tModelType(self):
raise RROMPyException("No trainedModel type assigned.")
def initializeModelData(self, datadict):
from .trained_model.trained_model_data import TrainedModelData
data = TrainedModelData(datadict["mu0"], datadict["mus"],
datadict.pop("projMat"),
datadict["scaleFactor"],
datadict.pop("parameterMap"))
- if hasattr(self.HFEngine, "_is_C_quadratic"):
- data._is_C_quadratic = self.HFEngine._is_C_quadratic
return (data, ["mu0", "scaleFactor", "mus"])
@property
def parameterList(self):
"""Value of parameterListSoft + parameterListCritical."""
return self.parameterListSoft + self.parameterListCritical
def _addParametersToList(self, whatSoft : strLst = [],
defaultSoft : ListAny = [],
whatCritical : strLst = [],
defaultCritical : ListAny = [],
toBeExcluded : strLst = []):
if not hasattr(self, "parameterToBeExcluded"):
self.parameterToBeExcluded = []
self.parameterToBeExcluded = toBeExcluded + self.parameterToBeExcluded
if not hasattr(self, "parameterListSoft"):
self.parameterListSoft = []
if not hasattr(self, "parameterDefaultSoft"):
self.parameterDefaultSoft = {}
if not hasattr(self, "parameterListCritical"):
self.parameterListCritical = []
if not hasattr(self, "parameterDefaultCritical"):
self.parameterDefaultCritical = {}
for j, what in enumerate(whatSoft):
if what not in self.parameterToBeExcluded:
self.parameterListSoft = [what] + self.parameterListSoft
self.parameterDefaultSoft[what] = defaultSoft[j]
for j, what in enumerate(whatCritical):
if what not in self.parameterToBeExcluded:
self.parameterListCritical = ([what]
+ self.parameterListCritical)
self.parameterDefaultCritical[what] = defaultCritical[j]
def _postInit(self):
if self.depth == 0:
vbMng(self, "DEL", "Done initializing.", 10)
del self.depth
else: self.depth -= 1
def name(self) -> str:
return self.__class__.__name__
def __str__(self) -> str:
return self.name()
def __repr__(self) -> str:
return self.__str__() + " at " + hex(id(self))
def setupSampling(self, reset_samples : bool = True):
"""Setup sampling engine."""
RROMPyAssert(self._mode, message = "Cannot setup sampling engine.")
if not hasattr(self, "_POD") or self._POD is None: return
if self.POD == 1:
sEng = SamplingEnginePOD
elif self.POD == 1/2:
sEng = SamplingEngineNormalize
else:
sEng = SamplingEngine
self.samplingEngine = sEng(self.HFEngine, verbosity = self.verbosity)
if reset_samples: self.resetSamples()
@property
def HFEngine(self):
"""Value of HFEngine."""
return self._HFEngine
@HFEngine.setter
def HFEngine(self, HFEngine):
raise RROMPyException("Cannot change HFEngine.")
@property
def mu0(self):
"""Value of mu0."""
return self._mu0
@mu0.setter
def mu0(self, mu0):
mu0 = checkParameter(mu0)
if not hasattr(self, "_mu0") or mu0 != self.mu0:
self.resetSamples()
self._mu0 = mu0
@property
def npar(self):
"""Number of parameters."""
return self.mu0.shape[1]
def checkParameterList(self, mu:paramList,
check_if_single : bool = False) -> paramList:
return checkParameterList(mu, self.npar, check_if_single)
def mapParameterList(self, *args, **kwargs):
return self.HFEngine.mapParameterList(*args, **kwargs)
@property
def approxParameters(self):
"""Value of approximant parameters."""
return self._approxParameters
@approxParameters.setter
def approxParameters(self, approxParams):
if not hasattr(self, "approxParameters"):
self._approxParameters = {}
approxParameters = purgeDict(approxParams, self.parameterList,
dictname = self.name() + ".approxParameters",
baselevel = 1)
keyList = list(approxParameters.keys())
for key in self.parameterListCritical:
if key in keyList:
setattr(self, "_" + key, self.parameterDefaultCritical[key])
for key in self.parameterListSoft:
if key in keyList:
setattr(self, "_" + key, self.parameterDefaultSoft[key])
fragile = False
for key in self.parameterListCritical:
if key in keyList:
val = approxParameters[key]
else:
val = getattr(self, "_" + key, None)
if val is None:
fragile = True
val = self.parameterDefaultCritical[key]
if self._mode == RROMPy_FRAGILE:
setattr(self, "_" + key, val)
self.approxParameters[key] = val
else:
getattr(self.__class__, key, None).fset(self, val)
for key in self.parameterListSoft:
if key in keyList:
val = approxParameters[key]
else:
val = getattr(self, "_" + key, None)
if val is None:
val = self.parameterDefaultSoft[key]
if self._mode == RROMPy_FRAGILE:
setattr(self, "_" + key, val)
self.approxParameters[key] = val
else:
getattr(self.__class__, key, None).fset(self, val)
if fragile: self._mode = RROMPy_FRAGILE
@property
def POD(self):
"""Value of POD."""
return self._POD
@POD.setter
def POD(self, POD):
if hasattr(self, "_POD"): PODold = self.POD
else: PODold = -1
if POD not in [0, 1/2, 1]:
raise RROMPyException("POD must be either 0, 1/2, or 1.")
self._POD = POD
self._approxParameters["POD"] = self.POD
if PODold != self.POD:
self.samplingEngine = None
self.resetSamples()
@property
def scaleFactorDer(self):
"""Value of scaleFactorDer."""
if self._scaleFactorDer == "NONE": return 1.
if self._scaleFactorDer == "AUTO": return self.scaleFactor
return self._scaleFactorDer
@scaleFactorDer.setter
def scaleFactorDer(self, scaleFactorDer):
if isinstance(scaleFactorDer, (str,)):
scaleFactorDer = scaleFactorDer.upper()
elif isinstance(scaleFactorDer, Iterable):
scaleFactorDer = list(scaleFactorDer)
self._scaleFactorDer = scaleFactorDer
self._approxParameters["scaleFactorDer"] = self._scaleFactorDer
@property
def scaleFactorRel(self):
"""Value of scaleFactorDer / scaleFactor."""
if self._scaleFactorDer == "AUTO": return None
try:
return np.divide(self.scaleFactorDer, self.scaleFactor)
except:
raise RROMPyException(("Error in computation of relative scaling "
"factor. Make sure that scaleFactor is "
"properly initialized.")) from None
@property
def S(self):
"""Value of S."""
return self._S
@S.setter
def S(self, S):
if S <= 0: raise RROMPyException("S must be positive.")
if hasattr(self, "_S") and self._S is not None: Sold = self.S
else: Sold = -1
self._S = S
self._approxParameters["S"] = self.S
if Sold != self.S: self.resetSamples()
@property
def trainedModel(self):
"""Value of trainedModel."""
return self._trainedModel
@trainedModel.setter
def trainedModel(self, trainedModel):
self._trainedModel = trainedModel
if self._trainedModel is not None:
self._trainedModel.reset()
self.lastSolvedApproxReduced = emptyParameterList()
self.lastSolvedApprox = emptyParameterList()
self.uApproxReduced = emptySampleList()
self.uApprox = emptySampleList()
def resetSamples(self):
if hasattr(self, "samplingEngine") and self.samplingEngine is not None:
self.samplingEngine.resetHistory()
else:
self.setupSampling()
self._mode = RROMPy_READY
def plotSamples(self, *args, **kwargs) -> List[str]:
"""
Do some nice plots of the samples.
Returns:
Output filenames.
"""
RROMPyAssert(self._mode, message = "Cannot plot samples.")
return self.samplingEngine.plotSamples(*args, **kwargs)
def outParaviewSamples(self, *args, **kwargs) -> List[str]:
"""
Output samples to ParaView file.
Returns:
Output filenames.
"""
RROMPyAssert(self._mode, message = "Cannot output samples.")
return self.samplingEngine.outParaviewSamples(*args, **kwargs)
def outParaviewTimeDomainSamples(self, *args, **kwargs) -> List[str]:
"""
Output samples to ParaView file, converted to time domain.
Returns:
Output filenames.
"""
RROMPyAssert(self._mode, message = "Cannot output samples.")
return self.samplingEngine.outParaviewTimeDomainSamples(*args,
**kwargs)
def setTrainedModel(self, model):
"""Deepcopy approximation from trained model."""
if hasattr(model, "storeTrainedModel"):
verb = model.verbosity
model.verbosity = 0
fileOut = model.storeTrainedModel()
model.verbosity = verb
else:
try:
fileOut = getNewFilename("trained_model", "pkl")
pickleDump(model.data.__dict__, fileOut)
except:
raise RROMPyException(("Failed to store model data. Parameter "
"model must have either "
"storeTrainedModel or "
"data.__dict__ properties.")) from None
self.loadTrainedModel(fileOut)
osrm(fileOut)
@abstractmethod
def setupApprox(self) -> int:
"""
Setup approximant. (ABSTRACT)
Any specialization should include something like
self.trainedModel = ...
self.trainedModel.data = ...
self.trainedModel.data.approxParameters = copy(
self.approxParameters)
Returns > 0 if error was encountered, < 0 if no computation was
necessary.
"""
if self.checkComputedApprox(): return -1
RROMPyAssert(self._mode, message = "Cannot setup approximant.")
vbMng(self, "INIT", "Setting up {}.". format(self.name()), 5)
pass
vbMng(self, "DEL", "Done setting up approximant.", 5)
return 0
def checkComputedApprox(self) -> bool:
"""
Check if setup of new approximant is not needed.
Returns:
True if new setup is not needed. False otherwise.
"""
return self._mode == RROMPy_FRAGILE or (self.trainedModel is not None
and self.trainedModel.data.approxParameters == self.approxParameters
and len(self.mus) == len(self.trainedModel.data.mus))
def _pruneBeforeEval(self, mu:paramList, field:str, append:bool,
prune:bool) -> Tuple[paramList, Np1D, Np1D, bool]:
mu = self.checkParameterList(mu)
idx = np.empty(len(mu), dtype = np.int)
if prune:
jExtra = np.zeros(len(mu), dtype = bool)
muExtra = emptyParameterList()
lastSolvedMus = getattr(self, "lastSolved" + field)
if (len(mu) > 0 and len(mu) == len(lastSolvedMus)
and mu == lastSolvedMus):
idx = np.arange(len(mu), dtype = np.int)
return muExtra, jExtra, idx, True
muKeep = copy(muExtra)
for j in range(len(mu)):
jPos = lastSolvedMus.find(mu[j])
if jPos is not None:
idx[j] = jPos
muKeep.append(mu[j])
else:
jExtra[j] = True
muExtra.append(mu[j])
if len(muKeep) > 0 and not append:
lastSolvedu = getattr(self, "u" + field)
idx[~jExtra] = getattr(self.__class__, "set" + field)(self,
muKeep, lastSolvedu[idx[~jExtra]], append)
append = True
else:
jExtra = np.ones(len(mu), dtype = bool)
muExtra = mu
return muExtra, jExtra, idx, append
def _setObject(self, mu:paramList, field:str, object:sampList,
append:bool) -> List[int]:
newMus = self.checkParameterList(mu)
newObj = sampleList(object)
if append:
getattr(self, "lastSolved" + field).append(newMus)
getattr(self, "u" + field).append(newObj)
Ltot = len(getattr(self, "u" + field))
return list(range(Ltot - len(newObj), Ltot))
setattr(self, "lastSolved" + field, copy(newMus))
setattr(self, "u" + field, copy(newObj))
return list(range(len(getattr(self, "u" + field))))
def setHF(self, muHF:paramList, uHF:sampleList,
append : bool = False) -> List[int]:
"""Assign high fidelity solution."""
return self._setObject(muHF, "HF", uHF, append)
def evalHF(self, mu:paramList, append : bool = False,
prune : bool = True) -> List[int]:
"""
Find high fidelity solution with original parameters and arbitrary
parameter.
Args:
mu: Target parameter.
append(optional): Whether to append new HF solutions to old ones.
prune(optional): Whether to remove duplicates of already appearing
HF solutions.
"""
muExtra, jExtra, idx, append = self._pruneBeforeEval(mu, "HF", append,
prune)
if len(muExtra) > 0:
vbMng(self, "INIT", "Solving HF model for mu = {}.".format(mu),
15)
- if (hasattr(self.HFEngine, "_is_C_quadratic")
- and self.HFEngine._is_C_quadratic):
- newuHFs = [self.HFEngine.solve(muE) for muE in muExtra]
- else:
- newuHFs = self.HFEngine.solve(muExtra)
+ newuHFs = self.HFEngine.solve(muExtra)
vbMng(self, "DEL", "Done solving HF model.", 15)
idx[jExtra] = self.setHF(muExtra, newuHFs, append)
return list(idx)
def setApproxReduced(self, muApproxR:paramList, uApproxR:sampleList,
append : bool = False) -> List[int]:
"""Assign high fidelity solution."""
return self._setObject(muApproxR, "ApproxReduced", uApproxR, append)
def evalApproxReduced(self, mu:paramList, append : bool = False,
prune : bool = False) -> List[int]:
"""
Evaluate reduced representation of approximant at arbitrary parameter.
Args:
mu: Target parameter.
append(optional): Whether to append new HF solutions to old ones.
prune(optional): Whether to remove duplicates of already appearing
HF solutions.
"""
self.setupApprox()
muExtra, jExtra, idx, append = self._pruneBeforeEval(mu,
"ApproxReduced",
append, prune)
if len(muExtra) > 0:
newuApproxs = self.trainedModel.getApproxReduced(muExtra)
idx[jExtra] = self.setApproxReduced(muExtra, newuApproxs, append)
return list(idx)
def setApprox(self, muApprox:paramList, uApprox:sampleList,
append : bool = False) -> List[int]:
"""Assign high fidelity solution."""
return self._setObject(muApprox, "Approx", uApprox, append)
def evalApprox(self, mu:paramList, append : bool = False,
prune : bool = False) -> List[int]:
"""
Evaluate approximant at arbitrary parameter.
Args:
mu: Target parameter.
append(optional): Whether to append new HF solutions to old ones.
prune(optional): Whether to remove duplicates of already appearing
HF solutions.
"""
self.setupApprox()
muExtra, jExtra, idx, append = self._pruneBeforeEval(mu, "Approx",
append, prune)
if len(muExtra) > 0:
newuApproxs = self.trainedModel.getApprox(muExtra)
idx[jExtra] = self.setApprox(muExtra, newuApproxs, append)
return list(idx)
def getHF(self, *args, **kwargs) -> sampList:
"""
Get HF solution at arbitrary parameter.
Returns:
HFsolution.
"""
idx = self.evalHF(*args, **kwargs)
return self.uHF(idx)
def getRHS(self, mu:paramList) -> sampList:
"""
Get linear system RHS at arbitrary parameter.
Args:
mu: Target parameter.
Returns:
Linear system RHS.
"""
return self.HFEngine.residual(mu, None)
def getApproxReduced(self, *args, **kwargs) -> sampList:
"""
Get approximant at arbitrary parameter.
Returns:
Reduced approximant.
"""
idx = self.evalApproxReduced(*args, **kwargs)
return self.uApproxReduced(idx)
def getApprox(self, *args, **kwargs) -> sampList:
"""
Get approximant at arbitrary parameter.
Returns:
Approximant.
"""
idx = self.evalApprox(*args, **kwargs)
return self.uApprox(idx)
def getRes(self, mu:paramList, *args, **kwargs) -> sampList:
"""
Get residual at arbitrary parameter.
Args:
mu: Target parameter.
Returns:
Approximant residual.
"""
if not self.HFEngine.isCEye:
raise RROMPyException(("Residual of solution with non-scalar C "
"not computable."))
return self.HFEngine.residual(mu, self.getApprox(mu, *args, **kwargs)
/ self.HFEngine.C(mu))
def getErr(self, *args, **kwargs) -> sampList:
"""
Get error at arbitrary parameter.
Returns:
Approximant error.
"""
return self.getApprox(*args, **kwargs) - self.getHF(*args, **kwargs)
def normApprox(self, mu:paramList, *args, **kwargs) -> float:
"""
Compute norm of approximant at arbitrary parameter.
Args:
mu: Target parameter.
Returns:
Target norm of approximant.
"""
return self.HFEngine.norm(self.getApprox(mu), *args, **kwargs)
def getPoles(self, *args, **kwargs) -> Np1D:
"""
Obtain approximant poles.
Returns:
Numpy complex vector of poles.
"""
self.setupApprox()
vbMng(self, "INIT", "Computing poles of model.", 20)
poles = self.trainedModel.getPoles(*args, **kwargs)
vbMng(self, "DEL", "Done computing poles.", 20)
return poles
def storeSamples(self, filenameBase : str = "samples",
forceNewFile : bool = True) -> str:
"""Store samples to file."""
filename = filenameBase + "_" + self.name()
if forceNewFile: filename = getNewFilename(filename, "pkl")[: - 4]
return self.samplingEngine.store(filename, False)
def storeTrainedModel(self, filenameBase : str = "trained_model",
forceNewFile : bool = True) -> str:
"""Store trained reduced model to file."""
self.setupApprox()
filename = None
if masterCore():
vbMng(self, "INIT", "Storing trained model to file.", 20)
if forceNewFile:
filename = getNewFilename(filenameBase, "pkl")
else:
filename = "{}.pkl".format(filenameBase)
pickleDump(self.trainedModel.data.__dict__, filename)
vbMng(self, "DEL", "Done storing trained model.", 20)
filename = bcast(filename)
return filename
def loadTrainedModel(self, filename:str):
"""Load trained reduced model from file."""
vbMng(self, "INIT", "Loading pre-trained model from file.", 20)
datadict = pickleLoad(filename)
self.mu0 = datadict["mu0"]
self.scaleFactor = datadict["scaleFactor"]
self.mus = datadict["mus"]
self.trainedModel = self.tModelType()
self.trainedModel.verbosity = self.verbosity
self.trainedModel.timestamp = self.timestamp
data, selfkeys = self.initializeModelData(datadict)
for key in selfkeys: setattr(self, key, datadict.pop(key))
approxParameters = datadict.pop("approxParameters")
data.approxParameters = copy(approxParameters)
for apkey in data.approxParameters.keys():
self._approxParameters[apkey] = approxParameters.pop(apkey)
setattr(self, "_" + apkey, self._approxParameters[apkey])
for key in datadict: setattr(data, key, datadict[key])
self.trainedModel.data = data
self._mode = RROMPy_FRAGILE
vbMng(self, "DEL", "Done loading pre-trained model.", 20)
diff --git a/rrompy/reduction_methods/base/trained_model/trained_model.py b/rrompy/reduction_methods/base/trained_model/trained_model.py
index 96640a2..d805323 100644
--- a/rrompy/reduction_methods/base/trained_model/trained_model.py
+++ b/rrompy/reduction_methods/base/trained_model/trained_model.py
@@ -1,149 +1,120 @@
# Copyright (C) 2018-2020 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 .
#
from abc import abstractmethod
import numpy as np
from rrompy.utilities.base.types import Np1D, List, paramList, sampList
from rrompy.utilities.expression import expressionEvaluator
from rrompy.parameter import checkParameterList
from rrompy.sampling import sampleList
from rrompy.utilities.exception_manager import RROMPyWarning
__all__ = ['TrainedModel']
class TrainedModel:
"""
ABSTRACT
ROM approximant evaluation.
Attributes:
Data: dictionary with all that can be pickled.
"""
def name(self) -> str:
return self.__class__.__name__
def __str__(self) -> str:
return self.name()
def __repr__(self) -> str:
return self.__str__() + " at " + hex(id(self))
def reset(self):
self.lastSolvedApproxReduced = None
self.lastSolvedApprox = None
def compress(self, collapse : bool = False, tol : float = 0.):
if collapse:
self.data.projMat = 1.
self.data._collapsed = True
if tol > 0.: self.data._compressTol = tol
@property
def npar(self):
"""Number of parameters."""
return self.data.mu0.shape[1]
def checkParameterList(self, mu:paramList,
check_if_single : bool = False) -> paramList:
return checkParameterList(mu, self.data.npar, check_if_single)
def mapParameterList(self, mu:paramList, direct : str = "F",
idx : List[int] = None) -> paramList:
if idx is None: idx = np.arange(self.npar)
muMapped = checkParameterList(mu, len(idx))
for j, d in enumerate(idx):
muMapped.data[:, j] = expressionEvaluator(
self.data.parameterMap[direct][d],
muMapped(j)).flatten()
return muMapped
@abstractmethod
def getApproxReduced(self, mu : paramList = []) -> sampList:
"""
Evaluate reduced representation of approximant at arbitrary parameter.
(ABSTRACT)
Args:
mu: Target parameter.
"""
pass
- def _setupQuadMapping(self, N2 : int = None):
- if not (hasattr(self.data, "_is_C_quadratic")
- and self.data._is_C_quadratic):
- RROMPyWarning(("Quadratic mapping is useless if output is not "
- "quadratic. Aborting."))
- return
- if N2 is None: N2 = self.data.projMat.shape[1]
- if hasattr(self, "quad_mapping") and self.quad_mapping.shape[1] == N2:
- return
- idx = np.arange(N2)
- if self.data._is_C_quadratic == 1:
- base = np.floor(idx ** .5)
- above = base * (base + 1) - idx
- li = base - above * (above > 0)
- ri = base + above * (above < 0)
- else: #if self.data._is_C_quadratic == 2:
- li = np.floor((.25 + 2 * idx) ** .5 - .5)
- ri = li * (li + 3) // 2 - idx
- self.quad_mapping = np.vstack([li, ri]).astype(int)
-
def getApprox(self, mu : paramList = []) -> sampList:
"""
Evaluate approximant at arbitrary parameter.
Args:
mu: Target parameter.
"""
mu = self.checkParameterList(mu)
if (not hasattr(self, "lastSolvedApprox")
or self.lastSolvedApprox != mu):
uApproxR = self.getApproxReduced(mu)
if self.data._collapsed:
self.uApprox = uApproxR
else:
for i, uApR in enumerate(uApproxR):
- if not (hasattr(self.data, "_is_C_quadratic")
- and self.data._is_C_quadratic):
- uApREff = uApR
- else:
- self._setupQuadMapping()
- uApREff = (uApR[self.quad_mapping[0]]
- * uApR[self.quad_mapping[1]].conj())
+ uApREff = uApR
uAp = self.data.projMat.dot(uApREff)
- if (hasattr(self.data, "_is_C_quadratic")
- and self.data._is_C_quadratic == 2):
- uAp = np.real(uAp)
if i == 0:
uApprox = np.empty((len(uAp), len(uApproxR)),
dtype = uAp.dtype)
uApprox[:, i] = uAp
self.uApprox = sampleList(uApprox)
self.lastSolvedApprox = mu
return self.uApprox
@abstractmethod
def getPoles(self) -> Np1D:
"""
Obtain approximant poles.
Returns:
Numpy complex vector of poles.
"""
pass
diff --git a/rrompy/reduction_methods/pivoted/generic_pivoted_approximant.py b/rrompy/reduction_methods/pivoted/generic_pivoted_approximant.py
index 438a6f8..674fe65 100644
--- a/rrompy/reduction_methods/pivoted/generic_pivoted_approximant.py
+++ b/rrompy/reduction_methods/pivoted/generic_pivoted_approximant.py
@@ -1,818 +1,835 @@
# Copyright (C) 2018-2020 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 .
#
from abc import abstractmethod
from os import mkdir, remove, rmdir
+from numbers import Number
import numpy as np
from collections.abc import Iterable
from copy import deepcopy as copy
from rrompy.reduction_methods.base.generic_approximant import (
GenericApproximant)
from .trained_model.convert_trained_model_pivoted import (
convertTrainedModelPivoted)
from rrompy.utilities.base.data_structures import purgeDict, getNewFilename
from rrompy.utilities.poly_fitting.polynomial import polybases as ppb
from rrompy.utilities.poly_fitting.radial_basis import polybases as rbpb
from rrompy.utilities.poly_fitting.piecewise_linear import sparsekinds as sk
from rrompy.utilities.base.types import Np2D, paramList, List, ListAny
from rrompy.utilities.base import verbosityManager as vbMng
from rrompy.utilities.numerical.degree import reduceDegreeN
from rrompy.utilities.exception_manager import RROMPyException, RROMPyWarning
from rrompy.parameter import checkParameterList
from rrompy.utilities.parallel import poolRank, bcast
__all__ = ['GenericPivotedApproximantNoMatch',
'GenericPivotedApproximantPoleMatch']
class GenericPivotedApproximantBase(GenericApproximant):
def __init__(self, directionPivot:ListAny, *args,
storeAllSamples : bool = False, **kwargs):
self._preInit()
if len(directionPivot) > 1:
raise RROMPyException(("Exactly 1 pivot parameter allowed in pole "
"matching."))
from rrompy.parameter.parameter_sampling import (EmptySampler as ES,
SparseGridSampler as SG)
self._addParametersToList(["radialDirectionalWeightsMarginal"], [1.],
["samplerPivot", "SMarginal",
"samplerMarginal"],
[ES(), 1, SG([[-1.], [1.]])],
toBeExcluded = ["sampler"])
self._directionPivot = directionPivot
self.storeAllSamples = storeAllSamples
if not hasattr(self, "_output_lvl"): self._output_lvl = []
self._output_lvl += [1 / 2]
super().__init__(*args, **kwargs)
self._postInit()
def setupSampling(self): super().setupSampling(False)
def initializeModelData(self, datadict):
if "directionPivot" in datadict.keys():
from .trained_model.trained_model_pivoted_data import (
TrainedModelPivotedData)
data = TrainedModelPivotedData(datadict["mu0"], datadict["mus"],
datadict.pop("projMat"),
datadict["scaleFactor"],
datadict.pop("parameterMap"),
datadict["directionPivot"])
- if hasattr(self.HFEngine, "_is_C_quadratic") and not (
- hasattr(self, "matchState") and self.matchState):
- data._is_C_quadratic = self.HFEngine._is_C_quadratic
return (data, ["mu0", "scaleFactor", "directionPivot", "mus"])
else:
return super().initializeModelData(datadict)
@property
def npar(self):
"""Number of parameters."""
if hasattr(self, "_temporaryPivot"): return self.nparPivot
return super().npar
def checkParameterListPivot(self, mu:paramList,
check_if_single : bool = False) -> paramList:
return checkParameterList(mu, self.nparPivot, check_if_single)
def checkParameterListMarginal(self, mu:paramList,
check_if_single : bool = False) -> paramList:
return checkParameterList(mu, self.nparMarginal, check_if_single)
def mapParameterList(self, *args, **kwargs):
if hasattr(self, "_temporaryPivot"):
return self.mapParameterListPivot(*args, **kwargs)
return super().mapParameterList(*args, **kwargs)
def mapParameterListPivot(self, mu:paramList, direct : str = "F",
idx : List[int] = None):
if idx is None:
idx = self.directionPivot
else:
idx = [self.directionPivot[j] for j in idx]
return super().mapParameterList(mu, direct, idx)
def mapParameterListMarginal(self, mu:paramList, direct : str = "F",
idx : List[int] = None):
if idx is None:
idx = self.directionMarginal
else:
idx = [self.directionMarginal[j] for j in idx]
return super().mapParameterList(mu, direct, idx)
@property
def mu0(self):
"""Value of mu0."""
if hasattr(self, "_temporaryPivot"):
return self.checkParameterListPivot(self._mu0(self.directionPivot))
return self._mu0
@mu0.setter
def mu0(self, mu0):
GenericApproximant.mu0.fset(self, mu0)
@property
def mus(self):
"""Value of mus. Its assignment may reset snapshots."""
return self._mus
@mus.setter
def mus(self, mus):
mus = self.checkParameterList(mus)
musOld = copy(self.mus) if hasattr(self, '_mus') else None
if (musOld is None or len(mus) != len(musOld) or not mus == musOld):
self.resetSamples()
self._mus = mus
@property
def musMarginal(self):
"""Value of musMarginal. Its assignment may reset snapshots."""
return self._musMarginal
@musMarginal.setter
def musMarginal(self, musMarginal):
musMarginal = self.checkParameterListMarginal(musMarginal)
if hasattr(self, '_musMarginal'):
musMOld = copy(self.musMarginal)
else:
musMOld = None
if (musMOld is None or len(musMarginal) != len(musMOld)
or not musMarginal == musMOld):
self.resetSamples()
self._musMarginal = musMarginal
@property
def SMarginal(self):
"""Value of SMarginal."""
return self._SMarginal
@SMarginal.setter
def SMarginal(self, SMarginal):
if SMarginal <= 0:
raise RROMPyException("SMarginal must be positive.")
if hasattr(self, "_SMarginal") and self._SMarginal is not None:
Sold = self.SMarginal
else: Sold = -1
self._SMarginal = SMarginal
self._approxParameters["SMarginal"] = self.SMarginal
if Sold != self.SMarginal: self.resetSamples()
@property
def radialDirectionalWeightsMarginal(self):
"""Value of radialDirectionalWeightsMarginal."""
return self._radialDirectionalWeightsMarginal
@radialDirectionalWeightsMarginal.setter
def radialDirectionalWeightsMarginal(self, radialDirWeightsMarg):
if isinstance(radialDirWeightsMarg, Iterable):
radialDirWeightsMarg = list(radialDirWeightsMarg)
else:
radialDirWeightsMarg = [radialDirWeightsMarg]
self._radialDirectionalWeightsMarginal = radialDirWeightsMarg
self._approxParameters["radialDirectionalWeightsMarginal"] = (
self.radialDirectionalWeightsMarginal)
@property
def directionPivot(self):
"""Value of directionPivot. Its assignment may reset snapshots."""
return self._directionPivot
@directionPivot.setter
def directionPivot(self, directionPivot):
if hasattr(self, '_directionPivot'):
directionPivotOld = copy(self.directionPivot)
else:
directionPivotOld = None
if (directionPivotOld is None
or len(directionPivot) != len(directionPivotOld)
or not directionPivot == directionPivotOld):
self.resetSamples()
self._directionPivot = directionPivot
@property
def directionMarginal(self):
return [x for x in range(self.HFEngine.npar) \
if x not in self.directionPivot]
@property
def nparPivot(self):
return len(self.directionPivot)
@property
def nparMarginal(self):
return self.npar - self.nparPivot
@property
def muBounds(self):
"""Value of muBounds."""
return self.samplerPivot.lims
@property
def muBoundsMarginal(self):
"""Value of muBoundsMarginal."""
return self.samplerMarginal.lims
@property
def sampler(self):
"""Proxy of samplerPivot."""
return self._samplerPivot
@property
def samplerPivot(self):
"""Value of samplerPivot."""
return self._samplerPivot
@samplerPivot.setter
def samplerPivot(self, samplerPivot):
if 'generatePoints' not in dir(samplerPivot):
raise RROMPyException("Pivot sampler type not recognized.")
if hasattr(self, '_samplerPivot') and self._samplerPivot is not None:
samplerOld = self.samplerPivot
self._samplerPivot = samplerPivot
self._approxParameters["samplerPivot"] = self.samplerPivot
if not 'samplerOld' in locals() or samplerOld != self.samplerPivot:
self.resetSamples()
@property
def samplerMarginal(self):
"""Value of samplerMarginal."""
return self._samplerMarginal
@samplerMarginal.setter
def samplerMarginal(self, samplerMarginal):
if 'generatePoints' not in dir(samplerMarginal):
raise RROMPyException("Marginal sampler type not recognized.")
if (hasattr(self, '_samplerMarginal')
and self._samplerMarginal is not None):
samplerOld = self.samplerMarginal
self._samplerMarginal = samplerMarginal
self._approxParameters["samplerMarginal"] = self.samplerMarginal
if not 'samplerOld' in locals() or samplerOld != self.samplerMarginal:
self.resetSamples()
def computeScaleFactor(self):
"""Compute parameter rescaling factor."""
self.scaleFactorPivot = .5 * np.abs((
self.mapParameterListPivot(self.muBounds[0])
- self.mapParameterListPivot(self.muBounds[1]))[0])
self.scaleFactorMarginal = .5 * np.abs((
self.mapParameterListMarginal(self.muBoundsMarginal[0])
- self.mapParameterListMarginal(self.muBoundsMarginal[1]))[0])
self.scaleFactor = np.empty(self.npar)
self.scaleFactor[self.directionPivot] = self.scaleFactorPivot
self.scaleFactor[self.directionMarginal] = self.scaleFactorMarginal
def _setupTrainedModel(self, pMat:Np2D, pMatUpdate : bool = False,
pMatOld : Np2D = None, forceNew : bool = False):
if forceNew or self.trainedModel is None:
self.trainedModel = self.tModelType()
self.trainedModel.verbosity = self.verbosity
self.trainedModel.timestamp = self.timestamp
datadict = {"mu0": self.mu0, "mus": copy(self.mus),
"projMat": pMat, "scaleFactor": self.scaleFactor,
"parameterMap": self.HFEngine.parameterMap,
"directionPivot": self.directionPivot}
self.trainedModel.data = self.initializeModelData(datadict)[0]
else:
self.trainedModel = self.trainedModel
if pMatUpdate:
self.trainedModel.data.projMat = np.hstack(
(self.trainedModel.data.projMat, pMat))
else:
self.trainedModel.data.projMat = copy(pMat)
self.trainedModel.data.mus = copy(self.mus)
self.trainedModel.data.musMarginal = copy(self.musMarginal)
def normApprox(self, mu:paramList) -> float:
_PODOld, self._POD = self.POD, 0
result = super().normApprox(mu)
self._POD = _PODOld
return result
@property
def storedSamplesFilenames(self) -> List[str]:
if not hasattr(self, "_sampleBaseFilename"): return []
return [self._sampleBaseFilename
+ "{}_{}.pkl" .format(idx + 1, self.name())
for idx in range(len(self.musMarginal))]
def purgeStoredSamples(self):
if not hasattr(self, "_sampleBaseFilename"): return
for file in self.storedSamplesFilenames: remove(file)
rmdir(self._sampleBaseFilename[: -8])
def storeSamples(self, idx : int = None):
"""Store samples to file."""
if not hasattr(self, "_sampleBaseFilename"):
filenameBase = None
if poolRank() == 0:
foldername = getNewFilename(self.name(), "samples")
mkdir(foldername)
filenameBase = foldername + "/sample_"
self._sampleBaseFilename = bcast(filenameBase, force = True)
if idx is not None:
super().storeSamples(self._sampleBaseFilename + str(idx + 1),
False)
def loadTrainedModel(self, filename:str):
"""Load trained reduced model from file."""
super().loadTrainedModel(filename)
self._musMarginal = self.trainedModel.data.musMarginal
def setTrainedModel(self, model):
"""Deepcopy approximation from trained model."""
super().setTrainedModel(model)
self.trainedModel = convertTrainedModelPivoted(self.trainedModel,
self.tModelType, self,
True)
+ self._preliminaryMarginalFinalization()
self._finalizeMarginalization()
self.trainedModel.data.approxParameters = self.approxParameters
class GenericPivotedApproximantNoMatch(GenericPivotedApproximantBase):
"""
ROM pivoted approximant (without pole matching) computation for parametric
problems (ABSTRACT).
Args:
HFEngine: HF problem solver.
mu0(optional): Default parameter. Defaults to 0.
directionPivot(optional): Pivot components. Defaults to [0].
approxParameters(optional): Dictionary containing values for main
parameters of approximant. Recognized keys are:
- 'POD': kind of snapshots orthogonalization; allowed values
include 0, 1/2, and 1; defaults to 1, i.e. POD;
- 'scaleFactorDer': scaling factors for derivative computation;
defaults to 'AUTO';
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator;
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant; defaults to 1.
Defaults to empty dict.
verbosity(optional): Verbosity level. Defaults to 10.
Attributes:
HFEngine: HF problem solver.
mu0: Default parameter.
directionPivot: Pivot components.
mus: Array of snapshot parameters.
musMarginal: Array of marginal snapshot parameters.
approxParameters: Dictionary containing values for main parameters of
approximant. Recognized keys are in parameterList.
parameterListSoft: Recognized keys of soft approximant parameters:
- 'POD': kind of snapshots orthogonalization;
- 'scaleFactorDer': scaling factors for derivative computation;
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant.
parameterListCritical: Recognized keys of critical approximant
parameters:
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator.
verbosity: Verbosity level.
POD: Kind of snapshots orthogonalization.
scaleFactorDer: Scaling factors for derivative computation.
S: Total number of pivot samples current approximant relies upon.
samplerPivot: Pivot sample point generator.
SMarginal: Total number of marginal samples current approximant relies
upon.
samplerMarginal: Marginal sample point generator.
radialDirectionalWeightsMarginal: Radial basis weights for marginal
interpolant.
muBounds: list of bounds for pivot parameter values.
muBoundsMarginal: list of bounds for marginal parameter values.
samplingEngine: Sampling engine.
uHF: High fidelity solution(s) with parameter(s) lastSolvedHF as
sampleList.
lastSolvedHF: Parameter(s) corresponding to last computed high fidelity
solution(s) as parameterList.
uApproxReduced: Reduced approximate solution(s) with parameter(s)
lastSolvedApprox as sampleList.
lastSolvedApproxReduced: Parameter(s) corresponding to last computed
reduced approximate solution(s) as parameterList.
uApprox: Approximate solution(s) with parameter(s) lastSolvedApprox as
sampleList.
lastSolvedApprox: Parameter(s) corresponding to last computed
approximate solution(s) as parameterList.
"""
@property
def tModelType(self):
from .trained_model.trained_model_pivoted_rational_nomatch import (
TrainedModelPivotedRationalNoMatch)
return TrainedModelPivotedRationalNoMatch
def _finalizeMarginalization(self):
self.trainedModel.setupMarginalInterp(
[self.radialDirectionalWeightsMarginal])
self.trainedModel.data.approxParameters = copy(self.approxParameters)
def _preliminaryMarginalFinalization(self):
pass
class GenericPivotedApproximantPoleMatch(GenericPivotedApproximantBase):
"""
ROM pivoted approximant (with pole matching) computation for parametric
problems (ABSTRACT).
Args:
HFEngine: HF problem solver.
mu0(optional): Default parameter. Defaults to 0.
directionPivot(optional): Pivot components. Defaults to [0].
approxParameters(optional): Dictionary containing values for main
parameters of approximant. Recognized keys are:
- 'POD': kind of snapshots orthogonalization; allowed values
include 0, 1/2, and 1; defaults to 1, i.e. POD;
- 'scaleFactorDer': scaling factors for derivative computation;
defaults to 'AUTO';
- 'matchState': whether to match the system state rather than the
system output; defaults to False;
- 'matchingWeight': weight for pole matching optimization; defaults
to 1;
- 'matchingChordalRadius': radius to be used in chordal metric for
poles and residues; if <= 0, Euclidean metric is used; if
'AUTO', automatically selected; defaults to -1;
- 'matchingShared': required ratio of marginal points to share
resonance; defaults to 1.;
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator;
- 'polybasisMarginal': type of polynomial basis for marginal
interpolation; allowed values include 'MONOMIAL_*',
'CHEBYSHEV_*', 'LEGENDRE_*', 'NEARESTNEIGHBOR', and
'PIECEWISE_LINEAR_*'; defaults to 'MONOMIAL';
- 'paramsMarginal': dictionary of parameters for marginal
interpolation; include:
. 'MMarginal': degree of marginal interpolant; defaults to
'AUTO', i.e. maximum allowed; not for 'NEARESTNEIGHBOR' or
'PIECEWISE_LINEAR_*';
. 'nNeighborsMarginal': number of marginal nearest neighbors;
defaults to 1; only for 'NEARESTNEIGHBOR';
. 'polydegreetypeMarginal': type of polynomial degree for
marginal; defaults to 'TOTAL'; not for 'NEARESTNEIGHBOR' or
'PIECEWISE_LINEAR_*';
. 'interpTolMarginal': tolerance for marginal interpolation;
defaults to None; not for 'NEARESTNEIGHBOR' or
'PIECEWISE_LINEAR_*';
. 'radialDirectionalWeightsMarginalAdapt': bounds for adaptive
rescaling of marginal radial basis weights; only for
radial basis.
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant; defaults to 1.
Defaults to empty dict.
verbosity(optional): Verbosity level. Defaults to 10.
Attributes:
HFEngine: HF problem solver.
mu0: Default parameter.
directionPivot: Pivot components.
mus: Array of snapshot parameters.
musMarginal: Array of marginal snapshot parameters.
approxParameters: Dictionary containing values for main parameters of
approximant. Recognized keys are in parameterList.
parameterListSoft: Recognized keys of soft approximant parameters:
- 'POD': kind of snapshots orthogonalization;
- 'scaleFactorDer': scaling factors for derivative computation;
- 'matchState': whether to match the system state rather than the
system output;
- 'matchingWeight': weight for pole matching optimization;
- 'matchingChordalRadius': radius to be used in chordal metric for
poles and residues;
- 'matchingShared': required ratio of marginal points to share
resonance;
- 'polybasisMarginal': type of polynomial basis for marginal
interpolation;
- 'paramsMarginal': dictionary of parameters for marginal
interpolation; include:
. 'MMarginal': degree of marginal interpolant;
. 'nNeighborsMarginal': number of marginal nearest neighbors;
. 'polydegreetypeMarginal': type of polynomial degree for
marginal;
. 'interpTolMarginal': tolerance for marginal interpolation;
. 'radialDirectionalWeightsMarginalAdapt': bounds for adaptive
rescaling of marginal radial basis weights.
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant.
parameterListCritical: Recognized keys of critical approximant
parameters:
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator.
verbosity: Verbosity level.
POD: Kind of snapshots orthogonalization.
scaleFactorDer: Scaling factors for derivative computation.
matchState: Whether to match the system state rather than the system
output.
matchingWeight: Weight for pole matching optimization.
matchingChordalRadius: Radius to be used in chordal metric for poles
and residues.
matchingShared: Required ratio of marginal points to share resonance.
S: Total number of pivot samples current approximant relies upon.
samplerPivot: Pivot sample point generator.
SMarginal: Total number of marginal samples current approximant relies
upon.
samplerMarginal: Marginal sample point generator.
polybasisMarginal: Type of polynomial basis for marginal interpolation.
paramsMarginal: Dictionary of parameters for marginal interpolation.
radialDirectionalWeightsMarginal: Radial basis weights for marginal
interpolant.
muBounds: list of bounds for pivot parameter values.
muBoundsMarginal: list of bounds for marginal parameter values.
samplingEngine: Sampling engine.
uHF: High fidelity solution(s) with parameter(s) lastSolvedHF as
sampleList.
lastSolvedHF: Parameter(s) corresponding to last computed high fidelity
solution(s) as parameterList.
uApproxReduced: Reduced approximate solution(s) with parameter(s)
lastSolvedApprox as sampleList.
lastSolvedApproxReduced: Parameter(s) corresponding to last computed
reduced approximate solution(s) as parameterList.
uApprox: Approximate solution(s) with parameter(s) lastSolvedApprox as
sampleList.
lastSolvedApprox: Parameter(s) corresponding to last computed
approximate solution(s) as parameterList.
"""
def __init__(self, *args, **kwargs):
self._preInit()
self._addParametersToList(["matchState", "matchingWeight",
"matchingChordalRadius", "matchingShared",
"polybasisMarginal", "paramsMarginal"],
[False, 1., [-1, -1], 1., "MONOMIAL", {}])
self.parameterMarginalList = ["MMarginal", "nNeighborsMarginal",
"polydegreetypeMarginal",
"interpTolMarginal",
"radialDirectionalWeightsMarginalAdapt"]
super().__init__(*args, **kwargs)
self._postInit()
@property
def tModelType(self):
from .trained_model.trained_model_pivoted_rational_polematch import (
TrainedModelPivotedRationalPoleMatch)
return TrainedModelPivotedRationalPoleMatch
@property
def matchState(self):
"""Value of matchState."""
return self._matchState
@matchState.setter
def matchState(self, matchState):
self._matchState = matchState
self._approxParameters["matchState"] = self.matchState
@property
def matchingWeight(self):
"""Value of matchingWeight."""
return self._matchingWeight
@matchingWeight.setter
def matchingWeight(self, matchingWeight):
self._matchingWeight = matchingWeight
self._approxParameters["matchingWeight"] = self.matchingWeight
@property
def matchingChordalRadius(self):
"""Value of matchingChordalRadius."""
return self._matchingChordalRadius
@matchingChordalRadius.setter
def matchingChordalRadius(self, matchingChordalRadius):
+ if not hasattr(matchingChordalRadius, "__len__"):
+ matchingChordalRadius = [matchingChordalRadius] * 2
+ if len(matchingChordalRadius) > 2:
+ matchingChordalRadius = matchingChordalRadius[: 2]
+ for j in range(2):
+ if isinstance(matchingChordalRadius[j], (str,)):
+ matchingChordalRadius[j] = (
+ matchingChordalRadius[j].upper().strip().replace(" ",""))
+ if self.POD != 1 and (matchingChordalRadius[1] == "AUTO"
+ or (isinstance(matchingChordalRadius[1], (Number,))
+ and matchingChordalRadius[1] > 0)):
+ RROMPyWarning(("Riemann interpolation of residues without POD "
+ "may lead to unreliable results due to metric "
+ "differences."))
self._matchingChordalRadius = matchingChordalRadius
self._approxParameters["matchingChordalRadius"] = (
self.matchingChordalRadius)
@property
def matchingShared(self):
"""Value of matchingShared."""
return self._matchingShared
@matchingShared.setter
def matchingShared(self, matchingShared):
if matchingShared > 1.:
RROMPyWarning("Shared ratio too large. Clipping to 1.")
matchingShared = 1.
elif matchingShared < 0.:
RROMPyWarning("Shared ratio too small. Clipping to 0.")
matchingShared = 0.
self._matchingShared = matchingShared
self._approxParameters["matchingShared"] = self.matchingShared
@property
def polybasisMarginal(self):
"""Value of polybasisMarginal."""
return self._polybasisMarginal
@polybasisMarginal.setter
def polybasisMarginal(self, polybasisMarginal):
try:
polybasisMarginal = polybasisMarginal.upper().strip().replace(" ",
"")
if polybasisMarginal not in ppb + rbpb + ["NEARESTNEIGHBOR"] + sk:
raise RROMPyException(
"Prescribed marginal polybasis not recognized.")
self._polybasisMarginal = polybasisMarginal
except:
RROMPyWarning(("Prescribed marginal polybasis not recognized. "
"Overriding to 'MONOMIAL'."))
self._polybasisMarginal = "MONOMIAL"
self._approxParameters["polybasisMarginal"] = self.polybasisMarginal
@property
def paramsMarginal(self):
"""Value of paramsMarginal."""
return self._paramsMarginal
@paramsMarginal.setter
def paramsMarginal(self, paramsMarginal):
paramsMarginal = purgeDict(paramsMarginal, self.parameterMarginalList,
dictname = self.name() + ".paramsMarginal",
baselevel = 1)
keyList = list(paramsMarginal.keys())
if not hasattr(self, "_paramsMarginal"): self._paramsMarginal = {}
if "MMarginal" in keyList:
MMarg = paramsMarginal["MMarginal"]
elif ("MMarginal" in self.paramsMarginal
and not hasattr(self, "_MMarginal_isauto")):
MMarg = self.paramsMarginal["MMarginal"]
else:
MMarg = "AUTO"
if isinstance(MMarg, str):
MMarg = MMarg.strip().replace(" ","")
if "-" not in MMarg: MMarg = MMarg + "-0"
self._MMarginal_isauto = True
self._MMarginal_shift = int(MMarg.split("-")[-1])
MMarg = 0
if MMarg < 0:
raise RROMPyException("MMarginal must be non-negative.")
self._paramsMarginal["MMarginal"] = MMarg
if "nNeighborsMarginal" in keyList:
self._paramsMarginal["nNeighborsMarginal"] = max(1,
paramsMarginal["nNeighborsMarginal"])
elif "nNeighborsMarginal" not in self.paramsMarginal:
self._paramsMarginal["nNeighborsMarginal"] = 1
if "polydegreetypeMarginal" in keyList:
try:
polydegtypeM = paramsMarginal["polydegreetypeMarginal"]\
.upper().strip().replace(" ","")
if polydegtypeM not in ["TOTAL", "FULL"]:
raise RROMPyException(("Prescribed polydegreetypeMarginal "
"not recognized."))
self._paramsMarginal["polydegreetypeMarginal"] = polydegtypeM
except:
RROMPyWarning(("Prescribed polydegreetypeMarginal not "
"recognized. Overriding to 'TOTAL'."))
self._paramsMarginal["polydegreetypeMarginal"] = "TOTAL"
elif "polydegreetypeMarginal" not in self.paramsMarginal:
self._paramsMarginal["polydegreetypeMarginal"] = "TOTAL"
if "interpTolMarginal" in keyList:
self._paramsMarginal["interpTolMarginal"] = (
paramsMarginal["interpTolMarginal"])
elif "interpTolMarginal" not in self.paramsMarginal:
self._paramsMarginal["interpTolMarginal"] = -1
if "radialDirectionalWeightsMarginalAdapt" in keyList:
self._paramsMarginal["radialDirectionalWeightsMarginalAdapt"] = (
paramsMarginal["radialDirectionalWeightsMarginalAdapt"])
elif "radialDirectionalWeightsMarginalAdapt" not in self.paramsMarginal:
self._paramsMarginal["radialDirectionalWeightsMarginalAdapt"] = [
-1., -1.]
self._approxParameters["paramsMarginal"] = self.paramsMarginal
def _setMMarginalAuto(self):
if (self.polybasisMarginal not in ppb + rbpb
or "MMarginal" not in self.paramsMarginal
or "polydegreetypeMarginal" not in self.paramsMarginal):
raise RROMPyException(("Cannot set MMarginal if "
"polybasisMarginal does not allow it."))
self.paramsMarginal["MMarginal"] = max(0, reduceDegreeN(
len(self.musMarginal), len(self.musMarginal),
self.nparMarginal,
self.paramsMarginal["polydegreetypeMarginal"])
- self._MMarginal_shift)
vbMng(self, "MAIN", ("Automatically setting MMarginal to {}.").format(
self.paramsMarginal["MMarginal"]), 25)
def purgeparamsMarginal(self):
self.paramsMarginal = {}
paramsMbadkeys = []
if self.polybasisMarginal in ppb + rbpb + sk:
paramsMbadkeys += ["nNeighborsMarginal"]
if self.polybasisMarginal not in rbpb:
paramsMbadkeys += ["radialDirectionalWeightsMarginalAdapt"]
if self.polybasisMarginal in ["NEARESTNEIGHBOR"] + sk:
paramsMbadkeys += ["MMarginal", "polydegreetypeMarginal",
"interpTolMarginal"]
if hasattr(self, "_MMarginal_isauto"): del self._MMarginal_isauto
if hasattr(self, "_MMarginal_shift"): del self._MMarginal_shift
for key in paramsMbadkeys:
if key in self._paramsMarginal: del self._paramsMarginal[key]
self._approxParameters["paramsMarginal"] = self.paramsMarginal
def _finalizeMarginalization(self):
vbMng(self, "INIT", "Checking shared ratio.", 10)
msg = self.trainedModel.checkShared(self.matchingShared)
vbMng(self, "DEL", "Done checking." + msg, 10)
if self.polybasisMarginal in rbpb + ["NEARESTNEIGHBOR"]:
self.computeScaleFactor()
rDWMEff = np.array([w * f for w, f in zip(
self.radialDirectionalWeightsMarginal,
self.scaleFactorMarginal)])
if self.polybasisMarginal in ppb + rbpb + sk:
interpPars = [self.polybasisMarginal]
if self.polybasisMarginal in ppb + rbpb:
if self.polybasisMarginal in rbpb: interpPars += [rDWMEff]
interpPars += [self.verbosity >= 5,
self.paramsMarginal["polydegreetypeMarginal"] == "TOTAL"]
if self.polybasisMarginal in ppb:
interpPars += [{}]
else: # if self.polybasisMarginal in rbpb:
interpPars += [{"optimizeScalingBounds":self.paramsMarginal[
"radialDirectionalWeightsMarginalAdapt"]}]
interpPars += [
{"rcond":self.paramsMarginal["interpTolMarginal"]}]
extraPar = hasattr(self, "_MMarginal_isauto")
else: # if self.polybasisMarginal in sk:
idxEff = [x for x in range(self.samplerMarginal.npoints)
if not hasattr(self.trainedModel, "_idxExcl")
or x not in self.trainedModel._idxExcl]
extraPar = self.samplerMarginal.depth[idxEff]
else: # if self.polybasisMarginal == "NEARESTNEIGHBOR":
interpPars = [self.paramsMarginal["nNeighborsMarginal"], rDWMEff]
extraPar = None
self.trainedModel.setupMarginalInterp(self, interpPars, extraPar)
self.trainedModel.data.approxParameters = copy(self.approxParameters)
def _preliminaryMarginalFinalization(self):
vbMng(self, "INIT", "Compressing and matching poles.", 10)
+ if (self.matchingChordalRadius[1] == "AUTO"
+ or self.matchingChordalRadius[1] > 0):
+ if self.HFEngine.isCEye:
+ if not hasattr(self.trainedModel.data, "projGramian"):
+ projG = self.HFEngine.innerProduct(
+ self.trainedModel.data.projMat,
+ self.trainedModel.data.projMat,
+ is_state = False)
+ else:
+ Sold = self.trainedModel.data.projGramian.shape[0]
+ S = self.trainedModel.data.projMat.shape[1]
+ if Sold > S:
+ projG = self.trainedModel.data.projGramian[: S, : S]
+ else:
+ projG = np.pad(self.trainedModel.data.projGramian,
+ (0, S - Sold), "constant")
+ projG[: Sold, Sold :] = self.HFEngine.innerProduct(
+ self.trainedModel.data.projMat[:, Sold :],
+ self.trainedModel.data.projMat[:, : Sold],
+ is_state = False)
+ projG[Sold :, : Sold] = projG[: Sold, Sold :].T.conj()
+ projG[Sold :, Sold :] = self.HFEngine.innerProduct(
+ self.trainedModel.data.projMat[:, Sold :],
+ self.trainedModel.data.projMat[:, Sold :],
+ is_state = False)
+ else:
+ projG = None
+ self.trainedModel.data.projGramian = projG
self.trainedModel.initializeFromRational(self.matchingWeight,
self.HFEngine,
self.matchState,
self.matchingChordalRadius)
vbMng(self, "DEL", "Done compressing and matching poles.", 10)
def _postApplyC(self):
if self.POD == 1 and not (
hasattr(self.HFEngine.C, "is_mu_independent")
and self.HFEngine.C.is_mu_independent in self._output_lvl):
raise RROMPyException(("Cannot apply mu-dependent C to "
"orthonormalized samples."))
- applyCglob = (hasattr(self.HFEngine, "_is_C_quadratic")
- and self.HFEngine._is_C_quadratic)
vbMng(self, "INIT", "Extracting system output from state.", 35)
- if applyCglob:
- pMat, dirM = None, self.trainedModel.data.directionMarginal
- for muM in self.trainedModel.data.musMarginal:
- idx = np.where([np.allclose(mu(dirM)[0], muM[0])
- for mu in self.trainedModel.data.mus])[0]
- pMati = self.trainedModel.data.projMat[:, idx]
- musi = self.trainedModel.data.mus[idx]
- pMati = self.HFEngine.applyC(pMati, musi)
- if pMat is None:
- pMat = np.array(pMati)
- else:
- pMat = np.append(pMat, pMati, axis = 1)
- else:
- pMat = None
- for j, mu in enumerate(musi):
- pMij = self.trainedModel.data.projMat[:, j]
- pMij = np.expand_dims(self.HFEngine.applyC(pMij, mu), -1)
- if pMati is None:
- pMat = np.array(pMij)
- else:
- pMat = np.append(pMat, pMij, axis = 1)
+ pMat = None
+ for j, mu in enumerate(self.trainedModel.data.mus):
+ pMatj = self.trainedModel.data.projMat[:, j]
+ pMatj = np.expand_dims(self.HFEngine.applyC(pMatj, mu), -1)
+ if pMat is None:
+ pMat = np.array(pMatj)
+ else:
+ pMat = np.append(pMat, pMatj, axis = 1)
vbMng(self, "DEL", "Done extracting system output.", 35)
self.trainedModel.data.projMat = pMat
- if hasattr(self.HFEngine, "_is_C_quadratic"):
- self.trainedModel.data._is_C_quadratic = (
- self.HFEngine._is_C_quadratic)
-
- def setTrainedModel(self, model):
- """Deepcopy approximation from trained model."""
- super().setTrainedModel(model)
- self._preliminaryMarginalFinalization()
- self.trainedModel.data.approxParameters = self.approxParameters
@abstractmethod
def setupApprox(self, *args, **kwargs) -> int:
if self.checkComputedApprox(): return -1
self.purgeparamsMarginal()
setupOK = super().setupApprox(*args, **kwargs)
if self.matchState: self._postApplyC()
return setupOK
diff --git a/rrompy/reduction_methods/pivoted/greedy/generic_pivoted_greedy_approximant.py b/rrompy/reduction_methods/pivoted/greedy/generic_pivoted_greedy_approximant.py
index b732007..627c9ca 100644
--- a/rrompy/reduction_methods/pivoted/greedy/generic_pivoted_greedy_approximant.py
+++ b/rrompy/reduction_methods/pivoted/greedy/generic_pivoted_greedy_approximant.py
@@ -1,663 +1,648 @@
# Copyright (C) 2018-2020 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 .
#
from abc import abstractmethod
from copy import deepcopy as copy
import numpy as np
from collections.abc import Iterable
from matplotlib import pyplot as plt
from rrompy.reduction_methods.pivoted.generic_pivoted_approximant import (
GenericPivotedApproximantBase,
GenericPivotedApproximantPoleMatch)
from rrompy.reduction_methods.pivoted.gather_pivoted_approximant import (
gatherPivotedApproximant)
from rrompy.utilities.base.types import (Np1D, Np2D, Tuple, List, paramVal,
paramList, ListAny)
from rrompy.utilities.base import verbosityManager as vbMng
-from rrompy.utilities.numerical.point_matching import (pointMatching,
- buildResiduesForDistance)
+from rrompy.utilities.numerical import dot
+from rrompy.utilities.numerical.point_matching import pointMatching
from rrompy.utilities.numerical.point_distances import doubleDistanceMatrix
from rrompy.utilities.exception_manager import (RROMPyException, RROMPyAssert,
RROMPyWarning)
from rrompy.parameter import emptyParameterList
from rrompy.utilities.parallel import (masterCore, indicesScatter,
arrayGatherv, isend)
__all__ = ['GenericPivotedGreedyApproximantPoleMatch']
class GenericPivotedGreedyApproximantBase(GenericPivotedApproximantBase):
_allowedEstimatorKindsMarginal = ["LOOK_AHEAD", "LOOK_AHEAD_RECOVER",
"NONE"]
def __init__(self, *args, **kwargs):
self._preInit()
self._addParametersToList(["matchingWeightError",
"errorEstimatorKindMarginal",
"greedyTolMarginal", "maxIterMarginal"],
[0., "NONE", 1e-1, 1e2])
super().__init__(*args, **kwargs)
self._postInit()
@property
def scaleFactorDer(self):
"""Value of scaleFactorDer."""
if self._scaleFactorDer == "NONE": return 1.
if self._scaleFactorDer == "AUTO": return self._scaleFactorOldPivot
return self._scaleFactorDer
@scaleFactorDer.setter
def scaleFactorDer(self, scaleFactorDer):
if isinstance(scaleFactorDer, (str,)):
scaleFactorDer = scaleFactorDer.upper()
elif isinstance(scaleFactorDer, Iterable):
scaleFactorDer = list(scaleFactorDer)
self._scaleFactorDer = scaleFactorDer
self._approxParameters["scaleFactorDer"] = self._scaleFactorDer
@property
def samplerMarginal(self):
"""Value of samplerMarginal."""
return self._samplerMarginal
@samplerMarginal.setter
def samplerMarginal(self, samplerMarginal):
if 'refine' not in dir(samplerMarginal):
raise RROMPyException("Marginal sampler type not recognized.")
GenericPivotedApproximantBase.samplerMarginal.fset(self,
samplerMarginal)
@property
def errorEstimatorKindMarginal(self):
"""Value of errorEstimatorKindMarginal."""
return self._errorEstimatorKindMarginal
@errorEstimatorKindMarginal.setter
def errorEstimatorKindMarginal(self, errorEstimatorKindMarginal):
errorEstimatorKindMarginal = errorEstimatorKindMarginal.upper()
if errorEstimatorKindMarginal not in (
self._allowedEstimatorKindsMarginal):
RROMPyWarning(("Marginal error estimator kind not recognized. "
"Overriding to 'NONE'."))
errorEstimatorKindMarginal = "NONE"
self._errorEstimatorKindMarginal = errorEstimatorKindMarginal
self._approxParameters["errorEstimatorKindMarginal"] = (
self.errorEstimatorKindMarginal)
@property
def matchingWeightError(self):
"""Value of matchingWeightError."""
return self._matchingWeightError
@matchingWeightError.setter
def matchingWeightError(self, matchingWeightError):
self._matchingWeightError = matchingWeightError
self._approxParameters["matchingWeightError"] = (
self.matchingWeightError)
@property
def greedyTolMarginal(self):
"""Value of greedyTolMarginal."""
return self._greedyTolMarginal
@greedyTolMarginal.setter
def greedyTolMarginal(self, greedyTolMarginal):
if greedyTolMarginal < 0:
raise RROMPyException("greedyTolMarginal must be non-negative.")
if (hasattr(self, "_greedyTolMarginal")
and self.greedyTolMarginal is not None):
greedyTolMarginalold = self.greedyTolMarginal
else:
greedyTolMarginalold = -1
self._greedyTolMarginal = greedyTolMarginal
self._approxParameters["greedyTolMarginal"] = self.greedyTolMarginal
if greedyTolMarginalold != self.greedyTolMarginal:
self.resetSamples()
@property
def maxIterMarginal(self):
"""Value of maxIterMarginal."""
return self._maxIterMarginal
@maxIterMarginal.setter
def maxIterMarginal(self, maxIterMarginal):
if maxIterMarginal <= 0:
raise RROMPyException("maxIterMarginal must be positive.")
if (hasattr(self, "_maxIterMarginal")
and self.maxIterMarginal is not None):
maxIterMarginalold = self.maxIterMarginal
else:
maxIterMarginalold = -1
self._maxIterMarginal = maxIterMarginal
self._approxParameters["maxIterMarginal"] = self.maxIterMarginal
if maxIterMarginalold != self.maxIterMarginal:
self.resetSamples()
def resetSamples(self):
"""Reset samples."""
super().resetSamples()
if not hasattr(self, "_temporaryPivot"):
self._mus = emptyParameterList()
self._musMarginal = emptyParameterList()
if hasattr(self, "samplerMarginal"): self.samplerMarginal.reset()
if hasattr(self, "samplingEngine") and self.samplingEngine is not None:
self.samplingEngine.resetHistory()
def _getDistanceApp(self, polesEx:Np1D, resEx:Np2D, muTest:paramVal,
foci:Tuple[float, float], ground:float) -> float:
polesAp = self.trainedModel.interpolateMarginalPoles(muTest)[0]
if self.matchingWeightError != 0:
resAp = self.trainedModel.interpolateMarginalCoeffs(muTest)[0][
: len(polesAp), :]
- if (hasattr(self.trainedModel.data, "_is_C_quadratic")
- and self.trainedModel.data._is_C_quadratic):
- self.trainedModel._setupQuadMapping()
- projMapping = self.quad_mapping
- projMappingReal = self.data._is_C_quadratic == 2
- else:
- projMapping, projMappingReal = None, False
- resEx = buildResiduesForDistance(resEx,
- self.trainedModel.data.projMat,
- 0, projMapping, projMappingReal)
- resAp = buildResiduesForDistance(resAp,
- self.trainedModel.data.projMat,
- 0, projMapping, projMappingReal)
+ resEx = dot(self.trainedModel.data.projMat, resEx)
+ resAp = dot(self.trainedModel.data.projMat, resAp)
else:
resAp = None
dist = doubleDistanceMatrix(polesEx, polesAp, self.matchingWeightError,
resEx, resAp, self.HFEngine, False,
self.trainedModel.data.chordalRadius)
pmR, pmC = pointMatching(dist)
return np.mean(dist[pmR, pmC])
def getErrorEstimatorMarginalLookAhead(self) -> Np1D:
if not hasattr(self.trainedModel, "_musMExcl"):
err = np.zeros(0)
err[:] = np.inf
self._musMarginalTestIdxs = np.zeros(0, dtype = int)
return err
self._musMarginalTestIdxs = np.array(self.trainedModel._idxExcl,
dtype = int)
idx, sizes = indicesScatter(len(self.trainedModel._musMExcl),
return_sizes = True)
err = []
if len(idx) > 0:
self.verbosity -= 25
self.trainedModel.verbosity -= 25
foci = self.samplerPivot.normalFoci()
ground = self.samplerPivot.groundPotential()
for j in idx:
muTest = self.trainedModel._musMExcl[j]
HITest = self.trainedModel._HIsExcl[j]
polesEx = HITest.poles
idxGood = np.logical_not(np.logical_or(np.isinf(polesEx),
np.isnan(polesEx)))
polesEx = polesEx[idxGood]
if self.matchingWeightError != 0:
resEx = HITest.coeffs[np.where(idxGood)[0]]
else:
resEx = None
if len(polesEx) == 0:
err += [0.]
continue
err += [self._getDistanceApp(polesEx, resEx, muTest,
foci, ground)]
self.verbosity += 25
self.trainedModel.verbosity += 25
return arrayGatherv(np.array(err), sizes)
def getErrorEstimatorMarginalNone(self) -> Np1D:
nErr = len(self.trainedModel.data.musMarginal)
self._musMarginalTestIdxs = np.arange(nErr)
return (1. + self.greedyTolMarginal) * np.ones(nErr)
def errorEstimatorMarginal(self, return_max : bool = False) -> Np1D:
vbMng(self.trainedModel, "INIT",
"Evaluating error estimator at mu = {}.".format(
self.trainedModel.data.musMarginal), 10)
if self.errorEstimatorKindMarginal == "NONE":
nErr = len(self.trainedModel.data.musMarginal)
self._musMarginalTestIdxs = np.arange(nErr)
err = (1. + self.greedyTolMarginal) * np.ones(nErr)
else:#if self.errorEstimatorKindMarginal[: 10] == "LOOK_AHEAD":
err = self.getErrorEstimatorMarginalLookAhead()
vbMng(self.trainedModel, "DEL", "Done evaluating error estimator.", 10)
if not return_max: return err
idxMaxEst = np.where(err > self.greedyTolMarginal)[0]
maxErr = err[idxMaxEst]
if self.errorEstimatorKindMarginal == "NONE": maxErr = None
return err, idxMaxEst, maxErr
def plotEstimatorMarginal(self, est:Np1D, idxMax:List[int],
estMax:List[float]):
if self.errorEstimatorKindMarginal == "NONE": return
if (not (np.any(np.isnan(est)) or np.any(np.isinf(est)))
and masterCore() and hasattr(self.trainedModel, "_musMExcl")):
fig = plt.figure(figsize = plt.figaspect(1. / self.nparMarginal))
for jpar in range(self.nparMarginal):
ax = fig.add_subplot(1, self.nparMarginal, 1 + jpar)
musre = np.real(self.trainedModel._musMExcl)
if len(idxMax) > 0 and estMax is not None:
maxrej = musre[idxMax, jpar]
errCP = copy(est)
idx = np.delete(np.arange(self.nparMarginal), jpar)
while len(musre) > 0:
if self.nparMarginal == 1:
currIdx = np.arange(len(musre))
else:
currIdx = np.where(np.isclose(np.sum(
np.abs(musre[:, idx] - musre[0, idx]), 1),
0., atol = 1e-15))[0]
currIdxSorted = currIdx[np.argsort(musre[currIdx, jpar])]
ax.semilogy(musre[currIdxSorted, jpar],
errCP[currIdxSorted], 'k.-', linewidth = 1)
musre = np.delete(musre, currIdx, 0)
errCP = np.delete(errCP, currIdx)
ax.semilogy(self.musMarginal.re(jpar),
(self.greedyTolMarginal,) * len(self.musMarginal),
'*m')
if len(idxMax) > 0 and estMax is not None:
ax.semilogy(maxrej, estMax, 'xr')
ax.set_xlim(*list(self.samplerMarginal.lims.re(jpar)))
ax.grid()
plt.tight_layout()
plt.show()
def _addMarginalSample(self, mus:paramList):
mus = self.checkParameterListMarginal(mus)
if len(mus) == 0: return
self._nmusOld, nmus = len(self.musMarginal), len(mus)
if (hasattr(self, "trainedModel") and self.trainedModel is not None
and hasattr(self.trainedModel, "_musMExcl")):
self._nmusOld += len(self.trainedModel._musMExcl)
vbMng(self, "MAIN",
("Adding marginal sample point{} no. {}{} at {} to training "
"set.").format("s" * (nmus > 1), self._nmusOld + 1,
"--{}".format(self._nmusOld + nmus) * (nmus > 1),
mus), 3)
self.musMarginal.append(mus)
self.setupApproxPivoted(mus)
self._preliminaryMarginalFinalization()
del self._nmusOld
if (self.errorEstimatorKindMarginal[: 10] == "LOOK_AHEAD"
and not self.firstGreedyIterM):
ubRange = len(self.trainedModel.data.musMarginal)
if hasattr(self.trainedModel, "_idxExcl"):
shRange = len(self.trainedModel._musMExcl)
else:
shRange = 0
testIdxs = list(range(ubRange + shRange - len(mus),
ubRange + shRange))
for j in testIdxs[::-1]:
self.musMarginal.pop(j - shRange)
if hasattr(self.trainedModel, "_idxExcl"):
testIdxs = self.trainedModel._idxExcl + testIdxs
self._updateTrainedModelMarginalSamples(testIdxs)
self._finalizeMarginalization()
self._SMarginal = len(self.musMarginal)
self._approxParameters["SMarginal"] = self.SMarginal
self.trainedModel.data.approxParameters["SMarginal"] = self.SMarginal
def greedyNextSampleMarginal(self, muidx:List[int],
plotEst : str = "NONE") \
-> Tuple[Np1D, List[int], float, paramVal]:
RROMPyAssert(self._mode, message = "Cannot add greedy sample.")
muidx = self._musMarginalTestIdxs[muidx]
if (self.errorEstimatorKindMarginal[: 10] == "LOOK_AHEAD"
and not self.firstGreedyIterM):
if not hasattr(self.trainedModel, "_idxExcl"):
raise RROMPyException(("Sample index to be added not present "
"in trained model."))
testIdxs = copy(self.trainedModel._idxExcl)
skippedIdx = 0
for cj, j in enumerate(self.trainedModel._idxExcl):
if j in muidx:
testIdxs.pop(skippedIdx)
self.musMarginal.insert(self.trainedModel._musMExcl[cj],
j - skippedIdx)
else:
skippedIdx += 1
if len(self.trainedModel._idxExcl) < (len(muidx)
+ len(testIdxs)):
raise RROMPyException(("Sample index to be added not present "
"in trained model."))
self._updateTrainedModelMarginalSamples(testIdxs)
self._SMarginal = len(self.musMarginal)
self._approxParameters["SMarginal"] = self.SMarginal
self.trainedModel.data.approxParameters["SMarginal"] = (
self.SMarginal)
self.firstGreedyIterM = False
idxAdded = self.samplerMarginal.refine(muidx)[0]
self._addMarginalSample(self.samplerMarginal.points[idxAdded])
errorEstTest, muidx, maxErrorEst = self.errorEstimatorMarginal(True)
if plotEst == "ALL":
self.plotEstimatorMarginal(errorEstTest, muidx, maxErrorEst)
return (errorEstTest, muidx, maxErrorEst,
self.samplerMarginal.points[muidx])
def _preliminaryTrainingMarginal(self):
"""Initialize starting snapshots of solution map."""
RROMPyAssert(self._mode, message = "Cannot start greedy algorithm.")
if np.sum(self.samplingEngine.nsamples) > 0: return
self.resetSamples()
self._addMarginalSample(self.samplerMarginal.generatePoints(
self.SMarginal))
def _preSetupApproxPivoted(self, mus:paramList) \
-> Tuple[ListAny, ListAny, ListAny]:
self.computeScaleFactor()
if self.trainedModel is None:
self._setupTrainedModel(np.zeros((0, 0)))
self.trainedModel.data.Qs, self.trainedModel.data.Ps = [], []
self.trainedModel.data.Psupp = []
self._trainedModelOld = copy(self.trainedModel)
self._scaleFactorOldPivot = copy(self.scaleFactor)
self.scaleFactor = self.scaleFactorPivot
self._temporaryPivot = 1
self._musLoc = copy(self.mus)
idx, sizes = indicesScatter(len(mus), return_sizes = True)
emptyCores = np.where(np.logical_not(sizes))[0]
self.verbosity -= 10
self.samplingEngine.verbosity -= 10
return idx, sizes, emptyCores
def _postSetupApproxPivoted(self, mus:Np2D, pMat:Np2D, Ps:ListAny,
Qs:ListAny, sizes:ListAny):
self.scaleFactor = self._scaleFactorOldPivot
del self._scaleFactorOldPivot, self._temporaryPivot
pMat, Ps, Qs, mus, nsamples = gatherPivotedApproximant(pMat, Ps, Qs,
mus, sizes,
self.polybasis)
if len(self._musLoc) > 0:
self._mus = self.checkParameterList(self._musLoc)
self._mus.append(mus)
else:
self._mus = self.checkParameterList(mus)
self.trainedModel = self._trainedModelOld
del self._trainedModelOld
padLeft = self.trainedModel.data.projMat.shape[1]
suppNew = np.append(0, np.cumsum(nsamples))
self._setupTrainedModel(pMat, padLeft > 0)
self.trainedModel.data.Qs += Qs
self.trainedModel.data.Ps += Ps
self.trainedModel.data.Psupp += list(padLeft + suppNew[: -1])
self.trainedModel.data.approxParameters = copy(self.approxParameters)
self.verbosity += 10
self.samplingEngine.verbosity += 10
def _localPivotedResult(self, pMat:Np2D, req:ListAny, emptyCores:ListAny,
mus:Np2D) -> Tuple[Np2D, ListAny, Np2D]:
pMati = self.samplingEngine.projectionMatrix
musi = self.samplingEngine.mus
if not hasattr(self, "matchState") or not self.matchState:
if self.POD == 1 and not (
hasattr(self.HFEngine.C, "is_mu_independent")
and self.HFEngine.C.is_mu_independent in self._output_lvl):
raise RROMPyException(("Cannot apply mu-dependent C "
"to orthonormalized samples."))
vbMng(self, "INIT", "Extracting system output from state.", 35)
- if (hasattr(self.HFEngine, "_is_C_quadratic")
- and self.HFEngine._is_C_quadratic):
- pMati = self.HFEngine.applyC(pMati, musi)
- else:
- pMatiEff = None
- for j, mu in enumerate(musi):
- pMij = np.expand_dims(self.HFEngine.applyC(pMati[:, j],
- mu), -1)
- if pMatiEff is None:
- pMatiEff = np.array(pMij)
- else:
- pMatiEff = np.append(pMatiEff, pMij, axis = 1)
- pMati = pMatiEff
+ pMatiEff = None
+ for j, mu in enumerate(musi):
+ pMij = np.expand_dims(self.HFEngine.applyC(pMati[:, j], mu),
+ -1)
+ if pMatiEff is None:
+ pMatiEff = np.array(pMij)
+ else:
+ pMatiEff = np.append(pMatiEff, pMij, axis = 1)
+ pMati = pMatiEff
vbMng(self, "DEL", "Done extracting system output.", 35)
if pMat is None:
mus = copy(musi.data)
pMat = copy(pMati)
if masterCore():
for dest in emptyCores:
req += [isend((len(pMat), pMat.dtype, mus.dtype),
dest = dest, tag = dest)]
else:
mus = np.vstack((mus, musi.data))
pMat = np.hstack((pMat, pMati))
return pMat, req, mus
@abstractmethod
def setupApproxPivoted(self, mus:paramList) -> int:
if self.checkComputedApproxPivoted(): return -1
RROMPyAssert(self._mode, message = "Cannot setup approximant.")
vbMng(self, "INIT", "Setting up pivoted approximant.", 10)
self._preSetupApproxPivoted()
data = []
pass
self._postSetupApproxPivoted(mus, data)
vbMng(self, "DEL", "Done setting up pivoted approximant.", 10)
return 0
def setupApprox(self, plotEst : str = "NONE") -> int:
"""Compute greedy snapshots of solution map."""
if self.checkComputedApprox(): return -1
RROMPyAssert(self._mode, message = "Cannot start greedy algorithm.")
vbMng(self, "INIT", "Starting computation of snapshots.", 3)
max2ErrorEst, self.firstGreedyIterM = np.inf, True
self._preliminaryTrainingMarginal()
if self.errorEstimatorKindMarginal == "NONE":
muidx = []
else:#if self.errorEstimatorKindMarginal[: 10] == "LOOK_AHEAD":
muidx = np.arange(len(self.trainedModel.data.musMarginal))
self._musMarginalTestIdxs = np.array(muidx)
while self.firstGreedyIterM or (max2ErrorEst > self.greedyTolMarginal
and self.samplerMarginal.npoints < self.maxIterMarginal):
errorEstTest, muidx, maxErrorEst, mu = \
self.greedyNextSampleMarginal(muidx, plotEst)
if maxErrorEst is None:
max2ErrorEst = 1. + self.greedyTolMarginal
else:
if len(maxErrorEst) > 0:
max2ErrorEst = np.max(maxErrorEst)
else:
max2ErrorEst = np.max(errorEstTest)
vbMng(self, "MAIN", ("Uniform testing error estimate "
"{:.4e}.").format(max2ErrorEst), 3)
if plotEst == "LAST":
self.plotEstimatorMarginal(errorEstTest, muidx, maxErrorEst)
vbMng(self, "DEL", ("Done computing snapshots (final snapshot count: "
"{}).").format(len(self.mus)), 3)
if (self.errorEstimatorKindMarginal == "LOOK_AHEAD_RECOVER"
and hasattr(self.trainedModel, "_idxExcl")
and len(self.trainedModel._idxExcl) > 0):
vbMng(self, "INIT", "Recovering {} test models.".format(
len(self.trainedModel._idxExcl)), 7)
for j, mu in zip(self.trainedModel._idxExcl,
self.trainedModel._musMExcl):
self.musMarginal.insert(mu, j)
+ self._preliminaryMarginalFinalization()
self._updateTrainedModelMarginalSamples()
self._finalizeMarginalization()
self._SMarginal = len(self.musMarginal)
self._approxParameters["SMarginal"] = self.SMarginal
self.trainedModel.data.approxParameters["SMarginal"] = (
self.SMarginal)
vbMng(self, "DEL", "Done recovering test models.", 7)
return 0
def checkComputedApproxPivoted(self) -> bool:
return (super().checkComputedApprox()
and len(self.musMarginal) == len(self.trainedModel.data.musMarginal))
class GenericPivotedGreedyApproximantPoleMatch(
GenericPivotedGreedyApproximantBase,
GenericPivotedApproximantPoleMatch):
"""
ROM pivoted greedy interpolant computation for parametric problems (with
pole matching) (ABSTRACT).
Args:
HFEngine: HF problem solver.
mu0(optional): Default parameter. Defaults to 0.
directionPivot(optional): Pivot components. Defaults to [0].
approxParameters(optional): Dictionary containing values for main
parameters of approximant. Recognized keys are:
- 'POD': kind of snapshots orthogonalization; allowed values
include 0, 1/2, and 1; defaults to 1, i.e. POD;
- 'scaleFactorDer': scaling factors for derivative computation;
defaults to 'AUTO';
- 'matchState': whether to match the system state rather than the
system output; defaults to False;
- 'matchingWeight': weight for pole matching optimization; defaults
to 1;
- 'matchingChordalRadius': radius to be used in chordal metric for
poles and residues; if <= 0, Euclidean metric is used; if
'AUTO', automatically selected; defaults to -1;
- 'matchingShared': required ratio of marginal points to share
resonance; defaults to 1.;
- 'matchingWeightError': weight for pole matching optimization in
error estimation; defaults to 0;
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': number of starting marginal samples;
- 'samplerMarginal': marginal sample point generator via sparse
grid;
- 'errorEstimatorKindMarginal': kind of marginal error estimator;
available values include 'LOOK_AHEAD', 'LOOK_AHEAD_RECOVER',
and 'NONE'; defaults to 'NONE';
- 'polybasisMarginal': type of polynomial basis for marginal
interpolation; allowed values include 'MONOMIAL_*',
'CHEBYSHEV_*', 'LEGENDRE_*', 'NEARESTNEIGHBOR', and
'PIECEWISE_LINEAR_*'; defaults to 'MONOMIAL';
- 'paramsMarginal': dictionary of parameters for marginal
interpolation; include:
. 'MMarginal': degree of marginal interpolant; defaults to
'AUTO', i.e. maximum allowed; not for 'NEARESTNEIGHBOR' or
'PIECEWISE_LINEAR_*';
. 'nNeighborsMarginal': number of marginal nearest neighbors;
defaults to 1; only for 'NEARESTNEIGHBOR';
. 'polydegreetypeMarginal': type of polynomial degree for
marginal; defaults to 'TOTAL'; not for 'NEARESTNEIGHBOR' or
'PIECEWISE_LINEAR_*';
. 'interpTolMarginal': tolerance for marginal interpolation;
defaults to None; not for 'NEARESTNEIGHBOR' or
'PIECEWISE_LINEAR_*';
. 'radialDirectionalWeightsMarginalAdapt': bounds for adaptive
rescaling of marginal radial basis weights; only for
radial basis.
- 'greedyTolMarginal': uniform error tolerance for marginal greedy
algorithm; defaults to 1e-1;
- 'maxIterMarginal': maximum number of marginal greedy steps;
defaults to 1e2;
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant; defaults to 1.
Defaults to empty dict.
verbosity(optional): Verbosity level. Defaults to 10.
Attributes:
HFEngine: HF problem solver.
mu0: Default parameter.
directionPivot: Pivot components.
mus: Array of snapshot parameters.
musMarginal: Array of marginal snapshot parameters.
approxParameters: Dictionary containing values for main parameters of
approximant. Recognized keys are in parameterList.
parameterListSoft: Recognized keys of soft approximant parameters:
- 'POD': kind of snapshots orthogonalization;
- 'scaleFactorDer': scaling factors for derivative computation;
- 'matchState': whether to match the system state rather than the
system output;
- 'matchingWeight': weight for pole matching optimization;
- 'matchingChordalRadius': radius to be used in chordal metric for
poles and residues;
- 'matchingShared': required ratio of marginal points to share
resonance;
- 'matchingWeightError': weight for pole matching optimization in
error estimation;
- 'errorEstimatorKindMarginal': kind of marginal error estimator;
- 'polybasisMarginal': type of polynomial basis for marginal
interpolation;
- 'paramsMarginal': dictionary of parameters for marginal
interpolation; include:
. 'MMarginal': degree of marginal interpolant;
. 'nNeighborsMarginal': number of marginal nearest neighbors;
. 'polydegreetypeMarginal': type of polynomial degree for
marginal;
. 'interpTolMarginal': tolerance for marginal interpolation;
. 'radialDirectionalWeightsMarginalAdapt': bounds for adaptive
rescaling of marginal radial basis weights.
- 'greedyTolMarginal': uniform error tolerance for marginal greedy
algorithm;
- 'maxIterMarginal': maximum number of marginal greedy steps;
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant.
parameterListCritical: Recognized keys of critical approximant
parameters:
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator via sparse
grid.
verbosity: Verbosity level.
POD: Kind of snapshots orthogonalization.
scaleFactorDer: Scaling factors for derivative computation.
matchState: Whether to match the system state rather than the system
output.
matchingWeight: Weight for pole matching optimization.
matchingChordalRadius: Radius to be used in chordal metric for poles
and residues.
matchingShared: Required ratio of marginal points to share resonance.
matchingWeightError: Weight for pole matching optimization in error
estimation.
S: Total number of pivot samples current approximant relies upon.
samplerPivot: Pivot sample point generator.
SMarginal: Total number of marginal samples current approximant relies
upon.
samplerMarginal: Marginal sample point generator via sparse grid.
errorEstimatorKindMarginal: Kind of marginal error estimator.
polybasisMarginal: Type of polynomial basis for marginal interpolation.
paramsMarginal: Dictionary of parameters for marginal interpolation.
greedyTolMarginal: Uniform error tolerance for marginal greedy
algorithm.
maxIterMarginal: Maximum number of marginal greedy steps.
radialDirectionalWeightsMarginal: Radial basis weights for marginal
interpolant.
muBounds: list of bounds for pivot parameter values.
muBoundsMarginal: list of bounds for marginal parameter values.
samplingEngine: Sampling engine.
uHF: High fidelity solution(s) with parameter(s) lastSolvedHF as
sampleList.
lastSolvedHF: Parameter(s) corresponding to last computed high fidelity
solution(s) as parameterList.
uApproxReduced: Reduced approximate solution(s) with parameter(s)
lastSolvedApprox as sampleList.
lastSolvedApproxReduced: Parameter(s) corresponding to last computed
reduced approximate solution(s) as parameterList.
uApprox: Approximate solution(s) with parameter(s) lastSolvedApprox as
sampleList.
lastSolvedApprox: Parameter(s) corresponding to last computed
approximate solution(s) as parameterList.
"""
def _updateTrainedModelMarginalSamples(self, idx : ListAny = []):
self.trainedModel.updateEffectiveSamples(idx, self.matchingWeight,
self.HFEngine, False,
self.matchingChordalRadius)
def setupApprox(self, *args, **kwargs) -> int:
if self.checkComputedApprox(): return -1
self.purgeparamsMarginal()
_polybasisMarginal = self.polybasisMarginal
self._polybasisMarginal = ("PIECEWISE_LINEAR_"
+ self.samplerMarginal.kind)
setupOK = super().setupApprox(*args, **kwargs)
self._polybasisMarginal = _polybasisMarginal
- self._finalizeMarginalization()
if self.matchState: self._postApplyC()
return setupOK
diff --git a/rrompy/reduction_methods/pivoted/rational_interpolant_greedy_pivoted.py b/rrompy/reduction_methods/pivoted/rational_interpolant_greedy_pivoted.py
index b4a2568..c516b91 100644
--- a/rrompy/reduction_methods/pivoted/rational_interpolant_greedy_pivoted.py
+++ b/rrompy/reduction_methods/pivoted/rational_interpolant_greedy_pivoted.py
@@ -1,576 +1,572 @@
# Copyright (C) 2018-2020 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 .
#
from copy import deepcopy as copy
import numpy as np
from .generic_pivoted_approximant import (GenericPivotedApproximantBase,
GenericPivotedApproximantNoMatch,
GenericPivotedApproximantPoleMatch)
from .gather_pivoted_approximant import gatherPivotedApproximant
from rrompy.reduction_methods.standard.greedy.rational_interpolant_greedy \
import RationalInterpolantGreedy
from rrompy.reduction_methods.standard.greedy.generic_greedy_approximant \
import pruneSamples
from rrompy.utilities.base.types import Np1D
from rrompy.utilities.base import verbosityManager as vbMng
from rrompy.utilities.poly_fitting.polynomial import polyvander as pv
from rrompy.utilities.exception_manager import RROMPyException, RROMPyAssert
from rrompy.parameter import emptyParameterList, parameterList
from rrompy.utilities.parallel import poolRank, indicesScatter, isend, recv
__all__ = ['RationalInterpolantGreedyPivotedNoMatch',
'RationalInterpolantGreedyPivotedPoleMatch']
class RationalInterpolantGreedyPivotedBase(GenericPivotedApproximantBase,
RationalInterpolantGreedy):
def __init__(self, *args, **kwargs):
self._preInit()
super().__init__(*args, **kwargs)
if self.nparPivot > 1: self.HFEngine._ignoreResidues = 1
self._postInit()
@property
def tModelType(self):
if hasattr(self, "_temporaryPivot"):
return RationalInterpolantGreedy.tModelType.fget(self)
return super().tModelType
def _polyvanderAuxiliary(self, mus, deg, *args):
degEff = [0] * self.npar
degEff[self.directionPivot[0]] = deg
return pv(mus, degEff, *args)
def _marginalizeMiscellanea(self, forward:bool):
if forward:
self._m_selfmus = copy(self.mus)
self._m_HFEparameterMap = copy(self.HFEngine.parameterMap)
self._mus = self.checkParameterListPivot(
self.mus(self.directionPivot))
self.HFEngine.parameterMap = {
"F": [self.HFEngine.parameterMap["F"][self.directionPivot[0]]],
"B": [self.HFEngine.parameterMap["B"][self.directionPivot[0]]]}
else:
self._mus = self._m_selfmus
self.HFEngine.parameterMap = self._m_HFEparameterMap
del self._m_selfmus, self._m_HFEparameterMap
def _marginalizeTrainedModel(self, forward:bool):
if forward:
del self._temporaryPivot
self.trainedModel.data.mu0 = self.mu0
self.trainedModel.data.scaleFactor = [1.] * self.npar
self.trainedModel.data.scaleFactor[self.directionPivot[0]] = (
self.scaleFactor[0])
self.trainedModel.data.parameterMap = self.HFEngine.parameterMap
self._m_musUniqueCN = copy(self._musUniqueCN)
musUniqueCNAux = np.zeros((self.S, self.npar),
dtype = self._musUniqueCN.dtype)
musUniqueCNAux[:, self.directionPivot[0]] = self._musUniqueCN(0)
self._musUniqueCN = self.checkParameterList(musUniqueCNAux)
self._m_derIdxs = copy(self._derIdxs)
for j in range(len(self._derIdxs)):
for l in range(len(self._derIdxs[j])):
derjl = self._derIdxs[j][l][0]
self._derIdxs[j][l] = [0] * self.npar
self._derIdxs[j][l][self.directionPivot[0]] = derjl
self.trainedModel.data.Q._dirPivot = self.directionPivot[0]
self.trainedModel.data.P._dirPivot = self.directionPivot[0]
# tell greedy error estimator that operator / RHS is pivot-affine
if hasattr(self.HFEngine.A, "is_affine"):
self._A_is_affine = self.HFEngine.A.is_affine
else:
self._A_is_affine = 0
if hasattr(self.HFEngine.b, "is_affine"):
self._b_is_affine = self.HFEngine.b.is_affine
else:
self._b_is_affine = 0
if self._A_is_affine >= 1 / 2 and self._b_is_affine >= 1 / 2:
self._affine_lvl += [1 / 2]
else:
self._temporaryPivot = 1
self.trainedModel.data.mu0 = self.checkParameterListPivot(
self.mu0(self.directionPivot))
self.trainedModel.data.scaleFactor = self.scaleFactor
self.trainedModel.data.parameterMap = {
"F": [self.HFEngine.parameterMap["F"][self.directionPivot[0]]],
"B": [self.HFEngine.parameterMap["B"][self.directionPivot[0]]]}
self._musUniqueCN = copy(self._m_musUniqueCN)
self._derIdxs = copy(self._m_derIdxs)
del self._m_musUniqueCN, self._m_derIdxs
del self.trainedModel.data.Q._dirPivot
del self.trainedModel.data.P._dirPivot
if self._A_is_affine >= 1 / 2 and self._b_is_affine >= 1 / 2:
self._affine_lvl.pop()
del self._A_is_affine, self._b_is_affine
self.trainedModel.data.npar = self.npar
def errorEstimator(self, mus:Np1D, return_max : bool = False) -> Np1D:
"""Standard residual-based error estimator."""
setupOK = self.setupApproxLocal()
if setupOK > 0:
err = np.empty(len(mus))
err[:] = np.nan
if not return_max: return err
return err, [- setupOK], np.nan
self._marginalizeTrainedModel(True)
errRes = super().errorEstimator(mus, return_max)
self._marginalizeTrainedModel(False)
return errRes
def _preliminaryTraining(self):
"""Initialize starting snapshots of solution map."""
RROMPyAssert(self._mode, message = "Cannot start greedy algorithm.")
self._S = self._setSampleBatch(self.S)
self.resetSamples()
self.samplingEngine.scaleFactor = self.scaleFactorDer
musPivot = self.trainSetGenerator.generatePoints(self.S)
while len(musPivot) > self.S: musPivot.pop()
muTestPivot = self.samplerPivot.generatePoints(self.nTestPoints, False)
idxPop = pruneSamples(self.mapParameterListPivot(muTestPivot),
self.mapParameterListPivot(musPivot),
1e-10 * self.scaleFactorPivot[0])
self._mus = emptyParameterList()
self.mus.reset((self.S, self.npar + len(self.musMargLoc)))
muTestBase = emptyParameterList()
muTestBase.reset((len(muTestPivot), self.npar + len(self.musMargLoc)))
for k in range(self.S):
muk = np.empty_like(self.mus[0])
muk[self.directionPivot] = musPivot[k]
muk[self.directionMarginal] = self.musMargLoc
self.mus[k] = muk
for k in range(len(muTestPivot)):
muk = np.empty_like(muTestBase[0])
muk[self.directionPivot] = muTestPivot[k]
muk[self.directionMarginal] = self.musMargLoc
muTestBase[k] = muk
muTestBase.pop(idxPop)
muLast = copy(self.mus[-1])
self.mus.pop()
if len(self.mus) > 0:
vbMng(self, "MAIN",
("Adding first {} sample point{} at {} to training "
"set.").format(self.S - 1, "" + "s" * (self.S > 2),
self.mus), 3)
self.samplingEngine.iterSample(self.mus)
self._S = len(self.mus)
self._approxParameters["S"] = self.S
self.muTest = parameterList(muTestBase)
self.muTest.append(muLast)
self.M, self.N = ("AUTO",) * 2
def setupApproxLocal(self) -> int:
"""Compute rational interpolant."""
self._marginalizeMiscellanea(True)
setupOK = super().setupApproxLocal()
self._marginalizeMiscellanea(False)
return setupOK
def setupApprox(self, *args, **kwargs) -> int:
"""Compute rational interpolant."""
if self.checkComputedApprox(): return -1
RROMPyAssert(self._mode, message = "Cannot setup approximant.")
vbMng(self, "INIT", "Setting up {}.". format(self.name()), 5)
self.computeScaleFactor()
self._musMarginal = self.samplerMarginal.generatePoints(self.SMarginal)
while len(self.musMarginal) > self.SMarginal: self.musMarginal.pop()
S0 = copy(self.S)
idx, sizes = indicesScatter(len(self.musMarginal), return_sizes = True)
pMat, Ps, Qs, mus = None, [], [], None
req, emptyCores = [], np.where(np.logical_not(sizes))[0]
if len(idx) == 0:
vbMng(self, "MAIN", "Idling.", 25)
if self.storeAllSamples: self.storeSamples()
pL, pT, mT = recv(source = 0, tag = poolRank())
pMat = np.empty((pL, 0), dtype = pT)
mus = np.empty((0, self.mu0.shape[1]), dtype = mT)
else:
_scaleFactorOldPivot = copy(self.scaleFactor)
self.scaleFactor = self.scaleFactorPivot
self._temporaryPivot = 1
for i in idx:
self.musMargLoc = self.musMarginal[i]
vbMng(self, "MAIN",
"Building marginal model no. {} at {}.".format(i + 1,
self.musMargLoc), 5)
self.samplingEngine.resetHistory()
self.trainedModel = None
self.verbosity -= 5
self.samplingEngine.verbosity -= 10
RationalInterpolantGreedy.setupApprox(self, *args, **kwargs)
self.verbosity += 5
self.samplingEngine.verbosity += 10
if self.storeAllSamples: self.storeSamples(i)
musi = self.samplingEngine.mus
pMati = self.samplingEngine.projectionMatrix
if not hasattr(self, "matchState") or not self.matchState:
if self.POD == 1 and not (
hasattr(self.HFEngine.C, "is_mu_independent")
and self.HFEngine.C.is_mu_independent in self._output_lvl):
raise RROMPyException(("Cannot apply mu-dependent C "
"to orthonormalized samples."))
vbMng(self, "INIT", "Extracting system output from state.",
35)
- if (hasattr(self.HFEngine, "_is_C_quadratic")
- and self.HFEngine._is_C_quadratic):
- pMati = self.HFEngine.applyC(pMati, musi)
- else:
- pMatiEff = None
- for j, mu in enumerate(musi):
- pMij = np.expand_dims(self.HFEngine.applyC(
+ pMatiEff = None
+ for j, mu in enumerate(musi):
+ pMij = np.expand_dims(self.HFEngine.applyC(
pMati[:, j], mu), -1)
- if pMatiEff is None:
- pMatiEff = np.array(pMij)
- else:
- pMatiEff = np.append(pMatiEff, pMij, axis = 1)
- pMati = pMatiEff
+ if pMatiEff is None:
+ pMatiEff = np.array(pMij)
+ else:
+ pMatiEff = np.append(pMatiEff, pMij, axis = 1)
+ pMati = pMatiEff
vbMng(self, "DEL", "Done extracting system output.", 35)
if pMat is None:
mus = copy(musi.data)
pMat = copy(pMati)
if i == 0:
for dest in emptyCores:
req += [isend((len(pMat), pMat.dtype, mus.dtype),
dest = dest, tag = dest)]
else:
mus = np.vstack((mus, musi.data))
pMat = np.hstack((pMat, pMati))
Ps += [copy(self.trainedModel.data.P)]
Qs += [copy(self.trainedModel.data.Q)]
self._S = S0
del self._temporaryPivot, self.musMargLoc
self.scaleFactor = _scaleFactorOldPivot
for r in req: r.wait()
pMat, Ps, Qs, mus, nsamples = gatherPivotedApproximant(pMat, Ps, Qs,
mus, sizes,
self.polybasis)
self._mus = self.checkParameterList(mus)
Psupp = np.append(0, np.cumsum(nsamples))
self._setupTrainedModel(pMat, forceNew = True)
self.trainedModel.data.Qs, self.trainedModel.data.Ps = Qs, Ps
self.trainedModel.data.Psupp = list(Psupp[: -1])
self._preliminaryMarginalFinalization()
self._finalizeMarginalization()
vbMng(self, "DEL", "Done setting up approximant.", 5)
return 0
class RationalInterpolantGreedyPivotedNoMatch(
RationalInterpolantGreedyPivotedBase,
GenericPivotedApproximantNoMatch):
"""
ROM pivoted rational interpolant (without pole matching) computation for
parametric problems.
Args:
HFEngine: HF problem solver.
mu0(optional): Default parameter. Defaults to 0.
directionPivot(optional): Pivot components. Defaults to [0].
approxParameters(optional): Dictionary containing values for main
parameters of approximant. Recognized keys are:
- 'POD': kind of snapshots orthogonalization; allowed values
include 0, 1/2, and 1; defaults to 1, i.e. POD;
- 'scaleFactorDer': scaling factors for derivative computation;
defaults to 'AUTO';
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator;
- 'polybasis': type of polynomial basis for pivot
interpolation; defaults to 'MONOMIAL';
- 'greedyTol': uniform error tolerance for greedy algorithm;
defaults to 1e-2;
- 'collinearityTol': collinearity tolerance for greedy algorithm;
defaults to 0.;
- 'maxIter': maximum number of greedy steps; defaults to 1e2;
- 'nTestPoints': number of test points; defaults to 5e2;
- 'trainSetGenerator': training sample points generator; defaults
to sampler;
- 'errorEstimatorKind': kind of error estimator; available values
include 'AFFINE', 'DISCREPANCY', 'LOOK_AHEAD',
'LOOK_AHEAD_RES', and 'NONE'; defaults to 'NONE';
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant; defaults to 1;
- 'functionalSolve': strategy for minimization of denominator
functional; allowed values include 'NORM', 'DOMINANT', 'NODAL',
'BARYCENTRIC_NORM', and 'BARYCENTRIC[_AVERAGE]' (check pdf in
main folder for meaning); defaults to 'NORM';
- 'interpTol': tolerance for pivot interpolation; defaults to
None;
- 'QTol': tolerance for robust rational denominator management;
defaults to 0.
Defaults to empty dict.
verbosity(optional): Verbosity level. Defaults to 10.
Attributes:
HFEngine: HF problem solver.
mu0: Default parameter.
directionPivot: Pivot components.
mus: Array of snapshot parameters.
musMarginal: Array of marginal snapshot parameters.
approxParameters: Dictionary containing values for main parameters of
approximant. Recognized keys are in parameterList.
parameterListSoft: Recognized keys of soft approximant parameters:
- 'POD': kind of snapshots orthogonalization;
- 'scaleFactorDer': scaling factors for derivative computation;
- 'polybasis': type of polynomial basis for pivot
interpolation;
- 'greedyTol': uniform error tolerance for greedy algorithm;
- 'collinearityTol': collinearity tolerance for greedy algorithm;
- 'maxIter': maximum number of greedy steps;
- 'nTestPoints': number of test points;
- 'trainSetGenerator': training sample points generator;
- 'errorEstimatorKind': kind of error estimator;
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant;
- 'functionalSolve': strategy for minimization of denominator
functional;
- 'interpTol': tolerance for pivot interpolation;
- 'QTol': tolerance for robust rational denominator management.
parameterListCritical: Recognized keys of critical approximant
parameters:
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator.
verbosity: Verbosity level.
POD: Kind of snapshots orthogonalization.
scaleFactorDer: Scaling factors for derivative computation.
S: Total number of pivot samples current approximant relies upon.
samplerPivot: Pivot sample point generator.
SMarginal: Total number of marginal samples current approximant relies
upon.
samplerMarginal: Marginal sample point generator.
polybasis: Type of polynomial basis for pivot interpolation.
greedyTol: uniform error tolerance for greedy algorithm.
collinearityTol: Collinearity tolerance for greedy algorithm.
maxIter: maximum number of greedy steps.
nTestPoints: number of starting training points.
trainSetGenerator: training sample points generator.
errorEstimatorKind: kind of error estimator.
radialDirectionalWeightsMarginal: Radial basis weights for marginal
interpolant.
functionalSolve: Strategy for minimization of denominator functional.
interpTol: Tolerance for pivot interpolation.
QTol: Tolerance for robust rational denominator management.
muBounds: list of bounds for pivot parameter values.
muBoundsMarginal: list of bounds for marginal parameter values.
samplingEngine: Sampling engine.
uHF: High fidelity solution(s) with parameter(s) lastSolvedHF as
sampleList.
lastSolvedHF: Parameter(s) corresponding to last computed high fidelity
solution(s) as parameterList.
uApproxReduced: Reduced approximate solution(s) with parameter(s)
lastSolvedApprox as sampleList.
lastSolvedApproxReduced: Parameter(s) corresponding to last computed
reduced approximate solution(s) as parameterList.
uApprox: Approximate solution(s) with parameter(s) lastSolvedApprox as
sampleList.
lastSolvedApprox: Parameter(s) corresponding to last computed
approximate solution(s) as parameterList.
Q: Numpy 1D vector containing complex coefficients of approximant
denominator.
P: Numpy 2D vector whose columns are FE dofs of coefficients of
approximant numerator.
"""
class RationalInterpolantGreedyPivotedPoleMatch(
RationalInterpolantGreedyPivotedBase,
GenericPivotedApproximantPoleMatch):
"""
ROM pivoted rational interpolant (with pole matching) computation for
parametric problems.
Args:
HFEngine: HF problem solver.
mu0(optional): Default parameter. Defaults to 0.
directionPivot(optional): Pivot components. Defaults to [0].
approxParameters(optional): Dictionary containing values for main
parameters of approximant. Recognized keys are:
- 'POD': kind of snapshots orthogonalization; allowed values
include 0, 1/2, and 1; defaults to 1, i.e. POD;
- 'scaleFactorDer': scaling factors for derivative computation;
defaults to 'AUTO';
- 'matchState': whether to match the system state rather than the
system output; defaults to False;
- 'matchingWeight': weight for pole matching optimization; defaults
to 1;
- 'matchingChordalRadius': radius to be used in chordal metric for
poles and residues; if <= 0, Euclidean metric is used; if
'AUTO', automatically selected; defaults to -1;
- 'matchingShared': required ratio of marginal points to share
resonance; defaults to 1.;
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator;
- 'polybasis': type of polynomial basis for pivot
interpolation; defaults to 'MONOMIAL';
- 'polybasisMarginal': type of polynomial basis for marginal
interpolation; allowed values include 'MONOMIAL_*',
'CHEBYSHEV_*', 'LEGENDRE_*', 'NEARESTNEIGHBOR', and
'PIECEWISE_LINEAR_*'; defaults to 'MONOMIAL';
- 'paramsMarginal': dictionary of parameters for marginal
interpolation; include:
. 'MMarginal': degree of marginal interpolant; defaults to
'AUTO', i.e. maximum allowed; not for 'NEARESTNEIGHBOR' or
'PIECEWISE_LINEAR_*';
. 'nNeighborsMarginal': number of marginal nearest neighbors;
defaults to 1; only for 'NEARESTNEIGHBOR';
. 'polydegreetypeMarginal': type of polynomial degree for
marginal; defaults to 'TOTAL'; not for 'NEARESTNEIGHBOR' or
'PIECEWISE_LINEAR_*';
. 'interpTolMarginal': tolerance for marginal interpolation;
defaults to None; not for 'NEARESTNEIGHBOR' or
'PIECEWISE_LINEAR_*';
. 'radialDirectionalWeightsMarginalAdapt': bounds for adaptive
rescaling of marginal radial basis weights; only for
radial basis.
- 'greedyTol': uniform error tolerance for greedy algorithm;
defaults to 1e-2;
- 'collinearityTol': collinearity tolerance for greedy algorithm;
defaults to 0.;
- 'maxIter': maximum number of greedy steps; defaults to 1e2;
- 'nTestPoints': number of test points; defaults to 5e2;
- 'trainSetGenerator': training sample points generator; defaults
to sampler;
- 'errorEstimatorKind': kind of error estimator; available values
include 'AFFINE', 'DISCREPANCY', 'LOOK_AHEAD',
'LOOK_AHEAD_RES', and 'NONE'; defaults to 'NONE';
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant; defaults to 1;
- 'functionalSolve': strategy for minimization of denominator
functional; allowed values include 'NORM', 'DOMINANT', 'NODAL',
'BARYCENTRIC_NORM', and 'BARYCENTRIC[_AVERAGE]' (check pdf in
main folder for meaning); defaults to 'NORM';
- 'interpTol': tolerance for pivot interpolation; defaults to None;
- 'QTol': tolerance for robust rational denominator management;
defaults to 0.
Defaults to empty dict.
verbosity(optional): Verbosity level. Defaults to 10.
Attributes:
HFEngine: HF problem solver.
mu0: Default parameter.
directionPivot: Pivot components.
mus: Array of snapshot parameters.
musMarginal: Array of marginal snapshot parameters.
approxParameters: Dictionary containing values for main parameters of
approximant. Recognized keys are in parameterList.
parameterListSoft: Recognized keys of soft approximant parameters:
- 'POD': kind of snapshots orthogonalization;
- 'scaleFactorDer': scaling factors for derivative computation;
- 'matchState': whether to match the system state rather than the
system output;
- 'matchingWeight': weight for pole matching optimization;
- 'matchingChordalRadius': radius to be used in chordal metric for
poles and residues;
- 'matchingShared': required ratio of marginal points to share
resonance;
- 'polybasis': type of polynomial basis for pivot
interpolation;
- 'polybasisMarginal': type of polynomial basis for marginal
interpolation;
- 'paramsMarginal': dictionary of parameters for marginal
interpolation; include:
. 'MMarginal': degree of marginal interpolant;
. 'nNeighborsMarginal': number of marginal nearest neighbors;
. 'polydegreetypeMarginal': type of polynomial degree for
marginal;
. 'interpTolMarginal': tolerance for marginal interpolation;
. 'radialDirectionalWeightsMarginalAdapt': bounds for adaptive
rescaling of marginal radial basis weights.
- 'greedyTol': uniform error tolerance for greedy algorithm;
- 'collinearityTol': collinearity tolerance for greedy algorithm;
- 'maxIter': maximum number of greedy steps;
- 'nTestPoints': number of test points;
- 'trainSetGenerator': training sample points generator;
- 'errorEstimatorKind': kind of error estimator;
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant;
- 'functionalSolve': strategy for minimization of denominator
functional;
- 'interpTol': tolerance for pivot interpolation;
- 'QTol': tolerance for robust rational denominator management.
parameterListCritical: Recognized keys of critical approximant
parameters:
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator.
verbosity: Verbosity level.
POD: Kind of snapshots orthogonalization.
scaleFactorDer: Scaling factors for derivative computation.
matchState: Whether to match the system state rather than the system
output.
matchingWeight: Weight for pole matching optimization.
matchingChordalRadius: Radius to be used in chordal metric for poles
and residues.
matchingShared: Required ratio of marginal points to share resonance.
S: Total number of pivot samples current approximant relies upon.
samplerPivot: Pivot sample point generator.
SMarginal: Total number of marginal samples current approximant relies
upon.
samplerMarginal: Marginal sample point generator.
polybasis: Type of polynomial basis for pivot interpolation.
polybasisMarginal: Type of polynomial basis for marginal interpolation.
paramsMarginal: Dictionary of parameters for marginal interpolation.
greedyTol: uniform error tolerance for greedy algorithm.
collinearityTol: Collinearity tolerance for greedy algorithm.
maxIter: maximum number of greedy steps.
nTestPoints: number of starting training points.
trainSetGenerator: training sample points generator.
errorEstimatorKind: kind of error estimator.
radialDirectionalWeightsMarginal: Radial basis weights for marginal
interpolant.
functionalSolve: Strategy for minimization of denominator functional.
interpTol: Tolerance for pivot interpolation.
QTol: Tolerance for robust rational denominator management.
muBounds: list of bounds for pivot parameter values.
muBoundsMarginal: list of bounds for marginal parameter values.
samplingEngine: Sampling engine.
uHF: High fidelity solution(s) with parameter(s) lastSolvedHF as
sampleList.
lastSolvedHF: Parameter(s) corresponding to last computed high fidelity
solution(s) as parameterList.
uApproxReduced: Reduced approximate solution(s) with parameter(s)
lastSolvedApprox as sampleList.
lastSolvedApproxReduced: Parameter(s) corresponding to last computed
reduced approximate solution(s) as parameterList.
uApprox: Approximate solution(s) with parameter(s) lastSolvedApprox as
sampleList.
lastSolvedApprox: Parameter(s) corresponding to last computed
approximate solution(s) as parameterList.
Q: Numpy 1D vector containing complex coefficients of approximant
denominator.
P: Numpy 2D vector whose columns are FE dofs of coefficients of
approximant numerator.
"""
def setupApprox(self, *args, **kwargs) -> int:
if self.checkComputedApprox(): return -1
self.purgeparamsMarginal()
setupOK = super().setupApprox(*args, **kwargs)
if self.matchState: self._postApplyC()
return setupOK
diff --git a/rrompy/reduction_methods/pivoted/rational_interpolant_pivoted.py b/rrompy/reduction_methods/pivoted/rational_interpolant_pivoted.py
index 38b8170..d3cec18 100644
--- a/rrompy/reduction_methods/pivoted/rational_interpolant_pivoted.py
+++ b/rrompy/reduction_methods/pivoted/rational_interpolant_pivoted.py
@@ -1,491 +1,487 @@
# Copyright (C) 2018-2020 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 .
#
import numpy as np
from collections.abc import Iterable
from copy import deepcopy as copy
from .generic_pivoted_approximant import (GenericPivotedApproximantBase,
GenericPivotedApproximantNoMatch,
GenericPivotedApproximantPoleMatch)
from .gather_pivoted_approximant import gatherPivotedApproximant
from rrompy.reduction_methods.standard.rational_interpolant import (
RationalInterpolant)
from rrompy.utilities.base import verbosityManager as vbMng
from rrompy.utilities.numerical.hash_derivative import nextDerivativeIndices
from rrompy.utilities.exception_manager import (RROMPyException, RROMPyAssert,
RROMPyWarning)
from rrompy.parameter import emptyParameterList
from rrompy.utilities.parallel import poolRank, indicesScatter, isend, recv
__all__ = ['RationalInterpolantPivotedNoMatch',
'RationalInterpolantPivotedPoleMatch']
class RationalInterpolantPivotedBase(GenericPivotedApproximantBase,
RationalInterpolant):
def __init__(self, *args, **kwargs):
self._preInit()
self._addParametersToList(toBeExcluded = ["polydegreetype"])
super().__init__(*args, **kwargs)
if self.nparPivot > 1: self.HFEngine._ignoreResidues = 1
self._postInit()
@property
def scaleFactorDer(self):
"""Value of scaleFactorDer."""
if self._scaleFactorDer == "NONE": return 1.
if self._scaleFactorDer == "AUTO": return self.scaleFactorPivot
return self._scaleFactorDer
@scaleFactorDer.setter
def scaleFactorDer(self, scaleFactorDer):
if isinstance(scaleFactorDer, (str,)):
scaleFactorDer = scaleFactorDer.upper()
elif isinstance(scaleFactorDer, Iterable):
scaleFactorDer = list(scaleFactorDer)
self._scaleFactorDer = scaleFactorDer
self._approxParameters["scaleFactorDer"] = self._scaleFactorDer
@property
def polydegreetype(self):
"""Value of polydegreetype."""
return "TOTAL"
@polydegreetype.setter
def polydegreetype(self, polydegreetype):
RROMPyWarning(("polydegreetype is used just to simplify inheritance, "
"and its value cannot be changed from 'TOTAL'."))
def _setupInterpolationIndices(self):
"""Setup parameters for polyvander."""
RROMPyAssert(self._mode,
message = "Cannot setup interpolation indices.")
if (self._musUniqueCN is None
or len(self._reorder) != len(self.musPivot)):
try:
muPC = self.trainedModel.centerNormalizePivot(self.musPivot)
except:
muPC = self.trainedModel.centerNormalize(self.musPivot)
self._musUniqueCN, musIdxsTo, musIdxs, musCount = (muPC.unique(
return_index = True, return_inverse = True,
return_counts = True))
self._musUnique = self.musPivot[musIdxsTo]
self._derIdxs = [None] * len(self._musUniqueCN)
self._reorder = np.empty(len(musIdxs), dtype = int)
filled = 0
for j, cnt in enumerate(musCount):
self._derIdxs[j] = nextDerivativeIndices([], self.nparPivot,
cnt)
jIdx = np.nonzero(musIdxs == j)[0]
self._reorder[jIdx] = np.arange(filled, filled + cnt)
filled += cnt
def setupApprox(self) -> int:
"""Compute rational interpolant."""
if self.checkComputedApprox(): return -1
RROMPyAssert(self._mode, message = "Cannot setup approximant.")
vbMng(self, "INIT", "Setting up {}.". format(self.name()), 5)
self.computeScaleFactor()
self.resetSamples()
self.samplingEngine.scaleFactor = self.scaleFactorDer
self.musPivot = self.samplerPivot.generatePoints(self.S)
while len(self.musPivot) > self.S: self.musPivot.pop()
self._musMarginal = self.samplerMarginal.generatePoints(self.SMarginal)
while len(self.musMarginal) > self.SMarginal: self.musMarginal.pop()
self._mus = emptyParameterList()
self.mus.reset((self.S * self.SMarginal, self.HFEngine.npar))
for j, muMarg in enumerate(self.musMarginal):
for k in range(j * self.S, (j + 1) * self.S):
muk = np.empty_like(self.mus[0])
muk[self.directionPivot] = self.musPivot[k - j * self.S]
muk[self.directionMarginal] = muMarg
self.mus[k] = muk
N0 = copy(self.N)
self._setupTrainedModel(np.zeros((0, 0)), forceNew = True)
idx, sizes = indicesScatter(len(self.musMarginal), return_sizes = True)
pMat, Ps, Qs = None, [], []
req, emptyCores = [], np.where(np.logical_not(sizes))[0]
if len(idx) == 0:
vbMng(self, "MAIN", "Idling.", 30)
if self.storeAllSamples: self.storeSamples()
pL, pT = recv(source = 0, tag = poolRank())
pMat = np.empty((pL, 0), dtype = pT)
else:
_scaleFactorOldPivot = copy(self.scaleFactor)
self.scaleFactor = self.scaleFactorPivot
self._temporaryPivot = 1
for i in idx:
musi = self.mus[self.S * i : self.S * (i + 1)]
vbMng(self, "MAIN",
"Building marginal model no. {} at {}.".format(i + 1,
self.musMarginal[i]), 5)
vbMng(self, "INIT", "Starting computation of snapshots.", 10)
self.samplingEngine.resetHistory()
self.samplingEngine.iterSample(musi)
vbMng(self, "DEL", "Done computing snapshots.", 10)
self.verbosity -= 5
self.samplingEngine.verbosity -= 10
self._setupRational(self._setupDenominator())
self.verbosity += 5
self.samplingEngine.verbosity += 10
if self.storeAllSamples: self.storeSamples(i)
pMati = self.samplingEngine.projectionMatrix
if not hasattr(self, "matchState") or not self.matchState:
if self.POD == 1 and not (
hasattr(self.HFEngine.C, "is_mu_independent")
and self.HFEngine.C.is_mu_independent in self._output_lvl):
raise RROMPyException(("Cannot apply mu-dependent C "
"to orthonormalized samples."))
vbMng(self, "INIT", "Extracting system output from state.",
35)
- if (hasattr(self.HFEngine, "_is_C_quadratic")
- and self.HFEngine._is_C_quadratic):
- pMati = self.HFEngine.applyC(pMati, musi)
- else:
- pMatiEff = None
- for j, mu in enumerate(musi):
- pMij = np.expand_dims(self.HFEngine.applyC(
+ pMatiEff = None
+ for j, mu in enumerate(musi):
+ pMij = np.expand_dims(self.HFEngine.applyC(
pMati[:, j], mu), -1)
- if pMatiEff is None:
- pMatiEff = np.array(pMij)
- else:
- pMatiEff = np.append(pMatiEff, pMij, axis = 1)
- pMati = pMatiEff
+ if pMatiEff is None:
+ pMatiEff = np.array(pMij)
+ else:
+ pMatiEff = np.append(pMatiEff, pMij, axis = 1)
+ pMati = pMatiEff
vbMng(self, "DEL", "Done extracting system output.", 35)
if pMat is None:
pMat = copy(pMati)
if i == 0:
for dest in emptyCores:
req += [isend((len(pMat), pMat.dtype), dest = dest,
tag = dest)]
else:
pMat = np.hstack((pMat, pMati))
Ps += [copy(self.trainedModel.data.P)]
Qs += [copy(self.trainedModel.data.Q)]
del self.trainedModel.data.Q, self.trainedModel.data.P
self.N = N0
del self._temporaryPivot
self.scaleFactor = _scaleFactorOldPivot
for r in req: r.wait()
pMat, Ps, Qs, _, _ = gatherPivotedApproximant(pMat, Ps, Qs,
self.mus.data, sizes,
self.polybasis, False)
self._setupTrainedModel(pMat)
self.trainedModel.data.Qs, self.trainedModel.data.Ps = Qs, Ps
Psupp = np.arange(0, len(self.musMarginal) * self.S, self.S)
self.trainedModel.data.Psupp = list(Psupp)
self._preliminaryMarginalFinalization()
self._finalizeMarginalization()
vbMng(self, "DEL", "Done setting up approximant.", 5)
return 0
class RationalInterpolantPivotedNoMatch(RationalInterpolantPivotedBase,
GenericPivotedApproximantNoMatch):
"""
ROM pivoted rational interpolant (without pole matching) computation for
parametric problems.
Args:
HFEngine: HF problem solver.
mu0(optional): Default parameter. Defaults to 0.
directionPivot(optional): Pivot components. Defaults to [0].
approxParameters(optional): Dictionary containing values for main
parameters of approximant. Recognized keys are:
- 'POD': kind of snapshots orthogonalization; allowed values
include 0, 1/2, and 1; defaults to 1, i.e. POD;
- 'scaleFactorDer': scaling factors for derivative computation;
defaults to 'AUTO';
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator;
- 'polybasis': type of polynomial basis for pivot
interpolation; defaults to 'MONOMIAL';
- 'M': degree of rational interpolant numerator; defaults to
'AUTO', i.e. maximum allowed;
- 'N': degree of rational interpolant denominator; defaults to
'AUTO', i.e. maximum allowed;
- 'radialDirectionalWeights': radial basis weights for pivot
numerator; defaults to 1;
- 'radialDirectionalWeightsAdapt': bounds for adaptive rescaling of
radial basis weights; defaults to [-1, -1];
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant; defaults to 1;
- 'functionalSolve': strategy for minimization of denominator
functional; allowed values include 'NORM', 'DOMINANT', 'NODAL',
'BARYCENTRIC_NORM', and 'BARYCENTRIC[_AVERAGE]' (check pdf in
main folder for meaning); defaults to 'NORM';
- 'interpTol': tolerance for pivot interpolation; defaults to
None;
- 'QTol': tolerance for robust rational denominator management;
defaults to 0.
Defaults to empty dict.
verbosity(optional): Verbosity level. Defaults to 10.
Attributes:
HFEngine: HF problem solver.
mu0: Default parameter.
directionPivot: Pivot components.
mus: Array of snapshot parameters.
musPivot: Array of pivot snapshot parameters.
musMarginal: Array of marginal snapshot parameters.
approxParameters: Dictionary containing values for main parameters of
approximant. Recognized keys are in parameterList.
parameterListSoft: Recognized keys of soft approximant parameters:
- 'POD': kind of snapshots orthogonalization;
- 'scaleFactorDer': scaling factors for derivative computation;
- 'polybasis': type of polynomial basis for pivot
interpolation;
- 'M': degree of rational interpolant numerator;
- 'N': degree of rational interpolant denominator;
- 'radialDirectionalWeights': radial basis weights for pivot
numerator;
- 'radialDirectionalWeightsAdapt': bounds for adaptive rescaling of
radial basis weights;
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant;
- 'functionalSolve': strategy for minimization of denominator
functional;
- 'interpTol': tolerance for pivot interpolation;
- 'QTol': tolerance for robust rational denominator management.
parameterListCritical: Recognized keys of critical approximant
parameters:
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator.
verbosity: Verbosity level.
POD: Kind of snapshots orthogonalization.
scaleFactorDer: Scaling factors for derivative computation.
S: Total number of pivot samples current approximant relies upon.
samplerPivot: Pivot sample point generator.
SMarginal: Total number of marginal samples current approximant relies
upon.
samplerMarginal: Marginal sample point generator.
polybasis: Type of polynomial basis for pivot interpolation.
M: Numerator degree of approximant.
N: Denominator degree of approximant.
radialDirectionalWeights: Radial basis weights for pivot numerator.
radialDirectionalWeightsAdapt: Bounds for adaptive rescaling of radial
basis weights.
radialDirectionalWeightsMarginal: Radial basis weights for marginal
interpolant.
functionalSolve: Strategy for minimization of denominator functional.
interpTol: Tolerance for pivot interpolation.
QTol: Tolerance for robust rational denominator management.
muBounds: list of bounds for pivot parameter values.
muBoundsMarginal: list of bounds for marginal parameter values.
samplingEngine: Sampling engine.
uHF: High fidelity solution(s) with parameter(s) lastSolvedHF as
sampleList.
lastSolvedHF: Parameter(s) corresponding to last computed high fidelity
solution(s) as parameterList.
uApproxReduced: Reduced approximate solution(s) with parameter(s)
lastSolvedApprox as sampleList.
lastSolvedApproxReduced: Parameter(s) corresponding to last computed
reduced approximate solution(s) as parameterList.
uApprox: Approximate solution(s) with parameter(s) lastSolvedApprox as
sampleList.
lastSolvedApprox: Parameter(s) corresponding to last computed
approximate solution(s) as parameterList.
Q: Numpy 1D vector containing complex coefficients of approximant
denominator.
P: Numpy 2D vector whose columns are FE dofs of coefficients of
approximant numerator.
"""
class RationalInterpolantPivotedPoleMatch(RationalInterpolantPivotedBase,
GenericPivotedApproximantPoleMatch):
"""
ROM pivoted rational interpolant (with pole matching) computation for
parametric problems.
Args:
HFEngine: HF problem solver.
mu0(optional): Default parameter. Defaults to 0.
directionPivot(optional): Pivot components. Defaults to [0].
approxParameters(optional): Dictionary containing values for main
parameters of approximant. Recognized keys are:
- 'POD': kind of snapshots orthogonalization; allowed values
include 0, 1/2, and 1; defaults to 1, i.e. POD;
- 'scaleFactorDer': scaling factors for derivative computation;
defaults to 'AUTO';
- 'matchState': whether to match the system state rather than the
system output; defaults to False;
- 'matchingWeight': weight for pole matching optimization; defaults
to 1;
- 'matchingChordalRadius': radius to be used in chordal metric for
poles and residues; if <= 0, Euclidean metric is used; if
'AUTO', automatically selected; defaults to -1;
- 'matchingShared': required ratio of marginal points to share
resonance; defaults to 1.;
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator;
- 'polybasis': type of polynomial basis for pivot
interpolation; defaults to 'MONOMIAL';
- 'polybasisMarginal': type of polynomial basis for marginal
interpolation; allowed values include 'MONOMIAL_*',
'CHEBYSHEV_*', 'LEGENDRE_*', 'NEARESTNEIGHBOR', and
'PIECEWISE_LINEAR_*'; defaults to 'MONOMIAL';
- 'paramsMarginal': dictionary of parameters for marginal
interpolation; include:
. 'MMarginal': degree of marginal interpolant; defaults to
'AUTO', i.e. maximum allowed; not for 'NEARESTNEIGHBOR' or
'PIECEWISE_LINEAR_*';
. 'nNeighborsMarginal': number of marginal nearest neighbors;
defaults to 1; only for 'NEARESTNEIGHBOR';
. 'polydegreetypeMarginal': type of polynomial degree for
marginal; defaults to 'TOTAL'; not for 'NEARESTNEIGHBOR' or
'PIECEWISE_LINEAR_*';
. 'interpTolMarginal': tolerance for marginal interpolation;
defaults to None; not for 'NEARESTNEIGHBOR' or
'PIECEWISE_LINEAR_*';
. 'radialDirectionalWeightsMarginalAdapt': bounds for adaptive
rescaling of marginal radial basis weights; only for
radial basis.
- 'M': degree of rational interpolant numerator; defaults to
'AUTO', i.e. maximum allowed;
- 'N': degree of rational interpolant denominator; defaults to
'AUTO', i.e. maximum allowed;
- 'radialDirectionalWeights': radial basis weights for pivot
numerator; defaults to 1;
- 'radialDirectionalWeightsAdapt': bounds for adaptive rescaling of
radial basis weights; defaults to [-1, -1];
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant; defaults to 1;
- 'functionalSolve': strategy for minimization of denominator
functional; allowed values include 'NORM', 'DOMINANT', 'NODAL',
'BARYCENTRIC_NORM', and 'BARYCENTRIC[_AVERAGE]' (check pdf in
main folder for meaning); defaults to 'NORM';
- 'interpTol': tolerance for pivot interpolation; defaults to None;
- 'QTol': tolerance for robust rational denominator management;
defaults to 0.
Defaults to empty dict.
verbosity(optional): Verbosity level. Defaults to 10.
Attributes:
HFEngine: HF problem solver.
mu0: Default parameter.
directionPivot: Pivot components.
mus: Array of snapshot parameters.
musPivot: Array of pivot snapshot parameters.
musMarginal: Array of marginal snapshot parameters.
approxParameters: Dictionary containing values for main parameters of
approximant. Recognized keys are in parameterList.
parameterListSoft: Recognized keys of soft approximant parameters:
- 'POD': kind of snapshots orthogonalization;
- 'scaleFactorDer': scaling factors for derivative computation;
- 'matchState': whether to match the system state rather than the
system output;
- 'matchingWeight': weight for pole matching optimization;
- 'matchingChordalRadius': radius to be used in chordal metric for
poles and residues;
- 'matchingShared': required ratio of marginal points to share
resonance;
- 'polybasis': type of polynomial basis for pivot
interpolation;
- 'polybasisMarginal': type of polynomial basis for marginal
interpolation;
- 'paramsMarginal': dictionary of parameters for marginal
interpolation; include:
. 'MMarginal': degree of marginal interpolant;
. 'nNeighborsMarginal': number of marginal nearest neighbors;
. 'polydegreetypeMarginal': type of polynomial degree for
marginal;
. 'interpTolMarginal': tolerance for marginal interpolation;
. 'radialDirectionalWeightsMarginalAdapt': bounds for adaptive
rescaling of marginal radial basis weights.
- 'M': degree of rational interpolant numerator;
- 'N': degree of rational interpolant denominator;
- 'radialDirectionalWeights': radial basis weights for pivot
numerator;
- 'radialDirectionalWeightsAdapt': bounds for adaptive rescaling of
radial basis weights;
- 'radialDirectionalWeightsMarginal': radial basis weights for
marginal interpolant;
- 'functionalSolve': strategy for minimization of denominator
functional;
- 'interpTol': tolerance for pivot interpolation;
- 'QTol': tolerance for robust rational denominator management.
parameterListCritical: Recognized keys of critical approximant
parameters:
- 'S': total number of pivot samples current approximant relies
upon;
- 'samplerPivot': pivot sample point generator;
- 'SMarginal': total number of marginal samples current approximant
relies upon;
- 'samplerMarginal': marginal sample point generator.
verbosity: Verbosity level.
POD: Kind of snapshots orthogonalization.
scaleFactorDer: Scaling factors for derivative computation.
matchState: Whether to match the system state rather than the system
output.
matchingWeight: Weight for pole matching optimization.
matchingChordalRadius: Radius to be used in chordal metric for poles
and residues.
matchingShared: Required ratio of marginal points to share resonance.
S: Total number of pivot samples current approximant relies upon.
samplerPivot: Pivot sample point generator.
SMarginal: Total number of marginal samples current approximant relies
upon.
samplerMarginal: Marginal sample point generator.
polybasis: Type of polynomial basis for pivot interpolation.
polybasisMarginal: Type of polynomial basis for marginal interpolation.
paramsMarginal: Dictionary of parameters for marginal interpolation.
M: Numerator degree of approximant.
N: Denominator degree of approximant.
radialDirectionalWeights: Radial basis weights for pivot numerator.
radialDirectionalWeightsAdapt: Bounds for adaptive rescaling of radial
basis weights.
radialDirectionalWeightsMarginal: Radial basis weights for marginal
interpolant.
functionalSolve: Strategy for minimization of denominator functional.
interpTol: Tolerance for pivot interpolation.
QTol: Tolerance for robust rational denominator management.
muBounds: list of bounds for pivot parameter values.
muBoundsMarginal: list of bounds for marginal parameter values.
samplingEngine: Sampling engine.
uHF: High fidelity solution(s) with parameter(s) lastSolvedHF as
sampleList.
lastSolvedHF: Parameter(s) corresponding to last computed high fidelity
solution(s) as parameterList.
uApproxReduced: Reduced approximate solution(s) with parameter(s)
lastSolvedApprox as sampleList.
lastSolvedApproxReduced: Parameter(s) corresponding to last computed
reduced approximate solution(s) as parameterList.
uApprox: Approximate solution(s) with parameter(s) lastSolvedApprox as
sampleList.
lastSolvedApprox: Parameter(s) corresponding to last computed
approximate solution(s) as parameterList.
Q: Numpy 1D vector containing complex coefficients of approximant
denominator.
P: Numpy 2D vector whose columns are FE dofs of coefficients of
approximant numerator.
"""
def setupApprox(self, *args, **kwargs) -> int:
if self.checkComputedApprox(): return -1
self.purgeparamsMarginal()
setupOK = super().setupApprox(*args, **kwargs)
if self.matchState: self._postApplyC()
return setupOK
diff --git a/rrompy/reduction_methods/pivoted/trained_model/convert_trained_model_pivoted.py b/rrompy/reduction_methods/pivoted/trained_model/convert_trained_model_pivoted.py
index 5a4a5eb..f4ede6c 100644
--- a/rrompy/reduction_methods/pivoted/trained_model/convert_trained_model_pivoted.py
+++ b/rrompy/reduction_methods/pivoted/trained_model/convert_trained_model_pivoted.py
@@ -1,68 +1,68 @@
# Copyright (C) 2018-2020 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 .
#
from .trained_model_pivoted_rational_nomatch import (
TrainedModelPivotedRationalNoMatch)
from .trained_model_pivoted_rational_polematch import (
TrainedModelPivotedRationalPoleMatch)
from rrompy.utilities.base import verbosityManager as vbMng
from rrompy.utilities.exception_manager import RROMPyException, RROMPyWarning
__all__ = ['convertTrainedModelPivoted']
def convertTrainedModelPivoted(model, outType, verbObj = None,
muteWarnings : bool = False):
if isinstance(model, outType): return model
if ((isinstance(model, TrainedModelPivotedRationalNoMatch)
and outType == TrainedModelPivotedRationalPoleMatch)
or (isinstance(model, TrainedModelPivotedRationalPoleMatch)
and outType == TrainedModelPivotedRationalNoMatch)):
return convertTrainedModelPivotedMatchUnmatch(model, outType, verbObj,
muteWarnings)
raise RROMPyException(("Model type or conversion output type not "
"recognized."))
def convertTrainedModelPivotedMatchUnmatch(model, outType, verbObj = None,
muteWarnings : bool = False):
if verbObj is not None:
sf, st = ["NoMatch", "PoleMatch"]
if outType == TrainedModelPivotedRationalPoleMatch:
msgw = "match poles, set up marginalInterp,"
else: #if outType == TrainedModelPivotedRationalNoMatch:
st, sf = sf, st
msgw = "set up marginalInterp"
vbMng(verbObj, "INIT",
"Starting model conversion from {} to {} model.".format(sf, st),
10)
excludeDataKey = ["marginalInterp", "approxParameters"]
if outType == TrainedModelPivotedRationalPoleMatch:
modelC = TrainedModelPivotedRationalPoleMatch()
msgw = "match poles, set up marginalInterp,"
else: #if outType == TrainedModelPivotedRationalNoMatch:
modelC = TrainedModelPivotedRationalNoMatch()
msgw = "set up marginalInterp"
excludeDataKey += ["HIs", "suppEffPts", "suppEffIdx", "coeffsEff",
- "polesEff"]
+ "polesEff", "projGramian"]
for key in model.__dict__.keys(): setattr(modelC, key, model.__dict__[key])
for key in excludeDataKey: delattr(modelC.data, key)
if verbObj is not None:
vbMng(verbObj, "DEL", "Finished model conversion.", 10)
if not muteWarnings:
RROMPyWarning(("Model conversion result not yet fuctional: must stil "
"{} and assign approxParameters.").format(msgw))
return modelC
diff --git a/rrompy/reduction_methods/pivoted/trained_model/trained_model_pivoted_rational_nomatch.py b/rrompy/reduction_methods/pivoted/trained_model/trained_model_pivoted_rational_nomatch.py
index e9aa8ba..3eb6414 100644
--- a/rrompy/reduction_methods/pivoted/trained_model/trained_model_pivoted_rational_nomatch.py
+++ b/rrompy/reduction_methods/pivoted/trained_model/trained_model_pivoted_rational_nomatch.py
@@ -1,293 +1,252 @@
# Copyright (C) 2018-2020 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 .
#
import numpy as np
from collections.abc import Iterable
from rrompy.reduction_methods.standard.trained_model.trained_model_rational \
import TrainedModelRational
from rrompy.utilities.base.types import (Np1D, Np2D, List, ListAny, paramVal,
paramList, sampList)
from rrompy.utilities.base import verbosityManager as vbMng, freepar as fp
from rrompy.utilities.numerical import dot
from rrompy.utilities.numerical.compress_matrix import compressMatrix
from rrompy.utilities.poly_fitting.heaviside import rational2heaviside
from rrompy.utilities.poly_fitting.nearest_neighbor import (
NearestNeighborInterpolator as NNI)
from rrompy.utilities.exception_manager import (RROMPyException, RROMPyAssert,
RROMPyWarning)
from rrompy.parameter import checkParameterList
from rrompy.sampling import emptySampleList
__all__ = ['TrainedModelPivotedRationalNoMatch']
class TrainedModelPivotedRationalNoMatch(TrainedModelRational):
"""
ROM approximant evaluation for pivoted approximants based on interpolation
of rational approximants (without pole matching).
Attributes:
Data: dictionary with all that can be pickled.
"""
- @property
- def supportEnd(self):
- # for a quadratic output, it is different from data.projMat.shape[1]
- lookAtExcl = hasattr(self, "_PsuppExcl") and len(self._PsuppExcl) > 0
- idxIncl = np.argmax(self.data.Psupp)
- if lookAtExcl: idxExcl = np.argmax(self._PsuppExcl)
- if (not lookAtExcl
- or self.data.Psupp[idxIncl] >= self._PsuppExcl[idxExcl]):
- return self.data.Psupp[idxIncl] + self.data.Ps[idxIncl].shape[0]
- return self._PsuppExcl[idxExcl] + self._PsExcl[idxExcl].shape[0]
-
def checkParameterListPivot(self, mu:paramList,
check_if_single : bool = False) -> paramList:
return checkParameterList(mu, self.data.nparPivot, check_if_single)
def checkParameterListMarginal(self, mu:paramList,
check_if_single : bool = False) -> paramList:
return checkParameterList(mu, self.data.nparMarginal, check_if_single)
def compress(self, collapse : bool = False, tol : float = 0.,
returnRMat : bool = False, **compressMatrixkwargs):
if not collapse and tol <= 0.: return
- if hasattr(self.data, "_is_C_quadratic") and self.data._is_C_quadratic:
- raise RROMPyException(("Cannot compress model with quadratic "
- "output."))
RMat = self.data.projMat
if not collapse:
if hasattr(self.data, "_compressTol"):
RROMPyWarning(("Recompressing already compressed model is "
"ineffective. Aborting."))
return
self.data.projMat, RMat, _ = compressMatrix(RMat, tol,
**compressMatrixkwargs)
if hasattr(self.data, "Ps"):
for obj, suppj in zip(self.data.Ps, self.data.Psupp):
obj.postmultiplyTensorize(RMat.T[suppj : suppj + obj.shape[0]])
if hasattr(self, "_PsExcl"):
for obj, suppj in zip(self._PsExcl, self._PsuppExcl):
obj.postmultiplyTensorize(RMat.T[suppj : suppj + obj.shape[0]])
self._PsuppExcl = [0] * len(self._PsuppExcl)
self.data.Psupp = [0] * len(self.data.Psupp)
super(TrainedModelRational, self).compress(collapse, tol)
if returnRMat: return RMat
def centerNormalizePivot(self, mu : paramList = [],
mu0 : paramVal = None) -> paramList:
"""
Compute normalized parameter to be plugged into approximant.
Args:
mu: Parameter(s) 1.
mu0: Parameter(s) 2. If None, set to self.data.mu0Pivot.
Returns:
Normalized parameter.
"""
mu = self.checkParameterListPivot(mu)
if mu0 is None:
mu0 = self.checkParameterListPivot(
self.data.mu0(0, self.data.directionPivot))
return (self.mapParameterList(mu, idx = self.data.directionPivot)
- self.mapParameterList(mu0, idx = self.data.directionPivot)
) / [self.data.scaleFactor[x] for x in self.data.directionPivot]
def setupMarginalInterp(self, interpPars:ListAny):
self.data.marginalInterp = NNI()
self.data.marginalInterp.setupByInterpolation(self.data.musMarginal,
np.arange(len(self.data.musMarginal)),
1, *interpPars)
def updateEffectiveSamples(self, exclude:List[int]):
if hasattr(self, "_idxExcl"):
for j, excl in enumerate(self._idxExcl):
self.data.musMarginal.insert(self._musMExcl[j], excl)
self.data.Ps.insert(excl, self._PsExcl[j])
self.data.Qs.insert(excl, self._QsExcl[j])
self.data.Psupp.insert(excl, self._PsuppExcl[j])
self._idxExcl, self._musMExcl = list(np.sort(exclude)), []
self._PsExcl, self._QsExcl, self._PsuppExcl = [], [], []
for excl in self._idxExcl[::-1]:
self._musMExcl = [self.data.musMarginal[excl]] + self._musMExcl
self.data.musMarginal.pop(excl)
self._PsExcl = [self.data.Ps.pop(excl)] + self._PsExcl
self._QsExcl = [self.data.Qs.pop(excl)] + self._QsExcl
self._PsuppExcl = [self.data.Psupp.pop(excl)] + self._PsuppExcl
- def _setupQuadMapping(self, N2 : List[int] = None):
- if not (hasattr(self.data, "_is_C_quadratic")
- and self.data._is_C_quadratic):
- RROMPyWarning(("Quadratic mapping is useless if output is not "
- "quadratic. Aborting."))
- return
- if N2 is None:
- N2 = np.array([P.shape[0] for P in self.data.Ps], dtype = int)
- if self.data._is_C_quadratic == 1:
- N2 = N2 ** 2
- else: # if self.data._is_C_quadratic == 2:
- N2 = N2 * (N2 + 1) // 2
- if (hasattr(self, "quad_mapping")
- and self.quad_mapping.shape[1] == np.sum(N2)): return
- quad_mapping = np.empty((2, 0), dtype = int)
- for Nloc, suppj in zip(N2, self.data.Psupp):
- self.quad_mapping = np.empty((0, 0))
- super()._setupQuadMapping(Nloc)
- quad_mapping = np.append(quad_mapping, self.quad_mapping + suppj,
- axis = 1)
- self.quad_mapping = quad_mapping
-
def getPVal(self, mu : paramList = []) -> sampList:
"""
Evaluate rational numerator at arbitrary parameter.
Args:
mu: Target parameter.
"""
RROMPyAssert(self.data.nparPivot, 1, "Number of pivot parameters")
mu = self.checkParameterList(mu)
muP = self.centerNormalizePivot(mu(self.data.directionPivot))
muM = self.checkParameterListMarginal(mu(self.data.directionMarginal))
idxMUnique, idxMmap = np.unique(self.data.marginalInterp(muM),
return_inverse = True)
idxMUnique = np.array(idxMUnique, dtype = int)
p = emptySampleList()
vbMng(self, "INIT", "Evaluating numerator at mu = {}.".format(mu), 17)
for i, iM in enumerate(idxMUnique):
idx = np.where(idxMmap == i)[0]
Pval, supp = self.data.Ps[iM](muP[idx]), self.data.Psupp[iM]
if i == 0:
- p.reset((self.supportEnd, len(mu)), dtype = Pval.dtype)
+ p.reset((self.data.projMat.shape[1], len(mu)),
+ dtype = Pval.dtype)
p.data[:] = 0.
p.data[supp : supp + len(Pval), idx] = Pval
vbMng(self, "DEL", "Done evaluating numerator.", 17)
return p
def getQVal(self, mu:Np1D, der : List[int] = None,
scl : Np1D = None) -> Np1D:
"""
Evaluate rational denominator at arbitrary parameter.
Args:
mu: Target parameter.
der(optional): Derivatives to take before evaluation.
"""
RROMPyAssert(self.data.nparPivot, 1, "Number of pivot parameters")
mu = self.checkParameterList(mu)
muP = self.centerNormalizePivot(mu(self.data.directionPivot))
muM = self.checkParameterListMarginal(mu(self.data.directionMarginal))
if der is None:
derP, derM = 0, [0]
else:
derP = der[self.data.directionPivot[0]]
derM = [der[x] for x in self.data.directionMarginal]
if np.any(np.array(derM) != 0):
raise RROMPyException(("Derivatives of Q with respect to marginal "
"parameters not allowed."))
sclP = 1 if scl is None else scl[self.data.directionPivot[0]]
idxMUnique, idxMmap = np.unique(self.data.marginalInterp(muM),
return_inverse = True)
idxMUnique = np.array(idxMUnique, dtype = int)
vbMng(self, "INIT", "Evaluating denominator at mu = {}.".format(mu),
17)
for i, iM in enumerate(idxMUnique):
idx = np.where(idxMmap == i)[0]
Qval = self.data.Qs[iM](muP[idx], derP, sclP)
if i == 0: q = np.empty(len(mu), dtype = Qval.dtype)
q[idx] = Qval
vbMng(self, "DEL", "Done evaluating denominator.", 17)
return q
def getPoles(self, *args, **kwargs) -> Np1D:
"""
Obtain approximant poles.
Returns:
Numpy complex vector of poles.
"""
RROMPyAssert(self.data.nparPivot, 1, "Number of pivot parameters")
if len(args) + len(kwargs) > 1:
raise RROMPyException(("Wrong number of parameters passed. "
"Only 1 available."))
elif len(args) + len(kwargs) == 1:
if len(args) == 1:
mVals = args[0]
else:
mVals = kwargs["marginalVals"]
if not isinstance(mVals, Iterable): mVals = [mVals]
mVals = list(mVals)
else:
mVals = [fp]
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."))
if rDim != self.data.directionPivot[0]:
raise RROMPyException(("'freepar' entry in marginalVals must "
"coincide with pivot direction."))
mVals[rDim] = self.data.mu0(rDim)[0]
muM = self.checkParameterListMarginal([mVals[j]
for j in range(len(mVals)) if j != rDim])
iM = int(self.data.marginalInterp(muM))
roots = self.data.scaleFactor[rDim] * self.data.Qs[iM].roots()
return self.mapParameterList(self.mapParameterList(self.data.mu0(rDim),
idx = [rDim])(0, 0)
+ roots, "B", [rDim])(0)
def getResidues(self, *args, **kwargs) -> Np2D:
"""
Obtain approximant residues.
Returns:
Numpy matrix with residues as columns.
"""
RROMPyAssert(self.data.nparPivot, 1, "Number of pivot parameters")
if len(args) + len(kwargs) > 1:
raise RROMPyException(("Wrong number of parameters passed. "
"Only 1 available."))
elif len(args) + len(kwargs) == 1:
if len(args) == 1:
mVals = args[0]
else:
mVals = kwargs["marginalVals"]
if not isinstance(mVals, Iterable): mVals = [mVals]
mVals = list(mVals)
else:
mVals = [fp]
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."))
if rDim != self.data.directionPivot[0]:
raise RROMPyException(("'freepar' entry in marginalVals must "
"coincide with pivot direction."))
mVals[rDim] = self.data.mu0(rDim)[0]
muM = self.checkParameterListMarginal([mVals[j]
for j in range(len(mVals)) if j != rDim])
iM = int(self.data.marginalInterp(muM))
res, pls, basis = rational2heaviside(self.data.Ps[iM],
self.data.Qs[iM])
res = res[: len(pls), :].T
- if hasattr(self.data, "_is_C_quadratic") and self.data._is_C_quadratic:
- self._setupQuadMapping()
- res = res[self.quad_mapping[0]] * res[self.quad_mapping[1]].conj()
if not self.data._collapsed: res = dot(self.data.projMat, res).T
- if (hasattr(self.data, "_is_C_quadratic")
- and self.data._is_C_quadratic == 2):
- res = np.real(res)
return pls, res
diff --git a/rrompy/reduction_methods/pivoted/trained_model/trained_model_pivoted_rational_polematch.py b/rrompy/reduction_methods/pivoted/trained_model/trained_model_pivoted_rational_polematch.py
index 69f9653..4b3a960 100644
--- a/rrompy/reduction_methods/pivoted/trained_model/trained_model_pivoted_rational_polematch.py
+++ b/rrompy/reduction_methods/pivoted/trained_model/trained_model_pivoted_rational_polematch.py
@@ -1,557 +1,570 @@
# Copyright (C) 2018-2020 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 .
#
import warnings
import numpy as np
from scipy.special import factorial as fact
-from scipy.sparse import spmatrix, csr_matrix, hstack, SparseEfficiencyWarning
+from scipy.sparse import csr_matrix, hstack, SparseEfficiencyWarning
from collections.abc import Iterable
from copy import deepcopy as copy
from itertools import combinations
-from rrompy.reduction_methods.standard.trained_model.trained_model_rational \
- import TrainedModelRational
from .trained_model_pivoted_rational_nomatch import (
TrainedModelPivotedRationalNoMatch)
from rrompy.utilities.base.types import (Tuple, Np1D, Np2D, List, ListAny,
paramVal, paramList, sampList, HFEng)
from rrompy.utilities.base import verbosityManager as vbMng, freepar as fp
from rrompy.utilities.numerical import dot
from rrompy.utilities.numerical.point_matching import rationalFunctionMatching
from rrompy.utilities.numerical.degree import reduceDegreeN
from rrompy.utilities.poly_fitting.polynomial import (polybases as ppb,
PolynomialInterpolator as PI)
from rrompy.utilities.poly_fitting.radial_basis import (polybases as rbpb,
RadialBasisInterpolator as RBI)
from rrompy.utilities.poly_fitting.heaviside import (rational2heaviside,
heavisideUniformShape,
HeavisideInterpolator as HI)
from rrompy.utilities.poly_fitting.nearest_neighbor import (
NearestNeighborInterpolator as NNI)
from rrompy.utilities.poly_fitting.piecewise_linear import (sparsekinds,
PiecewiseLinearInterpolator as PLI)
from rrompy.utilities.exception_manager import RROMPyException, RROMPyAssert
from rrompy.sampling import sampleList, emptySampleList
__all__ = ['TrainedModelPivotedRationalPoleMatch']
-def getChordalScaling(x:Np2D, r:float) -> Tuple[Np2D, Np2D]:
+def getChordalScaling(x:Np2D, r:float, projGramian : Np2D = 1.,
+ projHalfGramian : Np2D = None) -> Tuple[Np2D, Np2D]:
goodX = np.where(np.logical_not(np.isinf(x[:, 0])))[0]
normX = np.empty(len(x))
- normX[goodX] = np.linalg.norm(x[goodX], axis = 1) / r
+ if projGramian is None:
+ normX[goodX] = np.sum(np.abs(dot(projHalfGramian, x[goodX].T)) ** 2.,
+ axis = 0) ** .5
+ else:
+ normX[goodX] = np.abs(np.sum(dot(projGramian, x[goodX].T)
+ * x[goodX].T.conj(), axis = 0)) ** .5
scale = np.ones((len(normX), 1))
- scale[goodX, 0] = 1. / (normX[goodX] ** 2. + 1.)
+ scale[goodX, 0] = 1. / ((normX[goodX] / r) ** 2. + 1.)
xscaled = np.zeros_like(x)
for j in goodX: xscaled[j] = x[j] * scale[j]
return xscaled, r * (1 - scale)
-def normalizeChordal(x:Np2D, r:float) -> Np2D:
+def normalizeChordal(x:Np2D, r:float, projGramian : Np2D = 1.,
+ projHalfGramian : Np2D = None) -> Np2D:
for j in range(x.shape[0]):
x[j, -1] -= .5 * r
- if isinstance(x, (spmatrix,)):
- normxj = np.linalg.norm(x[j].todense())
+ if projGramian is None:
+ norm2xj = np.sum(np.abs(dot(projHalfGramian, x[j, : -1])) ** 2.)
else:
- normxj = np.linalg.norm(x[j])
+ norm2xj = np.abs(np.sum(dot(projGramian, x[j, : -1])
+ * x[j, : -1].conj()))
+ normxj = (norm2xj + np.abs(x[j, -1]) ** 2.) ** .5
if normxj < 1e-15: normxj += np.finfo(float).eps
x[j] *= .5 * r / normxj
x[j, -1] += .5 * r
return x
def pullbackChordal(x:Np2D, r:float) -> Np2D:
y = copy(x[:, : -1])
for j, p in enumerate(x[:, -1]):
scalexj = 1. - p / r
y[j] = np.inf if scalexj < 1e-15 else y[j] / scalexj
return y
class TrainedModelPivotedRationalPoleMatch(TrainedModelPivotedRationalNoMatch):
"""
ROM approximant evaluation for pivoted approximants based on interpolation
of rational approximants (with pole matching).
Attributes:
Data: dictionary with all that can be pickled.
"""
def compress(self, collapse : bool = False, tol : float = 0.,
returnRMat : bool = False, **compressMatrixkwargs):
Psupp = copy(self.data.Psupp)
RMat = super().compress(collapse, tol, True, **compressMatrixkwargs)
if RMat is None: return
for obj, suppj in zip(self.data.HIs, Psupp):
obj.postmultiplyTensorize(RMat.T[suppj : suppj + obj.shape[0]])
if hasattr(self, "_HIsExcl"):
for obj, suppj in zip(self._HIsExcl, Psupp):
obj.postmultiplyTensorize(RMat.T[suppj : suppj + obj.shape[0]])
if not hasattr(self, "_PsExcl"):
self._PsuppExcl = [0] * len(self._PsuppExcl)
if returnRMat: return RMat
def centerNormalizeMarginal(self, mu : paramList = [],
mu0 : paramVal = None) -> paramList:
"""
Compute normalized parameter to be plugged into approximant.
Args:
mu: Parameter(s) 1.
mu0: Parameter(s) 2. If None, set to self.data.mu0Marginal.
Returns:
Normalized parameter.
"""
mu = self.checkParameterListMarginal(mu)
if mu0 is None:
mu0 = self.checkParameterListMarginal(
self.data.mu0(0, self.data.directionMarginal))
return (self.mapParameterList(mu, idx = self.data.directionMarginal)
- self.mapParameterList(mu0, idx = self.data.directionMarginal)
) / [self.data.scaleFactor[x]
for x in self.data.directionMarginal]
def setupMarginalInterp(self, approx, interpPars:ListAny, extraPar = None):
vbMng(self, "INIT", "Starting computation of marginal interpolator.",
12)
musMCN = self.centerNormalizeMarginal(self.data.musMarginal)
nM, pbM = len(musMCN), approx.polybasisMarginal
if pbM in ppb + rbpb:
if extraPar: approx._setMMarginalAuto()
_MMarginalEff = approx.paramsMarginal["MMarginal"]
if pbM in ppb:
p = PI()
elif pbM in rbpb:
p = RBI()
else: # if pbM in sparsekinds + ["NEARESTNEIGHBOR"]:
if pbM == "NEARESTNEIGHBOR":
p = NNI()
else: # if pbM in sparsekinds:
pllims = [[-1.] * self.data.nparMarginal,
[1.] * self.data.nparMarginal]
p = PLI()
for ipts, pts in enumerate(self.data.suppEffPts):
if len(pts) == 0:
raise RROMPyException("Empty list of support points.")
musMCNEff, valsEff = musMCN[pts], np.eye(len(pts))
if pbM in ppb + rbpb:
if extraPar:
if ipts > 0:
verb = approx.verbosity
approx.verbosity = 0
_musM = approx.musMarginal
approx.musMarginal = musMCNEff
approx._setMMarginalAuto()
approx.musMarginal = _musM
approx.verbosity = verb
else:
approx.paramsMarginal["MMarginal"] = reduceDegreeN(
_MMarginalEff, len(musMCNEff), self.data.nparMarginal,
approx.paramsMarginal["polydegreetypeMarginal"])
MMEff = approx.paramsMarginal["MMarginal"]
while MMEff >= 0:
wellCond, msg = p.setupByInterpolation(musMCNEff, valsEff,
MMEff, *interpPars)
vbMng(self, "MAIN", msg, 30)
if wellCond: break
vbMng(self, "MAIN",
("Polyfit is poorly conditioned. Reducing "
"MMarginal by 1."), 35)
MMEff -= 1
if MMEff < 0:
raise RROMPyException(("Instability in computation of "
"interpolant. Aborting."))
if (pbM in rbpb and len(interpPars) > 4
and "optimizeScalingBounds" in interpPars[4].keys()):
interpPars[4]["optimizeScalingBounds"] = [-1., -1.]
elif pbM == "NEARESTNEIGHBOR":
if ipts > 0: interpPars[0] = 1
p.setupByInterpolation(musMCNEff, valsEff, *interpPars)
elif ipts == 0: # and pbM in sparsekinds:
p.setupByInterpolation(musMCNEff, valsEff, pllims,
extraPar[pts], *interpPars)
if ipts == 0:
self.data.marginalInterp = copy(p)
self.data.coeffsEff, self.data.polesEff = [], []
N = len(self.data.suppEffIdx)
goodIdx = np.where(self.data.suppEffIdx != -1)[0]
for hi, sup in zip(self.data.HIs, self.data.Psupp):
pEff, cEff = hi.poles.reshape(-1, 1), hi.coeffs
if self.data.chordalRadius[0] > 0.:
pEff = np.hstack(getChordalScaling(pEff,
self.data.chordalRadius[0]))
if self.data.chordalRadius[1] > 0.:
+ if self.data.projGramian is None:
+ projGramian = None
+ projHalfGramian = self.data.projMat[:,
+ sup : sup + cEff.shape[1]]
+ else:
+ projGramian = self.data.projGramian[
+ sup : sup + cEff.shape[1]][:,
+ sup : sup + cEff.shape[1]]
+ projHalfGramian = None
cEff, cEffH = getChordalScaling(cEff,
- self.data.chordalRadius[1])
+ self.data.chordalRadius[1],
+ projGramian, projHalfGramian)
else:
cEffH = np.empty((cEff.shape[0], 0))
if (self.data._collapsed
- or self.supportEnd == cEff.shape[1]):
+ or self.data.projMat.shape[1] == cEff.shape[1]):
cEff = np.hstack([cEff, cEffH])
else:
- supC = self.supportEnd - sup - cEff.shape[1]
+ supC = self.data.projMat.shape[1] - sup - cEff.shape[1]
cEff = hstack((csr_matrix((len(cEff), sup)),
csr_matrix(cEff),
csr_matrix((len(cEff), supC)),
cEffH), "csr")
goodIdxC = np.append(goodIdx, np.arange(N, cEff.shape[0]))
self.data.coeffsEff += [cEff[goodIdxC, :]]
self.data.polesEff += [pEff[goodIdx]]
else:
ptsBad = [i for i in range(nM) if i not in pts]
- idxBad = np.where(self.data.suppEffIdx == ipts)[0]
+ idxBad = np.where(self.data.suppEffIdx[goodIdx] == ipts)[0]
warnings.simplefilter('ignore', SparseEfficiencyWarning)
if pbM in sparsekinds:
for ij, j in enumerate(ptsBad):
nearest = pts[np.argmin(np.sum(np.abs(musMCNEff.data
- np.tile(musMCN[j], [len(pts), 1])
), axis = 1).flatten())]
self.data.coeffsEff[j][idxBad] = copy(
self.data.coeffsEff[nearest][idxBad])
self.data.polesEff[j][idxBad] = copy(
self.data.polesEff[nearest][idxBad])
else:
if (self.data._collapsed
- or self.supportEnd == cEff.shape[1]):
+ or self.data.projMat.shape[1] == cEff.shape[1]):
cfBase = np.zeros((len(idxBad), cEff.shape[1]),
dtype = cEff.dtype)
else:
cfBase = csr_matrix((len(idxBad),
self.data.coeffsEff[0].shape[1]),
dtype = cEff.dtype)
valMuMBad = p(musMCN[ptsBad])
for ijb, jb in enumerate(ptsBad):
self.data.coeffsEff[jb][idxBad] = copy(cfBase)
self.data.polesEff[jb][idxBad] = 0.
for ij, j in enumerate(pts):
val = valMuMBad[ij][ijb]
if not np.isclose(val, 0., atol = 1e-15):
self.data.coeffsEff[jb][idxBad] += (val
* self.data.coeffsEff[j][idxBad])
self.data.polesEff[jb][idxBad] += (val
* self.data.polesEff[j][idxBad])
if self.data.chordalRadius[0] > 0:
self.data.polesEff[jb][idxBad] = normalizeChordal(
self.data.polesEff[jb][idxBad],
self.data.chordalRadius[0])
if self.data.chordalRadius[1] > 0:
self.data.coeffsEff[jb][idxBad] = normalizeChordal(
self.data.coeffsEff[jb][idxBad],
- self.data.chordalRadius[1])
+ self.data.chordalRadius[1],
+ self.data.projGramian,
+ self.data.projMat)
warnings.filters.pop(0)
if pbM in ppb + rbpb:
approx.paramsMarginal["MMarginal"] = _MMarginalEff
vbMng(self, "DEL", "Done computing marginal interpolator.", 12)
def updateEffectiveSamples(self, exclude:List[int], *args, **kwargs):
if hasattr(self, "_idxExcl"):
for j, excl in enumerate(self._idxExcl):
self.data.HIs.insert(excl, self._HIsExcl[j])
super().updateEffectiveSamples(exclude)
self._HIsExcl = []
for excl in self._idxExcl[::-1]:
self._HIsExcl = [self.data.HIs.pop(excl)] + self._HIsExcl
poles = [hi.poles for hi in self.data.HIs]
coeffs = [hi.coeffs for hi in self.data.HIs]
self.initializeFromLists(poles, coeffs, self.data.Psupp,
self.data.HIs[0].polybasis, *args, **kwargs)
def initializeFromRational(self, *args, **kwargs):
"""Initialize Heaviside representation."""
RROMPyAssert(self.data.nparPivot, 1, "Number of pivot parameters")
poles, coeffs = [], []
for Q, P in zip(self.data.Qs, self.data.Ps):
cfs, pls, basis = rational2heaviside(P, Q)
poles += [pls]
coeffs += [cfs]
self.initializeFromLists(poles, coeffs, self.data.Psupp, basis, *args,
**kwargs)
def initializeFromLists(self, poles:ListAny, coeffs:ListAny, supps:ListAny,
basis:str, matchingWeight:float, HFEngine:HFEng,
is_state:bool, chordalRadius:Tuple[float, float]):
"""Initialize Heaviside representation."""
poles, coeffs = heavisideUniformShape(poles, coeffs)
N = len(poles[0])
if chordalRadius[0] == "AUTO": chordalRadius[0] = 1.
if chordalRadius[1] == "AUTO":
- chordalRadius[1] = np.mean([np.linalg.norm(c[: N], axis = 1)
- for c in coeffs])
- self.data.chordalRadius = chordalRadius
- if hasattr(self.data, "_is_C_quadratic") and self.data._is_C_quadratic:
- csizemax = np.max([len(c) for c in coeffs])
- #csizemax = np.max([c.shape[1] for c in coeffs])
- if self.data._is_C_quadratic == 1:
- csizemax = csizemax ** 2
- else: # if self.data._is_C_quadratic == 2:
- csizemax = csizemax * (csizemax + 1) // 2
- TrainedModelRational._setupQuadMapping(self, csizemax)
- projMapping = self.quad_mapping
- projMappingReal = self.data._is_C_quadratic == 2
- else:
- projMapping, projMappingReal = None, False
+ norm2s = 0.
+ for c, sup in zip(coeffs, self.data.Psupp):
+ if self.data.projGramian is None:
+ gramEff = self.data.projMat[:, sup : sup + c.shape[1]]
+ norm2s += np.sum(np.abs(dot(gramEff, c[: N].T)) ** 2.)
+ else:
+ gramEff = self.data.projGramian[sup : sup + c.shape[1]][:,
+ sup : sup + c.shape[1]]
+ norm2s += np.sum(np.abs(dot(gramEff, c[: N].T)
+ * c[: N].T.conj()))
+ chordalRadius[1] = (norm2s / N / len(coeffs)) ** .5
+ self.data.chordalRadius = copy(chordalRadius)
+ if is_state and chordalRadius[1] > 0: chordalRadius[1] = "AUTO"
poles, coeffs = rationalFunctionMatching(poles, coeffs,
self.data.musMarginal.data,
matchingWeight, supps,
self.data.projMat, HFEngine,
- is_state, None, chordalRadius,
- projMapping, projMappingReal)
+ is_state, None, chordalRadius)
self.data.HIs = []
for pls, cfs in zip(poles, coeffs):
hsi = HI()
hsi.poles = pls
if len(cfs) == len(pls):
cfs = np.pad(cfs, ((0, 1), (0, 0)), "constant")
hsi.coeffs = cfs
hsi.npar = 1
hsi.polybasis = basis
self.data.HIs += [hsi]
self.data.suppEffPts = [np.arange(len(self.data.HIs))]
self.data.suppEffIdx = np.zeros(len(poles[0]), dtype = int)
def checkShared(self, shared:float) -> str:
N = len(self.data.HIs[0].poles)
M = len(self.data.HIs)
goodLocPoles = np.array([np.logical_not(np.isinf(hi.poles)
) for hi in self.data.HIs])
self.data.suppEffPts = [np.arange(len(self.data.HIs))]
self.data.suppEffIdx = - np.ones(N, dtype = int)
goodGlobPoles = np.sum(goodLocPoles, axis = 0)
goodEnoughPoles = goodGlobPoles >= max(1., 1. * shared * M)
keepPole = np.where(goodEnoughPoles)[0]
halfPole = np.where(np.logical_and(goodEnoughPoles,
goodGlobPoles < M))[0]
self.data.suppEffIdx[keepPole] = 0
for idxR in halfPole:
pts = np.where(goodLocPoles[:, idxR])[0]
idxEff = len(self.data.suppEffPts)
for idEff, prevPts in enumerate(self.data.suppEffPts):
if len(prevPts) == len(pts):
if np.allclose(prevPts, pts):
idxEff = idEff
break
if idxEff == len(self.data.suppEffPts):
self.data.suppEffPts += [pts]
self.data.suppEffIdx[idxR] = idxEff
return (" Hard-erased {} pole".format(N - len(keepPole))
+ "s" * (N - len(keepPole) != 1)
+ " and soft-erased {} pole".format(len(halfPole))
+ "s" * (len(halfPole) != 1) + ".")
def getApproxReduced(self, mu : paramList = []) -> sampList:
"""
Evaluate reduced representation of approximant at arbitrary parameter.
Args:
mu: Target parameter.
"""
RROMPyAssert(self.data.nparPivot, 1, "Number of pivot parameters")
mu = self.checkParameterList(mu)
if (not hasattr(self, "lastSolvedApproxReduced")
or self.lastSolvedApproxReduced != mu):
vbMng(self, "INIT",
"Evaluating approximant at mu = {}.".format(mu), 12)
muP = self.centerNormalizePivot(mu(self.data.directionPivot))
muM = mu(self.data.directionMarginal)
his = self.interpolateMarginalInterpolator(muM)
for i, (mP, hi) in enumerate(zip(muP, his)):
uAppR = hi(mP)[:, 0]
if i == 0:
uApproxR = np.empty((len(uAppR), len(mu)),
dtype = uAppR.dtype)
uApproxR[:, i] = uAppR
self.uApproxReduced = sampleList(uApproxR)
vbMng(self, "DEL", "Done evaluating approximant.", 12)
self.lastSolvedApproxReduced = mu
return self.uApproxReduced
def interpolateMarginalInterpolator(self, mu : paramList = []) -> ListAny:
"""Obtain interpolated approximant interpolator."""
mu = self.checkParameterListMarginal(mu)
vbMng(self, "INIT",
"Interpolating marginal models at mu = {}.".format(mu), 95)
his = []
muC = self.centerNormalizeMarginal(mu)
mIvals = self.data.marginalInterp(muC)
verb, self.verbosity = self.verbosity, 0
poless = self.interpolateMarginalPoles(mu, mIvals)
coeffss = self.interpolateMarginalCoeffs(mu, mIvals)
self.verbosity = verb
for j in range(len(mu)):
his += [HI()]
his[-1].poles = poless[j]
his[-1].coeffs = coeffss[j]
his[-1].npar = 1
his[-1].polybasis = self.data.HIs[0].polybasis
vbMng(self, "DEL", "Done interpolating marginal models.", 95)
return his
def interpolateMarginalPoles(self, mu : paramList = [],
mIvals : Np2D = None) -> ListAny:
"""Obtain interpolated approximant poles."""
mu = self.checkParameterListMarginal(mu)
vbMng(self, "INIT",
"Interpolating marginal poles at mu = {}.".format(mu), 95)
intMPoles = np.zeros((len(mu),) + self.data.polesEff[0].shape,
dtype = self.data.polesEff[0].dtype)
if mIvals is None:
muC = self.centerNormalizeMarginal(mu)
mIvals = self.data.marginalInterp(muC)
for pEff, mI in zip(self.data.polesEff, mIvals):
for j, m in enumerate(mI): intMPoles[j] += m * pEff
rCP = self.data.chordalRadius[0]
if rCP > 0:
for j in range(len(mu)):
intMPoles[j, ..., 0] = pullbackChordal(
normalizeChordal(intMPoles[j], rCP),
rCP)[..., 0]
vbMng(self, "DEL", "Done interpolating marginal poles.", 95)
return intMPoles[..., 0]
def interpolateMarginalCoeffs(self, mu : paramList = [],
mIvals : Np2D = None) -> ListAny:
"""Obtain interpolated approximant coefficients."""
mu = self.checkParameterListMarginal(mu)
vbMng(self, "INIT",
"Interpolating marginal coefficients at mu = {}.".format(mu), 95)
intMCoeffs = np.zeros((len(mu),) + self.data.coeffsEff[0].shape,
dtype = self.data.coeffsEff[0].dtype)
if mIvals is None:
muC = self.centerNormalizeMarginal(mu)
mIvals = self.data.marginalInterp(muC)
for cEff, mI in zip(self.data.coeffsEff, mIvals):
for j, m in enumerate(mI): intMCoeffs[j] += m * cEff
rCC = self.data.chordalRadius[1]
if rCC > 0:
for j in range(len(mu)):
- intMCoeffs[j, ..., : -1] = pullbackChordal(normalizeChordal(
- intMCoeffs[j], rCC), rCC)
+ intMCoeffs[j, ..., : -1] = pullbackChordal(
+ normalizeChordal(intMCoeffs[j], rCC,
+ self.data.projGramian,
+ self.data.projMat),
+ rCC)
intMCoeffs = intMCoeffs[..., : -1]
vbMng(self, "DEL", "Done interpolating marginal coefficients.", 95)
return intMCoeffs
def getPVal(self, mu : paramList = []) -> sampList:
"""
Evaluate rational numerator at arbitrary parameter.
Args:
mu: Target parameter.
"""
RROMPyAssert(self.data.nparPivot, 1, "Number of pivot parameters")
mu = self.checkParameterList(mu)
p = emptySampleList()
muP = self.centerNormalizePivot(mu(self.data.directionPivot))
muM = mu(self.data.directionMarginal)
his = self.interpolateMarginalInterpolator(muM)
for i, (mP, hi) in enumerate(zip(muP, his)):
Pval = hi(mP) * np.prod(mP[0] - hi.poles)
if i == 0: p.reset((len(Pval), len(mu)), dtype = Pval.dtype)
p[i] = Pval
return p
def getQVal(self, mu:Np1D, der : List[int] = None,
scl : Np1D = None) -> Np1D:
"""
Evaluate rational denominator at arbitrary parameter.
Args:
mu: Target parameter.
der(optional): Derivatives to take before evaluation.
"""
RROMPyAssert(self.data.nparPivot, 1, "Number of pivot parameters")
mu = self.checkParameterList(mu)
muP = self.centerNormalizePivot(mu(self.data.directionPivot))
muM = mu(self.data.directionMarginal)
if der is None:
derP, derM = 0, [0]
else:
derP = der[self.data.directionPivot[0]]
derM = [der[x] for x in self.data.directionMarginal]
if np.any(np.array(derM) != 0):
raise RROMPyException(("Derivatives of Q with respect to marginal "
"parameters not allowed."))
sclP = 1 if scl is None else scl[self.data.directionPivot[0]]
derVal = np.zeros(len(mu), dtype = np.complex)
pls = self.interpolateMarginalPoles(muM)
for i, (mP, pl) in enumerate(zip(muP, pls)):
N = len(pl)
if derP == N: derVal[i] = 1.
elif derP >= 0 and derP < N:
plDist = mP[0] - pl
for terms in combinations(np.arange(N), N - derP):
derVal[i] += np.prod(plDist[list(terms)])
return sclP ** derP * fact(derP) * derVal
def getPoles(self, *args, **kwargs) -> Np1D:
"""
Obtain approximant poles.
Returns:
Numpy complex vector of poles.
"""
RROMPyAssert(self.data.nparPivot, 1, "Number of pivot parameters")
if len(args) + len(kwargs) > 1:
raise RROMPyException(("Wrong number of parameters passed. "
"Only 1 available."))
elif len(args) + len(kwargs) == 1:
if len(args) == 1:
mVals = args[0]
else:
mVals = kwargs["marginalVals"]
if not isinstance(mVals, Iterable): mVals = [mVals]
mVals = list(mVals)
else:
mVals = [fp]
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."))
if rDim != self.data.directionPivot[0]:
raise RROMPyException(("'freepar' entry in marginalVals must "
"coincide with pivot direction."))
mVals[rDim] = self.data.mu0(rDim)[0]
mMarg = [mVals[j] for j in range(len(mVals)) if j != rDim]
roots = (self.data.scaleFactor[rDim]
* self.interpolateMarginalPoles(mMarg)[0])
return self.mapParameterList(self.mapParameterList(self.data.mu0(rDim),
idx = [rDim])(0, 0)
+ roots, "B", [rDim])(0)
def getResidues(self, *args, **kwargs) -> Np2D:
"""
Obtain approximant residues.
Returns:
Numpy matrix with residues as columns.
"""
pls = self.getPoles(*args, **kwargs)
if len(args) == 1:
mVals = args[0]
elif len(args) == 0:
mVals = [None]
else:
mVals = kwargs["marginalVals"]
if not isinstance(mVals, Iterable): mVals = [mVals]
mVals = list(mVals)
rDim = mVals.index(fp)
mMarg = [mVals[j] for j in range(len(mVals)) if j != rDim]
res = self.interpolateMarginalCoeffs(mMarg)[0][: len(pls), :].T
- if hasattr(self.data, "_is_C_quadratic") and self.data._is_C_quadratic:
- self._setupQuadMapping()
- res = res[self.quad_mapping[0]] * res[self.quad_mapping[1]].conj()
if not self.data._collapsed: res = dot(self.data.projMat, res).T
- if (hasattr(self.data, "_is_C_quadratic")
- and self.data._is_C_quadratic == 2):
- res = np.real(res)
return pls, res
diff --git a/rrompy/reduction_methods/standard/generic_standard_approximant.py b/rrompy/reduction_methods/standard/generic_standard_approximant.py
index c6f7d81..daab2b0 100644
--- a/rrompy/reduction_methods/standard/generic_standard_approximant.py
+++ b/rrompy/reduction_methods/standard/generic_standard_approximant.py
@@ -1,221 +1,194 @@
# Copyright (C) 2018-2020 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 .
#
import numpy as np
from copy import deepcopy as copy
from rrompy.reduction_methods.base.generic_approximant import (
GenericApproximant)
from rrompy.utilities.base import verbosityManager as vbMng
from rrompy.utilities.base.types import Np2D
from rrompy.utilities.exception_manager import (RROMPyException, RROMPyAssert,
RROMPyWarning)
__all__ = ['GenericStandardApproximant']
class GenericStandardApproximant(GenericApproximant):
"""
ROM interpolant computation for parametric problems (ABSTRACT).
Args:
HFEngine: HF problem solver.
mu0(optional): Default parameter. Defaults to 0.
approxParameters(optional): Dictionary containing values for main
parameters of approximant. Recognized keys are:
- 'POD': kind of snapshots orthogonalization; allowed values
include 0, 1/2, and 1; defaults to 1, i.e. POD;
- 'scaleFactorDer': scaling factors for derivative computation;
defaults to 'AUTO';
- 'S': total number of samples current approximant relies upon;
- 'sampler': sample point generator.
Defaults to empty dict.
verbosity(optional): Verbosity level. Defaults to 10.
Attributes:
HFEngine: HF problem solver.
mu0: Default parameter.
mus: Array of snapshot parameters.
approxParameters: Dictionary containing values for main parameters of
approximant. Recognized keys are in parameterList.
parameterListSoft: Recognized keys of soft approximant parameters:
- 'POD': kind of snapshots orthogonalization;
- 'scaleFactorDer': scaling factors for derivative computation.
parameterListCritical: Recognized keys of critical approximant
parameters:
- 'S': total number of samples current approximant relies upon;
- 'sampler': sample point generator.
verbosity: Verbosity level.
POD: Kind of snapshots orthogonalization.
scaleFactorDer: Scaling factors for derivative computation.
S: Number of solution snapshots over which current approximant is
based upon.
sampler: Sample point generator.
muBounds: list of bounds for parameter values.
samplingEngine: Sampling engine.
uHF: High fidelity solution(s) with parameter(s) lastSolvedHF as
sampleList.
lastSolvedHF: Parameter(s) corresponding to last computed high fidelity
solution(s) as parameterList.
uApproxReduced: Reduced approximate solution(s) with parameter(s)
lastSolvedApprox as sampleList.
lastSolvedApproxReduced: Parameter(s) corresponding to last computed
reduced approximate solution(s) as parameterList.
uApprox: Approximate solution(s) with parameter(s) lastSolvedApprox as
sampleList.
lastSolvedApprox: Parameter(s) corresponding to last computed
approximate solution(s) as parameterList.
"""
def __init__(self, *args, **kwargs):
self._preInit()
from rrompy.parameter.parameter_sampling import EmptySampler as ES
self._addParametersToList([], [], ["sampler"], [ES()])
super().__init__(*args, **kwargs)
self._postInit()
@property
def mus(self):
"""Value of mus. Its assignment may reset snapshots."""
return self._mus
@mus.setter
def mus(self, mus):
mus = self.checkParameterList(mus)
musOld = copy(self.mus) if hasattr(self, '_mus') else None
if (musOld is None or len(mus) != len(musOld) or not mus == musOld):
self.resetSamples()
self._mus = mus
@property
def muBounds(self):
"""Value of muBounds."""
return self.sampler.lims
@property
def sampler(self):
"""Value of sampler."""
return self._sampler
@sampler.setter
def sampler(self, sampler):
if 'generatePoints' not in dir(sampler):
raise RROMPyException("Sampler type not recognized.")
if hasattr(self, '_sampler') and self._sampler is not None:
samplerOld = self.sampler
self._sampler = sampler
self._approxParameters["sampler"] = self.sampler
if not 'samplerOld' in locals() or samplerOld != self.sampler:
self.resetSamples()
def setSamples(self, samplingEngine, merge : bool = False):
"""Copy samplingEngine and samples."""
vbMng(self, "INIT", "Transfering samples.", 15)
if isinstance(samplingEngine, (str, list, tuple,)):
self.setupSampling()
self.samplingEngine.load(samplingEngine, merge)
elif merge:
try:
selfkeys = self.samplingEngine.feature_keys
for key in samplingEngine.feature_keys:
if key in selfkeys:
self.samplingEngine._mergeFeature(key,
samplingEngine.feature_vals[key])
except:
RROMPyWarning(("Sample merge failed. Falling back to complete "
"sampling engine replacement."))
self.samplingEngine = copy(samplingEngine)
else:
self.samplingEngine = copy(samplingEngine)
if self.POD != 0 and (self.samplingEngine.nsamples
!= len(self.samplingEngine.samples_normal)):
RROMPyWarning(("Assigning non-POD sampling engine to POD "
"approximant is unstable. Declassing local "
"POD to 0."))
self._POD = 0
self._mus = copy(self.samplingEngine.mus)
self.scaleFactor = self.samplingEngine.scaleFactor
vbMng(self, "DEL", "Done transfering samples.", 15)
def computeSnapshots(self):
"""Compute snapshots of solution map."""
RROMPyAssert(self._mode,
message = "Cannot start snapshot computation.")
if self.samplingEngine.nsamples != self.S:
self.computeScaleFactor()
self.samplingEngine.scaleFactor = self.scaleFactorDer
vbMng(self, "INIT", "Starting computation of snapshots.", 5)
self.mus = self.sampler.generatePoints(self.S)
while len(self.mus) > self.S: self.mus.pop()
self.samplingEngine.iterSample(self.mus)
vbMng(self, "DEL", "Done computing snapshots.", 5)
def computeScaleFactor(self):
"""Compute parameter rescaling factor."""
self.scaleFactor = .5 * np.abs((
self.mapParameterList(self.muBounds[0])
- self.mapParameterList(self.muBounds[1]))[0])
- def _setupTrainedModel(self, pMat:Np2D, pMatUpdate : bool = False,
- pMatOld : Np2D = None):
+ def _setupTrainedModel(self, pMat:Np2D, pMatUpdate : bool = False):
if self.POD == 1 and not (
hasattr(self.HFEngine.C, "is_mu_independent")
and self.HFEngine.C.is_mu_independent in self._output_lvl):
raise RROMPyException(("Cannot apply mu-dependent C to "
"orthonormalized samples."))
vbMng(self, "INIT", "Extracting system output from state.", 35)
- if pMatOld is None:
- pMat = self.HFEngine.applyC(pMat, self.mus)
- else:
- Snew, Sold = pMat.shape[1], pMatOld.shape[1]
- pMat0 = self.HFEngine.applyC(pMat, self.mus)
- pMat1 = self.HFEngine._applyCQuadratic(pMatOld, self.mus, pMat)
- if self.HFEngine._is_C_quadratic == 1:
- pMat2 = self.HFEngine._applyCQuadratic(pMat, self.mus, pMatOld)
- pMat = np.empty((len(pMat0),
- pMat0.shape[1] + 2 * pMat1.shape[1]),
- dtype = pMat1.dtype)
- else: # if self.HFEngine._is_C_quadratic == 2:
- pMat = np.empty((len(pMat0), pMat0.shape[1] + pMat1.shape[1]),
- dtype = pMat1.dtype)
- idG, idL = 0, 0
- for rowcol in range(Snew):
- if self.HFEngine._is_C_quadratic == 1:
- pMat[:, idG : idG + Sold] = pMat2[:, :, rowcol]
- idG += Sold
- diagsize = 2 * rowcol + 1
- else: # if self.HFEngine._is_C_quadratic == 2:
- diagsize = rowcol + 1
- pMat[:, idG : idG + diagsize] = pMat0[:, idL : idL + diagsize]
- idG += diagsize
- idL += diagsize
- pMat[:, idG : idG + Sold] = pMat1[:, rowcol, :]
- idG += Sold
+ pMat = self.HFEngine.applyC(pMat, self.mus)
vbMng(self, "DEL", "Done extracting system output.", 35)
if self.trainedModel is None:
self.trainedModel = self.tModelType()
self.trainedModel.verbosity = self.verbosity
self.trainedModel.timestamp = self.timestamp
datadict = {"mu0": self.mu0, "mus": copy(self.mus),
"projMat": pMat, "scaleFactor": self.scaleFactor,
"parameterMap": self.HFEngine.parameterMap}
self.trainedModel.data = self.initializeModelData(datadict)[0]
else:
self.trainedModel = self.trainedModel
if pMatUpdate:
self.trainedModel.data.projMat = np.hstack(
(self.trainedModel.data.projMat, pMat))
else:
self.trainedModel.data.projMat = copy(pMat)
self.trainedModel.data.mus = copy(self.mus)
diff --git a/rrompy/reduction_methods/standard/greedy/rational_interpolant_greedy.py b/rrompy/reduction_methods/standard/greedy/rational_interpolant_greedy.py
index 794e59a..4b6ce84 100644
--- a/rrompy/reduction_methods/standard/greedy/rational_interpolant_greedy.py
+++ b/rrompy/reduction_methods/standard/greedy/rational_interpolant_greedy.py
@@ -1,528 +1,518 @@
# Copyright (C) 2018-2020 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 .
#
from copy import deepcopy as copy
import numpy as np
from rrompy.hfengines.base.linear_affine_engine import checkIfAffine
from .generic_greedy_approximant import GenericGreedyApproximant
from rrompy.utilities.poly_fitting.polynomial import (polybases, polyfitname,
PolynomialInterpolator as PI,
polyvander)
from rrompy.utilities.numerical import dot
from rrompy.utilities.numerical.degree import totalDegreeN
from rrompy.utilities.expression import expressionEvaluator
from rrompy.reduction_methods.standard import RationalInterpolant
from rrompy.utilities.base.types import Np1D, Tuple, paramVal, List
from rrompy.utilities.base.verbosity_depth import (verbosityManager as vbMng,
getVerbosityDepth, setVerbosityDepth)
from rrompy.utilities.poly_fitting import customFit
from rrompy.utilities.exception_manager import (RROMPyWarning, RROMPyException,
RROMPyAssert, RROMPy_FRAGILE)
from rrompy.sampling import sampleList, emptySampleList
__all__ = ['RationalInterpolantGreedy']
class RationalInterpolantGreedy(GenericGreedyApproximant, RationalInterpolant):
"""
ROM greedy rational interpolant computation for parametric problems.
Args:
HFEngine: HF problem solver.
mu0(optional): Default parameter. Defaults to 0.
approxParameters(optional): Dictionary containing values for main
parameters of approximant. Recognized keys are:
- 'POD': kind of snapshots orthogonalization; allowed values
include 0, 1/2, and 1; defaults to 1, i.e. POD;
- 'scaleFactorDer': scaling factors for derivative computation;
defaults to 'AUTO';
- 'S': number of starting training points;
- 'sampler': sample point generator;
- 'greedyTol': uniform error tolerance for greedy algorithm;
defaults to 1e-2;
- 'collinearityTol': collinearity tolerance for greedy algorithm;
defaults to 0.;
- 'maxIter': maximum number of greedy steps; defaults to 1e2;
- 'nTestPoints': number of test points; defaults to 5e2;
- 'trainSetGenerator': training sample points generator; defaults
to sampler;
- 'polybasis': type of basis for interpolation; defaults to
'MONOMIAL';
- 'errorEstimatorKind': kind of error estimator; available values
include 'AFFINE', 'DISCREPANCY', 'LOOK_AHEAD',
'LOOK_AHEAD_RES', 'LOOK_AHEAD_OUTPUT', and 'NONE'; defaults to
'NONE';
- 'functionalSolve': strategy for minimization of denominator
functional; allowed values include 'NORM', 'DOMINANT', 'NODAL',
'BARYCENTRIC_NORM', and 'BARYCENTRIC[_AVERAGE]' (check pdf in
main folder for meaning); defaults to 'NORM';
- 'interpTol': tolerance for interpolation; defaults to None;
- 'QTol': tolerance for robust rational denominator management;
defaults to 0.
Defaults to empty dict.
verbosity(optional): Verbosity level. Defaults to 10.
Attributes:
HFEngine: HF problem solver.
mu0: Default parameter.
mus: Array of snapshot parameters.
approxParameters: Dictionary containing values for main parameters of
approximant. Recognized keys are in parameterList.
parameterListSoft: Recognized keys of soft approximant parameters:
- 'POD': kind of snapshots orthogonalization;
- 'scaleFactorDer': scaling factors for derivative computation;
- 'greedyTol': uniform error tolerance for greedy algorithm;
- 'collinearityTol': collinearity tolerance for greedy algorithm;
- 'maxIter': maximum number of greedy steps;
- 'nTestPoints': number of test points;
- 'trainSetGenerator': training sample points generator;
- 'errorEstimatorKind': kind of error estimator;
- 'functionalSolve': strategy for minimization of denominator
functional;
- 'interpTol': tolerance for interpolation;
- 'QTol': tolerance for robust rational denominator management.
parameterListCritical: Recognized keys of critical approximant
parameters:
- 'S': total number of samples current approximant relies upon;
- 'sampler': sample point generator.
verbosity: Verbosity level.
POD: Kind of snapshots orthogonalization.
scaleFactorDer: Scaling factors for derivative computation.
S: number of test points.
sampler: Sample point generator.
greedyTol: uniform error tolerance for greedy algorithm.
collinearityTol: Collinearity tolerance for greedy algorithm.
maxIter: maximum number of greedy steps.
nTestPoints: number of starting training points.
trainSetGenerator: training sample points generator.
errorEstimatorKind: kind of error estimator.
functionalSolve: Strategy for minimization of denominator functional.
interpTol: tolerance for interpolation.
QTol: tolerance for robust rational denominator management.
muBounds: list of bounds for parameter values.
samplingEngine: Sampling engine.
uHF: High fidelity solution(s) with parameter(s) lastSolvedHF as
sampleList.
lastSolvedHF: Parameter(s) corresponding to last computed high fidelity
solution(s) as parameterList.
uApproxReduced: Reduced approximate solution(s) with parameter(s)
lastSolvedApprox as sampleList.
lastSolvedApproxReduced: Parameter(s) corresponding to last computed
reduced approximate solution(s) as parameterList.
uApprox: Approximate solution(s) with parameter(s) lastSolvedApprox as
sampleList.
lastSolvedApprox: Parameter(s) corresponding to last computed
approximate solution(s) as parameterList.
"""
_allowedEstimatorKinds = ["AFFINE", "DISCREPANCY", "LOOK_AHEAD",
"LOOK_AHEAD_RES", "LOOK_AHEAD_OUTPUT", "NONE"]
def __init__(self, *args, **kwargs):
self._preInit()
self._addParametersToList(["errorEstimatorKind"], ["DISCREPANCY"],
toBeExcluded = ["M", "N", "polydegreetype",
"radialDirectionalWeights"])
super().__init__(*args, **kwargs)
self._postInit()
@property
def E(self):
"""Value of E."""
self._E = self.sampleBatchIdx - 1
return self._E
@E.setter
def E(self, E):
RROMPyWarning(("E is used just to simplify inheritance, and its value "
"cannot be changed from that of sampleBatchIdx - 1."))
def _setMAuto(self):
self.M = self.E
def _setNAuto(self):
self.N = self.E
@property
def polydegreetype(self):
"""Value of polydegreetype."""
return "TOTAL"
@polydegreetype.setter
def polydegreetype(self, polydegreetype):
RROMPyWarning(("polydegreetype is used just to simplify inheritance, "
"and its value cannot be changed from 'TOTAL'."))
@property
def polybasis(self):
"""Value of polybasis."""
return self._polybasis
@polybasis.setter
def polybasis(self, polybasis):
try:
polybasis = polybasis.upper().strip().replace(" ","")
if polybasis not in polybases:
raise RROMPyException("Sample type not recognized.")
self._polybasis = polybasis
except:
RROMPyWarning(("Prescribed polybasis not recognized. Overriding "
"to 'MONOMIAL'."))
self._polybasis = "MONOMIAL"
self._approxParameters["polybasis"] = self.polybasis
@property
def errorEstimatorKind(self):
"""Value of errorEstimatorKind."""
return self._errorEstimatorKind
@errorEstimatorKind.setter
def errorEstimatorKind(self, errorEstimatorKind):
errorEstimatorKind = errorEstimatorKind.upper()
if errorEstimatorKind not in self._allowedEstimatorKinds:
RROMPyWarning(("Error estimator kind not recognized. Overriding "
"to 'NONE'."))
errorEstimatorKind = "NONE"
self._errorEstimatorKind = errorEstimatorKind
self._approxParameters["errorEstimatorKind"] = self.errorEstimatorKind
def _polyvanderAuxiliary(self, mus, deg, *args):
return polyvander(mus, deg, *args)
def getErrorEstimatorDiscrepancy(self, mus:Np1D) -> Np1D:
"""Discrepancy-based residual estimator."""
checkIfAffine(self.HFEngine, "apply discrepancy-based error estimator",
False, self._affine_lvl)
mus = self.checkParameterList(mus)
muCTest = self.trainedModel.centerNormalize(mus)
tMverb, self.trainedModel.verbosity = self.trainedModel.verbosity, 0
QTest = self.trainedModel.getQVal(mus)
QTzero = np.where(QTest == 0.)[0]
if len(QTzero) > 0:
RROMPyWarning(("Adjusting estimator to avoid division by "
"numerically zero denominator."))
QTest[QTzero] = np.finfo(np.complex).eps / (1. + self.N)
self.HFEngine.buildA()
self.HFEngine.buildb()
nAs, nbs = self.HFEngine.nAs, self.HFEngine.nbs
muTrainEff = self.mapParameterList(self.mus)
muTestEff = self.mapParameterList(mus)
PTrain = self.trainedModel.getPVal(self.mus).data.T
QTrain = self.trainedModel.getQVal(self.mus)
QTzero = np.where(QTrain == 0.)[0]
if len(QTzero) > 0:
RROMPyWarning(("Adjusting estimator to avoid division by "
"numerically zero denominator."))
QTrain[QTzero] = np.finfo(np.complex).eps / (1. + self.N)
PTest = self.trainedModel.getPVal(mus).data
self.trainedModel.verbosity = tMverb
radiusAbTrain = np.empty((self.S, nAs * self.S + nbs),
dtype = np.complex)
radiusA = np.empty((self.S, nAs, len(mus)), dtype = np.complex)
radiusb = np.empty((nbs, len(mus)), dtype = np.complex)
for j, thA in enumerate(self.HFEngine.thAs):
idxs = j * self.S + np.arange(self.S)
radiusAbTrain[:, idxs] = expressionEvaluator(thA[0], muTrainEff,
(self.S, 1)) * PTrain
radiusA[:, j] = PTest * expressionEvaluator(thA[0], muTestEff,
(len(mus),))
for j, thb in enumerate(self.HFEngine.thbs):
idx = nAs * self.S + j
radiusAbTrain[:, idx] = QTrain * expressionEvaluator(thb[0],
muTrainEff, (self.S,))
radiusb[j] = QTest * expressionEvaluator(thb[0], muTestEff,
(len(mus),))
QRHSNorm2 = self._affineResidualMatricesContraction(radiusb)
vanTrain = self._polyvanderAuxiliary(self._musUniqueCN, self.E,
self.polybasis0, self._derIdxs,
self._reorder)
interpPQ = customFit(vanTrain, radiusAbTrain, rcond = self.interpTol)
vanTest = self._polyvanderAuxiliary(muCTest, self.E, self.polybasis0)
DradiusAb = vanTest.dot(interpPQ)
radiusA = (radiusA
- DradiusAb[:, : - nbs].reshape(len(mus), -1, self.S).T)
radiusb = radiusb - DradiusAb[:, - nbs :].T
ff, Lf, LL = self._affineResidualMatricesContraction(radiusb, radiusA)
err = np.abs((LL - 2. * np.real(Lf) + ff) / QRHSNorm2) ** .5
return err
def getErrorEstimatorLookAhead(self, mus:Np1D,
what : str = "") -> Tuple[Np1D, List[int]]:
"""Residual estimator based on look-ahead idea."""
errTest, QTest, idxMaxEst = self._EIMStep(mus)
mu_muTestS = mus[idxMaxEst]
app_muTestSample = self.getApproxReduced(mu_muTestS)
if self._mode == RROMPy_FRAGILE:
if what == "RES" and not self.HFEngine.isCEye:
raise RROMPyException(("Cannot compute LOOK_AHEAD_RES "
"estimator in fragile mode for "
"non-scalar C."))
app_muTestSample = dot(self.trainedModel.data.projMat[:,
: app_muTestSample.shape[0]],
app_muTestSample)
else:
app_muTestSample = dot(self.samplingEngine.projectionMatrix,
app_muTestSample)
app_muTestSample = sampleList(app_muTestSample)
if what == "RES":
errmu = self.HFEngine.residual(mu_muTestS, app_muTestSample,
post_c = False)
solmu = self.HFEngine.residual(mu_muTestS, None, post_c = False)
normSol = self.HFEngine.norm(solmu, dual = True)
normErr = self.HFEngine.norm(errmu, dual = True)
else:
- applyCglob = (hasattr(self.HFEngine, "_is_C_quadratic")
- and self.HFEngine._is_C_quadratic)
for j, mu in enumerate(mu_muTestS):
uEx = self.samplingEngine.nextSample(mu)
- if what == "OUTPUT" and not applyCglob:
+ if what == "OUTPUT":
uEx = self.HFEngine.applyC(uEx, mu)
app_muTS = self.HFEngine.applyC(app_muTestSample[j], mu)
if j == 0:
app_muTestS = emptySampleList()
app_muTestS.reset((len(app_muTS), len(mu_muTestS)),
dtype = app_muTS.dtype)
app_muTestS[j] = app_muTS
if j == 0:
solmu = emptySampleList()
solmu.reset((len(uEx), len(mu_muTestS)), dtype = uEx.dtype)
solmu[j] = uEx
- if what == "OUTPUT":
- if applyCglob:
- solmu = sampleList(self.HFEngine.applyC(solmu, mu_muTestS))
- app_muTestS = sampleList(self.HFEngine.applyC(
- app_muTestSample, mu_muTestS))
- app_muTestSample = app_muTestS
+ if what == "OUTPUT": app_muTestSample = app_muTestS
errmu = solmu - app_muTestSample
normSol = self.HFEngine.norm(solmu, is_state = what != "OUTPUT")
normErr = self.HFEngine.norm(errmu, is_state = what != "OUTPUT")
errsamples = normErr / normSol
musT = copy(self.mus)
musT.append(mu_muTestS)
musT = self.trainedModel.centerNormalize(musT)
musC = self.trainedModel.centerNormalize(mus)
errT = np.zeros((len(musT), len(mu_muTestS)), dtype = np.complex)
errT[np.arange(len(self.mus), len(musT)),
np.arange(len(mu_muTestS))] = errsamples * QTest[idxMaxEst]
vanT = self._polyvanderAuxiliary(musT, self.E + 1, self.polybasis)
fitOut = customFit(vanT, errT, full = True, rcond = self.interpTol)
vbMng(self, "MAIN",
("Fitting {} samples with degree {} through {}... Conditioning "
"of LS system: {:.4e}.").format(len(vanT), self.E + 1,
polyfitname(self.polybasis),
fitOut[1][2][0] / fitOut[1][2][-1]), 15)
vanC = self._polyvanderAuxiliary(musC, self.E + 1, self.polybasis)
err = np.sum(np.abs(vanC.dot(fitOut[0])), axis = -1) / QTest
return err, idxMaxEst
def getErrorEstimatorNone(self, mus:Np1D) -> Np1D:
"""EIM-based residual estimator."""
err = np.max(self._EIMStep(mus, True), axis = 1)
err *= self.greedyTol / np.mean(err)
return err
def _EIMStep(self, mus:Np1D,
only_one : bool = False) -> Tuple[Np1D, Np1D, List[int]]:
"""Residual estimator based on look-ahead idea."""
mus = self.checkParameterList(mus)
tMverb, self.trainedModel.verbosity = self.trainedModel.verbosity, 0
QTest = self.trainedModel.getQVal(mus)
QTzero = np.where(QTest == 0.)[0]
if len(QTzero) > 0:
RROMPyWarning(("Adjusting estimator to avoid division by "
"numerically zero denominator."))
QTest[QTzero] = np.finfo(np.complex).eps / (1. + self.N)
QTest = np.abs(QTest)
muCTest = self.trainedModel.centerNormalize(mus)
muCTrain = self.trainedModel.centerNormalize(self.mus)
self.trainedModel.verbosity = tMverb
vanTest = self._polyvanderAuxiliary(muCTest, self.E, self.polybasis)
vanTestNext = self._polyvanderAuxiliary(muCTest, self.E + 1,
self.polybasis)[:,
vanTest.shape[1] :]
idxsTest = np.arange(vanTestNext.shape[1])
basis = np.zeros((len(idxsTest), 0), dtype = float)
idxMaxEst = []
while len(idxsTest) > 0:
vanTrial = self._polyvanderAuxiliary(muCTrain, self.E,
self.polybasis)
vanTrialNext = self._polyvanderAuxiliary(muCTrain, self.E + 1,
self.polybasis)[:,
vanTrial.shape[1] :]
vanTrial = np.hstack((vanTrial, vanTrialNext.dot(basis).reshape(
len(vanTrialNext), basis.shape[1])))
valuesTrial = vanTrialNext[:, idxsTest]
vanTestEff = np.hstack((vanTest, vanTestNext.dot(basis).reshape(
len(vanTestNext), basis.shape[1])))
vanTestNextEff = vanTestNext[:, idxsTest]
coeffTest = np.linalg.solve(vanTrial, valuesTrial)
errTest = (np.abs(vanTestNextEff - vanTestEff.dot(coeffTest))
/ np.expand_dims(QTest, 1))
if only_one: return errTest
idxMaxErr = np.unravel_index(np.argmax(errTest), errTest.shape)
idxMaxEst += [idxMaxErr[0]]
muCTrain.append(muCTest[idxMaxErr[0]])
basis = np.pad(basis, [(0, 0), (0, 1)], "constant")
basis[idxsTest[idxMaxErr[1]], -1] = 1.
idxsTest = np.delete(idxsTest, idxMaxErr[1])
return errTest, QTest, idxMaxEst
def errorEstimator(self, mus:Np1D, return_max : bool = False) -> Np1D:
"""Standard residual-based error estimator."""
setupOK = self.setupApproxLocal()
if setupOK > 0:
err = np.empty(len(mus))
err[:] = np.nan
if not return_max: return err
return err, [- setupOK], np.nan
mus = self.checkParameterList(mus)
vbMng(self.trainedModel, "INIT",
"Evaluating error estimator at mu = {}.".format(mus), 10)
if self.errorEstimatorKind == "AFFINE":
err = self.getErrorEstimatorAffine(mus)
else:
self._setupInterpolationIndices()
if self.errorEstimatorKind == "DISCREPANCY":
err = self.getErrorEstimatorDiscrepancy(mus)
elif self.errorEstimatorKind[: 10] == "LOOK_AHEAD":
err, idxMaxEst = self.getErrorEstimatorLookAhead(mus,
self.errorEstimatorKind[11 :])
else: #if self.errorEstimatorKind == "NONE":
err = self.getErrorEstimatorNone(mus)
vbMng(self.trainedModel, "DEL", "Done evaluating error estimator.", 10)
if not return_max: return err
if self.errorEstimatorKind[: 10] != "LOOK_AHEAD":
idxMaxEst = np.empty(self.sampleBatchSize, dtype = int)
errCP = copy(err)
for j in range(self.sampleBatchSize):
k = np.argmax(errCP)
idxMaxEst[j] = k
if j + 1 < self.sampleBatchSize:
musZero = self.trainedModel.centerNormalize(mus, mus[k])
errCP *= np.linalg.norm(musZero.data, axis = 1)
return err, idxMaxEst, err[idxMaxEst]
def plotEstimator(self, *args, **kwargs):
super().plotEstimator(*args, **kwargs)
if self.errorEstimatorKind == "NONE":
vbMng(self, "MAIN",
("Warning! Error estimator has been arbitrarily normalized "
"before plotting."), 15)
def greedyNextSample(self, *args,
**kwargs) -> Tuple[Np1D, int, float, paramVal]:
"""Compute next greedy snapshot of solution map."""
RROMPyAssert(self._mode, message = "Cannot add greedy sample.")
self.sampleBatchIdx += 1
self.sampleBatchSize = totalDegreeN(self.npar - 1, self.sampleBatchIdx)
err, muidx, maxErr, muNext = super().greedyNextSample(*args, **kwargs)
if maxErr is not None and (np.any(np.isnan(maxErr))
or np.any(np.isinf(maxErr))):
self.sampleBatchIdx -= 1
self.sampleBatchSize = totalDegreeN(self.npar - 1,
self.sampleBatchIdx)
if (self.errorEstimatorKind == "NONE" and not np.isnan(maxErr)
and not np.isinf(maxErr)):
maxErr = None
return err, muidx, maxErr, muNext
def _setSampleBatch(self, maxS:int):
self.sampleBatchIdx, self.sampleBatchSize, S = -1, 0, 0
nextBatchSize = 1
while S + nextBatchSize <= maxS:
self.sampleBatchIdx += 1
self.sampleBatchSize = nextBatchSize
S += self.sampleBatchSize
nextBatchSize = totalDegreeN(self.npar - 1,
self.sampleBatchIdx + 1)
return S
def _preliminaryTraining(self):
"""Initialize starting snapshots of solution map."""
RROMPyAssert(self._mode, message = "Cannot start greedy algorithm.")
if self.samplingEngine.nsamples > 0: return
self._S = self._setSampleBatch(self.S)
super()._preliminaryTraining()
self.M, self.N = ("AUTO",) * 2
def setupApproxLocal(self) -> int:
"""Compute rational interpolant."""
if self.checkComputedApprox(): return -1
RROMPyAssert(self._mode, message = "Cannot setup approximant.")
self.verbosity -= 10
vbMng(self, "INIT", "Setting up local approximant.", 5)
- pMatOld, pMat = None, self.samplingEngine.projectionMatrix
+ pMat = self.samplingEngine.projectionMatrix
firstRun = self.trainedModel is None
- applyCglob = (hasattr(self.HFEngine, "_is_C_quadratic")
- and self.HFEngine._is_C_quadratic)
if not firstRun:
Sold = len(self.trainedModel.data.mus)
- if applyCglob: pMatOld = pMat[:, : Sold]
pMat = pMat[:, Sold :]
- self._setupTrainedModel(pMat, not firstRun, pMatOld)
+ self._setupTrainedModel(pMat, not firstRun)
self.catchInstability = 2
vbDepth = getVerbosityDepth()
unstable = 0
if self.E > 0:
try:
Q = self._setupDenominator()
except RROMPyException as RE:
if RE.critical: raise RE from None
setVerbosityDepth(vbDepth)
RROMPyWarning("Downgraded {}: {}".format(RE.__class__.__name__,
RE))
unstable = 1
else:
Q = PI()
Q.coeffs = np.ones((1,) * self.npar, dtype = np.complex)
Q.npar = self.npar
Q.polybasis = self.polybasis
if not unstable:
self.trainedModel.data.Q = copy(Q)
try:
P = copy(self._setupNumerator())
except RROMPyException as RE:
if RE.critical: raise RE from None
setVerbosityDepth(vbDepth)
RROMPyWarning("Downgraded {}: {}".format(RE.__class__.__name__,
RE))
unstable = 1
if not unstable:
self.trainedModel.data.P = copy(P)
self.trainedModel.data.approxParameters = copy(
self.approxParameters)
vbMng(self, "DEL", "Done setting up local approximant.", 5)
self.catchInstability = 0
self.verbosity += 10
return unstable
def setupApprox(self, plotEst : str = "NONE") -> int:
val = super().setupApprox(plotEst)
if val == 0:
if (self.errorEstimatorKind[:10] == "LOOK_AHEAD"
and len(self.mus) < self.samplingEngine.nsamples):
while len(self.mus) < self.samplingEngine.nsamples:
self.mus.append(self.samplingEngine.mus[len(self.mus)])
self.trainedModel = None
self._S = self._setSampleBatch(len(self.mus) + 1)
self.setupApproxLocal()
self._setupRational(self.trainedModel.data.Q,
self.trainedModel.data.P)
self.trainedModel.data.approxParameters = copy(
self.approxParameters)
return val
def loadTrainedModel(self, filename:str):
"""Load trained reduced model from file."""
super().loadTrainedModel(filename)
self._setSampleBatch(self.S + 1)
diff --git a/rrompy/reduction_methods/standard/greedy/reduced_basis_greedy.py b/rrompy/reduction_methods/standard/greedy/reduced_basis_greedy.py
index 080449a..f4efc07 100644
--- a/rrompy/reduction_methods/standard/greedy/reduced_basis_greedy.py
+++ b/rrompy/reduction_methods/standard/greedy/reduced_basis_greedy.py
@@ -1,149 +1,146 @@
# Copyright (C) 2018-2020 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 .
#
from copy import deepcopy as copy
from .generic_greedy_approximant import GenericGreedyApproximant
from rrompy.reduction_methods.standard import ReducedBasis
from rrompy.utilities.base import verbosityManager as vbMng
from rrompy.utilities.exception_manager import RROMPyWarning, RROMPyAssert
__all__ = ['ReducedBasisGreedy']
class ReducedBasisGreedy(GenericGreedyApproximant, ReducedBasis):
"""
ROM greedy RB approximant computation for parametric problems.
Args:
HFEngine: HF problem solver.
mu0(optional): Default parameter. Defaults to 0.
approxParameters(optional): Dictionary containing values for main
parameters of approximant. Recognized keys are:
- 'POD': kind of snapshots orthogonalization; allowed values
include 0, 1/2, and 1; defaults to 1, i.e. POD;
- 'scaleFactorDer': scaling factors for derivative computation;
defaults to 'AUTO';
- 'S': number of starting training points;
- 'sampler': sample point generator;
- 'greedyTol': uniform error tolerance for greedy algorithm;
defaults to 1e-2;
- 'collinearityTol': collinearity tolerance for greedy algorithm;
defaults to 0.;
- 'maxIter': maximum number of greedy steps; defaults to 1e2;
- 'nTestPoints': number of test points; defaults to 5e2;
- 'trainSetGenerator': training sample points generator; defaults
to sampler.
Defaults to empty dict.
verbosity(optional): Verbosity level. Defaults to 10.
Attributes:
HFEngine: HF problem solver.
mu0: Default parameter.
mus: Array of snapshot parameters.
approxParameters: Dictionary containing values for main parameters of
approximant. Recognized keys are in parameterList.
parameterListSoft: Recognized keys of soft approximant parameters:
- 'POD': kind of snapshots orthogonalization;
- 'scaleFactorDer': scaling factors for derivative computation;
- 'greedyTol': uniform error tolerance for greedy algorithm;
- 'collinearityTol': collinearity tolerance for greedy algorithm;
- 'maxIter': maximum number of greedy steps;
- 'nTestPoints': number of test points;
- 'trainSetGenerator': training sample points generator.
parameterListCritical: Recognized keys of critical approximant
parameters:
- 'S': total number of samples current approximant relies upon;
- 'sampler': sample point generator.
verbosity: Verbosity level.
POD: Kind of snapshots orthogonalization.
scaleFactorDer: Scaling factors for derivative computation.
S: number of test points.
sampler: Sample point generator.
greedyTol: uniform error tolerance for greedy algorithm.
collinearityTol: Collinearity tolerance for greedy algorithm.
maxIter: maximum number of greedy steps.
nTestPoints: number of starting training points.
trainSetGenerator: training sample points generator.
muBounds: list of bounds for parameter values.
samplingEngine: Sampling engine.
uHF: High fidelity solution(s) with parameter(s) lastSolvedHF as
sampleList.
lastSolvedHF: Parameter(s) corresponding to last computed high fidelity
solution(s) as parameterList.
uApproxReduced: Reduced approximate solution(s) with parameter(s)
lastSolvedApprox as sampleList.
lastSolvedApproxReduced: Parameter(s) corresponding to last computed
reduced approximate solution(s) as parameterList.
uApprox: Approximate solution(s) with parameter(s) lastSolvedApprox as
sampleList.
lastSolvedApprox: Parameter(s) corresponding to last computed
approximate solution(s) as parameterList.
As: List of sparse matrices (in CSC format) representing coefficients
of linear system matrix.
bs: List of numpy vectors representing coefficients of linear system
RHS.
ARBs: List of sparse matrices (in CSC format) representing coefficients
of compressed linear system matrix.
bRBs: List of numpy vectors representing coefficients of compressed
linear system RHS.
"""
def __init__(self, *args, **kwargs):
self._preInit()
self._addParametersToList(toBeExcluded = ["R", "PODTolerance"])
super().__init__(*args, **kwargs)
self._postInit()
@property
def PODTolerance(self):
"""Value of PODTolerance."""
self._PODTolerance = -1
return self._PODTolerance
@PODTolerance.setter
def PODTolerance(self, PODTolerance):
RROMPyWarning(("PODTolerance is used just to simplify inheritance, "
"and its value cannot be changed from -1."))
def setupApproxLocal(self) -> int:
"""Compute RB projection matrix."""
if self.checkComputedApprox(): return -1
RROMPyAssert(self._mode, message = "Cannot setup approximant.")
self.verbosity -= 10
vbMng(self, "INIT", "Setting up local approximant.", 5)
vbMng(self, "INIT", "Extracting projection matrix.", 15)
pMatOld, pMat = None, self.samplingEngine.projectionMatrix
firstRun = self.trainedModel is None
- applyCglob = (hasattr(self.HFEngine, "_is_C_quadratic")
- and self.HFEngine._is_C_quadratic)
if not firstRun:
Sold = len(self.trainedModel.data.mus)
_pMatOld, pMat = pMat[:, : Sold], pMat[:, Sold :]
- if applyCglob: pMatOld = _pMatOld
vbMng(self, "DEL", "Done extracting projection matrix.", 15)
- self._setupTrainedModel(pMat, not firstRun, pMatOld)
+ self._setupTrainedModel(pMat, not firstRun)
if firstRun:
self.trainedModel.data.affinePoly = self.HFEngine.affinePoly
self.trainedModel.data.thAs = self.HFEngine.thAs
self.trainedModel.data.thbs = self.HFEngine.thbs
- elif not applyCglob: pMatOld = _pMatOld
+ else: pMatOld = _pMatOld
ARBs, bRBs = self.assembleReducedSystem(pMat, pMatOld)
self.trainedModel.data.ARBs = ARBs
self.trainedModel.data.bRBs = bRBs
self.trainedModel.data.approxParameters = copy(self.approxParameters)
vbMng(self, "DEL", "Done setting up local approximant.", 5)
self.verbosity += 10
return 0
diff --git a/rrompy/reduction_methods/standard/trained_model/trained_model_nearest_neighbor.py b/rrompy/reduction_methods/standard/trained_model/trained_model_nearest_neighbor.py
index 3b8d49f..6a62c1e 100644
--- a/rrompy/reduction_methods/standard/trained_model/trained_model_nearest_neighbor.py
+++ b/rrompy/reduction_methods/standard/trained_model/trained_model_nearest_neighbor.py
@@ -1,106 +1,98 @@
# Copyright (C) 2018-2020 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 .
#
import numpy as np
from rrompy.reduction_methods.base.trained_model.trained_model import (
TrainedModel)
from rrompy.utilities.numerical.compress_matrix import compressMatrix
from rrompy.utilities.base.types import Np1D, paramList, sampList
from rrompy.utilities.base import verbosityManager as vbMng
-from rrompy.utilities.exception_manager import RROMPyException, RROMPyWarning
+from rrompy.utilities.exception_manager import RROMPyWarning
from rrompy.sampling import emptySampleList
__all__ = ['TrainedModelNearestNeighbor']
class TrainedModelNearestNeighbor(TrainedModel):
"""
ROM approximant evaluation for nearest neighbor approximant.
Attributes:
Data: dictionary with all that can be pickled.
"""
- @property
- def supportEnd(self):
- # for a quadratic output, it is different from data.projMat.shape[1]
- idxIncl = np.argmax(self.data.supp)
- return self.data.supp[idxIncl] + len(self.data.vals[idxIncl])
-
def compress(self, collapse : bool = False, tol : float = 0., *args,
**kwargs):
if not collapse and tol <= 0.: return
- if hasattr(self.data, "_is_C_quadratic") and self.data._is_C_quadratic:
- raise RROMPyException(("Cannot compress model with quadratic "
- "output."))
RMat = self.data.projMat
if not collapse:
if hasattr(self.data, "_compressTol"):
RROMPyWarning(("Recompressing already compressed model is "
"ineffective. Aborting."))
return
self.data.projMat, RMat, _ = compressMatrix(RMat, tol, *args,
**kwargs)
for j, suppj in enumerate(self.data.supp):
self.data.vals[j] = RMat[:, suppj : suppj + len(self.data.vals[j])
].dot(self.data.vals[j])
self.data.supp[j] = [0]
super().compress(collapse, tol)
def getNearestNeighbor(self, mu : paramList = []) -> Np1D:
"""
Find nearest neighbor to arbitrary parameter.
Args:
mu: Target parameter.
"""
mu = self.checkParameterList(mu)
idxUnique, idxmap = np.unique(self.data.NN(mu), return_inverse = True)
idxUnique = np.array(idxUnique, dtype = int)
vbMng(self, "INIT", "Finding nearest neighbor to mu = {}.".format(mu),
22)
nn = emptySampleList()
for i, iM in enumerate(idxUnique):
idx = np.where(idxmap == i)[0]
val, supp = self.data.vals[iM], self.data.supp[iM]
if i == 0:
- nn.reset((self.supportEnd, len(mu)), dtype = val.dtype)
+ nn.reset((self.data.projMat.shape[1], len(mu)),
+ dtype = val.dtype)
nn.data[:] = 0.
for i in idx: nn.data[supp : supp + len(val), i] = val
vbMng(self, "DEL", "Done finding nearest neighbor.", 22)
return nn
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",
"Evaluating approximant at mu = {}.".format(mu), 12)
self.uApproxReduced = self.getNearestNeighbor(mu)
vbMng(self, "DEL", "Done evaluating approximant.", 12)
self.lastSolvedApproxReduced = mu
return self.uApproxReduced
def getPoles(self, *args, **kwargs) -> Np1D:
"""Obtain approximant poles."""
return np.empty(0)
diff --git a/rrompy/reduction_methods/standard/trained_model/trained_model_rational.py b/rrompy/reduction_methods/standard/trained_model/trained_model_rational.py
index eb055c8..9e8aeab 100644
--- a/rrompy/reduction_methods/standard/trained_model/trained_model_rational.py
+++ b/rrompy/reduction_methods/standard/trained_model/trained_model_rational.py
@@ -1,197 +1,188 @@
# Copyright (C) 2018-2020 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 .
#
import numpy as np
from collections.abc import Iterable
from rrompy.reduction_methods.base.trained_model.trained_model import (
TrainedModel)
from rrompy.utilities.numerical import dot
from rrompy.utilities.numerical.compress_matrix import compressMatrix
from rrompy.utilities.base.types import (Np1D, Np2D, List, paramVal, paramList,
sampList)
from rrompy.utilities.base import verbosityManager as vbMng, freepar as fp
from rrompy.utilities.exception_manager import RROMPyException, RROMPyWarning
from rrompy.parameter import emptyParameterList
from rrompy.sampling import sampleList
__all__ = ['TrainedModelRational']
class TrainedModelRational(TrainedModel):
"""
ROM approximant evaluation for rational approximant.
Attributes:
Data: dictionary with all that can be pickled.
"""
def compress(self, collapse : bool = False, tol : float = 0., *args,
**kwargs):
if not collapse and tol <= 0.: return
- if hasattr(self.data, "_is_C_quadratic") and self.data._is_C_quadratic:
- raise RROMPyException(("Cannot compress model with quadratic "
- "output."))
RMat = self.data.projMat
if not collapse:
if hasattr(self.data, "_compressTol"):
RROMPyWarning(("Recompressing already compressed model is "
"ineffective. Aborting."))
return
self.data.projMat, RMat, _ = compressMatrix(RMat, tol, *args,
**kwargs)
self.data.P.postmultiplyTensorize(RMat.T)
super().compress(collapse, tol)
def centerNormalize(self, mu : paramList = [],
mu0 : paramVal = None) -> paramList:
"""
Compute normalized parameter to be plugged into approximant.
Args:
mu: Parameter(s) 1.
mu0: Parameter(s) 2. If None, set to self.data.mu0.
Returns:
Normalized parameter.
"""
mu = self.checkParameterList(mu)
if mu0 is None: mu0 = self.data.mu0
return (self.mapParameterList(mu)
- self.mapParameterList(mu0)) / self.data.scaleFactor
def getPVal(self, mu : paramList = []) -> sampList:
"""
Evaluate rational numerator at arbitrary parameter.
Args:
mu: Target parameter.
"""
mu = self.checkParameterList(mu)
vbMng(self, "INIT", "Evaluating numerator at mu = {}.".format(mu), 17)
p = sampleList(self.data.P(self.centerNormalize(mu)))
vbMng(self, "DEL", "Done evaluating numerator.", 17)
return p
def getQVal(self, mu:Np1D, der : List[int] = None,
scl : Np1D = None) -> Np1D:
"""
Evaluate rational denominator at arbitrary parameter.
Args:
mu: Target parameter.
der(optional): Derivatives to take before evaluation.
"""
mu = self.checkParameterList(mu)
vbMng(self, "INIT", "Evaluating denominator at mu = {}.".format(mu),
17)
q = self.data.Q(self.centerNormalize(mu), der, scl)
vbMng(self, "DEL", "Done evaluating denominator.", 17)
return q
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",
"Evaluating approximant at mu = {}.".format(mu), 12)
QV = self.getQVal(mu)
QVzero = np.where(QV == 0.)[0]
if len(QVzero) > 0:
QV[QVzero] = np.finfo(np.complex).eps / (1.
+ self.data.Q.deg[0])
self.uApproxReduced = self.getPVal(mu) / QV
vbMng(self, "DEL", "Done evaluating approximant.", 12)
self.lastSolvedApproxReduced = mu
return self.uApproxReduced
def getPoles(self, *args, **kwargs) -> Np1D:
"""
Obtain approximant poles.
Returns:
Numpy complex vector of poles.
"""
if len(args) + len(kwargs) > 1:
raise RROMPyException(("Wrong number of parameters passed. "
"Only 1 available."))
elif len(args) + len(kwargs) == 1:
if len(args) == 1:
mVals = args[0]
else:
mVals = kwargs["marginalVals"]
if not isinstance(mVals, Iterable): mVals = [mVals]
mVals = list(mVals)
else:
mVals = [fp]
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."))
mVals[rDim] = self.data.mu0(rDim)
mVals = list(self.centerNormalize(mVals).data.flatten())
mVals[rDim] = fp
roots = self.data.scaleFactor[rDim] * self.data.Q.roots(mVals)
return self.mapParameterList(self.mapParameterList(self.data.mu0(rDim),
idx = [rDim])(0, 0)
+ roots, "B", [rDim])(0)
def getResidues(self, *args, **kwargs) -> Np2D:
"""
Obtain approximant residues.
Returns:
Numpy matrix with residues as columns.
"""
pls = self.getPoles(*args, **kwargs)
if len(pls) == 0:
return pls, np.empty((0, 0), dtype = self.data.P.coeffs.dtype)
if len(args) == 1:
mVals = args[0]
elif len(args) == 0:
mVals = [None]
else:
mVals = kwargs["marginalVals"]
if not isinstance(mVals, Iterable): mVals = [mVals]
mVals = list(mVals)
rDim = mVals.index(fp)
poles = emptyParameterList()
poles.reset((len(pls), self.data.npar), dtype = pls.dtype)
for k, pl in enumerate(pls):
mValsLoc = list(mVals)
mValsLoc[rDim] = pl
poles[k] = mValsLoc
QV = self.getQVal(poles, list(1 * (np.arange(self.data.npar) == rDim)))
QVzero = np.where(QV == 0.)[0]
if len(QVzero) > 0:
RROMPyWarning(("Adjusting residuals to avoid division by "
"numerically zero denominator."))
QV[QVzero] = np.finfo(np.complex).eps / (1. + self.data.Q.deg[0])
res = self.getPVal(poles).data / QV
- if hasattr(self.data, "_is_C_quadratic") and self.data._is_C_quadratic:
- self._setupQuadMapping()
- res = res[self.quad_mapping[0]] * res[self.quad_mapping[1]].conj()
if not self.data._collapsed: res = dot(self.data.projMat, res).T
- if (hasattr(self.data, "_is_C_quadratic")
- and self.data._is_C_quadratic == 2):
- res = np.real(res)
return pls, res
diff --git a/rrompy/reduction_methods/standard/trained_model/trained_model_reduced_basis.py b/rrompy/reduction_methods/standard/trained_model/trained_model_reduced_basis.py
index f7d01d6..802237d 100644
--- a/rrompy/reduction_methods/standard/trained_model/trained_model_reduced_basis.py
+++ b/rrompy/reduction_methods/standard/trained_model/trained_model_reduced_basis.py
@@ -1,156 +1,153 @@
# Copyright (C) 2018-2020 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 .
#
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, "_is_C_quadratic") and self.data._is_C_quadratic:
- raise RROMPyException(("Cannot compress model with quadratic "
- "output."))
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)
diff --git a/rrompy/utilities/numerical/point_distances.py b/rrompy/utilities/numerical/point_distances.py
index cd0726a..a6adb03 100644
--- a/rrompy/utilities/numerical/point_distances.py
+++ b/rrompy/utilities/numerical/point_distances.py
@@ -1,80 +1,86 @@
# Copyright (C) 2018-2020 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 .
#
import numpy as np
-from rrompy.utilities.base.types import Tuple, Np1D, Np2D, HFEng
+from scipy.sparse import spmatrix
+from rrompy.utilities.base.types import Tuple, List, Np1D, Np2D, HFEng
__all__ = ['baseDistanceMatrix', 'vectorDistanceMatrix',
'doubleDistanceMatrix']
def baseDistanceMatrix(x:Np2D, y : Np2D = None, npar : int = None,
magnitude : bool = True, weights : Np1D = None) -> Np2D:
if npar is None: npar = x.shape[1] if x.ndim > 1 else 1
if y is None: y = x
if x.ndim != 3 or x.shape[1] != npar: x = x.reshape(-1, 1, npar)
if y.ndim != 2 or y.shape[1] != npar: y = y.reshape(-1, npar)
dist = np.repeat(x, len(y), axis = 1) - y
if weights is not None: dist *= np.array(weights).flatten()
if magnitude:
if dist.shape[2] == 1:
dist = np.abs(dist)[..., 0]
else:
dist = np.sum(np.abs(dist) ** 2., axis = 2) ** .5
return dist
def vectorDistanceMatrix(X:Np2D, Y:Np2D, HFEngine : HFEng = None,
is_state : bool = True, chordalRadius : float = -1,
- badValue : float = 0.) -> Np2D:
+ Xbad : List[bool] = None,
+ Ybad : List[bool] = None) -> Np2D:
if HFEngine is None:
innerT = np.real(Y.T.conj().dot(X))
- norm2X = np.sum(np.abs(X) ** 2., axis = 0)
- norm2Y = np.sum(np.abs(Y) ** 2., axis = 0)
+ if isinstance(X, (spmatrix,)):
+ norm2X = np.sum(np.abs(X.todense()) ** 2., axis = 0)
+ else:
+ norm2X = np.sum(np.abs(X) ** 2., axis = 0)
+ if isinstance(Y, (spmatrix,)):
+ norm2Y = np.sum(np.abs(Y.todense()) ** 2., axis = 0)
+ else:
+ norm2Y = np.sum(np.abs(Y) ** 2., axis = 0)
else:
innerT = np.real(HFEngine.innerProduct(X, Y, is_state = is_state))
norm2X = HFEngine.norm(X, is_state = is_state) ** 2.
norm2Y = HFEngine.norm(Y, is_state = is_state) ** 2.
+ if Xbad is None: Xbad = np.where(np.isinf(norm2X))[0]
+ if Ybad is None: Ybad = np.where(np.isinf(norm2Y))[0]
dist2T = (np.tile(norm2Y.reshape(-1, 1), len(norm2X))
+ norm2X.reshape(1, -1) - 2 * innerT)
- if np.isinf(badValue):
- xbad = np.where(np.isinf(norm2X))[0]
- ybad = np.where(np.isinf(norm2Y))[0]
- else:
- xbad = np.where(np.isclose(norm2X, badValue, atol = 1e-15))[0]
- ybad = np.where(np.isclose(norm2Y, badValue, atol = 1e-15))[0]
if chordalRadius <= 0:
- dist2T[:, xbad], dist2T[ybad, :] = np.inf, np.inf
+ dist2T[:, Xbad], dist2T[Ybad, :] = np.inf, np.inf
else:
- dist2T[:, xbad], dist2T[ybad, :] = 1., 1.
- dist2T[np.ix_(ybad, xbad)] = 0.
+ dist2T[:, Xbad], dist2T[Ybad, :] = 1., 1.
+ dist2T[np.ix_(Ybad, Xbad)] = 0.
dist2T[dist2T < 0.] = 0.
if chordalRadius <= 0: return dist2T.T ** .5
- norm2X[xbad], norm2Y[ybad] = 0., 0.
+ norm2X[Xbad], norm2Y[Ybad] = 0., 0.
norm2X, norm2Y = norm2X / chordalRadius ** 2., norm2Y / chordalRadius ** 2.
return ((dist2T / (norm2X + 1.)).T / (norm2Y + 1.)) ** .5
def doubleDistanceMatrix(x:Np1D, y:Np1D, w : float = 0, X : Np2D = None,
Y : Np2D = None, HFEngine : HFEng = None,
is_state : bool = True,
chordalRadius : Tuple[float, float] = [-1] * 2) \
-> Np2D:
+ Xbad, Ybad = np.where(np.isinf(x))[0], np.where(np.isinf(y))[0]
dist = vectorDistanceMatrix(np.reshape(x, [1, -1]), np.reshape(y, [1, -1]),
- chordalRadius = chordalRadius[0],
- badValue = np.inf)
+ chordalRadius = chordalRadius[0], Xbad = Xbad,
+ Ybad = Ybad)
if w == 0: return dist
- distAdj = vectorDistanceMatrix(X, Y, HFEngine, is_state, chordalRadius[1])
+ distAdj = vectorDistanceMatrix(X, Y, HFEngine, is_state, chordalRadius[1],
+ Xbad = Xbad, Ybad = Ybad)
return (dist + w * distAdj) / (1. + w)
diff --git a/rrompy/utilities/numerical/point_matching.py b/rrompy/utilities/numerical/point_matching.py
index 7288cf6..ffe0b42 100644
--- a/rrompy/utilities/numerical/point_matching.py
+++ b/rrompy/utilities/numerical/point_matching.py
@@ -1,132 +1,121 @@
# Copyright (C) 2018-2020 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 .
#
from copy import deepcopy as copy
import numpy as np
from scipy.optimize import linear_sum_assignment as LSA
+from .tensor_la import dot
from .point_distances import baseDistanceMatrix, doubleDistanceMatrix
from rrompy.utilities.base.types import Tuple, List, ListAny, Np1D, Np2D, HFEng
from rrompy.utilities.exception_manager import RROMPyAssert
__all__ = ['pointMatching', 'rationalFunctionMatching']
def pointMatching(distMatrix:Np2D) -> Tuple[Np1D, Np1D]:
return LSA(distMatrix)
-def buildResiduesForDistance(res:Np2D, projMat:Np2D, supp:int,
- projMapping : Np2D = None,
- projMappingReal : bool = False) -> Np2D:
- if projMapping is not None:
- badidx = np.where(projMapping >= len(res))
- if len(badidx[1]) > 0: projMapping = projMapping[:, : badidx[1][0]]
- res = res[projMapping[0]] * res[projMapping[1]].conj()
- if isinstance(projMat, (np.ndarray,)):
- res = projMat[:, supp : supp + len(res)].dot(res)
- if projMapping is not None and projMappingReal: res = np.real(res)
- return res
-
def rationalFunctionMatching(poles:List[Np1D], coeffs:List[Np2D],
featPts:Np2D, matchingWeight:float, supps:ListAny,
projMat:Np2D, HFEngine : HFEng = None,
is_state : bool = True, root : int = None,
- chordalRadius : Tuple[float, float] = [-1] * 2,
- projMapping : Np2D = None,
- projMappingReal : bool = False) \
+ chordalRadius : Tuple[float, float] = [-1] * 2) \
-> Tuple[List[Np1D], List[Np2D]]:
"""
Match poles and residues of a set of rational functions.
Args:
poles: List of (lists of) poles.
coeffs: List of (lists of) residues.
featPts: Marginal parameters corresponding to rational models.
matchingWeight: Matching weight in distance computation.
supps: Support indices for projection matrix.
projMat: Projection matrix for residues.
HFEngine(optional): Engine for distance evaluation. Defaults to None,
i.e. Euclidean metric.
is_state(optional): Whether residues are of system state. Defaults to
True.
root(optional): Root of search tree. Defaults to None, i.e.
automatically chosen.
chordalRadius(optional): Radius to be used in chordal metric. If <= 0,
Euclidean metric is used. Defaults to [-1, -1].
- projMapping(optional): Mapping for projection based on projMap. Should
- be assigned for nonlinear outputs. Defaults to None.
- projMappingReal(optional): Whether projection based on projMap is
- followed by collapse onto real part. Defaults to False.
Returns:
Matched list of (lists of) poles and list of (lists of) residues.
"""
M, N = len(featPts), len(poles[0])
RROMPyAssert(len(poles), M, "Number of rational functions to be matched")
RROMPyAssert(len(coeffs), M, "Number of rational functions to be matched")
if M <= 1: return poles, coeffs
featDist = baseDistanceMatrix(featPts)
free = list(range(M))
if root is None:
#start from sample point with closest neighbor,
#among those with no inf pole
notInfPls = np.where([np.logical_not(np.any(np.isinf(p)))
for p in poles])[0]
MEff = len(notInfPls)
if MEff == 1:
root = notInfPls[0]
else:
featDistEff = featDist[notInfPls][:, notInfPls]
root = notInfPls[np.argpartition(featDistEff.flatten(),
MEff)[MEff] % MEff]
polesC = copy(poles)
if matchingWeight != 0:
- resC = [buildResiduesForDistance(coeffs[j][: N].T, projMat, supps[j],
- projMapping, projMappingReal)
- for j in range(M)]
+ resC = [dot(projMat[:, supps[j] : supps[j] + coeffs[j].shape[1]],
+ coeffs[j][: N].T) for j in range(M)]
+ if chordalRadius[1] == "AUTO":
+ if HFEngine is None:
+ norm2S = [np.sum(np.abs(c) ** 2., axis = 0) for c in resC]
+ else:
+ norm2S = [HFEngine.norm(c, is_state = is_state) ** 2.
+ for c in resC]
+ chordalRadius[1] = np.mean(norm2S)
fixed = [free.pop(root)]
for j in range(M - 1, 0, -1):
#find closest point
idx = np.argmin(featDist[np.ix_(fixed, free)].flatten())
Ifix = fixed[idx // j]
fixed += [free.pop(idx % j)]
Ifree = fixed[-1]
plsfix, plsfree = polesC[Ifix], polesC[Ifree]
freeInf = np.where(np.isinf(plsfree))[0]
freeNotInf = np.where(np.logical_not(np.isinf(plsfree)))[0]
plsfree = plsfree[freeNotInf]
if matchingWeight == 0:
resfix, resfree = None, None
else:
resfix, resfree = resC[Ifix], resC[Ifree][:, freeNotInf]
#build assignment distance matrix
distj = doubleDistanceMatrix(plsfree, plsfix, matchingWeight, resfree,
resfix, HFEngine, is_state,
chordalRadius)
reordering = pointMatching(distj)[1]
reorderingInf = [x for x in range(N) if x not in reordering]
#reorder good poles
poles[Ifree][reordering], poles[Ifree][reorderingInf] = (
poles[Ifree][freeNotInf], poles[Ifree][freeInf])
coeffs[Ifree][reordering], coeffs[Ifree][reorderingInf] = (
coeffs[Ifree][freeNotInf], coeffs[Ifree][freeInf])
#transfer missing poles over
polesC[Ifree][reordering], polesC[Ifree][reorderingInf] = (
polesC[Ifree][freeNotInf], polesC[Ifix][reorderingInf])
if matchingWeight != 0:
resC[Ifree][:, reordering], resC[Ifree][:, reorderingInf] = (
resC[Ifree][:, freeNotInf], resC[Ifix][:, reorderingInf])
return poles, coeffs