diff --git a/modules/common/__init__.py b/modules/common/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/modules/common/cbor_raw.py b/modules/common/cbor_raw.py new file mode 100644 index 00000000..1cc5d8fc --- /dev/null +++ b/modules/common/cbor_raw.py @@ -0,0 +1,123 @@ +""" +Shared CBOR record reader for SleepyPod biometrics modules. + +The cbor2 C extension (_cbor2) reads files in internal 4096-byte chunks, +so cbor2.load(f) advances f.tell() by 4096 bytes regardless of the actual +record size. Since RAW file records are typically 17-5000 bytes, this causes +nearly every record to be skipped silently. + +This module provides a manual parser for the outer {seq, data} CBOR wrapper +that reads byte-by-byte using f.read(), keeping f.tell() accurate. + +Protocol contract: Pod firmware always emits the outer map with keys in the +order "seq" then "data". The seq value is encoded as a fixed-width uint32 +(0x1a) by the embedded CBOR encoder, regardless of value. This parser +validates that contract and raises ValueError on deviation. + +See: https://github.com/throwaway31265/free-sleep/pull/46 +""" + +import struct + +# Reject implausibly large data payloads (corrupt length field protection) +_MAX_DATA_LENGTH = 1_000_000 + + +def read_raw_record(f): + """Parse one outer {seq, data} CBOR record from file object *f*. + + Returns the raw inner data bytes, or ``None`` for empty placeholder + records (Pod firmware writes ``data=b''`` as sequence-number markers). + + Raises: + EOFError: End of file or incomplete record (not yet fully written). + ValueError: Malformed CBOR structure. + """ + b = f.read(1) + if not b: + raise EOFError + if b[0] != 0xa2: + raise ValueError('Expected outer map 0xa2, got 0x%02x' % b[0]) + + # "seq" key — text(3) "seq" + # Pod firmware always emits keys in order: seq, then data. + seq_key = f.read(4) + if len(seq_key) < 4: + raise EOFError + if seq_key != b'\x63\x73\x65\x71': + raise ValueError('Expected seq key') + + # seq value — Pod firmware uses fixed-width uint32 (0x1a), but we accept + # all valid CBOR unsigned integer encodings for forward compatibility. + hdr = f.read(1) + if not hdr: + raise EOFError + mt = hdr[0] >> 5 + ai = hdr[0] & 0x1f + if mt != 0: + raise ValueError('seq must be unsigned int, got major type %d' % mt) + if ai <= 23: + pass # inline value, no additional bytes + elif ai == 24: + if len(f.read(1)) < 1: + raise EOFError + elif ai == 25: + if len(f.read(2)) < 2: + raise EOFError + elif ai == 26: + if len(f.read(4)) < 4: + raise EOFError + elif ai == 27: + if len(f.read(8)) < 8: + raise EOFError + else: + raise ValueError('Unexpected seq encoding: 0x%02x' % hdr[0]) + + # "data" key — text(4) "data" + data_key = f.read(5) + if len(data_key) < 5: + raise EOFError + if data_key != b'\x64\x64\x61\x74\x61': + raise ValueError('Expected data key') + + # data value — must be a byte string (major type 2) + bs = f.read(1) + if not bs: + raise EOFError + if bs[0] >> 5 != 2: + raise ValueError('data must be a byte string, got major type %d' % (bs[0] >> 5)) + ai = bs[0] & 0x1f + if ai <= 23: + length = ai + elif ai == 24: + lb = f.read(1) + if not lb: + raise EOFError + length = lb[0] + elif ai == 25: + lb = f.read(2) + if len(lb) < 2: + raise EOFError + length = struct.unpack('>H', lb)[0] + elif ai == 26: + lb = f.read(4) + if len(lb) < 4: + raise EOFError + length = struct.unpack('>I', lb)[0] + elif ai == 27: + lb = f.read(8) + if len(lb) < 8: + raise EOFError + length = struct.unpack('>Q', lb)[0] + else: + raise ValueError('Unsupported data length encoding: %d' % ai) + + if length > _MAX_DATA_LENGTH: + raise ValueError('Implausibly large data field: %d bytes' % length) + + data = f.read(length) + if len(data) < length: + raise EOFError + if not data: # length==0: firmware placeholder record (sequence-number marker) + return None + return data diff --git a/modules/common/raw_follower.py b/modules/common/raw_follower.py new file mode 100644 index 00000000..79e0b679 --- /dev/null +++ b/modules/common/raw_follower.py @@ -0,0 +1,100 @@ +""" +Shared RAW file follower for SleepyPod biometrics modules. + +Tails the newest .RAW file in a directory, yielding decoded CBOR records +as they are appended by the hardware daemon. +""" + +import time +import logging +from pathlib import Path +from typing import Optional + +import cbor2 + +from common.cbor_raw import read_raw_record + +log = logging.getLogger(__name__) + +MAX_CONSECUTIVE_FAILURES = 5 + + +def _safe_mtime(p: Path) -> float: + """Return mtime, or 0.0 if the file was deleted between glob and stat.""" + try: + return p.stat().st_mtime + except FileNotFoundError: + return 0.0 + + +class RawFileFollower: + """Follow the newest .RAW file, yielding decoded CBOR records. + + Args: + data_dir: Directory containing .RAW files. + shutdown_event: A threading.Event that signals shutdown. + poll_interval: Seconds to sleep when no new data is available. + """ + + def __init__(self, data_dir: Path, shutdown_event, poll_interval: float = 0.01): + self.data_dir = data_dir + self._shutdown = shutdown_event + self._poll_interval = poll_interval + self._file = None + self._path = None + self._last_pos = 0 + self._consecutive_failures = 0 + + def _find_latest(self) -> Optional[Path]: + candidates = [p for p in self.data_dir.glob("*.RAW") if _safe_mtime(p) > 0] + candidates.sort(key=_safe_mtime, reverse=True) + return candidates[0] if candidates else None + + def read_records(self): + """Yield decoded CBOR records as they arrive, sleeping between poll attempts.""" + while not self._shutdown.is_set(): + latest = self._find_latest() + if latest is None: + time.sleep(1) + continue + + if latest != self._path: + log.info("Switched to RAW file: %s", latest.name) + if self._file: + self._file.close() + self._file = open(latest, "rb") + self._path = latest + self._last_pos = 0 + self._consecutive_failures = 0 + + try: + data_bytes = read_raw_record(self._file) + if data_bytes is None: + self._last_pos = self._file.tell() + self._consecutive_failures = 0 + continue # empty placeholder record + inner = cbor2.loads(data_bytes) + self._last_pos = self._file.tell() + self._consecutive_failures = 0 + yield inner + except (ValueError, cbor2.CBORDecodeError, OSError) as e: + self._consecutive_failures += 1 + if self._consecutive_failures >= MAX_CONSECUTIVE_FAILURES: + log.warning("Skipping past corrupt data at offset %d after %d failures: %s", + self._last_pos, self._consecutive_failures, e) + self._last_pos += 1 + self._consecutive_failures = 0 + else: + log.debug("Error reading RAW record (attempt %d): %s", + self._consecutive_failures, e) + self._file.seek(self._last_pos) + time.sleep(0.1) + except EOFError: + self._file.seek(self._last_pos) + self._consecutive_failures = 0 + time.sleep(self._poll_interval) + + # Clean up file handle on shutdown + if self._file: + self._file.close() + self._file = None diff --git a/modules/piezo-processor/main.py b/modules/piezo-processor/main.py index fc6e9fc5..8c818a62 100644 --- a/modules/piezo-processor/main.py +++ b/modules/piezo-processor/main.py @@ -27,7 +27,10 @@ from collections import deque from typing import Optional +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + import cbor2 +from common.raw_follower import RawFileFollower import numpy as np from scipy.signal import butter, filtfilt, welch import heartpy as hp @@ -98,17 +101,19 @@ def report_health(status: str, message: str) -> None: """Write module health to sleepypod.db system_health table.""" try: conn = sqlite3.connect(str(SLEEPYPOD_DB), timeout=2.0) - with conn: - conn.execute( - """INSERT INTO system_health (component, status, message, last_checked) - VALUES ('piezo-processor', ?, ?, ?) - ON CONFLICT(component) DO UPDATE SET - status=excluded.status, - message=excluded.message, - last_checked=excluded.last_checked""", - (status, message, int(time.time())), - ) - conn.close() + try: + with conn: + conn.execute( + """INSERT INTO system_health (component, status, message, last_checked) + VALUES ('piezo-processor', ?, ?, ?) + ON CONFLICT(component) DO UPDATE SET + status=excluded.status, + message=excluded.message, + last_checked=excluded.last_checked""", + (status, message, int(time.time())), + ) + finally: + conn.close() except Exception as e: log.warning("Could not write health status: %s", e) @@ -172,50 +177,6 @@ def compute_breathing_rate(samples: np.ndarray, fs: float = SAMPLE_RATE) -> Opti log.debug("Breathing rate computation failed: %s", e) return None -# --------------------------------------------------------------------------- -# RAW file follower -# --------------------------------------------------------------------------- - -class RawFileFollower: - """ - Follows the newest .RAW file in RAW_DATA_DIR, tailing it as new CBOR - records are appended by the hardware daemon. - """ - - def __init__(self, data_dir: Path): - self.data_dir = data_dir - self._file = None - self._path = None - - def _find_latest(self) -> Optional[Path]: - candidates = sorted(self.data_dir.glob("*.RAW"), key=lambda p: p.stat().st_mtime, reverse=True) - return candidates[0] if candidates else None - - def read_records(self): - """Yield decoded CBOR records as they arrive. Blocks between records.""" - while not _shutdown.is_set(): - latest = self._find_latest() - if latest is None: - time.sleep(1) - continue - - if latest != self._path: - log.info("Switched to RAW file: %s", latest.name) - if self._file: - self._file.close() - self._file = open(latest, "rb") - self._path = latest - - try: - record = cbor2.load(self._file) - inner = cbor2.loads(record["data"]) - yield inner - except EOFError: - # No new data yet — poll - time.sleep(0.01) - except Exception as e: - log.warning("Error reading RAW record: %s", e) - time.sleep(1) # --------------------------------------------------------------------------- # Per-side processor @@ -271,7 +232,7 @@ def main() -> None: db_conn = open_biometrics_db() left = SideProcessor("left", db_conn) right = SideProcessor("right", db_conn) - follower = RawFileFollower(RAW_DATA_DIR) + follower = RawFileFollower(RAW_DATA_DIR, _shutdown, poll_interval=0.01) report_health("healthy", "piezo-processor started") diff --git a/modules/sleep-detector/main.py b/modules/sleep-detector/main.py index f4abbf19..d560e75f 100644 --- a/modules/sleep-detector/main.py +++ b/modules/sleep-detector/main.py @@ -29,7 +29,10 @@ from dataclasses import dataclass, field from typing import Optional +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + import cbor2 +from common.raw_follower import RawFileFollower import numpy as np # --------------------------------------------------------------------------- @@ -120,17 +123,19 @@ def write_movement(conn: sqlite3.Connection, side: str, def report_health(status: str, message: str) -> None: try: conn = sqlite3.connect(str(SLEEPYPOD_DB), timeout=2.0) - with conn: - conn.execute( - """INSERT INTO system_health (component, status, message, last_checked) - VALUES ('sleep-detector', ?, ?, ?) - ON CONFLICT(component) DO UPDATE SET - status=excluded.status, - message=excluded.message, - last_checked=excluded.last_checked""", - (status, message, int(time.time())), - ) - conn.close() + try: + with conn: + conn.execute( + """INSERT INTO system_health (component, status, message, last_checked) + VALUES ('sleep-detector', ?, ?, ?) + ON CONFLICT(component) DO UPDATE SET + status=excluded.status, + message=excluded.message, + last_checked=excluded.last_checked""", + (status, message, int(time.time())), + ) + finally: + conn.close() except Exception as e: log.warning("Could not write health status: %s", e) @@ -255,48 +260,6 @@ def _flush_movement(self, ts: float) -> None: self._movement_buf = [] self._last_movement_write = ts -# --------------------------------------------------------------------------- -# RAW file follower (same pattern as piezo-processor) -# --------------------------------------------------------------------------- - -class RawFileFollower: - def __init__(self, data_dir: Path): - self.data_dir = data_dir - self._file = None - self._path = None - - def _find_latest(self): - candidates = sorted(self.data_dir.glob("*.RAW"), key=lambda p: p.stat().st_mtime, reverse=True) - return candidates[0] if candidates else None - - def read_records(self): - while not _shutdown.is_set(): - latest = self._find_latest() - if latest is None: - time.sleep(1) - continue - - if latest != self._path: - log.info("Switched to RAW file: %s", latest.name) - if self._file: - self._file.close() - self._file = open(latest, "rb") - self._path = latest - - try: - record = cbor2.load(self._file) - inner = cbor2.loads(record["data"]) - yield inner - except EOFError: - time.sleep(0.5) - except Exception as e: - log.warning("Error reading RAW record: %s", e) - time.sleep(1) - - # Clean up file handle on shutdown - if self._file: - self._file.close() - self._file = None # --------------------------------------------------------------------------- # Main loop @@ -312,7 +275,7 @@ def main() -> None: db_conn = open_biometrics_db() left = SessionTracker(side="left", db=db_conn) right = SessionTracker(side="right", db=db_conn) - follower = RawFileFollower(RAW_DATA_DIR) + follower = RawFileFollower(RAW_DATA_DIR, _shutdown, poll_interval=0.5) report_health("healthy", "sleep-detector started")