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