ezmsg.sigproc.filter#

Core IIR/FIR filtering infrastructure with BA and SOS coefficient support.

Functions

filtergen(axis, coefs, coef_type)[source]#

Filter data using the provided coefficients.

Returns:

FilterTransformer.

Parameters:
Return type:

FilterTransformer

Classes

class BaseFilterByDesignTransformerUnit(*args, settings=None, **kwargs)[source]#

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

Parameters:

settings (Settings | None)

class Filter(*args, settings=None, **kwargs)[source]#

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

Parameters:

settings (Settings | None)

SETTINGS#

alias of FilterSettings

class FilterBaseSettings(axis: str | None = None, coef_type: str = 'ba', use_mlx_metal: bool = True)[source]#

Bases: Settings

Parameters:

  • axis (str | None)

  • coef_type (str)

  • use_mlx_metal (bool)

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”.

use_mlx_metal: bool = True#

If True (default), SOS filtering on MLX inputs runs on the GPU via the bundled Metal kernel (sosfilt_mlx_metal) instead of round-tripping through scipy. Set to False to fall back to the scipy path (bit-exact with numpy at the cost of CPU round-trips and ~5-8x slowdown).

__init__(axis=None, coef_type='ba', use_mlx_metal=True)#
Parameters:

  • axis (str | None)

  • coef_type (str)

  • use_mlx_metal (bool)

Return type:

None

class FilterByDesignState[source]#

Bases: object

filter: FilterTransformer | None = None#
needs_redesign: bool = False#
class FilterByDesignTransformer(*args, **kwargs)[source]#

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

Abstract base class for filter design transformers.

classmethod get_message_type(dir)[source]#
Parameters:

dir (str)

Return type:

type[AxisArray]

abstractmethod get_design_function()[source]#

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

Return type:

Callable[[float], FilterCoefsType | None]

update_settings(new_settings=None, **kwargs)[source]#

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: ndarray = <factory>, a: ndarray = <factory>)[source]#

Bases: object

Parameters:
b: ndarray#
a: ndarray#
__init__(b=<factory>, a=<factory>)#
Parameters:
Return type:

None

class FilterSettings(axis: str | None = None, coef_type: str = 'ba', use_mlx_metal: bool = True, coefs: FilterCoefficients | None = None)[source]#

Bases: FilterBaseSettings

Parameters:
coefs: FilterCoefficients | None = None#

The pre-calculated filter coefficients.

__init__(axis=None, coef_type='ba', use_mlx_metal=True, coefs=None)#
Parameters:
Return type:

None

class FilterState[source]#

Bases: object

zi: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None#
fir_b: Any | None = None#
fir_b_1d: Any | None = None#
fir_method: str | None = None#
sos_method: str | None = None#
sos_mx: Any | None = None#
class FilterTransformer(*args, **kwargs)[source]#

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

Filter data using the provided coefficients.

update_coefficients(coefs, coef_type=None)[source]#

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:
Return type:

None