Page Menu
Home
c4science
Search
Configure Global Search
Log In
Files
F60993172
generic_approximant.py
No One
Temporary
Actions
Download File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Award Token
Subscribers
None
File Metadata
Details
File Info
Storage
Attached
Created
Fri, May 3, 20:42
Size
35 KB
Mime Type
text/x-python
Expires
Sun, May 5, 20:42 (1 d, 23 h)
Engine
blob
Format
Raw Data
Handle
17450435
Attached To
R6746 RationalROMPy
generic_approximant.py
View Options
# 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 <http://www.gnu.org/licenses/>.
#
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
,
getargs
=
{},
normargs
=
{})
->
Np1D
:
uV
=
getattr
(
self
.
__class__
,
"get"
+
fieldName
)(
self
,
mu
,
**
getargs
)
normargs
[
"is_state"
]
=
False
val
=
self
.
HFEngine
.
norm
(
uV
,
**
normargs
)
return
val
setattr
(
self
.
__class__
,
"norm"
+
fieldName
,
objFunc
)
def
addNormDualFieldToClass
(
self
,
fieldName
):
def
objFunc
(
self
,
mu
:
paramList
,
getargs
=
{},
normargs
=
{})
->
Np1D
:
uV
=
getattr
(
self
.
__class__
,
"get"
+
fieldName
)(
self
,
mu
,
**
getargs
)
normargs
[
"is_state"
]
=
True
if
"dual"
not
in
normargs
.
keys
():
normargs
[
"dual"
]
=
True
val
=
self
.
HFEngine
.
norm
(
uV
,
**
normargs
)
return
val
setattr
(
self
.
__class__
,
"norm"
+
fieldName
,
objFunc
)
def
addPlotFieldToClass
(
self
,
fieldName
):
def
objFunc
(
self
,
mu
:
paramList
,
getargs
=
{},
plotargs
=
{}):
uV
=
getattr
(
self
.
__class__
,
"get"
+
fieldName
)(
self
,
mu
,
**
getargs
)
uV
=
listScatter
(
uV
)[
0
]
.
T
filesOut
=
[]
if
len
(
uV
)
>
0
:
if
"name"
in
plotargs
.
keys
():
nameBase
=
copy
(
plotargs
[
"name"
])
for
j
,
u
in
enumerate
(
uV
):
if
"name"
in
plotargs
.
keys
()
and
len
(
uV
)
>
1
:
plotargs
[
"name"
]
=
nameBase
+
str
(
j
)
filesOut
+=
[
self
.
HFEngine
.
plot
(
u
,
**
plotargs
)]
if
"name"
in
plotargs
.
keys
():
plotargs
[
"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
,
getargs
=
{},
plotargs
=
{}):
uV
=
getattr
(
self
.
__class__
,
"get"
+
fieldName
)(
self
,
mu
,
**
getargs
)
uV
=
listScatter
(
uV
)[
0
]
.
T
filesOut
=
[]
if
len
(
uV
)
>
0
:
if
"name"
in
plotargs
.
keys
():
nameBase
=
copy
(
plotargs
[
"name"
])
for
j
,
u
in
enumerate
(
uV
):
if
"name"
in
plotargs
.
keys
()
and
len
(
uV
)
>
1
:
plotargs
[
"name"
]
=
nameBase
+
str
(
j
)
filesOut
+=
[
self
.
HFEngine
.
plot
(
u
,
**
plotargs
)]
if
"name"
in
plotargs
.
keys
():
plotargs
[
"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
,
getargs
=
{},
outargs
=
{}):
if
not
hasattr
(
self
.
HFEngine
,
"outParaview"
):
raise
RROMPyException
((
"High fidelity engine cannot output to "
"Paraview."
))
uV
=
getattr
(
self
.
__class__
,
"get"
+
fieldName
)(
self
,
mu
,
**
getargs
)
uV
=
listScatter
(
uV
)[
0
]
.
T
filesOut
=
[]
if
len
(
uV
)
>
0
:
if
"name"
in
outargs
.
keys
():
nameBase
=
copy
(
outargs
[
"name"
])
for
j
,
u
in
enumerate
(
uV
):
if
"name"
in
outargs
.
keys
()
and
len
(
uV
)
>
1
:
outargs
[
"name"
]
=
nameBase
+
str
(
j
)
filesOut
+=
[
self
.
HFEngine
.
outParaview
(
u
,
**
outargs
)]
if
"name"
in
outargs
.
keys
():
outargs
[
"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
,
getargs
=
{},
outargs
=
{}):
if
not
hasattr
(
self
.
HFEngine
,
"outParaviewTimeDomain"
):
raise
RROMPyException
((
"High fidelity engine cannot output to "
"Paraview."
))
uV
=
getattr
(
self
.
__class__
,
"get"
+
fieldName
)(
self
,
mu
,
**
getargs
)
uV
=
listScatter
(
uV
)[
0
]
.
T
filesOut
=
[]
if
len
(
uV
)
>
0
:
if
"omega"
not
in
outargs
.
keys
():
outargs
[
"omega"
]
=
np
.
real
(
mu
)
if
"name"
in
outargs
.
keys
():
nameBase
=
copy
(
outargs
[
"name"
])
filesOut
=
[]
for
j
,
u
in
enumerate
(
uV
):
if
"name"
in
outargs
.
keys
()
and
len
(
uV
)
>
1
:
outargs
[
"name"
]
=
nameBase
+
str
(
j
)
filesOut
+=
[
self
.
HFEngine
.
outParaviewTimeDomain
(
u
,
**
outargs
)]
if
"name"
in
outargs
.
keys
():
outargs
[
"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,Approx,Err} methods
"""
Compute norm of * at arbitrary parameter.
Args:
mu: Target parameter.
Returns:
Target norm of *.
"""
for
objName
in
[
"HF"
,
"Approx"
,
"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"
))
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
:
muExtra
=
self
.
checkParameterList
(
muExtra
)
vbMng
(
self
,
"INIT"
,
"Solving HF model for mu = {}."
.
format
(
muExtra
),
15
)
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
getPoles
(
self
,
*
args
,
**
kwargs
)
->
paramList
:
"""
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
compress
(
self
,
*
args
,
**
kwargs
):
"""Compress trained reduced model."""
return
self
.
trainedModel
.
compress
(
*
args
,
**
kwargs
)
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
)
Event Timeline
Log In to Comment