ezmsg.sigproc.filter

Functions

filtergen(axis, coefs, coef_type)

Filter data using the provided coefficients.

Returns:

FilterTransformer.

Parameters:

• axis (str)

• coefs (ndarray[tuple[Any, ...], dtype[_ScalarT]] | tuple[ndarray[tuple[Any, ...], dtype[_ScalarT]]] | None)

• coef_type (str)

Return type:

FilterTransformer

Classes

class BaseFilterByDesignTransformerUnit(*args, settings=None, **kwargs)

Bases: BaseTransformerUnit[SettingsType, AxisArray, AxisArray, FilterByDesignTransformer], Generic[SettingsType, TransformerType]

Parameters:

settings (Settings | None)

async on_settings(msg)

Receive a settings message, override self.SETTINGS, and re-create the processor. Child classes that wish to have fine-grained control over whether the core processor resets on settings changes should override this method.

Parameters:

msg (SettingsType) – a settings message.

Return type:

None

class Filter(*args, settings=None, **kwargs)

Bases: BaseTransformerUnit[FilterSettings, AxisArray, AxisArray, FilterTransformer]

Parameters:

settings (Settings | None)

SETTINGS

alias of FilterSettings

class FilterBaseSettings(axis: str | None = None, coef_type: str = 'ba')

Bases: Settings

Parameters:

• axis (str | None)

• coef_type (str)

axis: str | None = None

The name of the axis to operate on.

coef_type: str = 'ba'

The type of filter coefficients. One of “ba” or “sos”.

__init__(axis=None, coef_type='ba')

Parameters:

• axis (str | None)

• coef_type (str)

Return type:

None

class FilterByDesignState

Bases: object

filter: FilterTransformer | None = None

needs_redesign: bool = False

class FilterByDesignTransformer(*args, **kwargs)

Bases: BaseStatefulTransformer[SettingsType, AxisArray, AxisArray, FilterByDesignState], ABC, Generic[SettingsType, FilterCoefsType]

Abstract base class for filter design transformers.

classmethod get_message_type(dir)

Parameters:

dir (str)

Return type:

type[AxisArray]

abstractmethod get_design_function()

Return a function that takes sampling frequency and returns filter coefficients.

Return type:

Callable[[float], FilterCoefsType | None]

update_settings(new_settings=None, **kwargs)

Update settings and mark that filter coefficients need to be recalculated.

Parameters:

• new_settings (SettingsType | None) – Complete new settings object to replace current settings

• **kwargs – Individual settings to update

Return type:

None

class FilterCoefficients(b: numpy.ndarray = <factory>, a: numpy.ndarray = <factory>)

Bases: object

Parameters:

• b (ndarray)

• a (ndarray)

b: ndarray

a: ndarray

__init__(b=<factory>, a=<factory>)

Parameters:

• b (ndarray)

• a (ndarray)

Return type:

None

class FilterSettings(axis: str | None = None, coef_type: str = 'ba', coefs: FilterCoefficients | None = None)

Bases: FilterBaseSettings

Parameters:

• axis (str | None)

• coef_type (str)

• coefs (FilterCoefficients | None)

coefs: FilterCoefficients | None = None

The pre-calculated filter coefficients.

__init__(axis=None, coef_type='ba', coefs=None)

Parameters:

• axis (str | None)

• coef_type (str)

• coefs (FilterCoefficients | None)

Return type:

None

class FilterState

Bases: object

zi: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None

class FilterTransformer(*args, **kwargs)

Bases: BaseStatefulTransformer[FilterSettings, AxisArray, AxisArray, FilterState]

Filter data using the provided coefficients.

update_coefficients(coefs, coef_type=None)

Update filter coefficients.

If the new coefficients have the same length as the current ones, only the coefficients are updated. If the lengths differ, the filter state is also reset to handle the new filter order.

Parameters:

• coefs (FilterCoefficients | tuple[ndarray[tuple[Any, ...], dtype[_ScalarT]], ndarray[tuple[Any, ...], dtype[_ScalarT]]] | ndarray[tuple[Any, ...], dtype[_ScalarT]]) – New filter coefficients

• coef_type (str | None)

Return type:

None

class FilterCoefficients(b: numpy.ndarray = <factory>, a: numpy.ndarray = <factory>)

Bases: object

Parameters:

• b (ndarray)

• a (ndarray)

b: ndarray

a: ndarray

__init__(b=<factory>, a=<factory>)

Parameters:

• b (ndarray)

• a (ndarray)

Return type:

None

class FilterBaseSettings(axis: str | None = None, coef_type: str = 'ba')

Bases: Settings

Parameters:

• axis (str | None)

• coef_type (str)

axis: str | None = None

The name of the axis to operate on.

coef_type: str = 'ba'

The type of filter coefficients. One of “ba” or “sos”.

__init__(axis=None, coef_type='ba')

Parameters:

• axis (str | None)

• coef_type (str)

Return type:

None

class FilterSettings(axis: str | None = None, coef_type: str = 'ba', coefs: FilterCoefficients | None = None)

Bases: FilterBaseSettings

Parameters:

• axis (str | None)

• coef_type (str)

• coefs (FilterCoefficients | None)

coefs: FilterCoefficients | None = None

The pre-calculated filter coefficients.

__init__(axis=None, coef_type='ba', coefs=None)

Parameters:

• axis (str | None)

• coef_type (str)

• coefs (FilterCoefficients | None)

Return type:

None

class FilterState

Bases: object

zi: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None

class FilterTransformer(*args, **kwargs)

Bases: BaseStatefulTransformer[FilterSettings, AxisArray, AxisArray, FilterState]

Filter data using the provided coefficients.

update_coefficients(coefs, coef_type=None)

Update filter coefficients.

If the new coefficients have the same length as the current ones, only the coefficients are updated. If the lengths differ, the filter state is also reset to handle the new filter order.

Parameters:

• coefs (FilterCoefficients | tuple[ndarray[tuple[Any, ...], dtype[_ScalarT]], ndarray[tuple[Any, ...], dtype[_ScalarT]]] | ndarray[tuple[Any, ...], dtype[_ScalarT]]) – New filter coefficients

• coef_type (str | None)

Return type:

None

class Filter(*args, settings=None, **kwargs)

Bases: BaseTransformerUnit[FilterSettings, AxisArray, AxisArray, FilterTransformer]

Parameters:

settings (Settings | None)

SETTINGS

alias of FilterSettings

filtergen(axis, coefs, coef_type)

Filter data using the provided coefficients.

Returns:

FilterTransformer.

Parameters:

• axis (str)

• coefs (ndarray[tuple[Any, ...], dtype[_ScalarT]] | tuple[ndarray[tuple[Any, ...], dtype[_ScalarT]]] | None)

• coef_type (str)

Return type:

FilterTransformer

class FilterByDesignState

Bases: object

filter: FilterTransformer | None = None

needs_redesign: bool = False

class FilterByDesignTransformer(*args, **kwargs)

Bases: BaseStatefulTransformer[SettingsType, AxisArray, AxisArray, FilterByDesignState], ABC, Generic[SettingsType, FilterCoefsType]

Abstract base class for filter design transformers.

classmethod get_message_type(dir)

Parameters:

dir (str)

Return type:

type[AxisArray]

abstractmethod get_design_function()

Return a function that takes sampling frequency and returns filter coefficients.

Return type:

Callable[[float], FilterCoefsType | None]

update_settings(new_settings=None, **kwargs)

Update settings and mark that filter coefficients need to be recalculated.

Parameters:

• new_settings (SettingsType | None) – Complete new settings object to replace current settings

• **kwargs – Individual settings to update

Return type:

None

class BaseFilterByDesignTransformerUnit(*args, settings=None, **kwargs)

Bases: BaseTransformerUnit[SettingsType, AxisArray, AxisArray, FilterByDesignTransformer], Generic[SettingsType, TransformerType]

Parameters:

settings (Settings | None)

async on_settings(msg)

Receive a settings message, override self.SETTINGS, and re-create the processor. Child classes that wish to have fine-grained control over whether the core processor resets on settings changes should override this method.

Parameters:

msg (SettingsType) – a settings message.

Return type:

None