Add USBCANFD transport and visualize 0x391 door feedback

This commit is contained in:
zmv
2026-09-24 15:41:49 +08:00
parent 475fe2440b
commit 515a9f32d8
37 changed files with 1074 additions and 13 deletions
+7 -1
View File
@@ -106,6 +106,9 @@ def main():
for before, after, stem in zip(original, restored, rows(recorded)):
assert before['frame_index'] == after['frame_index']
assert before['capture'] == after['source_capture']
if 'can' in before['result']:
expected_can = dict(before['result']['can'], source='replay')
assert expected_can == after['result']['can'], 'Recorded CAN state changed during replay'
assert before['video_frame_index'] == after['video_frame_index']
matching_tracks += before['result']['track_ids'] == after['result']['track_ids']
matching_decisions += before['result']['risk']['decision'] == after['result']['risk']['decision']
@@ -124,7 +127,10 @@ def main():
'running_binary_sha256': running_hash, 'recorded_frames': len(original), 'replayed_frames': len(restored),
'frames_with_people': sum(bool(row['result']['track_ids']) for row in original),
'max_raw_pose_error_px': max_pose_error, 'matching_track_frames': matching_tracks, 'matching_decision_frames': matching_decisions,
'video_bytes': (recorded/'frames.mp4').stat().st_size, 'uncompressed_bgr_bytes': len(original)*int(probe['width'])*int(probe['height'])*3, 'sigterm_seconds': term_seconds, 'replay_sigint_seconds': interrupt_seconds, 'interrupted_frames': interrupted_count}
'video_bytes': (recorded/'frames.mp4').stat().st_size, 'uncompressed_bgr_bytes': len(original)*int(probe['width'])*int(probe['height'])*3, 'sigterm_seconds': term_seconds, 'replay_sigint_seconds': interrupt_seconds, 'interrupted_frames': interrupted_count,
'can_statuses': sorted({r['result'].get('can', {}).get('status', 'NOT_RECORDED') for r in original}),
'can_received': max(r['result'].get('can', {}).get('received', 0) for r in original),
'can_0x391_by_channel': [max((r['result'].get('can', {}).get('channels', [{}, {}])[ch].get('received', 0) for r in original)) for ch in range(2)]}
assert result['binary_sha256'] == running_hash
(args.output/'report.json').write_text(json.dumps(result, indent=2)+'\n')
print(json.dumps(result, indent=2))