Source code for stride.optimisation.pipelines.steps.norm_per_shot


import numpy as np

from ....core import Operator


[docs]class NormPerShot(Operator): """ Normalised a series of time traces to the maximum value of the set. Parameters ---------- amplitude : bool, optional Whether to keep the true amplitude of the modelled after normalisation. """ def __init__(self, **kwargs): super().__init__(**kwargs) self.amplitude = kwargs.pop('amplitude', False) self._num_traces = None
[docs] def forward(self, *traces, **kwargs): self._num_traces = len(traces) amplitude = kwargs.pop('amplitude', self.amplitude) if amplitude: scaling = self._norm(traces[0], **kwargs) else: scaling = 1. normed = tuple([self._apply(each, scaling=scaling, **kwargs) for each in traces]) if len(normed) == 1: normed = normed[0] return normed
[docs] def adjoint(self, *d_traces, **kwargs): d_traces = d_traces[:self._num_traces] if len(d_traces) == 1: d_traces = d_traces[0] return d_traces
def _norm(self, traces, **kwargs): norm_value = 0. for index in range(traces.extended_shape[0]): norm_value += np.sum(traces.extended_data[index] ** 2) norm_value = np.sqrt(norm_value / traces.extended_shape[0]) + 1e-31 return norm_value def _apply(self, traces, scaling, **kwargs): norm_value = self._norm(traces, **kwargs) out_traces = traces.alike(name='normed_%s' % traces.name) out_traces.extended_data[:] = scaling * traces.extended_data / norm_value return out_traces