Page Menu
Home
c4science
Search
Configure Global Search
Log In
Files
F84066966
sampling_engine_pivoted.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, Sep 20, 13:18
Size
5 KB
Mime Type
text/x-python
Expires
Sun, Sep 22, 13:18 (1 d, 23 h)
Engine
blob
Format
Raw Data
Handle
20887542
Attached To
R6746 RationalROMPy
sampling_engine_pivoted.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
copy
import
deepcopy
as
copy
import
numpy
as
np
from
rrompy.sampling.base.sampling_engine_base_pivoted
import
(
SamplingEngineBasePivoted
)
from
rrompy.hfengines.base
import
MarginalProxyEngine
from
rrompy.utilities.base.types
import
Np1D
,
paramVal
,
paramList
,
sampList
from
rrompy.utilities.base
import
verbosityManager
as
vbMng
,
freepar
as
fp
from
rrompy.utilities.exception_manager
import
RROMPyException
from
rrompy.utilities.numerical
import
dot
from
rrompy.utilities.numerical.hash_derivative
import
nextDerivativeIndices
from
rrompy.parameter
import
checkParameter
,
checkParameterList
__all__
=
[
'SamplingEnginePivoted'
]
class
SamplingEnginePivoted
(
SamplingEngineBasePivoted
):
def
preprocesssamples
(
self
,
idxs
:
Np1D
,
j
:
int
)
->
sampList
:
if
self
.
samples
[
j
]
is
None
or
len
(
self
.
samples
[
j
])
==
0
:
return
return
self
.
samples
[
j
](
idxs
)
def
postprocessu
(
self
,
u
:
sampList
,
j
:
int
,
overwrite
:
bool
=
False
):
self
.
setsample
(
u
,
j
,
overwrite
)
def
postprocessuBulk
(
self
,
j
:
int
):
pass
def
_getSampleConcurrence
(
self
,
mu
:
paramVal
,
j
:
int
,
previous
:
Np1D
)
->
sampList
:
if
not
(
self
.
sample_state
or
self
.
HFEngine
.
isCEye
):
raise
RROMPyException
((
"Derivatives of solution with non-scalar "
"C not computable."
))
if
len
(
previous
)
>=
len
(
self
.
_derIdxs
[
j
]):
self
.
_derIdxs
[
j
]
+=
nextDerivativeIndices
(
self
.
_derIdxs
[
j
],
self
.
nPivot
,
len
(
previous
)
+
1
-
len
(
self
.
_derIdxs
[
j
]))
derIdx
=
self
.
_derIdxs
[
j
][
len
(
previous
)]
mu
=
checkParameter
(
mu
,
self
.
nPivot
)
samplesOld
=
self
.
preprocesssamples
(
previous
,
j
)
RHS
=
self
.
scaleDer
(
derIdx
)
*
self
.
HFEngineMarginalized
.
b
(
mu
,
derIdx
)
for
j
,
derP
in
enumerate
(
self
.
_derIdxs
[
j
][:
len
(
previous
)]):
diffP
=
[
x
-
y
for
(
x
,
y
)
in
zip
(
derIdx
,
derP
)]
if
np
.
all
([
x
>=
0
for
x
in
diffP
]):
RHS
-=
self
.
scaleDer
(
diffP
)
*
dot
(
self
.
HFEngineMarginalized
.
A
(
mu
,
diffP
),
samplesOld
[
j
])
return
self
.
solveLS
(
mu
,
RHS
=
RHS
)
def
nextSample
(
self
,
mu
:
paramVal
,
j
:
int
,
overwrite
:
bool
=
False
,
postprocess
:
bool
=
True
)
->
Np1D
:
mu
=
checkParameter
(
mu
,
self
.
nPivot
)
muidxs
=
self
.
mus
[
j
]
.
findall
(
mu
[
0
])
if
len
(
muidxs
)
>
0
:
u
=
self
.
_getSampleConcurrence
(
mu
,
j
,
np
.
sort
(
muidxs
))
else
:
u
=
self
.
solveLS
(
mu
)
if
postprocess
:
self
.
postprocessu
(
u
,
j
,
overwrite
=
overwrite
)
else
:
self
.
setsample
(
u
,
j
,
overwrite
)
if
overwrite
:
self
.
mus
[
j
][
self
.
nsamples
[
j
]]
=
mu
[
0
]
else
:
self
.
mus
[
j
]
.
append
(
mu
)
self
.
nsamples
[
j
]
+=
1
return
self
.
samples
[
j
][
self
.
nsamples
[
j
]
-
1
]
def
iterSample
(
self
,
mus
:
paramList
,
musM
:
paramList
)
->
sampList
:
mus
=
checkParameterList
(
mus
,
self
.
nPivot
)[
0
]
musM
=
checkParameterList
(
musM
,
self
.
nMarginal
)[
0
]
vbMng
(
self
,
"INIT"
,
"Starting sampling iterations."
,
5
)
n
=
len
(
mus
)
m
=
len
(
musM
)
if
n
<=
0
:
raise
RROMPyException
(
"Number of samples must be positive."
)
if
m
<=
0
:
raise
RROMPyException
((
"Number of marginal samples must be "
"positive."
))
repeatedSamples
=
len
(
mus
.
unique
())
!=
n
for
j
in
range
(
m
):
muMEff
=
[
fp
]
*
self
.
HFEngine
.
npar
for
k
,
x
in
enumerate
(
self
.
directionMarginal
):
muMEff
[
x
]
=
musM
(
j
,
k
)
self
.
HFEngineMarginalized
=
MarginalProxyEngine
(
self
.
HFEngine
,
list
(
muMEff
))
if
repeatedSamples
:
for
k
in
range
(
n
):
vbMng
(
self
,
"MAIN"
,
"Computing sample {} / {} for marginal {} / {}."
\
.
format
(
k
+
1
,
n
,
j
,
m
),
10
)
self
.
nextSample
(
mus
[
k
],
j
,
overwrite
=
(
k
>
0
),
postprocess
=
False
)
if
n
>
1
and
k
==
0
:
self
.
preallocateSamples
(
self
.
samples
[
j
][
0
],
mus
[
0
],
n
,
j
)
else
:
self
.
setsample
(
self
.
solveLS
(
mus
),
j
,
overwrite
=
False
)
self
.
mus
[
j
]
=
copy
(
mus
)
self
.
nsamples
[
j
]
=
n
if
len
(
self
.
musMarginal
)
>
j
:
self
.
musMarginal
[
j
]
=
copy
(
musM
[
j
])
else
:
self
.
musMarginal
.
append
(
musM
[
j
])
self
.
postprocessuBulk
(
j
)
vbMng
(
self
,
"DEL"
,
"Finished sampling iterations."
,
5
)
return
self
.
samples
[
j
]
Event Timeline
Log In to Comment