ezmsg.sigproc.math.difference#

Take the difference between 2 signals or between a signal and a constant value.

Note

ConstDifferenceTransformer supports the Array API standard, enabling use with NumPy, CuPy, PyTorch, and other compatible array libraries. DifferenceProcessor (two-input difference) currently requires NumPy arrays.

Functions

const_difference(value=0.0, subtrahend=True)[source]#

result = (in_data - value) if subtrahend else (value - in_data) https://en.wikipedia.org/wiki/Template:Arithmetic_operations

Parameters:

  • value (float) – number to subtract or be subtracted from the input data

  • subtrahend (bool) – If True (default) then value is subtracted from the input data. If False, the input data is subtracted from value.

Return type:

ConstDifferenceTransformer

Returns: ConstDifferenceTransformer.

Classes

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

Bases: BaseTransformerUnit[ConstDifferenceSettings, AxisArray, AxisArray, ConstDifferenceTransformer]

Parameters:

settings (Settings | None)

SETTINGS#

alias of ConstDifferenceSettings

class ConstDifferenceSettings(value: float = 0.0, subtrahend: bool = True)[source]#

Bases: Settings

Parameters:
value: float = 0.0#

number to subtract or be subtracted from the input data

subtrahend: bool = True#

If True (default) then value is subtracted from the input data. If False, the input data is subtracted from value.

__init__(value=0.0, subtrahend=True)#
Parameters:
Return type:

None

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

Bases: BaseTransformer[ConstDifferenceSettings, AxisArray, AxisArray]

Parameters:

settings (SettingsType)

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

Bases: Unit

Subtract two signals (A - B).

Assumes compatible/similar axes/dimensions and aligned time spans. Messages are paired by arrival order (oldest from each queue).

OUTPUT = INPUT_SIGNAL_A - INPUT_SIGNAL_B

Parameters:

settings (Settings | None)

INPUT_SIGNAL_A = InputStream:unlocated[AxisArray]()#
INPUT_SIGNAL_B = InputStream:unlocated[AxisArray]()#
OUTPUT_SIGNAL = OutputStream:unlocated[AxisArray](self.num_buffers=32, self.force_tcp=False)#
async initialize()[source]#

Runs when the Unit is instantiated.

This is called from within the same process this unit will live in. This lifecycle hook can be overridden. It can be run as async functions by simply adding the async keyword when overriding.

This method is where you should initialize your unit’s state and prepare for message processing.

Return type:

None

async on_a(msg)[source]#
Parameters:

msg (AxisArray)

Return type:

None

async on_b(msg)[source]#
Parameters:

msg (AxisArray)

Return type:

None

async output()[source]#
Return type:

AsyncGenerator

class DifferenceProcessor[source]#

Bases: object

Processor that subtracts two AxisArray signals (A - B).

This processor maintains separate queues for two input streams and subtracts corresponding messages element-wise. It assumes both inputs have compatible shapes and aligned time spans.

__init__()[source]#
property state: DifferenceState#
push_a(msg)[source]#

Push a message to queue A (minuend).

Parameters:

msg (AxisArray)

Return type:

None

push_b(msg)[source]#

Push a message to queue B (subtrahend).

Parameters:

msg (AxisArray)

Return type:

None

class DifferenceState(queue_a=<factory>, queue_b=<factory>)[source]#

Bases: object

State for Difference processor with two input queues.

Parameters:
queue_a: Queue[AxisArray]#
queue_b: Queue[AxisArray]#
__init__(queue_a=<factory>, queue_b=<factory>)#
Parameters:
Return type:

None