ezmsg.baseproc.clock#

Clock generator for timing control.

Classes

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

Bases: BaseProducerUnit[ClockSettings, LinearAxis, ClockProducer]

Clock unit that produces ticks at a specified rate.

Output is a AxisArray.LinearAxis with: - gain: 1/dispatch_rate (seconds per tick) - offset: Wall clock timestamp

Parameters:: settings (Settings | None)

SETTINGS#: alias of ClockSettings

class ClockProducer(*args, **kwargs)[source]#

Bases: BaseStatefulProducer[ClockSettings, LinearAxis, ClockState]

Produces clock ticks at a specified rate.

Each tick outputs a AxisArray.LinearAxis containing: - gain: 1/dispatch_rate (seconds per tick), or 0.0 if dispatch_rate is infinite - offset: Wall clock timestamp (time.monotonic)

This output type allows downstream components (like Counter) to know both the timing of the tick and the nominal dispatch rate.

class ClockSettings(dispatch_rate=None)[source]#

Bases: Settings

Settings for ClockProducer.

Parameters:: dispatch_rate (float | None)

dispatch_rate: float | None = None#: Dispatch rate in Hz. - Finite value (e.g., 100.0): Dispatch 100 times per second. - None (default): Dispatch as fast as possible (no sleep). math.inf is

also accepted as an “unthrottled” sentinel for backward compatibility, but None is preferred (math.inf is not JSON-serializable, which breaks tools that reflect the settings schema, e.g. the ezmsg dashboard).

__init__(dispatch_rate=None)#

Parameters:: dispatch_rate (float | None)
Return type:: None

class ClockState[source]#

Bases: object

State for ClockProducer.

t_0: float#: Start time (monotonic).

n_dispatch: int = 0#: Number of dispatches since reset.