Skip to content

Fix biometrics on Pod 5: cbor2 C extension skips records + empty placeholder records break read loop #46

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
seanpasino wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Fix biometrics on Pod 5: cbor2 C extension skips records + empty placeholder records break read loop #46

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 77 additions & 3 deletions biometrics/load_raw_files.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import struct
import numpy as np
import traceback
from datetime import datetime, timedelta, timezone
Expand All @@ -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())
Expand Down Expand Up @@ -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)
Expand Down
105 changes: 93 additions & 12 deletions biometrics/stream/stream.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import sys
import platform
import struct
import cbor2
from datetime import datetime, timedelta

Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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()
Expand All @@ -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


Expand Down