Page MenuHomec4science
Files F61003459

sampling_engine_pivoted_pod.py
No OneTemporary
Actions

File Metadata

Created
Fri, May 3, 22:24

sampling_engine_pivoted_pod.py
View Options

# Copyright (C) 2018 by the RROMPy authors
#
# This file is part of RROMPy.
#
# RROMPy is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# RROMPy is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with RROMPy. If not, see <http://www.gnu.org/licenses/>.
#
import numpy as np
from copy import deepcopy as copy
from rrompy.sampling.base.pod_engine import PODEngine
from .sampling_engine_pivoted import SamplingEnginePivoted
from rrompy.utilities.base.types import Np1D, paramVal, sampList
from rrompy.utilities.base import verbosityManager as vbMng
from rrompy.sampling import sampleList, emptySampleList
__all__ = ['SamplingEnginePivotedPOD']
class SamplingEnginePivotedPOD(SamplingEnginePivoted):
"""HERE"""
def resetHistory(self, j : int = 1):
super().resetHistory(j)
self.samples_full = [None] * j
self.RPOD = [None] * j
def popSample(self, j:int):
if hasattr(self, "nsamples") and self.nsamples[j] > 1:
self.RPOD[j] = self.RPOD[j][: -1, : -1]
self.samples_full[j].pop()
super().popSample(j)
def coalesceSamples(self, tol : float = 1e-12):
super().coalesceSamples()
self.samplesCoalesced, RPODC = (
self.PODEngine.generalizedQR(self.samplesCoalesced))
self.RPODCoalesced = np.zeros((self.samplesCoalesced.shape[1],
self.samplesCoalesced.shape[1]),
dtype = self.RPOD[0].dtype)
self.samples_fullCoalesced = emptySampleList()
self.samples_fullCoalesced.reset((self.samples_full[0].shape[0],
self.samplesCoalesced.shape[1]),
self.samples_full[0].dtype)
ci = 0
for j, (Rloc, samp) in enumerate(zip(self.RPOD, self.samples_full)):
ri = 0
Rheg = Rloc.shape[1]
for k, Rloc2 in enumerate(self.RPOD[: j + 1]):
Rlen = Rloc2.shape[1]
self.RPODCoalesced[ri : ri + Rlen, ci : ci + Rheg] = (
RPODC[ri : ri + Rlen, ci : ci + Rheg].dot(Rloc))
ri += Rlen
self.samples_fullCoalesced.data[:, ci : ci + Rheg] = samp.data
ci += Rheg
RCdiag = np.abs(np.diag(self.RPODCoalesced))
RCdiag /= RCdiag[0]
ntrunc = np.nonzero(RCdiag < tol)[0]
if len(ntrunc) == 0: return
self.samplesCoalesced.data = self.samplesCoalesced.data[:, : ntrunc[0]]
self.RPODCoalesced = self.RPODCoalesced[: ntrunc[0], :]
@property
def HFEngine(self):
"""Value of HFEngine. Its assignment resets history."""
return self._HFEngine
@HFEngine.setter
def HFEngine(self, HFEngine):
self._HFEngine = HFEngine
self.resetHistory()
self.PODEngine = PODEngine(self._HFEngine)
def preprocesssamples(self, idxs:Np1D, j:int) -> sampList:
if self.samples_full[j] is None or len(self.samples_full[j]) == 0:
return
return self.samples_full[j](idxs)
def postprocessu(self, u:sampList, j:int,
overwrite : bool = False) -> Np1D:
ns = self.nsamples[j]
if overwrite:
self.samples_full[j][ns] = copy(u)
else:
if ns == 0:
self.samples_full[j] = sampleList(u)
else:
self.samples_full[j].append(u)
return u
def postprocessuBulk(self, u:sampList, j:int) -> sampList:
self.samples_full[j] = copy(u)
vbMng(self, "INIT",
"Starting orthogonalization for marginal {}.".format(j), 40)
u, self.RPOD[j] = self.PODEngine.generalizedQR(self.samples_full[j])
vbMng(self, "DEL", "Done orthogonalizing.", 40)
return u
def lastSampleManagement(self, j:int):
self.samples[j] = self.postprocessuBulk(self.samples_full[j], j)
def preallocateSamples(self, u:Np1D, mu:paramVal, n:int, j:int):
super().preallocateSamples(u, mu, n, j)
self.samples_full[j].reset((u.shape[0], n), u.dtype)
self.samples_full[j][0] = u

Event Timeline

c4science · Help