ezmsg.sigproc.resampleconcat#

Resample one stream onto another’s clock and concatenate them in one step.

This fuses ResampleProcessor and ConcatProcessor for the common case of combining two acquisition streams that share a nominal sampling rate but drift slightly:

  • INPUT_REFERENCE – the reference stream (its clock is the master timebase, and its data becomes one half of the concatenated output).

  • INPUT_SIGNAL – the stream that is resampled onto the reference clock.

  • OUTPUT_SIGNAL – the two streams concatenated along concat_axis.

Because the resampler emits the reference signal gathered onto the exact grid it resampled the signal onto (output_reference=True), the two halves share an identical alignment axis by construction. No second time-alignment (Align) is needed, so this collapses the otherwise-diamond graph

REF -> RESAMPLE.INPUT_REFERENCE SIG -> RESAMPLE.INPUT_SIGNAL REF -> MERGE.INPUT_SIGNAL_A RESAMPLE.OUTPUT -> MERGE.INPUT_SIGNAL_B

into a single linear two-input unit. All interpolation logic is reused from resample.py and all axis/attr-merge logic from concat.py; only the orchestration is new.

Classes

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

Bases: Unit

One-step resample-onto-reference + concatenate.

See module docstring. INPUT_REFERENCE is the master clock and side A; INPUT_SIGNAL is resampled onto it and becomes side B.

Parameters:

settings (Settings | None)

SETTINGS#

alias of ResampleConcatSettings

INPUT_SIGNAL = InputStream:unlocated[AxisArray]()#
INPUT_REFERENCE = InputStream:unlocated[AxisArray]()#
OUTPUT_SIGNAL = OutputStream:unlocated[AxisArray](self.num_buffers=32, self.force_tcp=None, self.allow_local=None)#
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_reference(message)[source]#
Parameters:

message (AxisArray)

Return type:

AsyncGenerator

async on_signal(message)[source]#
Parameters:

message (AxisArray)

Return type:

AsyncGenerator

class ResampleConcatProcessor(settings)[source]#

Bases: object

Resample INPUT_SIGNAL onto INPUT_REFERENCE’s clock, then concatenate.

Composes an unmodified ResampleProcessor (with output_reference=True) and an unmodified ConcatProcessor. The reference signal (A) and the resampled signal (B) are concatenated only when the resampler yields both on the same grid, so the two are aligned by construction.

Parameters:

settings (ResampleConcatSettings)

__init__(settings)[source]#
Parameters:

settings (ResampleConcatSettings)

property resample_state#

Expose resampler state for introspection / tests.

property concat_state#

Expose concat state for introspection / tests.

push_reference(message)[source]#

Feed a reference (clock + side-A) message.

Parameters:

message (AxisArray)

Return type:

None

push_signal(message)[source]#

Feed a signal message to be resampled onto the reference clock (side B).

Parameters:

message (AxisArray)

Return type:

None

class ResampleConcatSettings(axis: 'str' = 'time', concat_axis: 'str' = 'ch', buffer_duration: 'float' = 2.0, fill_value: 'str' = 'extrapolate', max_chunk_delay: 'float' = inf, buffer_update_strategy: 'UpdateStrategy' = 'immediate', reference_reset_after_chunks: 'float' = 3, relabel_axis: 'bool' = True, label_a: 'str' = '_a', label_b: 'str' = '_b', assert_identical_shared_axes: 'bool' = False, new_key: 'str | None' = None, auto_coerce_backend: 'bool' = False)[source]#

Bases: Settings

Parameters:

  • axis (str)

  • concat_axis (str)

  • buffer_duration (float)

  • fill_value (str)

  • max_chunk_delay (float)

  • buffer_update_strategy (Literal['immediate', 'threshold', 'on_demand'])

  • reference_reset_after_chunks (float)

  • relabel_axis (bool)

  • label_a (str)

  • label_b (str)

  • assert_identical_shared_axes (bool)

  • new_key (str | None)

  • auto_coerce_backend (bool)

axis: str = 'time'#

the reference stream’s clock that the signal is resampled onto.

Type:

Alignment axis

concat_axis: str = 'ch'#

Axis along which the reference and resampled signals are concatenated.

buffer_duration: float = 2.0#
fill_value: str = 'extrapolate'#
max_chunk_delay: float = inf#
buffer_update_strategy: Literal['immediate', 'threshold', 'on_demand'] = 'immediate'#
reference_reset_after_chunks: float = 3#

See ezmsg.sigproc.resample.ResampleSettings.reference_reset_after_chunks.

relabel_axis: bool = True#
label_a: str = '_a'#

Per-side label for the reference (A) side. See ConcatSettings.

label_b: str = '_b'#

Per-side label for the resampled-signal (B) side.

assert_identical_shared_axes: bool = False#
new_key: str | None = None#
auto_coerce_backend: bool = False#
__init__(axis='time', concat_axis='ch', buffer_duration=2.0, fill_value='extrapolate', max_chunk_delay=inf, buffer_update_strategy='immediate', reference_reset_after_chunks=3, relabel_axis=True, label_a='_a', label_b='_b', assert_identical_shared_axes=False, new_key=None, auto_coerce_backend=False)#
Parameters:

  • axis (str)

  • concat_axis (str)

  • buffer_duration (float)

  • fill_value (str)

  • max_chunk_delay (float)

  • buffer_update_strategy (Literal['immediate', 'threshold', 'on_demand'])

  • reference_reset_after_chunks (float)

  • relabel_axis (bool)

  • label_a (str)

  • label_b (str)

  • assert_identical_shared_axes (bool)

  • new_key (str | None)

  • auto_coerce_backend (bool)

Return type:

None