ezmsg.sigproc.util.binning#

Shared bin-boundary scheduling for fixed-duration binning.

A bin schedule answers one question, identically for every consumer that needs it: given an input sample rate, a target bin duration, and a stream of chunks (each of some length, with a partial bin carried across boundaries), where do the bin boundaries fall – and what output gain/offset label the resulting low-rate axis?

This is the single piece of arithmetic that historically diverged between ezmsg.sigproc.window.Window (which bins by a fixed int(bin_duration * fs) sample count) and ezmsg.event.rate.EventRate (which bins by a fractional bin_duration * fs with a carry accumulator). At a clean rate the two coincide; at an off-nominal rate (e.g. 30012 Hz) they diverge in both gain and bin count, so two such streams never share a grid.

BinSchedule makes the boundary rule a single source of truth. Any consumer (dense aggregation, event counting, window segmentation) that drives its boundaries through one BinSchedule, parameterized identically, is guaranteed to land on the same grid – by construction, not by coincidence.

The schedule deliberately holds only counts and indices; it never touches array data. Each consumer keeps its own data carry (raw samples for an arbitrary aggregator, a scalar partial-sum for a counter), because the right carry representation depends on the operation, but the boundaries are shared.

Boundary definition#

With spb samples per bin (fractional when fractional), the global sample index of the m-th bin boundary is B(m) = int(m * spb). Indexing by the global bin index m (rather than a per-chunk accumulator) makes the output identical regardless of how the input stream is chunked, and reproduces EventRate’s grid (gain, offset, bin count). The closed form int(m * spb) is also more numerically stable than an iterative carry accumulator: there is no drift to accumulate over a long stream.

Classes

class BinSchedule(bin_duration, fractional=True)[source]#

Bases: object

Stateful, backend-agnostic schedule of fixed-duration bin boundaries.

Construct once with the target bin duration and mode, then reset() with the input sample rate before streaming, and call advance() once per input chunk. The schedule tracks the global bin index and the carried sample count; it never holds array data.

Parameters:

  • bin_duration (float) – Output bin duration in seconds.

  • fractional (bool) – If True (default), bins span a fractional bin_duration * fs samples; bin widths track the nominal duration and the output gain is exactly bin_duration (matching EventRate). If False, bins span a fixed int(bin_duration * fs) samples and the output gain is int(bin_duration * fs) / fs (sample-locked, matching Window).

bin_duration: float#
fractional: bool = True#
fs: float | None = None#

Input sample rate, set by reset().

__init__(bin_duration, fractional=True)#
Parameters:
Return type:

None

spb: float = 0.0#

Bin width in input samples (fractional when fractional).

n_bins_done: int = 0#

Count of bins emitted so far across the stream (global bin index).

carry_count: int = 0#

Trailing input samples in the open partial bin, carried across chunks.

reset(fs)[source]#

(Re)initialize the schedule for an input stream at rate fs.

Emits the same sub-sample warnings the dense binner did, so callers that relied on those messages keep getting them from one place.

Parameters:

fs (float)

Return type:

None

property output_gain: float#

Gain (seconds per bin) of the output axis for this schedule.

advance(n_new, in_offset, gain_in)[source]#

Advance the schedule by a chunk of n_new new input samples.

Parameters:

  • n_new (int) – Number of new input samples in this chunk.

  • in_offset (float) – offset of the incoming chunk’s first new sample (i.e. message.axes[axis].offset).

  • gain_in (float) – gain of the input axis (1 / fs).

Returns:

A BinStep. The schedule’s n_bins_done and carry_count are updated to reflect the bins it reports closed.

Return type:

BinStep

class BinStep(cut_points, n_bins, output_gain, output_offset, carry_count)[source]#

Bases: object

Result of advancing a BinSchedule by one chunk.

cut_points are bin ends expressed as indices into the consumer’s work array – i.e. the concatenation of its carried partial-bin samples followed by the new chunk. Bin i spans work[starts[i]:cut_points[i]] where starts = [0] + cut_points[:-1]. The samples after the final cut (work[cut_points[-1]:]) are the new partial bin and become the next chunk’s carry.

Parameters:
cut_points: list[int]#

Bin-end indices into the [carry ++ new] work array (one per closed bin).

n_bins: int#

Number of bins that close in this chunk (== len(cut_points)).

output_gain: float#

Gain (seconds per bin) to label the output axis with.

output_offset: float#

Nominal start time of the first bin closing in this chunk. Derived from the global bin index, so it is identical regardless of chunking (and matches EventRate: stream_start + bins_before * output_gain).

carry_count: int#

Number of input samples in the still-open partial bin after the last cut. The consumer must keep exactly this many trailing work samples as carry.

__init__(cut_points, n_bins, output_gain, output_offset, carry_count)#
Parameters:
Return type:

None