ezmsg.baseproc.clockdriven

Clock-driven producer base classes for generating data synchronized to clock ticks.

Classes

class BaseClockDrivenProducer(*args, **kwargs)

Bases: BaseStatefulProcessor[ClockDrivenSettingsType, LinearAxis, AxisArray, StateType], Generic[ClockDrivenSettingsType, StateType]

Base class for clock-driven data producers.

Accepts clock ticks (LinearAxis) as input and produces AxisArray output. Handles all the timing/counter logic internally, so subclasses only need to implement the data generation logic.

This eliminates the need for the Clock → Counter → Generator pattern by combining the Counter functionality into the generator base class.

Subclasses must implement:

• _reset_state(time_axis): Initialize any state needed for production

• _produce(n_samples, time_axis): Generate the actual output data

Example:

@processor_state
class SinState(ClockDrivenState):
    ang_freq: float = 0.0

class SinProducer(BaseClockDrivenProducer[SinSettings, SinState]):
    def _reset_state(self, time_axis: AxisArray.TimeAxis) -> None:
        self._state.ang_freq = 2 * np.pi * self.settings.fs

    def _produce(self, n_samples: int, time_axis: AxisArray.TimeAxis) -> AxisArray:
        t = (np.arange(n_samples) + self._state.counter) * time_axis.gain
        data = np.sin(self._state.ang_freq * t)
        return AxisArray(data=data, dims=["time"], axes={"time": time_axis})

class ClockDrivenSettings(fs, n_time=None)

Bases: Settings

Base settings for clock-driven producers.

Subclass this to add your own settings while inheriting fs and n_time.

Example:

class SinGeneratorSettings(ClockDrivenSettings):
    freq: float = 1.0
    amp: float = 1.0

Parameters:

• fs (float)

• n_time (int | None)

fs: float

Output sampling rate in Hz.

n_time: int | None = None

Samples per block. - If specified: fixed chunk size (clock gain is ignored for determining chunk size) - If None: derived from clock gain (fs * clock.gain), with fractional sample tracking

__init__(fs, n_time=None)

Parameters:

• fs (float)

• n_time (int | None)

Return type:

None

class ClockDrivenState

Bases: object

Internal state for clock-driven producers.

Tracks sample counting and fractional sample accumulation. Subclasses should extend this if they need additional state.

counter: int = 0

Current sample counter (total samples produced).

fractional_samples: float = 0.0

Accumulated fractional samples for variable chunk mode.

class ClockDrivenSettings(fs, n_time=None)

Bases: Settings

Base settings for clock-driven producers.

Subclass this to add your own settings while inheriting fs and n_time.

Example:

class SinGeneratorSettings(ClockDrivenSettings):
    freq: float = 1.0
    amp: float = 1.0

Parameters:

• fs (float)

• n_time (int | None)

fs: float

Output sampling rate in Hz.

n_time: int | None = None

Samples per block. - If specified: fixed chunk size (clock gain is ignored for determining chunk size) - If None: derived from clock gain (fs * clock.gain), with fractional sample tracking

__init__(fs, n_time=None)

Parameters:

• fs (float)

• n_time (int | None)

Return type:

None

class ClockDrivenState

Bases: object

Internal state for clock-driven producers.

Tracks sample counting and fractional sample accumulation. Subclasses should extend this if they need additional state.

counter: int = 0

Current sample counter (total samples produced).

fractional_samples: float = 0.0

Accumulated fractional samples for variable chunk mode.

class BaseClockDrivenProducer(*args, **kwargs)

Bases: BaseStatefulProcessor[ClockDrivenSettingsType, LinearAxis, AxisArray, StateType], Generic[ClockDrivenSettingsType, StateType]

Base class for clock-driven data producers.

Accepts clock ticks (LinearAxis) as input and produces AxisArray output. Handles all the timing/counter logic internally, so subclasses only need to implement the data generation logic.

This eliminates the need for the Clock → Counter → Generator pattern by combining the Counter functionality into the generator base class.

Subclasses must implement:

• _reset_state(time_axis): Initialize any state needed for production

• _produce(n_samples, time_axis): Generate the actual output data

Example:

@processor_state
class SinState(ClockDrivenState):
    ang_freq: float = 0.0

class SinProducer(BaseClockDrivenProducer[SinSettings, SinState]):
    def _reset_state(self, time_axis: AxisArray.TimeAxis) -> None:
        self._state.ang_freq = 2 * np.pi * self.settings.fs

    def _produce(self, n_samples: int, time_axis: AxisArray.TimeAxis) -> AxisArray:
        t = (np.arange(n_samples) + self._state.counter) * time_axis.gain
        data = np.sin(self._state.ang_freq * t)
        return AxisArray(data=data, dims=["time"], axes={"time": time_axis})