# Copyright (C) 2020-2026 Micah G Allen and the Embodied Computation Group, Aarhus University
"""A recorder that synthesises a pulse signal instead of reading hardware.
Presents the surface the tasks already use, so it drops into
``parameters["oxiTask"]``. ``systole.serialSim`` substitutes for the serial port
instead, one layer lower, which still runs in real time and replays a fixed
trace. Here the rate is chosen, ``realtime=False`` returns immediately, and
``artefact_every`` injects windows with no detectable beats.
"""
from typing import Dict, List, Optional, Sequence, Union
import numpy as np
#: Trough of the synthetic pulse. Real PPG has a substantial floor; a
#: near-zero one is what let the rolling threshold collapse between beats.
PULSE_BASELINE = 20.0
#: Height of the pulse above that floor. The Nonin reports 0-255, and
#: baseline plus amplitude stays inside it.
PULSE_AMPLITUDE = 200.0
#: Exponent on the raised cosine. Higher is a sharper systolic peak; the
#: waveform stays continuous at any value, which is the property that matters.
PULSE_SHARPNESS = 3.0
#: Standard deviation of the noise added to every sample.
PULSE_NOISE = 0.5
[docs]class ReplayRecorder:
"""Stand-in for ``systole.recording.Oximeter``.
Parameters
----------
bpm :
Heart rate to synthesise, in beats per minute. A sequence is consumed
one entry per ``read`` call, which lets a test drive the rate over the
course of a session.
sfreq :
Sampling frequency, matching the Nonin default the tasks assume.
realtime :
When ``False`` (the default here, unlike real hardware) ``read``
returns as soon as it has generated the samples, rather than waiting
for them to arrive.
add_channels :
Number of auxiliary channels, for trigger codes.
rng :
Generator for the noise, so a seeded session replays identically.
artefact_every :
When set, every Nth ``read`` produces a signal with no detectable
beats, to exercise the task's retry path.
"""
[docs] def __init__(
self,
bpm: Union[float, Sequence[float]] = 60.0,
sfreq: int = 75,
realtime: bool = False,
add_channels: int = 1,
rng: Optional[np.random.Generator] = None,
artefact_every: Optional[int] = None,
):
self.sfreq = sfreq
self.realtime = realtime
self.rng = rng if rng is not None else np.random.default_rng()
self.artefact_every = artefact_every
self._bpm_sequence = (
[float(bpm)] if np.isscalar(bpm) else [float(b) for b in bpm] # type: ignore[arg-type, union-attr]
)
self._read_count = 0
self._phase = 0.0
self.recording: List[float] = []
self.times: List[float] = []
self.peaks: List[int] = []
self.instant_rr: List[float] = []
self.channels: Dict[str, List[int]] = {
f"Channel_{i}": [] for i in range(add_channels)
}
self.n_saves = 0
self.saved_paths: List[str] = []
self._archived_codes: List[int] = []
# -- the surface the tasks use --------------------------------------
def setup(self) -> "ReplayRecorder":
"""Clear the buffers, as ``Oximeter.setup`` does.
The markers written into the auxiliary channel are archived first. The
real recorder is reset at every break, which discards every trigger
written since the last one, and a test asking "did all five codes get
written" would otherwise only ever see the final block's worth.
"""
self._archived_codes.extend(int(v) for v in self.channels["Channel_0"] if v)
self.recording, self.times, self.peaks, self.instant_rr = [], [], [], []
for key in self.channels:
self.channels[key] = []
self._phase = 0.0
return self
def read(self, duration: float = 1.0) -> "ReplayRecorder":
"""Append ``duration`` seconds of signal."""
self._read_count += 1
n = int(round(duration * self.sfreq))
artefact = (
self.artefact_every is not None
and self._read_count % self.artefact_every == 0
)
bpm = self._bpm_sequence[min(self._read_count - 1, len(self._bpm_sequence) - 1)]
signal, peaks = self._synthesise(n, bpm, artefact)
start = len(self.times) / self.sfreq
self.recording.extend(signal.tolist())
self.peaks.extend(peaks.tolist())
self.times.extend((start + np.arange(n) / self.sfreq).tolist())
for key in self.channels:
self.channels[key].extend([0] * n)
if self.realtime: # pragma: no cover - only used against real timing
import time as _time
_time.sleep(duration)
return self
def readInWaiting(self) -> None:
"""Append the handful of samples that would have arrived meanwhile.
The real recorder appends whatever whole packets are waiting on the
serial port. Appending a small fixed number here keeps the trigger
idiom ``channels["Channel_0"][-1] = n`` working, which is what the task
relies on.
"""
n = max(int(self.sfreq * 0.02), 1)
bpm = self._bpm_sequence[min(self._read_count - 1, len(self._bpm_sequence) - 1)]
signal, peaks = self._synthesise(n, bpm, False)
start = len(self.times) / self.sfreq
self.recording.extend(signal.tolist())
self.peaks.extend(peaks.tolist())
self.times.extend((start + np.arange(n) / self.sfreq).tolist())
for key in self.channels:
self.channels[key].extend([0] * n)
def save(self, path: str) -> None:
"""Record that a save happened, without writing megabytes in a test."""
self.n_saves += 1
self.saved_paths.append(str(path))
def close(self) -> None:
pass
# -- helpers ---------------------------------------------------------
def _synthesise(self, n: int, bpm: float, artefact: bool):
"""A pulse waveform this package's own detector can measure, and its
peak mask.
The waveform is continuous, because a flat baseline is what broke the
previous one. It was a narrow spike on a near-zero baseline, and
systole thresholds at the mean plus standard deviation of a rolling
one-second window: whenever that window fell between spikes it saw
only noise, the threshold collapsed to about 0.5, and noise maxima
were detected as beats. That happens exactly when the cardiac period
exceeds the window, so every rate at or below 60 BPM came back wrong
-- 42 measured as 90, 55 as 85, 60 as 75 -- while 72 and 100 were
exact. Every test used the default 72, so nothing caught it.
A raised cosine has substantial variance everywhere, so the rolling
threshold stays meaningful at any rate, and it has one maximum per
cycle at the cycle boundary, which is where the returned peak mask
marks it. Measured against ``ppg_peaks`` the recovered rate is within
0.05 BPM from 40 to 180.
"""
if artefact:
# Noise with no periodic component, so peak detection finds nothing
# usable and the task has to decide what to do about it.
return self.rng.normal(0, 1, n), np.zeros(n, dtype=int)
f = bpm / 60.0
t = self._phase + np.arange(n) / self.sfreq
self._phase = float(t[-1] + 1 / self.sfreq) if n else self._phase
cycle = (t * f) % 1.0
shape = ((1 + np.cos(2 * np.pi * cycle)) / 2) ** PULSE_SHARPNESS
signal = PULSE_BASELINE + PULSE_AMPLITUDE * shape
signal = signal + self.rng.normal(0, PULSE_NOISE, n)
peaks = (np.diff(np.floor(t * f), prepend=np.floor(t[0] * f)) > 0).astype(int)
return signal, peaks
# -- conveniences for tests ------------------------------------------
@property
def trigger_codes(self) -> List[int]:
"""Every marker written, including those cleared by a reset."""
return self._archived_codes + [int(v) for v in self.channels["Channel_0"] if v]