From 7e25c0123c85b758e3142bbb51f1774288eccd75 Mon Sep 17 00:00:00 2001 From: Sean Passino Date: Sat, 14 Mar 2026 09:49:18 -0400 Subject: [PATCH] Fix biometrics on Pod 5: cbor2 C extension buffering skips records + empty placeholder records break loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two bugs combine to produce zero biometric data on Pod 5 (and degrade accuracy on Pod 3/4): Bug 1: cbor2 C extension reads in 4096-byte chunks When the _cbor2 C extension is installed, cbor2.load(f) advances f.tell() by 4096 bytes per call regardless of actual record size. RAW file records are 17-5000 bytes each, so only one record per 4096-byte block is ever decoded — the rest are silently skipped. On Pod 5 with ~2700-byte piezo-dual records, this skips the majority of data. Fix: replace cbor2.load(f) with _read_raw_record(f), a manual parser that uses f.read() byte-by-byte to parse the outer {seq: uint, data: bytes} CBOR wrapper, keeping f.tell() accurate after every record. Bug 2: Empty placeholder records raise EOFError mid-file Pod 5 firmware writes placeholder records with data=b'' (CBOR 0x40 = empty byte string) as sequence number markers between real records. cbor2.loads(b'') raises CBORDecodeEOF, which is a subclass of EOFError. Since both files catch EOFError to detect end-of-file and break the loop, any placeholder record mid-file terminates the entire read early — even with megabytes of valid data remaining. Fix: _read_raw_record() returns None for empty data; callers continue. Additionally, the original stream.py error handler called break on any exception, leaving the file position at an unknown offset. Changed to seek back to last_pos before breaking so recovery is possible. Tested on Pod 5, cbor2 5.6.5 with _cbor2 C extension, Python 3.10. Result: 846 vitals rows (HR, HRV, breathing rate) and 400 movement rows recorded in a single night. Stream runs stably with no OOM kills. --- biometrics/load_raw_files.py | 80 +++++++++++++++++++++++++- biometrics/stream/stream.py | 105 +++++++++++++++++++++++++++++++---- 2 files changed, 170 insertions(+), 15 deletions(-) diff --git a/biometrics/load_raw_files.py b/biometrics/load_raw_files.py index cfa23e58..596bb9a2 100644 --- a/biometrics/load_raw_files.py +++ b/biometrics/load_raw_files.py @@ -1,3 +1,4 @@ +import struct import numpy as np import traceback from datetime import datetime, timedelta, timezone @@ -15,6 +16,75 @@ logger = get_logger() +def _read_raw_record(f): + """ + Manually parse one outer {seq, data} CBOR record using f.read(). + + 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 function parses the outer {seq: uint, data: bytes} wrapper byte-by-byte + using f.read(), keeping f.tell() accurate after each record. + + Returns the raw inner data bytes, or None for empty placeholder records + (which the Pod firmware writes as sequence number markers with data=b''). + Raises EOFError at end of file, ValueError on malformed data. + """ + b = f.read(1) + if not b: + raise EOFError + if b[0] != 0xa2: + raise ValueError('Expected outer map 0xa2, got 0x%02x' % b[0]) + if f.read(4) != b'\x63\x73\x65\x71': + raise ValueError('Expected seq key') + hdr = f.read(1) + if not hdr: + raise EOFError + if hdr[0] == 0x1a: + seq_bytes = f.read(4) + if len(seq_bytes) < 4: + raise EOFError + elif hdr[0] == 0x1b: + seq_bytes = f.read(8) + if len(seq_bytes) < 8: + raise EOFError + else: + raise ValueError('Unexpected seq encoding: 0x%02x' % hdr[0]) + if f.read(5) != b'\x64\x64\x61\x74\x61': + raise ValueError('Expected data key') + bs = f.read(1) + if not bs: + raise EOFError + 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] + else: + raise ValueError('Unsupported length encoding: %d' % ai) + data = f.read(length) + if len(data) < length: + raise EOFError + if not data: + return None # empty placeholder record, caller should skip + return data + + def get_current_files(folder_path: str): return [ str(f.resolve()) @@ -75,9 +145,13 @@ def _decode_cbor_file(file_path: str, data: dict, start_time, end_time, side: Si while True: try: - # Decode the next CBOR object - row = cbor2.load(raw_data) - decoded_data = cbor2.loads(row['data']) + # Use manual reader instead of cbor2.load() to avoid the cbor2 + # C extension reading in 4096-byte chunks, which causes it to + # skip most records regardless of their actual size. + data_bytes = _read_raw_record(raw_data) + if data_bytes is None: + continue # empty placeholder record + decoded_data = cbor2.loads(data_bytes) if not decoded_data['type'] in load_raw_types: continue _delete_other_side(decoded_data, side, sensor_count) diff --git a/biometrics/stream/stream.py b/biometrics/stream/stream.py index c833ddbc..595f516a 100644 --- a/biometrics/stream/stream.py +++ b/biometrics/stream/stream.py @@ -19,6 +19,7 @@ import sys import platform +import struct import cbor2 from datetime import datetime, timedelta @@ -47,6 +48,75 @@ piezo_record_queue = queue.Queue() +def _read_raw_record(f): + """ + Manually parse one outer {seq, data} CBOR record using f.read(). + + 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 function parses the outer {seq: uint, data: bytes} wrapper byte-by-byte + using f.read(), keeping f.tell() accurate after each record. + + Returns the raw inner data bytes, or None for empty placeholder records + (which the Pod firmware writes as sequence number markers with data=b''). + Raises EOFError at end of file, ValueError on malformed data. + """ + b = f.read(1) + if not b: + raise EOFError + if b[0] != 0xa2: + raise ValueError('Expected outer map 0xa2, got 0x%02x' % b[0]) + if f.read(4) != b'\x63\x73\x65\x71': + raise ValueError('Expected seq key') + hdr = f.read(1) + if not hdr: + raise EOFError + if hdr[0] == 0x1a: + seq_bytes = f.read(4) + if len(seq_bytes) < 4: + raise EOFError + elif hdr[0] == 0x1b: + seq_bytes = f.read(8) + if len(seq_bytes) < 8: + raise EOFError + else: + raise ValueError('Unexpected seq encoding: 0x%02x' % hdr[0]) + if f.read(5) != b'\x64\x64\x61\x74\x61': + raise ValueError('Expected data key') + bs = f.read(1) + if not bs: + raise EOFError + 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] + else: + raise ValueError('Unsupported length encoding: %d' % ai) + data = f.read(length) + if len(data) < length: + raise EOFError + if not data: + return None # empty placeholder record, caller should skip + return data + + def _safe_getmtime(path: str) -> float: try: if os.path.exists(path): @@ -110,19 +180,28 @@ def follow_latest_file(self): while True: try: - # Decode CBOR object from a **single line** - row = cbor2.load(self.latest_file_obj) # Load the next CBOR object + # Use manual reader instead of cbor2.load() to avoid the cbor2 + # C extension reading in 4096-byte chunks, which causes it to + # skip most records regardless of their actual size. + data_bytes = _read_raw_record(self.latest_file_obj) + if data_bytes is None: + self.last_pos = self.latest_file_obj.tell() + continue # empty placeholder record + + decoded_data = cbor2.loads(data_bytes) one_minute_ago = datetime.now() - timedelta(minutes=2) - if 'data' in row: # Check if 'data' key exists - decoded_data = cbor2.loads(row['data']) - if decoded_data['type'] != 'piezo-dual': - continue - record_time = datetime.fromtimestamp(decoded_data['ts']) - if one_minute_ago > record_time: - continue - load_piezo_row(decoded_data, 'right') - piezo_record_queue.put(decoded_data) + if not isinstance(decoded_data, dict) or decoded_data.get('type') != 'piezo-dual': + self.last_pos = self.latest_file_obj.tell() + continue + + record_time = datetime.fromtimestamp(decoded_data['ts']) + if one_minute_ago > record_time: + self.last_pos = self.latest_file_obj.tell() + continue + + load_piezo_row(decoded_data, 'right') + piezo_record_queue.put(decoded_data) # Update last read position self.last_pos = self.latest_file_obj.tell() @@ -131,7 +210,9 @@ def follow_latest_file(self): # No more CBOR objects to read break except Exception as e: - logger.error(f"Error decoding CBOR: {e}") + logger.error(f"Error reading record: {e}") + # Seek back to last known good position to avoid cascading errors + self.latest_file_obj.seek(self.last_pos) break