Per-stream monotonic device→local time conversion

pycbsdk/pyproject.toml

@@ -49,6 +49,8 @@ lint = [
 test = [
     "pytest",
     "pytest-timeout",
+    "numpy",
+    "pylsl",
 ]
 
 [tool.setuptools.packages.find]
@@ -60,6 +62,7 @@ pycbsdk = ["*.dll", "*.so", "*.so.*", "*.dylib"]
 [tool.pytest.ini_options]
 markers = [
     "integration: tests requiring a running nPlayServer instance",
+    "lsl: nPlay+LSL tests; need an EXCLUSIVE nPlayServer (one shmem per device type). Auto-skipped when file-based nPlay tests are also collected — run with `pytest -m lsl`.",
 ]
 
 [tool.setuptools_scm]

pycbsdk/src/pycbsdk/_cdef.py

@@ -357,6 +357,10 @@
 cbsdk_result_t cbsdk_session_get_clock_offset(cbsdk_session_t session, int64_t* offset_ns);
 cbsdk_result_t cbsdk_session_get_clock_uncertainty(cbsdk_session_t session, int64_t* uncertainty_ns);
 cbsdk_result_t cbsdk_session_send_clock_probe(cbsdk_session_t session);
+cbsdk_result_t cbsdk_session_to_local_time(
+    cbsdk_session_t session, int64_t stream_id,
+    const uint64_t* device_ns, int64_t* out_steady_ns, size_t n);
+cbsdk_result_t cbsdk_session_reset_monotonic(cbsdk_session_t session, int64_t stream_id);
 
 // Utility
 int64_t cbsdk_get_steady_clock_ns(void);

pycbsdk/src/pycbsdk/session.py

@@ -1712,15 +1712,62 @@ def send_clock_probe(self):
             "Failed to send clock probe",
         )
 
-    def device_to_monotonic(self, device_time_ns: int) -> float:
+    def device_to_monotonic_batch(self, device_ns, stream_id: int = -1) -> list[float]:
+        """Convert a batch of device timestamps to ``time.monotonic()`` seconds.
+
+        The whole batch is converted inside the library against one consistent
+        clock snapshot (device_ns → steady_clock_ns), then the steady_clock →
+        monotonic offset is applied here.  This is the efficient path for
+        per-batch stamping and the only way to get monotonicity enforcement.
+
+        Args:
+            device_ns: Iterable of device timestamps in nanoseconds.
+            stream_id: Per-stream key controlling monotonicity.  When ``>= 0``,
+                each output is clamped to be non-decreasing relative to the
+                previous conversion request on the *same* ``stream_id`` — except
+                across a genuine clock re-sync, where the floor resets so the
+                timeline follows the new regime instead of stalling.  The
+                natural key is the sample group (e.g. one id for the 30 kHz raw
+                stream, another for 1 kHz); use a distinct id per derived stream.
+                ``-1`` (default) is a stateless conversion with no clamping —
+                use it for spikes and one-off lookups.  Monotonic streams assume
+                timestamps are submitted in intended-output order.
+
+        Returns:
+            List of ``time.monotonic()`` values in seconds, one per input.
+
+        Raises:
+            RuntimeError: If no clock sync data is available yet.
+        """
+        self._maybe_recalibrate_monotonic()
+        device_list = [int(t) for t in device_ns]
+        n = len(device_list)
+        if n == 0:
+            return []
+        inp = ffi.new("uint64_t[]", device_list)
+        out = ffi.new("int64_t[]", n)
+        result = _get_lib().cbsdk_session_to_local_time(
+            self._session, stream_id, inp, out, n
+        )
+        if result != 0:
+            raise RuntimeError("No clock sync data available")
+        k = self._mono_to_steady_offset_ns
+        return [(out[i] - k) / 1_000_000_000 for i in range(n)]
+
+    def device_to_monotonic(self, device_time_ns: int, stream_id: int = -1) -> float:
         """Convert a device timestamp to ``time.monotonic()`` seconds.
 
         Chains two offsets:
-        1. device_ns → steady_clock_ns  (via clock_offset_ns from device sync)
+        1. device_ns → steady_clock_ns  (via the device clock sync)
         2. steady_clock_ns → monotonic_ns  (via calibration at session creation)
 
         Args:
             device_time_ns: Device timestamp in nanoseconds (e.g., header.time).
+            stream_id: Per-stream monotonicity key; ``-1`` (default) is a
+                stateless conversion.  See :meth:`device_to_monotonic_batch` for
+                the monotonicity semantics; for a stream you want kept monotonic,
+                stamp each batch through that method (or pass a stable
+                ``stream_id`` here) rather than mixing ids.
 
         Returns:
             Corresponding ``time.monotonic()`` value in seconds.
@@ -1736,13 +1783,18 @@ def on_spike(header, data):
                 latency_ms = (time.monotonic() - t) * 1000
                 print(f"Spike latency: {latency_ms:.1f} ms")
         """
-        self._maybe_recalibrate_monotonic()
-        offset = self.clock_offset_ns
-        if offset is None:
-            raise RuntimeError("No clock sync data available")
-        steady_ns = device_time_ns - offset
-        mono_ns = steady_ns - self._mono_to_steady_offset_ns
-        return mono_ns / 1_000_000_000
+        return self.device_to_monotonic_batch([device_time_ns], stream_id)[0]
+
+    def reset_monotonic(self, stream_id: int) -> None:
+        """Drop the monotonic floor/epoch state for one ``stream_id``.
+
+        The next monotonic conversion on that stream starts fresh.  No-op if the
+        stream is unknown.
+        """
+        _check(
+            _get_lib().cbsdk_session_reset_monotonic(self._session, stream_id),
+            "Failed to reset monotonic state",
+        )
 
     # --- Commands ---
 

pycbsdk/tests/_lsl_generator.py

@@ -0,0 +1,72 @@
+"""Standalone 30 kHz int16 LSL stream generator for nPlay+LSL tests.
+
+Launched as a subprocess by the ``nplay_lsl`` fixture.  nPlayServer
+(``--device=LSL <name>``) resolves this stream by name and replays it onto the
+Cerebus protocol, deriving each packet's device timestamp from the stream's
+ORIGINATING time scaled to a 30 kHz sample counter.  We therefore pace pushes to
+real time so the originating timestamps advance at the true sample rate — the
+condition CereLink's clock sync must map to a monotonic host timeline.
+
+Usage:
+    python _lsl_generator.py <stream_name> [n_channels]
+
+Runs until killed (SIGTERM/SIGINT) by the fixture.
+"""
+
+import signal
+import sys
+import time
+
+import numpy as np
+import pylsl
+
+SRATE = 30000
+CHUNK_SAMPLES = 300  # 10 ms per push at 30 kHz
+
+
+def main() -> int:
+    name = sys.argv[1]
+    n_ch = int(sys.argv[2]) if len(sys.argv) > 2 else 8
+
+    info = pylsl.StreamInfo(
+        name=name,
+        type="EEG",
+        channel_count=n_ch,
+        nominal_srate=SRATE,
+        channel_format=pylsl.cf_int16,
+        source_id=f"{name}_src",
+    )
+    outlet = pylsl.StreamOutlet(info, chunk_size=CHUNK_SAMPLES, max_buffered=360)
+
+    # Exit cleanly on the fixture's terminate() so liblsl tears the outlet down.
+    running = {"go": True}
+
+    def _stop(*_a):
+        running["go"] = False
+
+    signal.signal(signal.SIGTERM, _stop)
+    signal.signal(signal.SIGINT, _stop)
+
+    # A slow sine across channels — content is irrelevant to the timing test,
+    # but non-constant data makes the stream easy to eyeball in a viewer.
+    phase = np.arange(CHUNK_SAMPLES, dtype=np.float64)
+    chan_offsets = np.linspace(0.0, np.pi, n_ch, dtype=np.float64)
+
+    t0 = time.monotonic()
+    idx = 0
+    while running["go"]:
+        angle = 2.0 * np.pi * (phase[:, None] + idx) / 3000.0 + chan_offsets[None, :]
+        chunk = (np.sin(angle) * 1000.0).astype(np.int16)
+        outlet.push_chunk(chunk)
+        idx += CHUNK_SAMPLES
+        # Pace to real time: sample idx should be emitted at t0 + idx/SRATE.
+        target = t0 + idx / SRATE
+        dt = target - time.monotonic()
+        if dt > 0:
+            time.sleep(dt)
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())