saftig.filtering ================ .. py:module:: saftig.filtering .. autoapi-nested-parse:: Implementations of filtering techniques with a common interface. Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/saftig/filtering/common/index /autoapi/saftig/filtering/lms/index /autoapi/saftig/filtering/polylms/index /autoapi/saftig/filtering/uwf/index /autoapi/saftig/filtering/wf/index Classes ------- .. autoapisummary:: saftig.filtering.FilterBase saftig.filtering.WienerFilter saftig.filtering.UpdatingWienerFilter saftig.filtering.LMSFilter saftig.filtering.PolynomialLMSFilter Package Contents ---------------- .. py:class:: FilterBase(n_filter, idx_target, n_channel = 1, _from_dict=None) Bases: :py:obj:`abc.ABC` common interface definition for Filter implementations :param n_filter: Length of the FIR filter (how many samples are in the input window per output sample) :param idx_target: Position of the prediction :param n_channel: Number of witness sensor channels .. py:attribute:: requires_apply_target :type: bool .. py:attribute:: n_filter :type: int .. py:attribute:: n_channel :type: int .. py:attribute:: idx_target :type: int .. py:attribute:: method_hash_value :type: bytes .. py:attribute:: supports_multi_sequence :value: True .. py:attribute:: filter_name :value: 'FilterBase' .. py:method:: supports_saving_loading() :staticmethod: Indicates whether saving and loading is supported Due to the way dataclasses work with inheritance, class values with default values don't work in the parent dataclass. Thus, this is a function .. py:method:: condition(witness, target) Use an input dataset to condition the filter :param witness: Witness sensor data :param target: Target sensor data .. py:method:: condition_multi_sequence(witness, target) :abstractmethod: Similar to condition(), but expects multiple sequences .. py:method:: apply(witness, target = None, pad = True, update_state = False) Apply the filter to input data :param witness: Witness sensor data (1D or 2D array) :param target: Target sensor data (1D array) :param pad: if True, apply padding zeros so that the length matches the target signal :param update_state: if True, the filter state will be changed. If false, the filter state will remain :return: prediction .. py:method:: apply_multi_sequence(witness, target, pad = True, update_state = False) :abstractmethod: Apply the filter to input data Similar to apply() but expects multiple sequences. .. py:method:: check_data_dimensions(witness, target = None) Check the dimensions of the provided input data and apply make_2d_array() :param witness: Witness sensor data :param target: Target sensor data :return: data as (target, witness) :raises: AssertionError .. py:method:: check_data_dimensions_multi_sequence(witness: collections.abc.Sequence | numpy.typing.NDArray, target: None) -> tuple[list[numpy.typing.NDArray], None] check_data_dimensions_multi_sequence(witness: collections.abc.Sequence | numpy.typing.NDArray, target: collections.abc.Sequence | numpy.typing.NDArray) -> tuple[list[numpy.typing.NDArray], list[numpy.typing.NDArray]] Check the dimensions of the provided input data and apply make_2d_array() :param witness: Witness sensor data :param target: Target sensor data :return: data as (target, witness) :raises: AssertionError .. py:method:: as_dict() Returns a dictionary that represents the state of this filter. .. py:method:: from_dict(input_dict) :classmethod: Create a filter instance from a dictionary that was created from as_dict() .. py:method:: make_filename(filename) :classmethod: Append the file type of save files for this class to the given filename, if it is not already present .. py:method:: save(filename, warn_incompatible = False) Save the filter state as a numpy file The given filename will be autocompleted with a "..npz" filename extension, unless a matching extension is detected. warn_incompatible: set to True to warn for object types might not compatible with np.save(allow_pickle=False) during development .. py:method:: load(filename) :classmethod: Load a filter state from the supplied filename. The given filename will be autocompleted with a "..npz" filename extension, unless a matching extension is detected. .. py:method:: _file_hash() :classmethod: Calculates a hash value based on the file in which this method was defined. .. py:property:: method_hash :type: bytes A hash of the method and parameters NOTE: This is not a hash of the conditioned filter! Thus, the same filter configuration applied to a different dataset will result in the same hash! .. py:property:: method_filename_part :type: str string that can be used in a file name .. py:class:: WienerFilter(n_filter, idx_target, n_channel = 1) Bases: :py:obj:`saftig.filtering.common.FilterBase` Satic Wiener filter implementation :param n_filter: Length of the FIR filter (how many samples are in the input window per output sample) :param idx_target: Position of the prediction :param n_channel: Number of witness sensor channels >>> import saftig as sg >>> n_filter = 128 >>> witness, target = sg.evaluation.TestDataGenerator(0.1).generate(int(1e5)) >>> filt = sg.filtering.WienerFilter(n_filter, 0, 1) >>> _coefficients, full_rank = filt.condition(witness, target) >>> full_rank True >>> prediction = filt.apply(witness, target) # check on the data used for conditioning >>> residual_rms = sg.evaluation.rms(target-prediction) >>> residual_rms > 0.05 and residual_rms < 0.15 # the expected RMS in this test scenario is 0.1 True .. py:attribute:: filter_state :type: numpy.typing.NDArray | None :value: None .. py:attribute:: filter_name :type: str :value: 'WF' .. py:attribute:: requires_apply_target :value: False .. py:method:: condition_multi_sequence(witness, target) Use an input dataset to condition the filter :param witness: Witness sensor data :param target: Target sensor data .. py:method:: apply_multi_sequence(witness, target = None, pad = True, update_state = False) Apply the filter to input data :param witness: Witness sensor data :param target: Target sensor data (is ignored) :param pad: if True, apply padding zeros so that the length matches the target signal :param update_state: ignored :return: prediction .. py:class:: UpdatingWienerFilter(n_filter, idx_target, n_channel = 1, context_pre = 0, context_post = 0) Bases: :py:obj:`saftig.filtering.common.FilterBase` Updating Wiener filter implementation :param n_filter: Length of the FIR filter (how many samples are in the input window per output sample) :param idx_target: Position of the prediction :param n_channel: Number of witness sensor channels :param context_pre: how many additional samples before the current block are used to update the filters :param context_post: how many additional samples after the current block are used to update the filters >>> import saftig as sg >>> n_filter = 128 >>> witness, target = sg.evaluation.TestDataGenerator(0.1).generate(int(1e5)) >>> filt = sg.filtering.UpdatingWienerFilter(n_filter, 0, 1, context_pre=20*n_filter, context_post=20*n_filter) >>> prediction = filt.apply(witness, target) # check on the data used for conditioning >>> residual_rms = sg.evaluation.rms(target-prediction) >>> residual_rms > 0.05 and residual_rms < 0.15 # the expected RMS in this test scenario is 0.1 True .. py:attribute:: context_pre :type: int .. py:attribute:: context_post :type: int .. py:attribute:: filter_name :type: str :value: 'UWF' .. py:attribute:: filter_state :type: numpy.typing.NDArray | None :value: None .. py:method:: supports_saving_loading() :staticmethod: Indicates whether saving and loading is supported. .. py:method:: condition_multi_sequence(witness, target, hide_warning = False) Placeholder for compatibility to other filters; does nothing! .. py:method:: apply(witness, target = None, pad = True, update_state = False) Apply the filter to input data :param witness: Witness sensor data :param target: Target sensor data (is ignored) :param pad: if True, apply padding zeros so that the length matches the target signal :param update_state: ignored :return: prediction, bool indicating if all WF updates had full rank .. py:method:: apply_multi_sequence(witness, target, pad = True, update_state = False) Apply the filter to input data Similar to apply() but expects multiple sequences. .. py:class:: LMSFilter(n_filter, idx_target, n_channel = 1, normalized = True, step_scale = 0.1, coefficient_clipping = np.nan) Bases: :py:obj:`saftig.filtering.common.FilterBase` LMS filter implementation :param n_filter: Length of the FIR filter (how many samples are in the input window per output sample) :param idx_target: Position of the prediction :param n_channel: Number of witness sensor channels :param normalized: if True: NLMS, else LMS :param coefficient_clipping: If set to a positive float, FIR filter coefficients will be limited to this value. This can increase filter stability. :param step_scale: the learning rate of the LMS filter >>> import saftig as sg >>> n_filter = 128 >>> witness, target = sg.evaluation.TestDataGenerator(0.1).generate(int(1e5)) >>> filt = sg.filtering.LMSFilter(n_filter, 0, 1) >>> filt.condition(witness, target) >>> prediction = filt.apply(witness, target) # check on the data used for conditioning >>> residual_rms = sg.evaluation.rms(target-prediction) >>> residual_rms > 0.05 and residual_rms < 0.15 # the expected RMS in this test scenario is 0.1 True .. py:attribute:: filter_state :type: numpy.typing.NDArray .. py:attribute:: normalized :type: bool .. py:attribute:: step_scale :type: float .. py:attribute:: coefficient_clipping :type: float .. py:attribute:: filter_name :type: str :value: 'LMS' .. py:method:: reset() reset the filter coefficients to zero .. py:method:: condition(witness, target) Use an input dataset to condition the filter :param witness: Witness sensor data :param target: Target sensor data .. py:method:: condition_multi_sequence(witness, target) Similar to condition(), but expects multiple sequences .. py:method:: apply(witness, target = None, pad = True, update_state = False) Apply the filter to input data :param witness: Witness sensor data :param target: Target sensor data (is ignored) :param pad: if True, apply padding zeros so that the length matches the target signal :param update_state: if True, the filter state will be changed. If false, the filter state will remain :return: prediction .. py:method:: apply_multi_sequence(witness, target = None, pad = True, update_state = False) Apply the filter to input data Similar to apply() but expects multiple sequences. .. py:class:: PolynomialLMSFilter(n_filter, idx_target, n_channel = 1, normalized = True, step_scale = 0.5, coefficient_clipping = np.nan, order = 1) Bases: :py:obj:`saftig.filtering.common.FilterBase` Experimental non-linear LMS-like filter implementation Implements: :math:`x[n] = \sum_p\sum_i\sum_t {w_i[n-t]}^pH_{it}` where p is the polynomial order, i the channel and t the index within the filter :param n_filter: Length of the FIR filter (how many samples are in the input window per output sample) :param idx_target: Position of the prediction :param n_channel: Number of witness sensor channels :param normalized: If True: NLMS, else LMS :param step_scale: The learning rate of the LMS filter :param coefficient_clipping: If set to a positive float, FIR filter coefficients will be limited to this value. This can increase filter stability. :param order: Polynomial order of the filter >>> import saftig as sg >>> n_filter = 128 >>> witness, target = sg.evaluation.TestDataGenerator(0.1).generate(int(1e5)) >>> filt = sg.filtering.PolynomialLMSFilter(n_filter, 0, 1, step_scale=0.1, order=2, coefficient_clipping=4) >>> filt.condition(witness, target) >>> prediction = filt.apply(witness, target) # check on the data used for conditioning >>> residual_rms = sg.evaluation.rms((target-prediction)[1000:]) >>> residual_rms > 0.05 and residual_rms < 0.15 # the expected RMS in this test scenario is 0.1 True .. py:attribute:: filter_state :type: numpy.typing.NDArray .. py:attribute:: normalized :type: bool .. py:attribute:: step_scale :type: float .. py:attribute:: coefficient_clipping :type: float .. py:attribute:: order :type: int .. py:attribute:: filter_name :type: str :value: 'PolyLMS' .. py:method:: reset() reset the filter coefficients to zero .. py:method:: condition(witness, target) Use an input dataset to condition the filter :param witness: Witness sensor data :param target: Target sensor data .. py:method:: condition_multi_sequence(witness, target) Similar to condition(), but expects multiple sequences .. py:method:: apply(witness, target = None, pad = True, update_state = False) Apply the filter to input data :param witness: Witness sensor data :param target: Target sensor data (is ignored) :param pad: if True, apply padding zeros so that the length matches the target signal :param update_state: if True, the filter state will be changed. If false, the filter state will remain :return: prediction .. py:method:: apply_multi_sequence(witness, target = None, pad = True, update_state = False) Apply the filter to input data Similar to apply() but expects multiple sequences.