# SPDX-FileCopyrightText: 2020-2026 ReproNim Team <info@repronim.org>
#
# SPDX-License-Identifier: MIT
"""
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.
"""
import logging
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Generator, List, Optional
import jsonlines
import pandas
from pydantic import BaseModel, Field
# initialize the logger
logger = logging.getLogger(__name__)
# logging.getLogger().addHandler(logging.StreamHandler(sys.stderr))
logger.debug(f"name={__name__}")
# Default tmap JSONL filename bundled alongside this module.
TMAP_FILENAME: str = "repronim_tmap.jsonl"
[docs]
def parse_jsonl_gen(path: str) -> Generator[dict, None, None]:
"""Yield each object from a JSON Lines file as a :class:`dict`.
:param path: Path to the ``.jsonl`` file to read.
:type path: str
:returns: Generator yielding one parsed dict per line.
:rtype: Generator[dict, None, None]
"""
with jsonlines.open(path) as reader:
for obj in reader:
yield obj
[docs]
def str_isotime(v: datetime) -> Optional[str]:
"""Format a :class:`datetime` as an ISO 8601 string with microseconds.
:param v: Datetime to format, or any falsy value (e.g. ``None``).
:type v: datetime
:returns: Formatted string ``YYYY-MM-DDTHH:MM:SS.ffffff``, or ``None``
when *v* is falsy.
:rtype: Optional[str]
"""
if not v:
return None
return v.strftime("%Y-%m-%dT%H:%M:%S.%f")
# define ReproNim clocks
[docs]
class Clock(str, Enum):
"""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 :class:`TMapRecord`.
"""
ISOTIME = "isotime" # Reference NTP clock
BIRCH = "birch" # Birch clock
DICOMS = "dicoms" # DICOMs clock
PSYCHOPY = "psychopy" # psychopy clock
QRINFO = "qrinfo" # QRInfo clock
REPROEVENTS = "reproevents" # ReproEvents clock
# ReproStim video clock, by now the same as QRInfo clock
REPROSTIM_VIDEO = "reprostim_video"
# Define period data model
[docs]
class TPeriodData(BaseModel):
"""Inter-mark period statistics computed by calibration jobs.
Each instance covers the gap between two consecutive :class:`TMapRecord`
marks. The ``avg_period`` attribute on :class:`TMapService` holds a
weighted average over all valid periods in the session.
"""
key: Optional[str] = Field(None, description="Unique tmap key for back reference")
duration: Optional[float] = Field(
0.0, description="Reference period duration in seconds"
)
deviation: Optional[float] = Field(
1.0, description="Reference period deviation related to master clock"
)
dicoms_duration: Optional[float] = Field(
0.0, description="DICOMs period duration in seconds"
)
dicoms_deviation: Optional[float] = Field(
1.0, description="DICOMs period deviation related to master clock"
)
dicoms_valid: Optional[bool] = Field(
False,
description="Specified DICOMs is valid and corresponds "
"to expected deviations",
)
# Define abstract timing map model
[docs]
class TMapRecord(BaseModel):
"""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``.
"""
isotime: Optional[datetime] = Field(
None, description="Reference time bound to NTP in isotime format"
)
duration: Optional[float] = Field(
None, description="Reference duration of the mark in seconds"
)
session_id: Optional[str] = Field(
None,
description="Unique session identifier generated by "
"collect_data.sh script, e.g. ses-20240528",
)
mark_id: Optional[str] = Field(
None,
description="Unique mark identifier across session only, e.g. mark_000025",
)
mark_name: Optional[str] = Field(
None, description="Optional name or description of the mark"
)
dicoms_id: Optional[str] = Field(
None, description="DICOMs dump unique identifier, e.g. dicoms-000025"
)
dicoms_isotime: Optional[datetime] = Field(
None, description="Corresponding DICOMs clock time in isotime format"
)
dicoms_offset: Optional[float] = Field(
0.0, description="DICOMs offset in seconds from isotime"
)
dicoms_duration: Optional[float] = Field(
None, description="Duration measured in DICOMs clock in seconds"
)
dicoms_deviation: Optional[float] = Field(
0.0,
description="Represents DICOMs time deviation ratio comparing "
"to master clock",
)
birch_id: Optional[str] = Field(
None, description="Birch dump identifier, e.g. birch-000027"
)
birch_isotime: Optional[datetime] = Field(
None, description="Corresponding birch clock time in isotime format"
)
birch_offset: Optional[float] = Field(
0.0, description="birch offset in seconds from isotime"
)
birch_duration: Optional[float] = Field(
None, description="Duration measured in Birch clock in seconds"
)
birch_deviation: Optional[float] = Field(
0.0,
description="Represents birch time deviation ratio comparing "
"to master clock",
)
psychopy_id: Optional[str] = Field(
None, description="Psychopy dump identifier, e.g. psychopy-000017"
)
psychopy_isotime: Optional[datetime] = Field(
None, description="Corresponding psychopy clock time in isotime format"
)
psychopy_offset: Optional[float] = Field(
0.0, description="psychopy offset in seconds from isotime"
)
psychopy_duration: Optional[float] = Field(
None, description="Duration measured in psychopy clock in seconds"
)
psychopy_deviation: Optional[float] = Field(
0.0,
description="Represents psychopy time deviation ratio "
"comparing to master clock",
)
reproevents_id: Optional[str] = Field(
None, description="ReproEvents dump identifier, e.g. reproevents-000012"
)
reproevents_isotime: Optional[datetime] = Field(
None, description="Corresponding ReproEvents clock time in isotime format"
)
reproevents_offset: Optional[float] = Field(
0.0, description="ReproEvents offset in seconds from isotime"
)
reproevents_duration: Optional[float] = Field(
None, description="Duration measured in ReproEvents clock in seconds"
)
reproevents_deviation: Optional[float] = Field(
0.0,
description="Represents ReproEvents time deviation ratio "
"comparing to master clock",
)
qrinfo_id: Optional[str] = Field(
None, description="QRInfo dump identifier, e.g. qrinfo-000125"
)
reprostim_video_isotime: Optional[datetime] = Field(
None,
description="Corresponding ReproStim video clock time in isotime format",
)
reprostim_video_offset: Optional[float] = Field(
0.0, description="ReproStim video offset in seconds from isotime"
)
reprostim_video_duration: Optional[float] = Field(
None, description="Duration measured in ReproStim video clock in seconds"
)
reprostim_video_deviation: Optional[float] = Field(
0.0,
description="Represents ReproStim video time deviation ratio "
"comparing to master clock",
)
# find tmap offset by clock
[docs]
def get_tmap_offset(clock: Clock, tmap: TMapRecord) -> float:
"""Return the clock offset (seconds from isotime) stored in a tmap record.
:param clock: Clock whose offset to retrieve.
:type clock: Clock
:param tmap: Tmap record to read from.
:type tmap: TMapRecord
:returns: Offset in seconds; ``0.0`` for :attr:`Clock.ISOTIME`.
:rtype: float
:raises ValueError: If *clock* is not a recognized :class:`Clock` member.
"""
if clock == Clock.ISOTIME:
return 0.0
if clock == Clock.BIRCH:
return tmap.birch_offset
elif clock == Clock.DICOMS:
return tmap.dicoms_offset
elif clock == Clock.PSYCHOPY:
return tmap.psychopy_offset
elif clock == Clock.QRINFO:
return tmap.reprostim_video_offset
elif clock == Clock.REPROEVENTS:
return tmap.reproevents_offset
elif clock == Clock.REPROSTIM_VIDEO:
return tmap.reprostim_video_offset
else:
raise ValueError(f"Unknown clock: {clock}")
# find tmap isotime by clock
[docs]
def get_tmap_isotime(clock: Clock, tmap: TMapRecord) -> datetime:
"""Return the isotime for the given clock stored in a tmap record.
:param clock: Clock whose isotime to retrieve.
:type clock: Clock
:param tmap: Tmap record to read from.
:type tmap: TMapRecord
:returns: Datetime value for the requested clock domain.
:rtype: datetime
:raises ValueError: If *clock* is not a recognized :class:`Clock` member.
"""
if clock == Clock.ISOTIME:
return tmap.isotime
if clock == Clock.BIRCH:
return tmap.birch_isotime
elif clock == Clock.DICOMS:
return tmap.dicoms_isotime
elif clock == Clock.PSYCHOPY:
return tmap.psychopy_isotime
elif clock == Clock.QRINFO:
return tmap.reprostim_video_isotime
elif clock == Clock.REPROEVENTS:
return tmap.reproevents_isotime
elif clock == Clock.REPROSTIM_VIDEO:
return tmap.reprostim_video_isotime
else:
raise ValueError(f"Unknown clock: {clock}")
# find tmap deviation by clock
[docs]
def 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.
:param clock: Clock whose deviation to retrieve.
:type clock: Clock
:param tmap: Tmap record to read from.
:type tmap: TMapRecord
:returns: Deviation ratio; ``1.0`` for :attr:`Clock.ISOTIME`.
:rtype: float
:raises ValueError: If *clock* is not a recognized :class:`Clock` member.
"""
if clock == Clock.ISOTIME:
return 1.0
if clock == Clock.BIRCH:
return tmap.birch_deviation
elif clock == Clock.DICOMS:
return tmap.dicoms_deviation
elif clock == Clock.PSYCHOPY:
return tmap.psychopy_deviation
elif clock == Clock.QRINFO:
return tmap.reprostim_video_deviation
elif clock == Clock.REPROEVENTS:
return tmap.reproevents_deviation
elif clock == Clock.REPROSTIM_VIDEO:
return tmap.reprostim_video_deviation
else:
raise ValueError(f"Unknown clock: {clock}")
# get tmap unique key
[docs]
def 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
:attr:`TMapService.periods`.
:param tmap: Tmap record to derive the key from.
:type tmap: TMapRecord
:returns: Composite key string.
:rtype: str
"""
return f"{tmap.session_id}|{tmap.mark_id}"
# Define ReproNim timing map service
[docs]
class TMapService:
"""Service that loads, indexes, and queries a ReproNim timing map.
On construction (or via :meth:`load`) the service ingests a JSONL tmap
file or a pre-parsed list of record dicts, sorts marks by isotime, and
calls :meth:`calc_periods` to build per-interval statistics. The primary
public entry point is :meth:`convert`, which translates a datetime from
one :class:`Clock` domain to another using the nearest preceding mark as
the calibration anchor.
"""
def __init__(self, path_or_marks: str | List = None):
"""Initialise the service, optionally loading marks immediately.
:param path_or_marks: Path to a ``.jsonl`` tmap file, or a list of
dicts (each representing one :class:`TMapRecord`). When ``None``
the service starts empty; call :meth:`load` later.
:type path_or_marks: str or list or None
"""
self.marks = []
self.periods = {}
self.avg_period = TPeriodData()
self._force_offset = {}
if path_or_marks:
self.load(path_or_marks)
# adjust clock offset using correction
[docs]
def adjust_offset(
self, offset: float, clock: Clock, dt: datetime, tmap: TMapRecord
) -> float:
"""Apply a drift correction to *offset* for the DICOMS clock.
For clocks other than :attr:`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``.
:param offset: Raw clock offset in seconds from the tmap anchor.
:type offset: float
:param clock: Clock domain of the offset being adjusted.
:type clock: Clock
:param dt: Datetime being converted (used to compute elapsed time).
:type dt: datetime
:param tmap: Anchor tmap record for this conversion.
:type tmap: TMapRecord
:returns: Adjusted offset in seconds.
:rtype: float
"""
# limit to DICOMs clock only atm
if clock != Clock.DICOMS:
return offset
# skip correction when offset manually is specified/hardcoded
if self._force_offset.get(clock.value) is not None:
return offset
tp: TPeriodData = self.get_period(tmap)
if not tp:
logger.debug("use avg period")
tp = self.avg_period
# for invalid periods use average deviation
# but in future this can be tuned up on demand
if not tp.dicoms_valid:
logger.debug("use avg period instead of invalid one")
tp = self.avg_period
# delta sec
d: float = (dt - tmap.isotime).total_seconds()
correction: float = d * tp.dicoms_deviation - d
adjusted_offset: float = offset + correction
return adjusted_offset
# calculate inter-series periods based on sequential
# and sorted marks data
[docs]
def calc_periods(self):
"""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 :attr:`avg_period`.
Called automatically by :meth:`load` after sorting.
"""
ap: TPeriodData = TPeriodData(
duration=0.0,
deviation=1.0,
dicoms_duration=0.0,
dicoms_deviation=1.0,
dicoms_valid=True,
)
prev_mark: TMapRecord = None
for mark in self.marks:
if prev_mark:
tp: TPeriodData = TPeriodData()
tp.key = get_tmap_key(prev_mark)
tp.duration = (mark.isotime - prev_mark.isotime).total_seconds()
tp.deviation = 1.0
tp.dicoms_duration = (
mark.dicoms_isotime - prev_mark.dicoms_isotime
).total_seconds()
if tp.duration and tp.duration != 0:
tp.dicoms_deviation = tp.dicoms_duration / tp.duration
expected_offset: float = (
prev_mark.dicoms_offset
+ tp.duration * tp.dicoms_deviation
- tp.dicoms_duration
)
offset_diff: float = expected_offset - mark.dicoms_offset
# detected clock correction, mark period as invalid
# tune this 30 sec interval later
tp.dicoms_valid = True if abs(offset_diff) < 30.0 else False
# logger.debug(f"expected_offset={expected_offset} /
# real={mark.dicoms_offset},
# diff={offset_diff}, duration={tp.duration},
# delta={tp.duration * tp.dicoms_deviation}")
# logger.debug(f"DELTA = {1000*(tp.dicoms_deviation-
# mark.dicoms_deviation)}")
self.periods[tp.key] = tp
# for valid periods calculate global average deviation
# where each valid deviation added proportionally to the
# period duration
if tp.dicoms_valid:
ap.duration += tp.duration
ap.dicoms_duration += tp.dicoms_duration
ap.dicoms_deviation += (
tp.dicoms_duration / 100
) * tp.dicoms_deviation
prev_mark = mark
if ap.dicoms_duration != 0:
ap.dicoms_deviation /= ap.dicoms_duration / 100
ap.duration = round(ap.duration, 1)
ap.dicoms_duration = round(ap.dicoms_duration, 1)
self.avg_period = ap
# convert datetime from one ReproNim clock to another
[docs]
def convert(
self, 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.
:param from_clock: Source clock domain.
:type from_clock: Clock
:param to_clock: Target clock domain.
:type to_clock: Clock
:param from_dt: Datetime in the *from_clock* domain to convert.
:type from_dt: datetime
: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.
:rtype: datetime
"""
# bypass conversion if clocks are the same
if from_clock == to_clock:
return from_dt
# skip empty datetime
if not from_dt:
return None
tmap: TMapRecord = self.find_tmap(from_clock, from_dt)
# bypass conversion if tmap is not found
if tmap is None:
logger.warning(f"tmap not found for {from_dt}")
return from_dt
# calculate offset
from_offset: float = self.get_offset(from_clock, tmap)
from_offset = self.adjust_offset(from_offset, from_clock, from_dt, tmap)
to_offset: float = self.get_offset(to_clock, tmap)
to_offset = self.adjust_offset(to_offset, to_clock, from_dt, tmap)
logger.debug(f"from_offset={from_offset}, to_offset={to_offset}")
offset: float = to_offset - from_offset
logger.debug(f"offset={offset}")
return from_dt + pandas.Timedelta(offset, unit="s")
# for debug purposes report tmap table, calculated periods and
# global average periods if any
[docs]
def dump_periods(self):
"""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.
"""
for i, m in enumerate(self.marks):
p: TPeriodData = self.get_period(m)
logger.info(f"[{i:03}] mark : {m.model_dump_json()}")
logger.info(f"[{i:03}] period : {p.model_dump_json() if p else None}")
logger.info(f"avg period : {self.avg_period.model_dump_json()}")
# find tmap record by datetime and clock in sorted
# list of marks
[docs]
def find_tmap(self, clock: Clock, dt: datetime) -> TMapRecord:
"""Find the last mark whose clock time is ≤ *dt*.
Performs a linear scan over the sorted :attr:`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.
:param clock: Clock domain to compare against *dt*.
:type clock: Clock
:param dt: Datetime to look up.
:type dt: datetime
:returns: Best-matching :class:`TMapRecord`, or ``None``.
:rtype: TMapRecord
"""
if not self.marks or len(self.marks) == 0:
return None
if len(self.marks) == 1:
return self.marks[0]
last_mark = self.marks[0]
for mark in self.marks:
# lookup based on provided clock using appropriate
# ***_isotime field
if get_tmap_isotime(clock, mark) > dt:
break
last_mark = mark
return last_mark
# force offset for certain clock
[docs]
def force_offset(self, 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 :meth:`get_offset` call on this clock.
Passing ``None`` removes a previously set override.
:param clock: Clock key string (e.g. ``"dicoms"``).
:type clock: str
:param offset: Fixed offset in seconds, or ``None`` to clear.
:type offset: float or None
"""
if offset is None:
if clock in self._force_offset:
del self._force_offset[clock]
else:
logger.debug("forcing offset for %s: %f", clock, offset)
self._force_offset[clock] = offset
# override clock offset if any
[docs]
def get_offset(self, clock: Clock, tmap: TMapRecord) -> float:
"""Return the effective offset for *clock*, respecting forced overrides.
If a forced offset was set via :meth:`force_offset`, that value is
returned; otherwise delegates to :func:`get_tmap_offset`.
:param clock: Clock domain.
:type clock: Clock
:param tmap: Tmap record to read from when no override is active.
:type tmap: TMapRecord
:returns: Effective offset in seconds.
:rtype: float
"""
if self._force_offset.get(clock.value) is not None:
logger.debug(
"overriding offset for %s: %f", clock, self._force_offset[clock.value]
)
return self._force_offset[clock.value]
return get_tmap_offset(clock, tmap)
# get period by tmap/mark
[docs]
def get_period(self, tmap: TMapRecord) -> Optional[TPeriodData]:
"""Return the :class:`TPeriodData` for the interval starting at *tmap*.
:param tmap: Tmap record whose key to look up.
:type tmap: TMapRecord
:returns: Period data if the key is present, ``None`` otherwise.
:rtype: Optional[TPeriodData]
"""
key: str = get_tmap_key(tmap)
return self.periods.get(key)
# load marks from file
[docs]
def load(self, path_or_marks: str | List):
"""Load tmap marks from a JSONL file path or a list of dicts.
Appends parsed :class:`TMapRecord` objects to :attr:`marks`, sorts
the full list by :attr:`~TMapRecord.isotime`, and recomputes periods
via :meth:`calc_periods`.
:param path_or_marks: JSONL file path, or list of dicts each
representing one :class:`TMapRecord`.
:type path_or_marks: str or list
"""
if isinstance(path_or_marks, str):
for obj in parse_jsonl_gen(path_or_marks):
self.marks.append(TMapRecord(**obj))
else:
for obj in path_or_marks:
self.marks.append(TMapRecord(**obj))
# sort by isotime
self.marks.sort(key=lambda x: x.isotime)
self.calc_periods()
[docs]
def to_label(self) -> 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.
:rtype: str
"""
# dump number of marks and each mark in format [N]=isotime
if len(self.marks) == 0:
return "TMap is empty"
return f"TMap marks count {len(self.marks)} : " + ", ".join(
[f"[{i}]={str_isotime(mark.isotime)}" for i, mark in enumerate(self.marks)]
)
_tmap_svc: TMapService = None
[docs]
def get_tmap_svc() -> TMapService:
"""Return the module-level :class:`TMapService` singleton.
Loads the tmap JSONL file on first call and caches the result.
When :data:`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 :data:`TMAP_FILENAME` to a fixture absolute path.
:returns: Initialised :class:`TMapService` instance.
:rtype: TMapService
"""
global _tmap_svc
if not _tmap_svc:
p = Path(TMAP_FILENAME)
if p.is_absolute():
path_tmap: str = str(p)
else:
cwd_candidate = Path.cwd() / TMAP_FILENAME
path_tmap: str = str(
cwd_candidate
if cwd_candidate.exists()
else Path(__file__).with_name(TMAP_FILENAME)
)
logger.info(f"Loading tmap : {path_tmap}")
_tmap_svc = TMapService(path_or_marks=path_tmap)
logger.info(f" : {_tmap_svc.to_label()}")
return _tmap_svc
