Page Menu
Home
c4science
Search
Configure Global Search
Log In
Files
F88635861
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
Sat, Oct 19, 21:35
Size
33 KB
Mime Type
text/x-python
Expires
Mon, Oct 21, 21:35 (1 d, 23 h)
Engine
blob
Format
Raw Data
Handle
21763967
Attached To
R6746 RationalROMPy
generic_approximant.py
View Options
# Copyright (C) 2018 by the RROMPy authors
#
# This file is part of RROMPy.
#
# RROMPy is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# RROMPy is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with RROMPy. If not, see <http://www.gnu.org/licenses/>.
#
from
abc
import
abstractmethod
import
numpy
as
np
from
itertools
import
product
as
iterprod
from
copy
import
deepcopy
as
copy
from
os
import
remove
as
osrm
from
rrompy.sampling.standard
import
(
SamplingEngineStandard
,
SamplingEngineStandardPOD
)
from
rrompy.utilities.base.types
import
(
Np1D
,
DictAny
,
HFEng
,
List
,
Tuple
,
ListAny
,
strLst
,
paramVal
,
paramList
,
sampList
)
from
rrompy.utilities.base
import
(
purgeDict
,
verbosityManager
as
vbMng
,
getNewFilename
)
from
rrompy.utilities.exception_manager
import
(
RROMPyException
,
RROMPyAssert
,
RROMPy_READY
,
RROMPy_FRAGILE
)
from
rrompy.utilities.base
import
pickleDump
,
pickleLoad
from
rrompy.parameter
import
(
emptyParameterList
,
checkParameter
,
checkParameterList
)
from
rrompy.sampling
import
sampleList
,
emptySampleList
__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
)
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
.
plot
(
u
,
*
args
,
**
kwargs
)]
if
"name"
in
kwargs
.
keys
():
kwargs
[
"name"
]
=
nameBase
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
)
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
.
plot
(
u
,
*
args
,
**
kwargs
)]
if
"name"
in
kwargs
.
keys
():
kwargs
[
"name"
]
=
nameBase
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
)
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
.
outParaview
(
u
,
*
args
,
**
kwargs
)]
if
"name"
in
kwargs
.
keys
():
kwargs
[
"name"
]
=
nameBase
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
)
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
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': whether to compute POD of snapshots; defaults to True;
- 'S': total number of samples current approximant relies upon.
Defaults to empty dict.
approx_state(optional): Whether to approximate state. Defaults to
False.
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': whether to compute POD of snapshots.
parameterListCritical: Recognized keys of critical approximant
parameters:
- 'S': total number of samples current approximant relies upon.
approx_state: Whether to approximate state.
verbosity: Verbosity level.
POD: Whether to compute POD of snapshots.
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
=
{},
approx_state
:
bool
=
False
,
verbosity
:
int
=
10
,
timestamp
:
bool
=
True
):
self
.
_preInit
()
self
.
_mode
=
RROMPy_READY
self
.
approx_state
=
approx_state
self
.
verbosity
=
verbosity
self
.
timestamp
=
timestamp
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"
],
[
True
],
[
"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
import
TrainedModelData
return
(
TrainedModelData
(
datadict
[
"mu0"
],
datadict
.
pop
(
"projMat"
),
datadict
[
"scaleFactor"
],
datadict
.
pop
(
"rescalingExp"
)),
[
"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
):
"""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
:
SamplingEngine
=
SamplingEngineStandardPOD
else
:
SamplingEngine
=
SamplingEngineStandard
self
.
samplingEngine
=
SamplingEngine
(
self
.
HFEngine
,
sample_state
=
self
.
approx_state
,
verbosity
=
self
.
verbosity
)
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
]
@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
:
val
=
self
.
parameterDefaultCritical
[
key
]
getattr
(
self
.
__class__
,
key
,
None
)
.
fset
(
self
,
val
)
fragile
=
fragile
or
val
is
None
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
]
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
self
.
_POD
=
POD
self
.
_approxParameters
[
"POD"
]
=
self
.
POD
if
PODold
!=
self
.
POD
:
self
.
samplingEngine
=
None
self
.
resetSamples
()
@property
def
approx_state
(
self
):
"""Value of approx_state."""
return
self
.
_approx_state
@approx_state.setter
def
approx_state
(
self
,
approx_state
):
if
hasattr
(
self
,
"_approx_state"
):
approx_stateold
=
self
.
approx_state
else
:
approx_stateold
=
-
1
self
.
_approx_state
=
approx_state
if
approx_stateold
!=
self
.
approx_state
:
self
.
samplingEngine
=
None
self
.
resetSamples
()
@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
setSamples
(
self
,
samplingEngine
):
"""Copy samplingEngine and samples."""
vbMng
(
self
,
"INIT"
,
"Transfering samples."
,
10
)
self
.
samplingEngine
=
copy
(
samplingEngine
)
vbMng
(
self
,
"DEL"
,
"Done transfering samples."
,
10
)
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."
))
self
.
loadTrainedModel
(
fileOut
)
osrm
(
fileOut
)
@abstractmethod
def
setupApprox
(
self
)
->
int
:
"""
Setup approximant. (ABSTRACT)
Any specialization should include something like
if self.checkComputedApprox():
return
RROMPyAssert(self._mode, message = "Cannot setup approximant.")
...
self.trainedModel = ...
self.trainedModel.data = ...
self.trainedModel.data.approxParameters = copy(
self.approxParameters)
Returns > 0 if error was encountered, < 0 if no computation was
necessary.
"""
pass
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
)
def
_pruneBeforeEval
(
self
,
mu
:
paramList
,
field
:
str
,
append
:
bool
,
prune
:
bool
)
->
Tuple
[
paramList
,
Np1D
,
Np1D
,
bool
]:
mu
=
checkParameterList
(
mu
,
self
.
npar
)[
0
]
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
=
checkParameterList
(
mu
,
self
.
npar
)[
0
]
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
)
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
=
True
)
->
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
=
True
)
->
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
,
mu
:
paramList
,
append
:
bool
=
False
,
prune
:
bool
=
True
)
->
sampList
:
"""
Get HF solution at arbitrary parameter.
Args:
mu: Target parameter.
Returns:
HFsolution.
"""
mu
=
checkParameterList
(
mu
,
self
.
npar
)[
0
]
idx
=
self
.
evalHF
(
mu
,
append
=
append
,
prune
=
prune
)
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
,
mu
:
paramList
,
append
:
bool
=
False
,
prune
:
bool
=
True
)
->
sampList
:
"""
Get approximant at arbitrary parameter.
Args:
mu: Target parameter.
Returns:
Reduced approximant.
"""
mu
=
checkParameterList
(
mu
,
self
.
npar
)[
0
]
idx
=
self
.
evalApproxReduced
(
mu
,
append
=
append
,
prune
=
prune
)
return
self
.
uApproxReduced
(
idx
)
def
getApprox
(
self
,
mu
:
paramList
,
append
:
bool
=
False
,
prune
:
bool
=
True
)
->
sampList
:
"""
Get approximant at arbitrary parameter.
Args:
mu: Target parameter.
Returns:
Approximant.
"""
mu
=
checkParameterList
(
mu
,
self
.
npar
)[
0
]
idx
=
self
.
evalApprox
(
mu
,
append
=
append
,
prune
=
prune
)
return
self
.
uApprox
(
idx
)
def
getRes
(
self
,
mu
:
paramList
)
->
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
)
/
self
.
HFEngine
.
C
)
def
getErr
(
self
,
mu
:
paramList
,
append
:
bool
=
False
,
prune
:
bool
=
True
)
->
sampList
:
"""
Get error at arbitrary parameter.
Args:
mu: Target parameter.
Returns:
Approximant error.
"""
return
(
self
.
getApprox
(
mu
,
append
=
append
,
prune
=
prune
)
-
self
.
getHF
(
mu
,
append
=
append
,
prune
=
prune
))
def
normApprox
(
self
,
mu
:
paramList
)
->
float
:
"""
Compute norm of approximant at arbitrary parameter.
Args:
mu: Target parameter.
Returns:
Target norm of approximant.
"""
if
not
(
self
.
POD
and
self
.
HFEngine
.
isCEye
):
return
self
.
HFEngine
.
norm
(
self
.
getApprox
(
mu
),
is_state
=
False
)
return
np
.
linalg
.
norm
(
self
.
HFEngine
.
C
*
self
.
getApproxReduced
(
mu
)
.
data
,
axis
=
0
)
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
storeTrainedModel
(
self
,
filenameBase
:
str
=
"trained_model"
,
forceNewFile
:
bool
=
True
)
->
str
:
"""Store trained reduced model to file."""
self
.
setupApprox
()
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
)
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"
]
trainedModel
=
self
.
tModelType
()
trainedModel
.
verbosity
=
self
.
verbosity
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
])
trainedModel
.
data
=
data
self
.
trainedModel
=
trainedModel
self
.
_mode
=
RROMPy_FRAGILE
vbMng
(
self
,
"DEL"
,
"Done loading pre-trained model."
,
20
)
Event Timeline
Log In to Comment