#!/usr/bin/env python3 """Independently compare a RideForm master with CSV and ZIP exports (stdlib only). Usage: verify-export.py --bundle SESSION.rideform --export-dir CSV --zip master.zip Optional --hashes-out snapshots all master bytes; --hashes-before checks immutability against a previous snapshot. No time alignment, interpolation, or master mutation. """ import argparse import csv import hashlib import json import math from pathlib import Path import struct import sys import zipfile def require(condition, message): if not condition: raise ValueError(message) def load_json(path): return json.loads(path.read_text(encoding='utf-8'), parse_constant=lambda x: (_ for _ in ()).throw(ValueError('nonfinite JSON: ' + x))) def load_rows(path): if not path.exists(): return [] rows = [] for i, line in enumerate(path.read_text(encoding='utf-8').splitlines(), 1): require(bool(line), f'{path}:{i}: empty JSONL row') rows.append(json.loads(line, parse_constant=lambda x: (_ for _ in ()).throw(ValueError('nonfinite JSON: ' + x)))) return rows def compare_json(a, b, context): if isinstance(a, dict): require(isinstance(b, dict) and a.keys() == b.keys(), context + ': keys differ') for key in a: compare_json(a[key], b[key], context + '.' + key) elif isinstance(a, list): require(isinstance(b, list) and len(a) == len(b), context + ': length differs') for i, (x, y) in enumerate(zip(a, b)): compare_json(x, y, f'{context}[{i}]') elif isinstance(a, (int, float)) and not isinstance(a, bool): require(isinstance(b, (int, float)) and not isinstance(b, bool), context + ': numeric type differs') if isinstance(a, int) and isinstance(b, int): require(a == b, context + ': integer differs') else: require(struct.pack('>d', float(a)) == struct.pack('>d', float(b)), context + ': binary64 differs') else: require(type(a) is type(b) and a == b, context + ': value differs') def motion_fields(sample): fields = timing_fields(sample) for key, cols in ( ('attitudeRadians', ('attitude_roll_rad', 'attitude_pitch_rad', 'attitude_yaw_rad')), ('attitudeQuaternion', tuple('attitude_quaternion_' + x for x in 'xyzw')), ('attitudeRotationMatrix', tuple('attitude_matrix_m' + str(r) + str(c) for r in range(1, 4) for c in range(1, 4))), ('rotationRate', tuple('rotation_rate_device_' + x + '_rad_s' for x in 'xyz')), ('gravity', tuple('gravity_device_' + x + '_g' for x in 'xyz')), ('userAcceleration', tuple('user_acceleration_device_' + x + '_g' for x in 'xyz')), ): axes = list('xyzw') if key == 'attitudeQuaternion' else [f'm{r}{c}' for r in range(1, 4) for c in range(1, 4)] if key == 'attitudeRotationMatrix' else list('xyz') fields.update(zip(cols, (sample[key][axis] for axis in axes))) for key, prefix, cols in ( ('magneticField', 'magnetic_field', tuple('magnetic_field_device_' + x + '_microtesla' for x in 'xyz')), ('headingRadians', 'heading', ('heading_reference_rad',)), ('headingAccuracyRadians', 'heading_accuracy', ('heading_accuracy_rad',)), ): require(len(sample[key]) == 1, key + ': invalid tagged measurement') status, body = next(iter(sample[key].items())) require(status in ('available', 'invalid', 'unavailable'), key + ': unknown status') fields[prefix + '_status'] = status fields[prefix + '_reason'] = body.get('reason', '') fields[prefix + '_raw_validity_code'] = body.get('rawValidityCode') value = body.get('value') values = [value[x] for x in 'xyz'] if isinstance(value, dict) else [value] * len(cols) fields.update(zip(cols, values)) return fields def timing_fields(sample): return {'sequence': sample['sequence'], 'sensor_timestamp_seconds_since_boot': sample['sensorTimestamp'], 'receipt_host_uptime_seconds': sample['receiptUptime']} def raw_fields(sample, kind): fields = timing_fields(sample) key, prefix, unit = ('acceleration', 'acceleration_device_', '_g') if kind == 'accelerometer' else ('rotationRate', 'rotation_rate_device_', '_rad_s') fields.update((prefix + x + unit, sample[key][x]) for x in 'xyz') return fields def event_fields(sample): return {column: sample.get(key) for column, key in ( ('id', 'id'), ('sequence', 'sequence'), ('kind', 'kind'), ('label', 'label'), ('host_uptime_seconds', 'hostUptime'), ('scheduled_host_uptime_seconds', 'scheduledUptime'), ('audio_dispatch_host_uptime_seconds', 'audioDispatchUptime'), ('note', 'note'))} def calibration_fields(sample): fields = {'id': sample['id'], 'actual_duration_seconds': sample['actualDurationSeconds'], 'dispersion_rad': sample.get('dispersionRadians'), 'verified': sample['verified'], 'method': sample['method'], 'configuration_json': sample['configuration']} fields.update(('neutral_quaternion_' + x, sample['neutralQuaternion'][x]) for x in 'xyzw') fields.update(('device_to_rider_basis_m' + str(r) + str(c), sample['deviceToRiderBasis'][f'm{r}{c}']) for r in range(1, 4) for c in range(1, 4)) return fields def stream_headers(kind): vector = dict.fromkeys('xyz', 0.0) sample = {'sequence': 0, 'sensorTimestamp': 0.0, 'receiptUptime': 0.0, 'attitudeRadians': vector, 'attitudeQuaternion': dict.fromkeys('xyzw', 0.0), 'attitudeRotationMatrix': {f'm{r}{c}': 0.0 for r in range(1, 4) for c in range(1, 4)}, 'rotationRate': vector, 'gravity': vector, 'userAcceleration': vector, 'acceleration': vector, 'magneticField': {'unavailable': {'reason': ''}}, 'headingRadians': {'unavailable': {'reason': ''}}, 'headingAccuracyRadians': {'unavailable': {'reason': ''}}} return set(motion_fields(sample) if kind == 'motion' else raw_fields(sample, kind)) def check_csv(path, expected, expected_headers=None): with path.open(encoding='utf-8', newline='') as handle: reader = csv.DictReader(handle, strict=True) headers = reader.fieldnames rows = list(reader) require(headers is not None and len(set(headers)) == len(headers), str(path) + ': duplicate/absent headers') require(len(rows) == len(expected), str(path) + ': row count differs') if expected_headers is not None: require(set(headers) == expected_headers, str(path) + ': unexpected/missing columns') count = 0 for i, (actual, fields) in enumerate(zip(rows, expected)): require(set(headers) == fields.keys(), str(path) + ': unexpected/missing columns') for column, value in fields.items(): cell = actual[column] context = f'{path.name}[{i}].{column}' if value is None: require(cell == '', context + ': unavailable value became a measurement') elif isinstance(value, bool): require(cell == str(value).lower(), context + ': bool differs') elif isinstance(value, int): if column == 'sequence' or column.endswith('raw_validity_code'): require(int(cell) == value, context + ': integer differs') else: require(struct.pack('>d', float(cell)) == struct.pack('>d', float(value)), context + ': binary64 differs') elif isinstance(value, float): require(math.isfinite(value) and struct.pack('>d', float(cell)) == struct.pack('>d', value), context + ': binary64 differs') elif isinstance(value, dict): compare_json(value, json.loads(cell), context) else: require(cell == value, context + ': string/Unicode differs') count += 1 return {'rows': len(rows), 'cells': count} def hashes(bundle): return {str(path.relative_to(bundle)): hashlib.sha256(path.read_bytes()).hexdigest() for path in sorted(bundle.rglob('*')) if path.is_file()} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--bundle', type=Path, required=True) parser.add_argument('--export-dir', type=Path, required=True) parser.add_argument('--zip', type=Path, required=True) parser.add_argument('--hashes-out', type=Path) parser.add_argument('--hashes-before', type=Path) args = parser.parse_args() before = hashes(args.bundle) manifest = load_json(args.bundle / 'manifest.json') require(manifest['schemaVersion'] == 1, 'unsupported schema version') compare_json(manifest, load_json(args.export_dir / 'manifest.json'), 'manifest') results = {} for kind in ('motion', 'accelerometer', 'gyroscope'): rows = load_rows(args.bundle / (kind + '.jsonl')) matches = [x for x in manifest['streams'] if x['kind'] == kind] require(len(matches) == 1, kind + ': missing/duplicate metadata') metadata = matches[0] require(metadata['writtenCount'] == len(rows), kind + ': manifest written count differs') require(metadata['acceptedCount'] == metadata['writtenCount'] + metadata['lossCount'], kind + ': accepted/written/loss do not reconcile') require(metadata['lossCount'] >= 0, kind + ': negative loss') if metadata['lossCount']: explained = [d for d in manifest['diagnostics'] if d.get('stream') == kind and d.get('code') == 'nonfinite_value'] require(len(explained) == metadata['lossCount'], kind + ': loss not explained by explicit nonfinite diagnostics') require(metadata['statistics']['count'] == len(rows), kind + ': statistics count differs') require(not rows or 'available' in metadata['capability'], kind + ': unavailable stream contains measurements') # Preserve each stream's original order/times; deliberately never join streams. fields = [motion_fields(row) if kind == 'motion' else raw_fields(row, kind) for row in rows] results[kind] = check_csv(args.export_dir / (kind + '.csv'), fields, stream_headers(kind)) events = load_rows(args.bundle / 'events.jsonl') require(manifest['eventCount'] == len(events), 'manifest event count differs') results['events'] = check_csv(args.export_dir / 'events.csv', [event_fields(row) for row in events], set(event_fields({}))) calibration_path = args.bundle / 'calibration.json' if calibration_path.exists(): calibration = load_json(calibration_path) require(calibration['id'] == manifest.get('calibrationID'), 'calibration ID differs') compare_json(calibration, load_json(args.export_dir / 'calibration.json'), 'calibration') results['calibration'] = check_csv(args.export_dir / 'calibration.csv', [calibration_fields(calibration)]) for name, key in (('neutral', 'neutralSamples'), ('left_gesture', 'leftGestureSamples'), ('forward_gesture', 'forwardGestureSamples')): results['calibration_' + name] = check_csv(args.export_dir / ('calibration_' + name + '.csv'), [motion_fields(row) for row in calibration[key]], stream_headers('motion')) else: require(manifest.get('calibrationID') is None, 'missing calibration file') with zipfile.ZipFile(args.zip) as archive: require(archive.testzip() is None, 'ZIP CRC validation failed') expected = {args.bundle.name + '/' + name for name in before} names = archive.namelist() require(set(names) == expected and len(names) == len(expected), 'ZIP missing/extra/duplicate entries') for name in names: require(archive.read(name) == (args.bundle / name.split('/', 1)[1]).read_bytes(), name + ': archive bytes differ') require(before == hashes(args.bundle), 'checker run changed master') if args.hashes_before: require(before == load_json(args.hashes_before), 'master bytes changed since snapshot') if args.hashes_out: args.hashes_out.write_text(json.dumps(before, indent=2) + '\n', encoding='utf-8') print(json.dumps({'result': 'PASS', 'bundle': str(args.bundle), 'archive_files': len(before), 'csv': results, 'master_sha256': before}, indent=2)) if __name__ == '__main__': try: main() except (ValueError, KeyError, OSError, csv.Error, zipfile.BadZipFile) as error: print('FAIL: ' + str(error), file=sys.stderr) sys.exit(1)