reprostim.qr.timing

ReproNim timing-map (tmap) subsystem. Provides tools to load, index, and query timing-map (tmap) records that anchor various device clocks (swimlines) to a single NTP reference (isotime). It implements models for tmap marks and inter-mark statistics, an enum of supported clock domains, and a service class for conversions between clock domains using recorded synchronization marks.

Functions

get_tmap_deviation(clock, tmap)	Return the deviation ratio for the given clock stored in a tmap record.
get_tmap_isotime(clock, tmap)	Return the isotime for the given clock stored in a tmap record.
get_tmap_key(tmap)	Return the unique string key for a tmap record.
get_tmap_offset(clock, tmap)	Return the clock offset (seconds from isotime) stored in a tmap record.
get_tmap_svc()	Return the module-level TMapService singleton.
parse_jsonl_gen(path)	Yield each object from a JSON Lines file as a dict.
str_isotime(v)	Format a datetime as an ISO 8601 string with microseconds.

Classes

Clock(value)	Enumeration of all ReproNim device clocks.
TMapRecord(*[, isotime, duration, ...])	A single timing-map mark anchoring all device clocks to NTP isotime.
TMapService([path_or_marks])	Service that loads, indexes, and queries a ReproNim timing map.
TPeriodData(*[, key, duration, deviation, ...])	Inter-mark period statistics computed by calibration jobs.

class reprostim.qr.timing.Clock(value)

Enumeration of all ReproNim device clocks.

Each value is a string key used to identify the clock domain in tmap records and conversion calls. ISOTIME is the NTP reference clock; all other clocks are expressed as offsets relative to it. QRINFO and REPROSTIM_VIDEO currently share the same underlying field (reprostim_video_*) in TMapRecord.

BIRCH = 'birch'

DICOMS = 'dicoms'

ISOTIME = 'isotime'

PSYCHOPY = 'psychopy'

QRINFO = 'qrinfo'

REPROEVENTS = 'reproevents'

REPROSTIM_VIDEO = 'reprostim_video'

class reprostim.qr.timing.TMapRecord(*, isotime: datetime | None = None, duration: float | None = None, session_id: str | None = None, mark_id: str | None = None, mark_name: str | None = None, dicoms_id: str | None = None, dicoms_isotime: datetime | None = None, dicoms_offset: float | None = 0.0, dicoms_duration: float | None = None, dicoms_deviation: float | None = 0.0, birch_id: str | None = None, birch_isotime: datetime | None = None, birch_offset: float | None = 0.0, birch_duration: float | None = None, birch_deviation: float | None = 0.0, psychopy_id: str | None = None, psychopy_isotime: datetime | None = None, psychopy_offset: float | None = 0.0, psychopy_duration: float | None = None, psychopy_deviation: float | None = 0.0, reproevents_id: str | None = None, reproevents_isotime: datetime | None = None, reproevents_offset: float | None = 0.0, reproevents_duration: float | None = None, reproevents_deviation: float | None = 0.0, qrinfo_id: str | None = None, reprostim_video_isotime: datetime | None = None, reprostim_video_offset: float | None = 0.0, reprostim_video_duration: float | None = None, reprostim_video_deviation: float | None = 0.0)

A single timing-map mark anchoring all device clocks to NTP isotime.

Each record represents one synchronization event captured by calibrartion jobs. Per-device fields follow the pattern <device>_isotime, <device>_offset, <device>_duration, and <device>_deviation; absent devices are None / 0.0.

birch_deviation: float | None

birch_duration: float | None

birch_id: str | None

birch_isotime: datetime | None

birch_offset: float | None

dicoms_deviation: float | None

dicoms_duration: float | None

dicoms_id: str | None

dicoms_isotime: datetime | None

dicoms_offset: float | None

duration: float | None

isotime: datetime | None

mark_id: str | None

mark_name: str | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

psychopy_deviation: float | None

psychopy_duration: float | None

psychopy_id: str | None

psychopy_isotime: datetime | None

psychopy_offset: float | None

qrinfo_id: str | None

reproevents_deviation: float | None

reproevents_duration: float | None

reproevents_id: str | None

reproevents_isotime: datetime | None

reproevents_offset: float | None

reprostim_video_deviation: float | None

reprostim_video_duration: float | None

reprostim_video_isotime: datetime | None

reprostim_video_offset: float | None

session_id: str | None

class reprostim.qr.timing.TMapService(path_or_marks: str | List | None = None)

Service that loads, indexes, and queries a ReproNim timing map.

On construction (or via load()) the service ingests a JSONL tmap file or a pre-parsed list of record dicts, sorts marks by isotime, and calls calc_periods() to build per-interval statistics. The primary public entry point is convert(), which translates a datetime from one Clock domain to another using the nearest preceding mark as the calibration anchor.

adjust_offset(offset: float, clock: Clock, dt: datetime, tmap: TMapRecord) → float

Apply a drift correction to offset for the DICOMS clock.

For clocks other than Clock.DICOMS, or when a forced offset is active, the original offset is returned unchanged. Otherwise the correction is computed from the period deviation between tmap and dt: correction = d * deviation - d where d is the elapsed seconds since tmap.isotime.

Parameters:

• offset (float) – Raw clock offset in seconds from the tmap anchor.

• clock (Clock) – Clock domain of the offset being adjusted.

• dt (datetime) – Datetime being converted (used to compute elapsed time).

• tmap (TMapRecord) – Anchor tmap record for this conversion.

Returns:

Adjusted offset in seconds.

Return type:

float

calc_periods()

Compute inter-mark period statistics from the sorted marks list.

Iterates consecutive mark pairs, computing duration, dicoms_duration, and dicoms_deviation for each interval. Periods where the observed DICOMS offset deviates by more than 30 s from the expected value are flagged as invalid (likely clock correction events). Valid periods contribute to a weighted average stored in avg_period.

Called automatically by load() after sorting.

convert(from_clock: Clock, to_clock: Clock, from_dt: datetime) → datetime

Convert a datetime from one ReproNim clock domain to another.

Finds the nearest preceding tmap mark for from_dt, computes the net offset between from_clock and to_clock (with drift correction applied for DICOMS), and returns the shifted datetime.

When no tmap data is available or no preceding mark exists, the original from_dt is returned with a warning logged.

Parameters:

• from_clock (Clock) – Source clock domain.

• to_clock (Clock) – Target clock domain.

• from_dt (datetime) – Datetime in the from_clock domain to convert.

Returns:

Converted datetime in the to_clock domain, or None when from_dt is falsy, or from_dt unchanged when no anchor mark is found.

Return type:

datetime

dump_periods()

Log all marks, their associated periods, and the average period.

Emits one INFO line per mark and per period (as JSON), followed by the weighted average period. Intended for debugging.

find_tmap(clock: Clock, dt: datetime) → TMapRecord

Find the last mark whose clock time is ≤ dt.

Performs a linear scan over the sorted marks list using the appropriate *_isotime field for clock. Returns the first mark when there is only one, or None when the list is empty.

Parameters:

• clock (Clock) – Clock domain to compare against dt.

• dt (datetime) – Datetime to look up.

Returns:

Best-matching TMapRecord, or None.

Return type:

TMapRecord

force_offset(clock: str, offset: float)

Pin or clear a manual offset override for clock.

When offset is not None the value overrides the tmap-derived offset for every subsequent get_offset() call on this clock. Passing None removes a previously set override.

Parameters:

• clock (str) – Clock key string (e.g. "dicoms").

• offset (float or None) – Fixed offset in seconds, or None to clear.

get_offset(clock: Clock, tmap: TMapRecord) → float

Return the effective offset for clock, respecting forced overrides.

If a forced offset was set via force_offset(), that value is returned; otherwise delegates to get_tmap_offset().

Parameters:

• clock (Clock) – Clock domain.

• tmap (TMapRecord) – Tmap record to read from when no override is active.

Returns:

Effective offset in seconds.

Return type:

float

get_period(tmap: TMapRecord) → TPeriodData | None

Return the TPeriodData for the interval starting at tmap.

Parameters:

tmap (TMapRecord) – Tmap record whose key to look up.

Returns:

Period data if the key is present, None otherwise.

Return type:

Optional[TPeriodData]

load(path_or_marks: str | List)

Load tmap marks from a JSONL file path or a list of dicts.

Appends parsed TMapRecord objects to marks, sorts the full list by isotime, and recomputes periods via calc_periods().

Parameters:

path_or_marks (str or list) – JSONL file path, or list of dicts each representing one TMapRecord.

to_label() → str

Return a compact human-readable summary of the loaded marks.

Returns:

String of the form "TMap marks count N : [0]=<isotime>, …" or "TMap is empty" when no marks are loaded.

Return type:

str

class reprostim.qr.timing.TPeriodData(*, key: str | None = None, duration: float | None = 0.0, deviation: float | None = 1.0, dicoms_duration: float | None = 0.0, dicoms_deviation: float | None = 1.0, dicoms_valid: bool | None = False)

Inter-mark period statistics computed by calibration jobs.

Each instance covers the gap between two consecutive TMapRecord marks. The avg_period attribute on TMapService holds a weighted average over all valid periods in the session.

deviation: float | None

dicoms_deviation: float | None

dicoms_duration: float | None

dicoms_valid: bool | None

duration: float | None

key: str | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

reprostim.qr.timing.get_tmap_deviation(clock: Clock, tmap: TMapRecord) → float

Return the deviation ratio for the given clock stored in a tmap record.

A value of 1.0 means the clock runs at the same rate as the NTP reference.

Parameters:

• clock (Clock) – Clock whose deviation to retrieve.

• tmap (TMapRecord) – Tmap record to read from.

Returns:

Deviation ratio; 1.0 for Clock.ISOTIME.

Return type:

float

Raises:

ValueError – If clock is not a recognized Clock member.

reprostim.qr.timing.get_tmap_isotime(clock: Clock, tmap: TMapRecord) → datetime

Return the isotime for the given clock stored in a tmap record.

Parameters:

• clock (Clock) – Clock whose isotime to retrieve.

• tmap (TMapRecord) – Tmap record to read from.

Returns:

Datetime value for the requested clock domain.

Return type:

datetime

Raises:

ValueError – If clock is not a recognized Clock member.

reprostim.qr.timing.get_tmap_key(tmap: TMapRecord) → str

Return the unique string key for a tmap record.

The key is "<session_id>|<mark_id>" and is used as a dict key in TMapService.periods.

Parameters:

tmap (TMapRecord) – Tmap record to derive the key from.

Returns:

Composite key string.

Return type:

str

reprostim.qr.timing.get_tmap_offset(clock: Clock, tmap: TMapRecord) → float

Return the clock offset (seconds from isotime) stored in a tmap record.

Parameters:

• clock (Clock) – Clock whose offset to retrieve.

• tmap (TMapRecord) – Tmap record to read from.

Returns:

Offset in seconds; 0.0 for Clock.ISOTIME.

Return type:

float

Raises:

ValueError – If clock is not a recognized Clock member.

reprostim.qr.timing.get_tmap_svc() → TMapService

Return the module-level TMapService singleton.

Loads the tmap JSONL file on first call and caches the result. When TMAP_FILENAME is an absolute path it is used directly. For a plain filename the lookup order is: current working directory first, then the directory containing this module. Tests may monkeypatch TMAP_FILENAME to a fixture absolute path.

Returns:

Initialised TMapService instance.

Return type:

TMapService

reprostim.qr.timing.parse_jsonl_gen(path: str) → Generator[dict, None, None]

Yield each object from a JSON Lines file as a dict.

Parameters:

path (str) – Path to the .jsonl file to read.

Returns:

Generator yielding one parsed dict per line.

Return type:

Generator[dict, None, None]

reprostim.qr.timing.str_isotime(v: datetime) → str | None

Format a datetime as an ISO 8601 string with microseconds.

Parameters:

v (datetime) – Datetime to format, or any falsy value (e.g. None).

Returns:

Formatted string YYYY-MM-DDTHH:MM:SS.ffffff, or None when v is falsy.

Return type:

Optional[str]