franc.filtering

Implementations of filtering techniques with a common interface.

Submodules

• franc.filtering.bypass
• franc.filtering.common
• franc.filtering.lms
• franc.filtering.polylms
• franc.filtering.uwf
• franc.filtering.wf

Classes

FilterBase	common interface definition for Filter implementations
WienerFilter	Satic Wiener filter implementation
UpdatingWienerFilter	Updating Wiener filter implementation
LMSFilter	LMS filter implementation
PolynomialLMSFilter	Experimental non-linear LMS-like filter implementation

Package Contents

class franc.filtering.FilterBase(n_channel, n_filter, idx_target, _from_dict=None)

Bases: franc.evaluation.FilterInterface

common interface definition for Filter implementations

Parameters:

• n_filter (int) – Length of the FIR filter (how many samples are in the input window per output sample)

• idx_target (int) – Position of the prediction

• n_channel (int) – Number of witness sensor channels

n_filter: int

idx_target: int

default_args = [None, None, None]

property method_filename_part: str

string that can be used in a file name

Return type:

str

class franc.filtering.WienerFilter(n_channel, n_filter, idx_target)

Bases: franc.filtering.common.FilterBase

Satic Wiener filter implementation

Parameters:

• n_channel (int) – Number of witness sensor channels

• n_filter (int) – Length of the FIR filter (how many samples are in the input window per output sample)

• idx_target (int) – Position of the prediction

>>> import franc as fnc
>>> n_filter = 128
>>> witness, target = fnc.evaluation.TestDataGenerator(0.1).generate(int(1e5))
>>> filt = fnc.filtering.WienerFilter(1, n_filter, 0)
>>> _coefficients, full_rank = filt.condition(witness, target)
>>> full_rank
True
>>> prediction = filt.apply(witness, target) # check on the data used for conditioning
>>> residual_rms = fnc.evaluation.rms(target-prediction)
>>> residual_rms > 0.05 and residual_rms < 0.15 # the expected RMS in this test scenario is 0.1
True

filter_state: numpy.typing.NDArray | None = None

filter_name: str = 'WF'

requires_apply_target = False

condition_multi_sequence(witness, target)

Use an input dataset to condition the filter

Parameters:

• witness (collections.abc.Sequence | collections.abc.Sequence[collections.abc.Sequence] | numpy.typing.NDArray) – Witness sensor data

• target (collections.abc.Sequence | numpy.typing.NDArray) – Target sensor data

Return type:

tuple[numpy.typing.NDArray, bool]

apply_multi_sequence(witness, target=None, pad=True, update_state=False)

Apply the filter to input data

Parameters:

• witness (collections.abc.Sequence | numpy.typing.NDArray) – Witness sensor data

• target (collections.abc.Sequence | numpy.typing.NDArray | None) – Target sensor data (is ignored)

• pad (bool) – if True, apply padding zeros so that the length matches the target signal

• update_state (bool) – ignored

Returns:

prediction

Return type:

list[numpy.typing.NDArray]

class franc.filtering.UpdatingWienerFilter(n_channel, n_filter, idx_target, context_pre=0, context_post=0)

Bases: franc.filtering.common.FilterBase

Updating Wiener filter implementation

Parameters:

• n_filter (int) – Length of the FIR filter (how many samples are in the input window per output sample)

• idx_target (int) – Position of the prediction

• n_channel (int) – Number of witness sensor channels

• context_pre (int) – how many additional samples before the current block are used to update the filters

• context_post (int) – how many additional samples after the current block are used to update the filters

>>> import franc as fnc
>>> n_filter = 128
>>> witness, target = fnc.evaluation.TestDataGenerator(0.1).generate(int(1e5))
>>> filt = fnc.filtering.UpdatingWienerFilter(1, n_filter, 0, context_pre=20*n_filter, context_post=20*n_filter)
>>> prediction = filt.apply(witness, target) # check on the data used for conditioning
>>> residual_rms = fnc.evaluation.rms(target-prediction)
>>> residual_rms > 0.05 and residual_rms < 0.15 # the expected RMS in this test scenario is 0.1
True

context_pre: int

context_post: int

filter_name: str = 'UWF'

filter_state: numpy.typing.NDArray | None = None

static supports_saving_loading()

Indicates whether saving and loading is supported.

Return type:

bool

condition_multi_sequence(witness, target, hide_warning=False)

Placeholder for compatibility to other filters; does nothing!

Parameters:

• witness (collections.abc.Sequence | numpy.typing.NDArray) –

• target (collections.abc.Sequence | numpy.typing.NDArray) –

• hide_warning (bool) –

Return type:

None

apply(witness, target=None, pad=True, update_state=False)

Apply the filter to input data

Parameters:

• witness (collections.abc.Sequence | numpy.typing.NDArray) – Witness sensor data

• target (collections.abc.Sequence | numpy.typing.NDArray | None) – Target sensor data (is ignored)

• pad (bool) – if True, apply padding zeros so that the length matches the target signal

• update_state (bool) – ignored

Returns:

prediction, bool indicating if all WF updates had full rank

Return type:

numpy.typing.NDArray

apply_multi_sequence(witness, target, pad=True, update_state=False)

Apply the filter to multiple sequences of input data.

Similar to apply() but expects multiple sequences. First index to the given data objects indicates the sequence. The last index indicates the time within a single sequence. Sequences must not have the same length.

Parameters:

• witness (collections.abc.Sequence | numpy.typing.NDArray) –

• target (collections.abc.Sequence | numpy.typing.NDArray | None) –

• pad (bool) –

• update_state (bool) –

Return type:

collections.abc.Sequence[numpy.typing.NDArray]

class franc.filtering.LMSFilter(n_channel, n_filter, idx_target, normalized=True, step_scale=0.1, coefficient_clipping=np.nan)

Bases: franc.filtering.common.FilterBase

LMS filter implementation

Parameters:

• n_filter (int) – Length of the FIR filter (how many samples are in the input window per output sample)

• idx_target (int) – Position of the prediction

• n_channel (int) – Number of witness sensor channels

• normalized (bool) – if True: NLMS, else LMS

• coefficient_clipping (float) – If set to a positive float, FIR filter coefficients will be limited to this value. This can increase filter stability.

• step_scale (float) – the learning rate of the LMS filter

>>> import franc as fnc
>>> n_filter = 128
>>> witness, target = fnc.evaluation.TestDataGenerator(0.1).generate(int(1e5))
>>> filt = fnc.filtering.LMSFilter(1, n_filter, 0)
>>> filt.condition(witness, target)
>>> prediction = filt.apply(witness, target) # check on the data used for conditioning
>>> residual_rms = fnc.evaluation.rms(target-prediction)
>>> residual_rms > 0.05 and residual_rms < 0.15 # the expected RMS in this test scenario is 0.1
True

filter_state: numpy.typing.NDArray

normalized: bool

step_scale: float

coefficient_clipping: float

filter_name: str = 'LMS'

reset()

reset the filter coefficients to zero

condition(witness, target)

Use an input dataset to condition the filter

Parameters:

• witness (collections.abc.Sequence | numpy.typing.NDArray) – Witness sensor data

• target (collections.abc.Sequence | numpy.typing.NDArray) – Target sensor data

Return type:

None

condition_multi_sequence(witness, target)

Similar to condition(), but expects multiple sequences

Parameters:

• witness (collections.abc.Sequence | collections.abc.Sequence[collections.abc.Sequence] | numpy.typing.NDArray) –

• target (collections.abc.Sequence | numpy.typing.NDArray) –

Return type:

None

apply(witness, target=None, pad=True, update_state=False)

Apply the filter to input data

Parameters:

• witness (collections.abc.Sequence | numpy.typing.NDArray) – Witness sensor data

• target (collections.abc.Sequence | numpy.typing.NDArray | None) – Target sensor data (is ignored)

• pad (bool) – if True, apply padding zeros so that the length matches the target signal

• update_state (bool) – if True, the filter state will be changed. If false, the filter state will remain

Returns:

prediction

Return type:

numpy.typing.NDArray

apply_multi_sequence(witness, target=None, pad=True, update_state=False)

Apply the filter to multiple sequences of input data.

Similar to apply() but expects multiple sequences. First index to the given data objects indicates the sequence. The last index indicates the time within a single sequence. Sequences must not have the same length.

Parameters:

• witness (collections.abc.Sequence | numpy.typing.NDArray) –

• target (collections.abc.Sequence | numpy.typing.NDArray | None) –

• pad (bool) –

• update_state (bool) –

Return type:

collections.abc.Sequence[numpy.typing.NDArray]

class franc.filtering.PolynomialLMSFilter(n_channel, n_filter, idx_target, normalized=True, step_scale=0.5, coefficient_clipping=np.nan, order=1)

Bases: franc.filtering.common.FilterBase

Experimental non-linear LMS-like filter implementation Implements: \(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

Parameters:

• n_filter (int) – Length of the FIR filter (how many samples are in the input window per output sample)

• idx_target (int) – Position of the prediction

• n_channel (int) – Number of witness sensor channels

• normalized (bool) – If True: NLMS, else LMS

• step_scale (float) – The learning rate of the LMS filter

• coefficient_clipping (float) – If set to a positive float, FIR filter coefficients will be limited to this value. This can increase filter stability.

• order (int) – Polynomial order of the filter

>>> import franc as fnc
>>> n_filter = 128
>>> witness, target = fnc.evaluation.TestDataGenerator(0.1).generate(int(1e5))
>>> filt = fnc.filtering.PolynomialLMSFilter(1, n_filter, 0, 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 = fnc.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

filter_state: numpy.typing.NDArray

normalized: bool

step_scale: float

coefficient_clipping: float

order: int

filter_name: str = 'PolyLMS'

reset()

reset the filter coefficients to zero

condition(witness, target)

Use an input dataset to condition the filter

Parameters:

• witness (collections.abc.Sequence | numpy.typing.NDArray) – Witness sensor data

• target (collections.abc.Sequence | numpy.typing.NDArray) – Target sensor data

Return type:

None

condition_multi_sequence(witness, target)

Similar to condition(), but expects multiple sequences

Parameters:

• witness (collections.abc.Sequence | collections.abc.Sequence[collections.abc.Sequence] | numpy.typing.NDArray) –

• target (collections.abc.Sequence | numpy.typing.NDArray) –

Return type:

None

apply(witness, target=None, pad=True, update_state=False)

Apply the filter to input data

Parameters:

• witness (collections.abc.Sequence | numpy.typing.NDArray) – Witness sensor data

• target (collections.abc.Sequence | numpy.typing.NDArray | None) – Target sensor data (is ignored)

• pad (bool) – if True, apply padding zeros so that the length matches the target signal

• update_state (bool) – if True, the filter state will be changed. If false, the filter state will remain

Returns:

prediction

Return type:

numpy.typing.NDArray

apply_multi_sequence(witness, target=None, pad=True, update_state=False)

Apply the filter to multiple sequences of input data.

Similar to apply() but expects multiple sequences. First index to the given data objects indicates the sequence. The last index indicates the time within a single sequence. Sequences must not have the same length.

Parameters:

• witness (collections.abc.Sequence | numpy.typing.NDArray) –

• target (collections.abc.Sequence | numpy.typing.NDArray | None) –

• pad (bool) –

• update_state (bool) –

Return type:

collections.abc.Sequence[numpy.typing.NDArray]