Source code for grond.targets.base

import copy

import numpy as num

from pyrocko import gf
from pyrocko.guts_array import Array
from pyrocko.guts import Object, Float, String, Dict, List, Choice, load, dump

from grond.analysers.base import AnalyserResult
from grond.meta import has_get_plot_classes, GrondError

guts_prefix = 'grond'

[docs]class TargetGroup(Object): normalisation_family = gf.StringID.T( optional=True, help='Group with common misfit normalisation') path = gf.StringID.T( optional=True, help=' will be prefixed with this path') weight = Float.T( default=1.0, help='Additional manual weight of the target group') interpolation = gf.InterpolationMethod.T( default='nearest_neighbor', help='Interpolation from pre-calculated GF store.') store_id = gf.StringID.T( optional=True, help='ID of the Green\'s function store for this TargetGroup.') def get_targets(self, ds, event, default_path='none'): if not self._targets: raise NotImplementedError()
[docs]class MisfitResult(Object): misfits = Array.T( shape=(None, 2), dtype=num.float64)
[docs]class MisfitConfig(Object): pass
[docs]@has_get_plot_classes class MisfitTarget(Object): manual_weight = Float.T( default=1.0, help='Relative weight of this target') analyser_results = Dict.T( gf.StringID.T(), AnalyserResult.T(), help='Dictionary of analyser results') normalisation_family = gf.StringID.T( optional=True, help='Normalisation family of this misfit target') path = gf.StringID.T( help='A path identifier used for plotting') misfit_config = MisfitConfig.T( default=MisfitConfig.D(), help='Misfit configuration') bootstrap_weights = Array.T( dtype=num.float64, serialize_as='base64', optional=True) bootstrap_residuals = Array.T( dtype=num.float64, serialize_as='base64', optional=True) can_bootstrap_weights = False can_bootstrap_residuals = False plot_misfits_cumulative = True def __init__(self, **kwargs): Object.__init__(self, **kwargs) self.parameters = [] self._ds = None self._result_mode = 'sparse' self._combined_weight = None self._target_parameters = None self._target_ranges = None self._combined_weight = None @classmethod def get_plot_classes(cls): return [] def set_dataset(self, ds): self._ds = ds def get_dataset(self): return self._ds def string_id(self): return str(self.path) def misfits_string_ids(self): raise NotImplementedError('%s does not implement misfits_string_id' % self.__class__.__name__) @property def nmisfits(self): return 1 def noise_weight_matrix(self): return num.array([[1]]) @property def nparameters(self): if self._target_parameters is None: return 0 return len(self._target_parameters) @property def target_parameters(self): if self._target_parameters is None: self._target_parameters = copy.deepcopy(self.parameters) for p in self._target_parameters: p.set_groups([self.string_id()]) return self._target_parameters @property def target_ranges(self): return {} def set_parameter_values(self, model): for i, p in enumerate(self.parameters): self.parameter_values[p.name_nogroups] = model[i] def set_result_mode(self, result_mode): self._result_mode = result_mode def post_process(self, engine, source, statics): raise NotImplementedError() def get_combined_weight(self): if self._combined_weight is None: w = self.manual_weight for analyser in self.analyser_results.values(): w *= analyser.weight self._combined_weight = num.array([w], dtype=float) return self._combined_weight def get_correlated_weights(self, nthreads=0): pass def set_bootstrap_weights(self, weights): self.bootstrap_weights = weights def get_bootstrap_weights(self): if self.bootstrap_weights is None: raise Exception('Bootstrap weights have not been set!') nbootstraps = self.bootstrap_weights.size // self.nmisfits return self.bootstrap_weights.reshape(nbootstraps, self.nmisfits) def init_bootstrap_residuals(self, nbootstrap, rstate=None, nthreads=0): raise NotImplementedError() def set_bootstrap_residuals(self, residuals): self.bootstrap_residuals = residuals def get_bootstrap_residuals(self): if self.bootstrap_residuals is None: raise Exception('Bootstrap residuals have not been set!') nbootstraps = self.bootstrap_residuals.size // self.nmisfits return self.bootstrap_residuals.reshape(nbootstraps, self.nmisfits)
[docs] def prepare_modelling(self, engine, source, targets): ''' Prepare modelling target This function shall return a list of :class:`` for forward modelling in the :class:``. ''' return [self]
[docs] def finalize_modelling( self, engine, source, modelling_targets, modelling_results): ''' Manipulate modelling before misfit calculation This function can be overloaded interact with the modelling results. ''' return modelling_results[0]
class MisfitResultError(Object): message = String.T() class MisfitResultCollection(Object): results = List.T(List.T( Choice.T([MisfitResult.T(), MisfitResultError.T()]))) def dump_misfit_result_collection(misfit_result_collection, path): dump(misfit_result_collection, filename=path) def load_misfit_result_collection(path): try: obj = load(filename=path) except OSError as e: raise GrondError( 'Failed to read ensemble misfit results from file "%s" (%s)' % ( path, e)) if not isinstance(obj, MisfitResultCollection): raise GrondError( 'File "%s" does not contain any misfit result collection.' % path) return obj __all__ = ''' TargetGroup MisfitTarget MisfitResult MisfitResultError dump_misfit_result_collection load_misfit_result_collection MisfitResultCollection '''.split()