Homec4science
Diffusion RationalROMPy (feature_greedy_coarsening) Browse

RROMPy/rrompy/sampling/linear_problem/sampling_engine_distributed.py
f972e1cdd3b2feature_greedy_coarseni…

Last Change
Raw File

sampling_engine_distributed.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 rrompy.sampling.base.sampling_engine_base import SamplingEngineBase
from rrompy.utilities.base.types import Np1D, Np2D
from rrompy.utilities.base import verbosityDepth
from rrompy.utilities.exception_manager import RROMPyException
__all__ = ['SamplingEngineDistributed']
class SamplingEngineDistributed(SamplingEngineBase):
"""HERE"""
nameBase = 1
def preprocesssamples(self, idxs:Np1D):
if self.samples is None: return
return self.samples[:, idxs]
def postprocessu(self, u:Np1D, overwrite : bool = False):
return u
def _getSampleConcurrence(self, mu:complex, previous:Np1D,
homogeneized : bool = False) -> Np1D:
samplesOld = self.preprocesssamples(previous)
RHS = self.HFEngine.b(mu, len(previous), homogeneized = homogeneized)
for i in range(1, len(previous) + 1):
RHS -= self.HFEngine.A(mu, i).dot(samplesOld[:, - i])
return self.solveLS(mu, RHS = RHS, homogeneized = homogeneized)
def nextSample(self, mu:complex, overwrite : bool = False,
homogeneized : bool = False) -> Np1D:
ns = self.nsamples
muidxs = np.nonzero(self.mus[:ns] == mu)[0]
if len(muidxs) > 0:
u = self._getSampleConcurrence(mu, np.sort(muidxs), homogeneized)
else:
u = self.solveLS(mu, homogeneized = homogeneized)
u = self.postprocessu(u, overwrite = overwrite)
if overwrite:
self.samples[:, ns] = u
self.mus[ns] = mu
else:
if ns == 0:
self.samples = u[:, None]
else:
self.samples = np.hstack((self.samples, u[:, None]))
self.mus = self.mus + [mu]
self.nsamples += 1
return u
def iterSample(self, mus:Np1D, homogeneized : bool = False) -> Np2D:
if self.verbosity >= 5:
verbosityDepth("INIT", "Starting sampling iterations.",
timestamp = self.timestamp)
n = mus.size
if n <= 0:
raise RROMPyException(("Number of samples must be positive."))
self.resetHistory()
if self.verbosity >= 7:
verbosityDepth("MAIN", "Computing sample {}/{}.".format(1, n),
timestamp = self.timestamp)
u = self.nextSample(mus[0], homogeneized = homogeneized)
if n > 1:
self.preallocateSamples(u, mus[0], n)
for j in range(1, n):
if self.verbosity >= 7:
verbosityDepth("MAIN",
"Computing sample {}/{}.".format(j + 1, n),
timestamp = self.timestamp)
self.nextSample(mus[j], overwrite = True,
homogeneized = homogeneized)
if self.verbosity >= 5:
verbosityDepth("DEL", "Finished sampling iterations.",
timestamp = self.timestamp)
return self.samples
c4science · Help