Homec4science
Diffusion RationalROMPy (master) Browse

RROMPy/rrompy/sampling/engines/sampling_engine_normalize.py
5c2ab7641888master

Last Change
Raw File

sampling_engine_normalize.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 .pod_engine import PODEngine
from .sampling_engine import SamplingEngine
from rrompy.utilities.base.types import (Np1D, Np2D, TupleAny, DictAny, Any,
paramVal, sampList)
from rrompy.utilities.base import verbosityManager as vbMng
from rrompy.sampling import sampleList, emptySampleList
__all__ = ['SamplingEngineNormalize']
class SamplingEngineNormalize(SamplingEngine):
@property
def HFEngine(self):
"""Value of HFEngine. Its assignment resets history."""
return self._HFEngine
@HFEngine.setter
def HFEngine(self, HFEngine):
SamplingEngine.HFEngine.fset(self, HFEngine)
self.PODEngine = PODEngine(self._HFEngine)
@property
def feature_keys(self) -> TupleAny:
return super().feature_keys + ["samples_normal", "Rscale"]
@property
def feature_vals(self) -> DictAny:
vals = super().feature_vals
vals["samples_normal"] = self.samples_normal
vals["Rscale"] = self.Rscale
return vals
def _mergeFeature(self, name:str, value:Any):
if name == "samples_normal":
self.samples_normal.append(value)
elif name == "Rscale":
self.Rscale = np.append(self.Rscale, value)
else:
super()._mergeFeature(name, value)
@property
def projectionMatrix(self) -> Np2D:
return self.samples_normal.data
def resetHistory(self):
super().resetHistory()
self.samples_normal = emptySampleList()
self.Rscale = np.zeros(0, dtype = np.complex)
def setsample_normal(self, u:sampList, overwrite : bool = False):
if overwrite:
self.samples_normal[self.nsamples] = u
else:
if self.nsamples == 0:
self.samples_normal = sampleList(u)
else:
self.samples_normal.append(u)
def popSample(self):
if hasattr(self, "nsamples") and self.nsamples > 1:
self.Rscale = self.Rscale[: -1]
self.samples_normal.pop()
super().popSample()
def preallocateSamples(self, u:Np1D, mu:paramVal, n:int):
super().preallocateSamples(u, mu, n)
self.samples_normal.reset((u.shape[0], n), u.dtype)
def postprocessu(self, u:sampList, overwrite : bool = False):
"""Postprocess by normalizing snapshot."""
self.setsample(u, overwrite)
vbMng(self, "INIT", "Starting normalization.", 20)
u, r = self.PODEngine.normalize(u, is_state = self.sample_state)
self.Rscale = np.append(self.Rscale, r)
vbMng(self, "DEL", "Done normalizing.", 20)
self.setsample_normal(u, overwrite)
def postprocessuBulk(self):
"""Postprocess by normalizing snapshots in bulk."""
vbMng(self, "INIT", "Starting normalization.", 10)
samples_normal, self.Rscale = self.PODEngine.normalize(self.samples,
is_state = self.sample_state)
vbMng(self, "DEL", "Done normalizing.", 10)
nsamples, self.nsamples = self.nsamples, 0
self.setsample_normal(samples_normal)
self.nsamples = nsamples
c4science · Help