ezmsg.sigproc.resample#
Resample signals to a target rate using interpolation.
Classes
- class ResampleProcessor(*args, **kwargs)[source]#
Bases:
BaseStatefulProcessor[ResampleSettings,AxisArray,AxisArray,ResampleState]
- class ResampleSettings(axis: str = 'time', resample_rate: float | None = None, max_chunk_delay: float = inf, fill_value: str = 'extrapolate', buffer_duration: float = 2.0, buffer_update_strategy: Literal['immediate', 'threshold', 'on_demand'] = 'immediate', output_reference: bool = False, reference_reset_after_chunks: float = 3)[source]#
Bases:
Settings- Parameters:
- resample_rate: float | None = None#
target resample rate in Hz. If None, the resample rate will be determined by the reference signal.
- max_chunk_delay: float = inf#
Maximum delay between outputs in seconds. If the delay exceeds this value, the transformer will extrapolate.
- fill_value: str = 'extrapolate'#
Value to use for out-of-bounds samples. If ‘extrapolate’, the transformer will extrapolate. If ‘last’, the transformer will use the last sample. See scipy.interpolate.interp1d for more options.
- buffer_update_strategy: Literal['immediate', 'threshold', 'on_demand'] = 'immediate'#
The buffer update strategy. See
ezmsg.sigproc.util.buffer.UpdateStrategy. If you expect to push data much more frequently than it is resampled, then “on_demand” might be more efficient. For most other scenarios, “immediate” is best.
- output_reference: bool = False#
If True, also buffer the data carried by INPUT_REFERENCE messages and, for each output chunk, gather the reference samples that sit on the exact same grid the signal was resampled onto. The gathered reference signal is exposed via
ResampleState.reference_output(and published onOUTPUT_REFERENCEbyResampleUnit). This is what lets a downstream consumer concatenate the reference stream with the resampled stream without a second time-alignment, because the two share an identical axis by construction.Only meaningful when
resample_rate is None(reference-driven mode); in prescribed- rate mode the reference grid is synthetic and carries no data.
- reference_reset_after_chunks: float = 3#
Robustness against a non-monotonic reference clock.
The resampler only emits at reference values greater than the last one it returned (a high-water mark). A sustained backward jump in the reference clock (e.g. a misbehaving source whose chunk offsets reset to an earlier time) would otherwise leave every incoming reference value behind the high-water mark forever, so the transformer would stop producing output.
If this many consecutive reference messages arrive entirely at or below the high-water mark, the jump is treated as a clock reset: the high-water mark is re-anchored to the new (lower) clock and a
RuntimeWarningis emitted. Output then resumes on the new clock. Small, self-correcting jitter (a few out-of-order samples) does not trigger this and is simply skipped, keeping the output monotonic.Set to
float("inf")to disable reset recovery (the transformer may then stall indefinitely on a backward clock jump). Output across a genuine reset is necessarily discontinuous; sanitising the reference timestamps upstream remains the robust fix.
- __init__(axis='time', resample_rate=None, max_chunk_delay=inf, fill_value='extrapolate', buffer_duration=2.0, buffer_update_strategy='immediate', output_reference=False, reference_reset_after_chunks=3)#
- class ResampleState[source]#
Bases:
object- src_buffer: HybridAxisArrayBuffer | None = None#
Buffer for the incoming signal data. This is the source for training the interpolation function. Its contents are rarely empty because we usually hold back some data to allow for accurate interpolation and optionally extrapolation.
- ref_axis_buffer: HybridAxisBuffer | None = None#
The buffer for the reference axis (usually a time axis). The interpolation function will be evaluated at the reference axis values. When resample_rate is None, this buffer will be filled with the axis from incoming _reference_ messages. When resample_rate is not None (i.e., prescribed float resample_rate), this buffer is filled with a synthetic axis that is generated from the incoming signal messages.
- last_ref_ax_val: float | None = None#
The last value of the reference axis that was returned. This helps us to know what the _next_ returned value should be, and to avoid returning the same value. TODO: We can eliminate this variable if we maintain “by convention” that the reference axis always has 1 value at its start that we exclude from the resampling.
- last_write_time: float = -inf#
Monotonic time of the last write to the signal buffer. This is used to determine if we need to extrapolate the reference axis if we have not received an update within max_chunk_delay.
- ref_data_buffer: HybridAxisArrayBuffer | None = None#
Buffer for the data carried by reference messages, used only when
output_referenceis set. Held in lockstep withref_axis_buffer(same writes, same seeks) so that the reference data at indexicorresponds to the reference axis value at indexi. Lets__next__()gather the reference signal on the exact output grid.
- reference_output: AxisArray | None = None#
The reference signal gathered onto the most recent output grid (only populated when
output_referenceis set and the last__next__()produced data). Shares its axis with the resampled output, so the two can be concatenated without re-aligning.
- stale_ref_pushes: int = 0#
Number of consecutive reference messages that arrived entirely at or below the high-water mark. Used to detect a sustained backward clock jump (reset). See
ResampleSettings.reference_reset_after_chunks.
- class ResampleUnit(*args, settings=None, **kwargs)[source]#
Bases:
BaseConsumerUnit[ResampleSettings,AxisArray,ResampleProcessor]- Parameters:
settings (Settings | None)
- SETTINGS#
alias of
ResampleSettings
- INPUT_REFERENCE = InputStream:unlocated[AxisArray]()#
- OUTPUT_SIGNAL = OutputStream:unlocated[AxisArray](self.num_buffers=32, self.force_tcp=None, self.allow_local=None)#
- OUTPUT_REFERENCE = OutputStream:unlocated[AxisArray](self.num_buffers=32, self.force_tcp=None, self.allow_local=None)#
The reference signal gathered onto the same grid as
OUTPUT_SIGNAL.Only published when
output_referenceis set. Because it shares an identical axis withOUTPUT_SIGNALby construction, a downstreamConcatcan combine the two without a second time-alignment (seeResampleConcat, which fuses both steps).