ezmsg.blackrock.clock#

Device-clock → host-clock conversion for CereLink streams.

Two things live here:

  • device_to_monotonic_offset() — the small, reusable conversion that the signal and spike producers reach through their _produce methods. It wraps pycbsdk.Session.device_to_monotonic() and leaves the no-sync fallback policy to the caller.

  • CbtimeToMonotonic / CbtimeToMonotonicTransformer — a standalone ezmsg unit that re-stamps a device-time AxisArray (one produced by a source with cbtime=True, whose time axis offset is device-ns / 1e9) onto time.monotonic(). It opens its own read-only client Session to the same device — CereLink’s shared memory lets a second session attach to a device a source already opened (same host).

Functions

device_to_monotonic_batch_offsets(session, device_ns, *, stream_id=-1)[source]#

Convert a whole batch of device timestamps to time.monotonic() seconds.

Like device_to_monotonic_offset() but for an iterable of device timestamps, returning None (instead of raising) when no clock sync is available yet.

Converting the full batch (rather than only its first sample) matters for monotonicity: the library advances the per-stream floor to the last converted value, so the next batch’s start is clamped to be >= this batch’s end. Converting only the first sample would protect batch-start offsets but leave the reconstructed per-sample grid free to step backward at a batch boundary when the clock offset slews between batches.

Parameters:

  • session (Session)

  • stream_id (int)

Return type:

list[float] | None

device_to_monotonic_offset(session, device_ns, *, stream_id=-1)[source]#

Convert a device timestamp (ns) to a time.monotonic() offset in seconds.

Thin wrapper over pycbsdk.Session.device_to_monotonic() that returns None instead of raising when no clock sync is available yet, leaving the fallback policy to the caller.

stream_id >= 0 engages the library’s per-stream monotonicity enforcement (host time kept non-decreasing, with a floor reset across a genuine clock re-sync); -1 is a stateless conversion. Use a stable id per logical stream — the scalar call routes through device_to_monotonic_batch under the hood, so a stable id is all that’s needed for enforcement.

Parameters:

  • session (Session)

  • device_ns (int)

  • stream_id (int)

Return type:

float | None

Classes

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

Bases: BaseTransformerUnit[CbtimeToMonotonicSettings, AxisArray, AxisArray, CbtimeToMonotonicTransformer]

ezmsg Unit wrapping CbtimeToMonotonicTransformer.

Parameters:

settings (Settings | None)

SETTINGS#

alias of CbtimeToMonotonicSettings

shutdown()[source]#

Runs when the Unit terminates.

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 clean up resources and perform any necessary shutdown procedures.

Return type:

None

class CbtimeToMonotonicSettings(device_type=None, stream_id=0)[source]#

Bases: Settings

Settings for CbtimeToMonotonic.

Input contract: the time axis offset is a device timestamp in seconds (device-ns / 1e9), as emitted by CereLinkSignalSource / CereLinkSpikeSource with cbtime=True.

Parameters:
device_type: DeviceType | None = None#

Device whose clock to attach to. None = idle (pure passthrough).

stream_id: int = 0#

Per-stream monotonicity key. >= 0 enforces a non-decreasing host timeline (each transformer owns its own Session, so the id namespace is private); -1 is a stateless conversion.

__init__(device_type=None, stream_id=0)#
Parameters:
Return type:

None

class CbtimeToMonotonicState[source]#

Bases: object

session: Session | None = None#
class CbtimeToMonotonicTransformer(*args, **kwargs)[source]#

Bases: BaseStatefulTransformer[CbtimeToMonotonicSettings, AxisArray, AxisArray, CbtimeToMonotonicState]

Re-stamp a device-time AxisArray onto time.monotonic().

The client Session is opened once, on the first message (default _hash_message returns 0). If it fails to open, or clock sync isn’t available yet, messages pass through unchanged (still device time).

close()[source]#

Synchronous teardown — for unit shutdown().

Return type:

None

device_to_monotonic_offset(session, device_ns, *, stream_id=-1)[source]#

Convert a device timestamp (ns) to a time.monotonic() offset in seconds.

Thin wrapper over pycbsdk.Session.device_to_monotonic() that returns None instead of raising when no clock sync is available yet, leaving the fallback policy to the caller.

stream_id >= 0 engages the library’s per-stream monotonicity enforcement (host time kept non-decreasing, with a floor reset across a genuine clock re-sync); -1 is a stateless conversion. Use a stable id per logical stream — the scalar call routes through device_to_monotonic_batch under the hood, so a stable id is all that’s needed for enforcement.

Parameters:

  • session (Session)

  • device_ns (int)

  • stream_id (int)

Return type:

float | None

device_to_monotonic_batch_offsets(session, device_ns, *, stream_id=-1)[source]#

Convert a whole batch of device timestamps to time.monotonic() seconds.

Like device_to_monotonic_offset() but for an iterable of device timestamps, returning None (instead of raising) when no clock sync is available yet.

Converting the full batch (rather than only its first sample) matters for monotonicity: the library advances the per-stream floor to the last converted value, so the next batch’s start is clamped to be >= this batch’s end. Converting only the first sample would protect batch-start offsets but leave the reconstructed per-sample grid free to step backward at a batch boundary when the clock offset slews between batches.

Parameters:

  • session (Session)

  • stream_id (int)

Return type:

list[float] | None

class CbtimeToMonotonicSettings(device_type=None, stream_id=0)[source]#

Bases: Settings

Settings for CbtimeToMonotonic.

Input contract: the time axis offset is a device timestamp in seconds (device-ns / 1e9), as emitted by CereLinkSignalSource / CereLinkSpikeSource with cbtime=True.

Parameters:
device_type: DeviceType | None = None#

Device whose clock to attach to. None = idle (pure passthrough).

stream_id: int = 0#

Per-stream monotonicity key. >= 0 enforces a non-decreasing host timeline (each transformer owns its own Session, so the id namespace is private); -1 is a stateless conversion.

__init__(device_type=None, stream_id=0)#
Parameters:
Return type:

None

class CbtimeToMonotonicState[source]#

Bases: object

session: Session | None = None#
class CbtimeToMonotonicTransformer(*args, **kwargs)[source]#

Bases: BaseStatefulTransformer[CbtimeToMonotonicSettings, AxisArray, AxisArray, CbtimeToMonotonicState]

Re-stamp a device-time AxisArray onto time.monotonic().

The client Session is opened once, on the first message (default _hash_message returns 0). If it fails to open, or clock sync isn’t available yet, messages pass through unchanged (still device time).

close()[source]#

Synchronous teardown — for unit shutdown().

Return type:

None

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

Bases: BaseTransformerUnit[CbtimeToMonotonicSettings, AxisArray, AxisArray, CbtimeToMonotonicTransformer]

ezmsg Unit wrapping CbtimeToMonotonicTransformer.

Parameters:

settings (Settings | None)

SETTINGS#

alias of CbtimeToMonotonicSettings

shutdown()[source]#

Runs when the Unit terminates.

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 clean up resources and perform any necessary shutdown procedures.

Return type:

None