This module contains nodes for signal filtering.

filters

class timeflux_dsp.nodes.filters.DropRows(factor, method=None)[source]

Bases: timeflux.core.node.Node

Decimate signal by an integer factor.

This node uses Pandas computationally efficient functions to drop rows. By default, it simply transfers one row out of factor and drops the others. If method is mean (resp. median), it applies a rolling window of length equals factor, computes the mean and returns one value per window. It maintains an internal state to ensure that every k’th sample is picked even across chunk boundaries.

Variables

i (Port) – Default input, expects DataFrame.
o (Port) – Default output, provides DataFrame.

Parameters

factor (int) – Decimation factor. Only every k’th sample will be
output. (transferred into the) –
method (str|None) – Method to use to drop rows. If None, the values are transferred as it. If mean (resp. median), the mean (resp. median) of the samples is taken.

Example

graphs:

  - id: example
    nodes:
    - id: random
      module: timeflux.nodes.random
      class: Random

    - id: droprows
      module: timeflux_dsp.nodes.filters
      class: DropRows
      params:
        factor: 2

    - id: display
      module: timeflux.nodes.debug
      class: Display
    edges:
    - source: random
      target: droprows
    - source: droprows
      target: display
    rate: 10

Example

In this exemple, we generate white noise to stream and we drop one sample out of two using DropRows, setting:

factor = 2
method = None (see orange trace) | method = “mean” (see green trace)

Notes

Note that this node is not supposed to dejitter the timestamps, so if the input chunk is not uniformly sampled, the output chunk won’t be either.

Also, this filter does not implement any anti-aliasing filter. Hence, it is recommended to precede this node by a low-pass filter (e.g., FIR or IIR) which cuts out below half of the new sampling rate.

Instantiate the node.

update()[source]: Update the input and output ports.

class timeflux_dsp.nodes.filters.Resample(factor, window=None)[source]

Bases: timeflux.core.node.Node

Resample signal.

This node calls the scipy.signal.resample function to decimate the signal using Fourier method.

Variables

i (Port) – Default input, expects DataFrame.
o (Port) – Default output, provides DataFrame.

Parameters

factor (int) – Decimation factor. Only every k’th sample will be transferred into the output.
window (str|list|float) – Specifies the window applied to the signal in the Fourier domain. Default: None.

Example

graphs:

  - id: example
    nodes:
    - id: random
      module: timeflux.nodes.random
      class: Random
    - id: window
      module: timeflux.nodes.window
      class: Window
      params:
        length: 2
    - id: resample
      module: timeflux_dsp.nodes.filters
      class: Resample
      params:
        factor: 2
        window: null

    - id: display
      module: timeflux.nodes.debug
      class: Display
    edges:
    - source: random
      target: window
    - source: window
      target: resample
    - source: resample
      target: display
    rate: 10

Notes

This node should be used after a buffer to assure that the FFT window has always the same length.

References

scipy.signal.resample

Instantiate the node.

update()[source]: Update the input and output ports.

class timeflux_dsp.nodes.filters.IIRFilter(frequencies=None, rate=None, filter_type='bandpass', sos=None, **kwargs)[source]

Bases: timeflux.core.node.Node

Apply IIR filter to signal.

If sos is None, this node uses adapted methods from mne.filters to design the filter coefficients based on the specified parameters. If no transition band is given, default is to use :

l_trans_bandwidth = min(max(l_freq * 0.25, 2), l_freq)
h_trans_bandwidth = min(max(h_freq * 0.25, 2.), rate / 2. - h_freq)

Else, it uses sos as filter coefficients.

Once the kernel has been estimated, the node applies the filtering to each columns in columns using scipy.signal.sosfilt to generate the output given the input, hence ensures continuity across chunk boundaries,

Variables

i (Port) – Default input, expects DataFrame.
o (Port) – Default output, provides DataFrame.

Parameters

rate (float) – Nominal sampling rate of the input data. If None, rate is get from the meta.
order (int, optional) – Filter order. Default: None.
frequencies (list|None) – Transition frequencies. Ignored when sos is given.
filter_type (str|None) – Filter mode (lowpass, highpass, bandstop, bandpass). Default: bandpass. Ignored when sos is given.
sos (array|None, optional) – Array of second-order sections (sos) representation, must have shape (n_sections, 6). Default: None.
kwargs – keyword arguments to pass to the filter constructor

Example

In this example, we generate a signal that is the sum of two sinus with respective periods of 1kHz and 15kHz and respective amplitudes of 1 and 0.5. We stream this signal using the IIRFilter node, designed for lowpass filtering at cutoff frequency 6kHz, order 3.

order = 3
freqs = [6000]
mode = ‘lowpass’

We plot the input signal, the output signal and the corresponding offline filtering.

Notes

This node ensures continuity across chunk boundaries, using a recursive algorithm, based on a cascade of biquads filters.

The filter is initialized to have a minimal step response, but needs a ‘warm up’ period for the filtering to be stable, leading to small artifacts on the first few chunks.

The IIR filter is faster than the FIR filter and delays the signal less but this delay is not constant and the stability not ensured.

References

Instantiate the node.

update()[source]: Update the input and output ports.

class timeflux_dsp.nodes.filters.IIRLineFilter(rate=None, edges_center=(50, 60, 100, 120), orders=(2, 1, 1, 1), edges_width=(3, 3, 3, 3))[source]

Bases: timeflux.core.node.Node

Apply multiple Notch IIR Filter in series.

Attributes: i (Port): Default input, expects DataFrame. o (Port): Default output, provides DataFrame.

Parameters

rate (float) – Nominal sampling rate of the input data. If None, rate is get from the meta.
edges_center – List with center of the filters.
orders (tuple|int|None) – List with orders of the filters. If int, the same order will be used for all filters. If None, order 2 will be used for all filters.
edges_width – List with orders of the filters. If int, the same order will be used for all filters. If None, width of 3 (Hz) will be used for all filters.

Instantiate the node.

update()[source]: Update the input and output ports.

class timeflux_dsp.nodes.filters.FIRFilter(frequencies, rate=None, columns='all', order=20, filter_type='bandpass', coeffs=None, **kwargs)[source]

Bases: timeflux.core.node.Node

Apply FIR filter to signal.

If coeffs is None, this node uses adapted methods from mne.filters to design the filter coefficients based on the specified parameters. If no transition band is given, default is to use:

l_trans_bandwidth = min(max(l_freq * 0.25, 2), l_freq)
h_trans_bandwidth = min(max(h_freq * 0.25, 2.), fs / 2. - h_freq)

Else, it uses coeffs as filter coefficients.

It applies the filtering to each columns in columns using scipy.signal.lfilter to generate the output given the input, hence ensures continuity across chunk boundaries,

The delay introduced is estimated and stored in the meta FIRFilter, delay.

Variables

i (Port) – Default input, expects DataFrame.
o (Port) – Default output, provides DataFrame and meta.

Parameters

rate (float) – Nominal sampling rate of the input data. If None, rate is get from the meta.
columns (list|'all', optional) – Columns to apply filter on. Default: all.
order (int) – Filter order.
frequencies (list) – Transition frequencies.
filter_type (str, optional) – Filter mode (lowpass, highpass, bandstop or bandpass). Default: bandpass.
coeffs (array|None, optional) – Custom coeffs to pass as b in signal.filter. Default: None.
kwargs – keyword arguments to pass to the filter constructor (window, phase,… )

Example

In this exemple, we generate a signal that is the sum of two sinus with respective periods of 1kHz and 15kHz and respective amplitudes of 1 and 0.5. We stream this signal using the FIRFilter node, designed for lowpass filtering at cutoff frequency 6kHz, order 20.

order = 20
freqs = [6000, 6100]
mode = ‘lowpass’

The FIR is a linear phase filter, so it allows one to correct for the introduced delay. Here, we retrieve the input sinus of period 1kHz. We plot the input signal, the output signal, the corresponding offline filtering and the output signal after delay correction.

Notes

The FIR filter ensures a linear phase response, but is computationnaly more costly than the IIR filter.

The filter is initialized to have a minimal step response, but needs a ‘warmup’ period for the filtering to be stable, leeding to small artifacts on the first few chunks.

Instantiate the node.

update()[source]: Update the input and output ports.

class timeflux_dsp.nodes.filters.Scaler(method='sklearn.preprocessing.StandardScaler', **kwargs)[source]

Bases: timeflux.core.node.Node

Apply a sklearn scaler

Variables

i (Port) – Default input, expects DataFrame.
o (Port) – Default output, provides DataFrame and meta.

Parameters

method (str) – Name of the scaler object (see https://scikit-learn.org/stable/modules/classes.html#module-sklearn.preprocessing)
**kwargs – keyword arguments to initialize the scaler.

Instantiate the node.

update()[source]: Update the input and output ports.

class timeflux_dsp.nodes.filters.AdaptiveScaler(length, method='sklearn.preprocessing.StandardScaler', dropna=False, **kwargs)[source]

Bases: timeflux.nodes.window.TimeWindow

Scale the data adaptively. This nodes transforms the data using a sklearn scaler object that is continuously fitted on a rolling window.

Variables

i (Port) – Default input, expects DataFrame.
o (Port) – Default output, provides DataFrame and meta.

Parameters

length (float) – The length of the window, in seconds.
method (str) – Name of the scaler object (see https://scikit-learn.org/stable/modules/classes.html#module-sklearn.preprocessing)
dropna (bool) – Whether or not NaN should be dropped before fitting the estimator. Default to False.
**kwargs – keyword arguments to initialize the scaler.

Instantiate the node.

update()[source]: Update the input and output ports.

class timeflux_dsp.nodes.filters.FilterBank(filters, method='IIRFilter', rate=None, **kwargs)[source]

Bases: timeflux.core.branch.Branch

Apply multiple IIR Filters to the signal and stack the components horizontally

Variables

i (Port) – Default input, expects DataFrame.
o (Port) – Default output, provides DataFrame.

Parameters

rate (float) – Nominal sampling rate of the input data. If None, rate is get from the meta.
filters (dict|None) – Define the iir filter to apply given its name and its params.

Instantiate the node.

update()[source]: Update the input and output ports.