From 8cea6724290efee2da58f5a0c03a794deef6200b Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Fri, 25 Sep 2026 14:53:21 -0500 Subject: [PATCH] Drive Statistics Implement background drive analysis and statistics with explicit worker and persistence ownership. Co-authored-by: Danny Co-authored-by: Prabhaav Pillai Co-authored-by: firestarsdog <229254897+firestarsdog@users.noreply.github.com> --- openpilot/starpilot/galaxy/drive_analysis.py | 443 ++++++++++++++ openpilot/starpilot/galaxy/drive_stats.py | 546 ++++++++++++++++++ .../galaxy/tests/test_drive_analysis.py | 280 +++++++++ .../galaxy/tests/test_drive_history.py | 179 ++++++ .../galaxy/tests/test_drive_stats.mjs | 92 +++ .../galaxy/tests/test_drive_stats.py | 538 +++++++++++++++++ .../tests/test_drive_stats_background.py | 91 +++ .../galaxy/tests/test_history_recovery.py | 93 +++ .../galaxy/tests/test_model_recording.py | 66 +++ .../starpilot/galaxy/tests/test_plots.mjs | 216 +++++++ .../galaxy/tests/test_record_history.mjs | 341 +++++++++++ openpilot/starpilot/tests/test_storage.py | 120 ++++ 12 files changed, 3005 insertions(+) create mode 100644 openpilot/starpilot/galaxy/drive_analysis.py create mode 100644 openpilot/starpilot/galaxy/drive_stats.py create mode 100644 openpilot/starpilot/galaxy/tests/test_drive_analysis.py create mode 100644 openpilot/starpilot/galaxy/tests/test_drive_history.py create mode 100644 openpilot/starpilot/galaxy/tests/test_drive_stats.mjs create mode 100644 openpilot/starpilot/galaxy/tests/test_drive_stats.py create mode 100644 openpilot/starpilot/galaxy/tests/test_drive_stats_background.py create mode 100644 openpilot/starpilot/galaxy/tests/test_history_recovery.py create mode 100644 openpilot/starpilot/galaxy/tests/test_model_recording.py create mode 100644 openpilot/starpilot/galaxy/tests/test_plots.mjs create mode 100644 openpilot/starpilot/galaxy/tests/test_record_history.mjs create mode 100644 openpilot/starpilot/tests/test_storage.py diff --git a/openpilot/starpilot/galaxy/drive_analysis.py b/openpilot/starpilot/galaxy/drive_analysis.py new file mode 100644 index 0000000000..6c3bf7c838 --- /dev/null +++ b/openpilot/starpilot/galaxy/drive_analysis.py @@ -0,0 +1,443 @@ +"""Measured statistics from a bounded, immutable local route recording.""" + +from __future__ import annotations + +from collections.abc import Callable +from contextlib import contextmanager +from datetime import UTC, datetime +import json +import math +import os +from pathlib import Path +import stat +import time + +from openpilot.starpilot.flm.local_logs import DIR_FLAGS, FILE_FLAGS, MAX_COMPRESSED_BYTES, LocalLogUnavailable +from openpilot.starpilot.flm.log_decode import LogDecodeError, decode_segment +from openpilot.starpilot.galaxy.drive_history import MAX_ROOT_ENTRIES, MAX_SEGMENT_ENTRIES, MAX_STATS_SEGMENTS_PER_ROUTE, SEGMENT_NAME +from openpilot.starpilot.models.catalog import model_entries +from openpilot.starpilot.models.receipt import logged_model_load +from openpilot.starpilot.models.status import ModelVariant + + +MAX_ANALYSIS_SECONDS = 30.0 +MAX_PAIR_SECONDS = 1.0 +MAX_SPEED_MPS = 70.0 +MAX_ROUTE_COMPRESSED_BYTES = 256 * 1024 * 1024 +ANALYSIS_VERSION = 3 +MAX_MODEL_METADATA_SECONDS = 4.0 +DISTRACTED_EVENT = 'driverDistracted2' +UNRESPONSIVE_EVENT = 'driverUnresponsive3' +LOGS = {'qlog.zst': 'zst', 'qlog.bz2': 'bz2', 'rlog.zst': 'zst', 'rlog.bz2': 'bz2'} + + +class _Incomplete(Exception): + pass + + +def _revision(info: os.stat_result) -> tuple[int, ...]: + return info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns, info.st_ctime_ns + + +@contextmanager +def _closed_source(root: Path, segment_name: str, expected: dict, allowed: Callable[[], bool], *, inspect_only: bool = False, + prefer_rlog: bool = False): + fds: list[int] = [] + try: + root_fd = os.open(root, DIR_FLAGS) + fds.append(root_fd) + segment_fd = os.open(segment_name, DIR_FLAGS, dir_fd=root_fd) + fds.append(segment_fd) + root_identity = os.fstat(root_fd).st_dev, os.fstat(root_fd).st_ino + segment_identity = os.fstat(segment_fd).st_dev, os.fstat(segment_fd).st_ino + found: list[str] = [] + with os.scandir(segment_fd) as entries: + for count, entry in enumerate(entries, 1): + if not allowed(): + raise _Incomplete('Analysis cancelled') + if count > MAX_SEGMENT_ENTRIES or entry.name.endswith('.lock'): + raise _Incomplete('Segment is open or unavailable') + if entry.name in LOGS: + found.append(entry.name) + preferred = [name for name in found if name.startswith('rlog.' if prefer_rlog else 'qlog.')] + if any(name.startswith('qlog.') for name in found) != (expected.get('qlog') is True) or \ + any(name.startswith('rlog.') for name in found) != (expected.get('rlog') is True): + raise _Incomplete('Route source changed') + if len(preferred) > 1 or (not preferred and len(found) != 1): + raise _Incomplete('One closed qlog or full rlog is required') + filename = preferred[0] if preferred else found[0] + fd = os.open(filename, FILE_FLAGS, dir_fd=segment_fd) + fds.append(fd) + initial = os.fstat(fd) + if not stat.S_ISREG(initial.st_mode) or not 0 < initial.st_size <= MAX_COMPRESSED_BYTES: + raise _Incomplete('Route log is empty, oversized or unavailable') + revision = _revision(initial) + evidence = (root_identity, segment_identity, filename, revision, tuple(sorted(found))) + + def current() -> bool: + try: + root_now = os.stat(root, follow_symlinks=False) + segment_now = os.stat(segment_name, dir_fd=root_fd, follow_symlinks=False) + source_now = os.stat(filename, dir_fd=segment_fd, follow_symlinks=False) + if not stat.S_ISDIR(root_now.st_mode) or not stat.S_ISDIR(segment_now.st_mode) or \ + not stat.S_ISREG(source_now.st_mode) or \ + (root_now.st_dev, root_now.st_ino) != root_identity or \ + (segment_now.st_dev, segment_now.st_ino) != segment_identity or \ + _revision(source_now) != revision or _revision(os.fstat(fd)) != revision: + return False + with os.scandir(segment_fd) as entries: + names = [] + for count, entry in enumerate(entries, 1): + if count > MAX_SEGMENT_ENTRIES or entry.name.endswith('.lock'): + return False + if entry.name in LOGS: + names.append(entry.name) + return sorted(names) == sorted(found) + except OSError: + return False + + if not current(): + raise _Incomplete('Route source changed') + if inspect_only: + yield None, LOGS[filename], current, evidence + return + chunks: list[bytes] = [] + remaining = initial.st_size + while remaining: + if not allowed(): + raise _Incomplete('Analysis cancelled') + chunk = os.read(fd, min(1024 * 1024, remaining)) + if not chunk: + raise _Incomplete('Route source changed') + chunks.append(chunk) + remaining -= len(chunk) + if os.read(fd, 1) or not current(): + raise _Incomplete('Route source changed') + yield b''.join(chunks), LOGS[filename], current, evidence + except OSError as error: + raise _Incomplete('Route source is unavailable') from error + finally: + for fd in reversed(fds): + os.close(fd) + + +def _route_names(root: Path, route_id: str, allowed: Callable[[], bool]) -> set[str]: + names: set[str] = set() + try: + with os.scandir(root) as entries: + for count, entry in enumerate(entries, 1): + if not allowed(): + raise _Incomplete('Analysis cancelled') + if count > MAX_ROOT_ENTRIES: + raise _Incomplete('Route inventory is incomplete') + matched = SEGMENT_NAME.fullmatch(entry.name) + if matched is not None and matched.group('route').replace('_', '|') == route_id: + names.add(entry.name) + except OSError as error: + raise _Incomplete('Route directory is unavailable') from error + return names + + +def _selected(route: dict) -> tuple[str, list[str]]: + if not isinstance(route, dict): + raise _Incomplete('Invalid route selection') + route_id = route.get('routeId') + segments = route.get('segments') + if type(route_id) is not str or type(segments) is not list or not 1 <= len(segments) <= MAX_STATS_SEGMENTS_PER_ROUTE or \ + route.get('segmentCount') != len(segments): + raise _Incomplete('Invalid route selection') + names: list[str] = [] + for index, segment in enumerate(segments): + if not isinstance(segment, dict) or segment.get('number') != index or \ + not isinstance(segment.get('files'), dict) or \ + not (segment['files'].get('qlog') is True or segment['files'].get('rlog') is True): + raise _Incomplete('Route has missing segments or logs') + name = segment.get('segmentName') + matched = SEGMENT_NAME.fullmatch(name) if type(name) is str and len(name) <= 180 else None + if matched is None or matched.group('route').replace('_', '|') != route_id or int(matched.group('number')) != index: + raise _Incomplete('Invalid route selection') + assert isinstance(name, str) + names.append(name) + if len(set(names)) != len(names): + raise _Incomplete('Duplicate route segment') + return route_id, names + + +def _route_wall_time(route_id: str) -> float | None: + token = route_id.rsplit('|', maxsplit=1)[-1] + try: + if len(token) != 20 or token[4] != '-' or token[10:12] != '--': + return None + value = datetime.strptime(token, '%Y-%m-%d--%H-%M-%S').replace(tzinfo=UTC).timestamp() + return value if 946684800 <= value <= datetime.now(UTC).timestamp() + 86400 else None + except ValueError: + return None + + +def _recorded_model(init_data) -> str | None: + try: + values = {str(entry.key): bytes(entry.value).decode('utf-8', 'replace').strip() + for entry in init_data.params.entries if str(entry.key) in ('DrivingModelName', 'DrivingModel', 'Model')} + except (AttributeError, TypeError, ValueError): + return None + for key in ('DrivingModelName', 'DrivingModel', 'Model'): + value = values.get(key, '') + if 0 < len(value) <= 100 and value.isprintable(): + return value + return None + + +def _model_label(outputs, loads, selected): + """Join actual output variants to recorded load boundaries; saved intent is not execution.""" + if not outputs: + return f'Selected: {selected}' if selected else None + loads = sorted(set(loads), key=lambda item: item[0]) + labels = [] + index = -1 + for stamp, big in sorted(set(outputs)): + while index + 1 < len(loads) and loads[index + 1][0] <= stamp: + index += 1 + load = loads[index][1] if index >= 0 else None + variant = 'Chestnut big' if big else 'Small' + if load is not None and (load.variant is ModelVariant.CHESTNUT) == big: + entry = model_entries().get(load.model_id) + label = f'{entry.name if entry is not None else load.model_id} ({variant})' + else: + label = f'{variant} model (identity not recorded)' + if label not in labels: + labels.append(label) + combined = ' → '.join(labels) + return combined if len(combined) <= 120 else 'Multiple recorded models' + + +def _rlog_model_loads(root, route, names, allowed): + """Optional exact identities. A missing/oversized log never prevents drive statistics.""" + deadline = time.monotonic() + MAX_MODEL_METADATA_SECONDS + def permitted(): + return time.monotonic() < deadline and allowed() + loads, evidence = [], [] + total = 0 + try: + for index, name in enumerate(names): + files = route['segments'][index]['files'] + if not files.get('rlog'): + return [] # Missing intervals cannot prove which model ran for the whole route. + with _closed_source(root, name, files, permitted, prefer_rlog=True) as (compressed, codec, current, revision): + if not current(): + return [] + total += len(compressed) + if total > MAX_ROUTE_COMPRESSED_BYTES: + return [] + for event in decode_segment(compressed, codec, cancelled=lambda: not permitted()): + if event.valid and event.which() == 'logMessage': + raw = str(event.logMessage) + if (load := logged_model_load(raw)) is not None and 0 < load.loaded_mono_ns <= event.logMonoTime: + loads.append((load.loaded_mono_ns, load)) + elif len(raw) <= 8192: + try: + record = json.loads(raw) + if isinstance(record, dict) and record.get('msg') == 'modeld init' and record.get('module') == 'modeld': + loads.append((int(event.logMonoTime), None)) + except (ValueError, TypeError, RecursionError): + pass + evidence.append(revision) + for index, name in enumerate(names): + with _closed_source(root, name, route['segments'][index]['files'], permitted, + inspect_only=True, prefer_rlog=True) as (_, _, current, revision): + if not current() or revision != evidence[index]: + return [] + return loads + except (_Incomplete, LocalLogUnavailable, LogDecodeError, OSError, ValueError): + return [] + + +def analyze_route(root: Path, route: dict, *, permitted: Callable[[], bool], prefer_rlog: bool = False) -> dict: + """Return a complete route only when every selected source and sample span is sound.""" + result = {'routeId': route.get('routeId') if isinstance(route, dict) else None, + 'startTime': None, 'endTime': None, 'distanceMeters': None, 'durationSeconds': None, + 'engagedSeconds': None, 'engagedPercent': None, 'model': None, + 'distractedMoments': None, 'unresponsiveMoments': None, + 'complete': False, 'reason': None, 'segmentCount': 0, 'analysisVersion': ANALYSIS_VERSION} + segment_count = len(route.get('segments', [])) if isinstance(route, dict) else 0 + deadline = time.monotonic() + min(120.0, max(60.0, segment_count * 14.0)) + def allowed() -> bool: + if time.monotonic() > deadline: + return False + try: + return permitted() + except Exception: + return False + try: + if not isinstance(root, Path) or not root.is_absolute() or '..' in root.parts: + raise _Incomplete('Invalid recording root') + if not allowed(): + raise _Incomplete('Analysis cancelled') + route_id, names = _selected(route) + result['segmentCount'] = len(names) + if _route_names(root, route_id, allowed) != set(names): + raise _Incomplete('Route segments changed or are incomplete') + + first_car = last_car = first_state = last_state = None + previous_car = previous_state = None + distance = engaged = 0.0 + car_pairs = state_pairs = 0 + coverage_gap = False + end_of_route = False + clocks_offset = None + model = None + model_outputs = [] + model_loads = [] + previous_events: set[str] = set() + distracted = unresponsive = 0 + first_attention = last_attention = None + attention_gap = False + compressed_total = 0 + evidence = [] + for index, name in enumerate(names): + if not allowed(): + raise _Incomplete('Analysis cancelled') + with _closed_source(root, name, route['segments'][index]['files'], allowed, prefer_rlog=prefer_rlog) as (compressed, codec, current, source_evidence): + if not current(): + raise _Incomplete('Route source changed') + evidence.append(source_evidence) + compressed_total += len(compressed) + if compressed_total > MAX_ROUTE_COMPRESSED_BYTES: + raise _Incomplete('Route source changed or exceeds analysis limit') + events = decode_segment(compressed, codec, cancelled=lambda: not allowed(), + retain=frozenset(('clocks', 'initData', 'logMessage', 'modelV2', 'drivingModelData', + 'carState', 'selfdriveState', 'onroadEvents', 'sentinel'))) + segment_end = False + for event in events: + if not allowed(): + raise _Incomplete('Analysis cancelled') + kind = event.which() + stamp = int(event.logMonoTime) + if stamp <= 0 or not event.valid: + continue + if kind == 'clocks': + wall = int(event.clocks.wallTimeNanos) + if 946684800 * 1_000_000_000 <= wall <= (datetime.now(UTC).timestamp() + 86400) * 1_000_000_000: + offset = (wall - stamp) / 1e9 + if clocks_offset is not None and abs(offset - clocks_offset) > 5: + raise _Incomplete('Conflicting logged wall clocks') + clocks_offset = offset if clocks_offset is None else clocks_offset + elif kind == 'initData' and model is None: + model = _recorded_model(event.initData) + elif kind == 'logMessage': + load = logged_model_load(str(event.logMessage)) + if load is not None and 0 < load.loaded_mono_ns <= stamp: + model_loads.append((load.loaded_mono_ns, load)) + elif len(str(event.logMessage)) <= 8192: + try: + marker = json.loads(str(event.logMessage)).get('msg', {}) + if marker == 'modeld init': + model_loads.append((stamp, None)) + if (isinstance(marker, dict) and marker.get('event') == 'modeld.identity-unavailable' and + type(marker.get('sinceMonoNs')) is int and 0 < marker['sinceMonoNs'] <= stamp): + model_loads.append((marker['sinceMonoNs'], None)) + except (ValueError, TypeError, AttributeError): + pass + elif kind in ('modelV2', 'drivingModelData'): + model_outputs.append((stamp, bool(getattr(event, kind).big))) + elif kind == 'carState': + speed = float(event.carState.vEgo) + if not math.isfinite(speed) or not -.5 <= speed <= MAX_SPEED_MPS or (last_car is not None and stamp <= last_car): + coverage_gap = True + previous_car = None + continue + speed = max(0.0, speed) + if previous_car is not None: + delta = (stamp - previous_car[0]) / 1e9 + if delta > MAX_PAIR_SECONDS: + coverage_gap = True + else: + distance += (previous_car[1] + speed) * .5 * delta + car_pairs += 1 + first_car = stamp if first_car is None else first_car + last_car = stamp + previous_car = stamp, speed + elif kind == 'selfdriveState': + if last_state is not None and stamp <= last_state: + coverage_gap = True + previous_state = None + continue + if previous_state is not None: + delta = (stamp - previous_state[0]) / 1e9 + if delta > MAX_PAIR_SECONDS: + coverage_gap = True + else: + engaged += delta if previous_state[1] else 0.0 + state_pairs += 1 + first_state = stamp if first_state is None else first_state + last_state = stamp + previous_state = stamp, bool(event.selfdriveState.enabled) + elif kind == 'onroadEvents': + if last_attention is not None and not 0 < (stamp - last_attention) / 1e9 <= 5: + attention_gap = True + first_attention = stamp if first_attention is None else first_attention + last_attention = stamp + active = {str(item.name).split('.')[-1] for item in event.onroadEvents} + distracted += DISTRACTED_EVENT in active and DISTRACTED_EVENT not in previous_events + unresponsive += UNRESPONSIVE_EVENT in active and UNRESPONSIVE_EVENT not in previous_events + previous_events = active + elif kind == 'sentinel': + sentinel_type = str(event.sentinel.type).split('.')[-1] + if sentinel_type == 'endOfRoute': + if index != len(names) - 1: + raise _Incomplete('Route ended before its last segment') + segment_end = True + if index == len(names) - 1: + end_of_route = segment_end + del events, compressed + if _route_names(root, route_id, allowed) != set(names): + raise _Incomplete('Route source changed') + for index, name in enumerate(names): + if not allowed(): + raise _Incomplete('Analysis cancelled') + with _closed_source(root, name, route['segments'][index]['files'], allowed, inspect_only=True, prefer_rlog=prefer_rlog) as (_, _, current, observed): + if evidence[index] != observed or not current(): + raise _Incomplete('Route source changed') + + if first_car is None or last_car is None or first_state is None or last_state is None: + raise _Incomplete('Route has no observed duration') + first = min(first_car, first_state) + last = max(last_car, last_state) + if last <= first: + raise _Incomplete('Route has no observed duration') + duration = (last - first) / 1e9 + if clocks_offset is not None: + start_time = first / 1e9 + clocks_offset + else: + start_time = _route_wall_time(route_id) + if start_time is not None: + result['startTime'] = start_time + result['endTime'] = start_time + duration + result['durationSeconds'] = round(duration, 3) + result['model'] = _model_label(model_outputs, model_loads, model) + if first_attention is not None and last_attention is not None and not attention_gap and \ + first_attention - first <= 5 * 1e9 and last - last_attention <= 5 * 1e9: + result['distractedMoments'] = distracted + result['unresponsiveMoments'] = unresponsive + if car_pairs and not coverage_gap: + result['distanceMeters'] = round(distance, 2) + if state_pairs and not coverage_gap: + result['engagedSeconds'] = round(min(engaged, duration), 3) + result['engagedPercent'] = round(min(engaged, duration) / duration * 100, 1) + if not end_of_route: + raise _Incomplete('End of route was not recorded') + if not car_pairs or not state_pairs or coverage_gap or \ + max(first_car - first, first_state - first, last - last_car, last - last_state) > MAX_PAIR_SECONDS * 1e9: + raise _Incomplete('Route samples are missing or gapped') + if start_time is None: + raise _Incomplete('Route has no logged or named wall time') + result['complete'] = True + if model_outputs: + result['model'] = _model_label(model_outputs, model_loads or _rlog_model_loads(root, route, names, allowed), model) + return result + except (_Incomplete, LocalLogUnavailable, LogDecodeError, OSError, ValueError) as error: + result['reason'] = str(error) or 'Route analysis unavailable' + if (not prefer_rlog and result['reason'] in ('Route samples are missing or gapped', 'Route has no observed duration') and + isinstance(route, dict) and route.get('segments') and + all(segment.get('files', {}).get('rlog') is True for segment in route['segments']) and permitted()): + return analyze_route(root, route, permitted=allowed, prefer_rlog=True) + return result diff --git a/openpilot/starpilot/galaxy/drive_stats.py b/openpilot/starpilot/galaxy/drive_stats.py new file mode 100644 index 0000000000..47aa660637 --- /dev/null +++ b/openpilot/starpilot/galaxy/drive_stats.py @@ -0,0 +1,546 @@ +"""Durable, bounded statistics for analyzed local Galaxy drives.""" + +from __future__ import annotations + +from collections import defaultdict +from collections.abc import Callable +from datetime import UTC, datetime, timedelta, tzinfo +import hashlib +import json +import math +import os +from pathlib import Path +import stat +import threading +import time + +from openpilot.starpilot.galaxy.drive_history import DriveHistory, MAX_ROOT_ENTRIES, MAX_STATS_SEGMENTS_PER_ROUTE +from openpilot.starpilot.galaxy.drive_analysis import ANALYSIS_VERSION +from openpilot.starpilot.storage import galaxy_storage_root + +SCHEMA_VERSION = 1 +MAX_CACHED_ROUTES = 2048 +MAX_CACHE_BYTES = 4_000_000 +MAX_SCAN_ROUTES = 100 +MAX_ANALYSES_PER_SCAN = 2 +SCAN_INTERVAL_SECONDS = 60 +RETRY_SECONDS = 300 +MAX_RETRY_ATTEMPTS = 5 +MAX_RETRY_AGE_SECONDS = 86400 +MAX_RETRY_DELAY_SECONDS = 3600 +RECENT_LIMIT = 5 + + +def _finite(value, minimum=0.0): + if type(value) not in (int, float): + return None + try: + if not math.isfinite(value) or value < minimum: + return None + except (OverflowError, ValueError): + return None + return float(value) + + +def _valid_record(value: object) -> bool: + if not isinstance(value, dict) or type(value.get('routeId')) is not str or len(value['routeId']) > 180: + return False + if type(value.get('complete')) is not bool or type(value.get('ignored')) is not bool: + return False + if type(value.get('segmentCount')) is not int or not 0 <= value['segmentCount'] <= MAX_STATS_SEGMENTS_PER_ROUTE: + return False + for key in ('startTime', 'endTime'): + if value.get(key) is not None and (_finite(value[key]) is None or value[key] > 253402300799): + return False + for key in ('distanceMeters', 'durationSeconds', 'engagedSeconds'): + if value.get(key) is not None and _finite(value[key]) is None: + return False + for key in ('distractedMoments', 'unresponsiveMoments'): + if value.get(key) is not None and (type(value[key]) is not int or value[key] < 0): + return False + if value.get('model') is not None and (type(value['model']) is not str or len(value['model']) > 120): + return False + if value.get('reason') is not None and (type(value['reason']) is not str or len(value['reason']) > 200): + return False + if value.get('engagedPercent') is not None and (_finite(value['engagedPercent']) is None or value['engagedPercent'] > 100): + return False + if type(value.get('sourceSignature')) is not str or len(value['sourceSignature']) > 128: + return False + if type(value.get('analysisVersion', 0)) is not int or not 0 <= value.get('analysisVersion', 0) < 2**31: + return False + manifest = value.get('sourceManifest') + if manifest is not None and (not isinstance(manifest, dict) or len(manifest) > MAX_STATS_SEGMENTS_PER_ROUTE * 2 or + any(type(key) is not str or len(key) > 220 or not isinstance(revision, list) or + len(revision) != 5 or any(type(part) is not int or part < 0 for part in revision) + for key, revision in manifest.items())): + return False + if value['complete'] and (value.get('startTime') is None or value.get('endTime') is None or + any(value.get(key) is None for key in ('distanceMeters', 'durationSeconds', 'engagedSeconds')) or + value['endTime'] < value['startTime'] or value['engagedSeconds'] > value['durationSeconds']): + return False + return True + + +def _route_signature(root: Path, route: dict) -> tuple[str, dict]: + parts = [] + manifest = {} + for segment in route.get('segments', []): + name = segment.get('segmentName', '') + files = segment.get('files', {}) + for filename in ('rlog.zst', 'rlog.bz2', 'qlog.zst', 'qlog.bz2'): + try: + info = (root / name / filename).stat(follow_symlinks=False) + if stat.S_ISREG(info.st_mode): + manifest[f'{name}/{filename}'] = [info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns, info.st_ctime_ns] + except OSError: + pass + parts.append((segment.get('number'), name, files)) + encoded = json.dumps(parts, sort_keys=True, separators=(',', ':')).encode() + source = hashlib.sha256(encoded + json.dumps(manifest, sort_keys=True, separators=(',', ':')).encode()).hexdigest() + return source, manifest + + +def _only_deleted_sources(previous: dict, current: dict) -> bool: + old = previous.get('sourceManifest') + return isinstance(old, dict) and all(old.get(key) == revision for key, revision in current.items()) + + +def _personal_records(drives: list[dict], zone: tzinfo) -> list[dict]: + if not drives: + return [] + ordered = sorted(drives, key=lambda r: (r['startTime'], r['routeId'])) + daily = defaultdict(lambda: {'duration': 0.0, 'engaged': 0.0}) + weekly = defaultdict(float) + longest = max(drives, key=lambda r: r['distanceMeters']) + undistracted = [] + clean_streak = best_clean_streak = 0 + for drive in ordered: + day = datetime.fromtimestamp(drive['startTime'], zone).date() + daily[day]['duration'] += drive['durationSeconds'] + daily[day]['engaged'] += drive['engagedSeconds'] + weekly[day - timedelta(days=day.weekday())] += drive['distanceMeters'] + if drive.get('distractedMoments') == 0: + undistracted.append(drive) + clean = drive.get('unresponsiveMoments') == 0 + if clean: + clean_streak += 1 + best_clean_streak = max(best_clean_streak, clean_streak) + else: + clean_streak = 0 + engaged_days = [(100 * data['engaged'] / data['duration'], day) for day, data in daily.items() if data['duration'] > 0] + best_day = max(engaged_days, default=(None, None)) + best_week = max(((distance, day) for day, distance in weekly.items()), default=(None, None)) + streak = best_streak = 0 + previous_day = None + for day in sorted(daily): + streak = streak + 1 if previous_day is not None and (day - previous_day).days == 1 else 1 + best_streak = max(best_streak, streak) + previous_day = day + best_undistracted = max(undistracted, key=lambda r: r['durationSeconds']) if undistracted else None + return [ + {'id': 'longestDrive', 'label': 'Longest drive', 'value': longest['distanceMeters'], 'unit': 'm'}, + {'id': 'mostEngagedDay', 'label': 'Most engaged day', 'value': round(best_day[0], 1) if best_day[0] is not None else None, 'unit': '%'}, + {'id': 'bestWeek', 'label': 'Best week', 'value': best_week[0], 'unit': 'm'}, + {'id': 'highestStreak', 'label': 'Highest streak', 'value': best_streak, 'unit': 'days'}, + {'id': 'longestUndistractedDrive', 'label': 'Longest undistracted drive', + 'value': best_undistracted['durationSeconds'] if best_undistracted else None, 'unit': 's'}, + {'id': 'cleanDriveStreak', 'label': 'Clean drive streak', 'value': best_clean_streak, 'unit': 'drives'}, + ] + + +class DriveStatsOwner: + def __init__(self, root: Path | None = None, store: Path | None = None, history: DriveHistory | None = None, + analyzer: Callable | None = None, permitted: Callable[[], bool] | None = None, + clock: Callable[[], float] | None = None, monotonic: Callable[[], float] | None = None, + metric: Callable[[], bool] | None = None): + self.root = Path(root) if root is not None else None + self.store = Path(store) if store is not None else galaxy_storage_root() / 'drive-stats.json' + self.history = history if history is not None else DriveHistory(self.root, max_segments_per_route=MAX_STATS_SEGMENTS_PER_ROUTE, + max_segments=MAX_ROOT_ENTRIES) + self.root = Path(self.history.root) if self.root is None else self.root + if analyzer is None: + from openpilot.starpilot.galaxy.drive_analysis import analyze_route + analyzer = analyze_route + self.analyzer = analyzer + self.permitted = permitted if permitted is not None else lambda: False + self.clock = clock if clock is not None else lambda: datetime.now(UTC).timestamp() + self.monotonic = monotonic if monotonic is not None else time.monotonic + self.metric = metric if metric is not None else lambda: True + self._lock = threading.RLock() + self._permit_lock = threading.Lock() + self._permit_checked_at = float('-inf') + self._permit_cached = False + self._closed = threading.Event() + self._worker: threading.Thread | None = None + self._scheduler: threading.Thread | None = None + self._next_scan = 0.0 + self._records: dict[str, dict] = {} + self._failures: dict[str, dict] = {} + self._scan_incomplete = False + self._scanned = False + self._truncated = False + self._corrupt = False + self._pending = 0 + self._cursor = 0 + self._load() + + def _load(self): + try: + with self.store.open('rb') as stream: + raw = stream.read(MAX_CACHE_BYTES + 1) + if len(raw) > MAX_CACHE_BYTES: + raise ValueError('Oversized cache') + data = json.loads(raw) + records = data['routes'] + if data.get('schemaVersion') != SCHEMA_VERSION or not isinstance(records, dict) or len(records) > MAX_CACHED_ROUTES: + raise ValueError('Invalid cache') + if any(key != value.get('routeId') or not _valid_record(value) for key, value in records.items()): + raise ValueError('Invalid route') + failures = data.get('failures', {}) + if not isinstance(failures, dict) or len(failures) > MAX_CACHED_ROUTES: + raise ValueError('Invalid retry cache') + for key, failure in failures.items(): + if (type(key) is not str or len(key) > 180 or not isinstance(failure, dict) or + type(failure.get('signature')) is not str or len(failure['signature']) > 128 or + type(failure.get('attempts')) is not int or not 1 <= failure['attempts'] <= MAX_RETRY_ATTEMPTS or + any(_finite(failure.get(field)) is None for field in ('firstAt', 'retryAt')) or + type(failure.get('reason')) is not str or len(failure['reason']) > 200): + raise ValueError('Invalid retry record') + self._failures = failures + self._records = records + self._scan_incomplete = data.get('scanIncomplete') is True + self._scanned = data.get('scanned') is True + self._truncated = data.get('truncated') is True + except FileNotFoundError: + return + except (OSError, ValueError, TypeError, KeyError, AttributeError): + self._corrupt = True + + def _save(self): + self.store.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + payload = {'schemaVersion': SCHEMA_VERSION, 'routes': self._records, + 'scanned': self._scanned, 'scanIncomplete': self._scan_incomplete, 'truncated': self._truncated, + 'failures': self._failures} + encoded = json.dumps(payload, separators=(',', ':'), allow_nan=False).encode() + if len(encoded) > MAX_CACHE_BYTES: + raise OSError('Drive statistics cache exceeds storage limit') + temporary = self.store.with_name(f'.{self.store.name}.{os.getpid()}.{threading.get_ident()}.tmp') + try: + fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, 'wb') as stream: + stream.write(encoded) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, self.store) + directory_fd = os.open(self.store.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + self._corrupt = False + finally: + try: + temporary.unlink() + except FileNotFoundError: + pass + + def _allowed(self) -> bool: + if self._closed.is_set(): + return False + with self._permit_lock: + if self._closed.is_set(): + return False + now = self.monotonic() + if now - self._permit_checked_at >= .1 or now < self._permit_checked_at: + try: + self._permit_cached = bool(self.permitted()) + except Exception: + self._permit_cached = False + self._permit_checked_at = now + return self._permit_cached + + def _run(self): + if not self._allowed(): + return + assert self.root is not None + try: + inventory = self.history.snapshot() + routes = inventory.get('routes', []) + if not isinstance(routes, list): + raise ValueError('Invalid route inventory') + with self._lock: + self._scanned = True + self._scan_incomplete = inventory.get('scanIncomplete') is True or len(routes) > MAX_SCAN_ROUTES + self._pending = 0 + route_ids = {route.get('routeId') for route in routes if isinstance(route, dict) and type(route.get('routeId')) is str} + self._failures = {key: value for key, value in self._failures.items() if key in route_ids} + candidates = [] + for route in routes[:MAX_SCAN_ROUTES]: + if not self._allowed(): + break + route_id = route.get('routeId') + if type(route_id) is not str or len(route_id) > 180: + continue + signature, manifest = _route_signature(self.root, route) + with self._lock: + existing = self._records.get(route_id) + failed = self._failures.get(route_id) + if (existing and existing.get('sourceSignature') == signature and existing.get('complete') and + existing.get('analysisVersion', 0) >= ANALYSIS_VERSION): + continue + if (existing and existing.get('complete') and manifest != existing.get('sourceManifest') and + _only_deleted_sources(existing, manifest)): + prior_signature = existing['sourceSignature'] + existing['sourceSignature'] = signature + try: + self._save() + except OSError: + existing['sourceSignature'] = prior_signature + continue + if failed and failed['signature'] == signature: + now = self.clock() + if (failed['attempts'] >= MAX_RETRY_ATTEMPTS or now - failed['firstAt'] >= MAX_RETRY_AGE_SECONDS or + failed['firstAt'] <= now < failed['retryAt']): + continue + candidates.append((route, route_id, signature, manifest)) + with self._lock: + self._pending = len(candidates) + if candidates: + start = self._cursor % len(candidates) + candidates = candidates[start:] + candidates[:start] + attempted = 0 + for offset, (route, route_id, signature, manifest) in enumerate(candidates): + if not self._allowed() or attempted >= MAX_ANALYSES_PER_SCAN: + break + with self._lock: + existing = self._records.get(route_id) + failed = self._failures.get(route_id) + if existing and existing.get('complete') and existing.get('sourceSignature') != signature: + prior = existing.copy() + existing['complete'] = False + existing['reason'] = 'sourceChanged' + try: + self._save() + except OSError: + self._records[route_id] = prior + raise + attempted += 1 + self._cursor = (start + offset + 1) % len(candidates) + try: + result = self.analyzer(self.root, route, permitted=self._allowed) + if _route_signature(self.root, route) != (signature, manifest): + raise ValueError('Route source changed during analysis') + if not isinstance(result, dict) or result.get('routeId') != route_id: + raise ValueError('Invalid route analysis') + if not self._allowed(): + break # A drive starting or shutdown is not an unreadable recording. + record = {key: result.get(key) for key in ('routeId', 'startTime', 'endTime', 'distanceMeters', + 'durationSeconds', 'engagedSeconds', 'engagedPercent', 'model', + 'distractedMoments', 'unresponsiveMoments', 'complete', 'reason', 'segmentCount')} + record['ignored'] = bool(existing and existing.get('ignored')) + record['analysisVersion'] = result.get('analysisVersion', ANALYSIS_VERSION) + record['sourceSignature'] = signature + record['sourceManifest'] = manifest + if not _valid_record(record): + raise ValueError('Invalid route analysis') + with self._lock: + if self._closed.is_set(): + break + current = self._records.get(route_id) + prior_records = self._records.copy() + prior_failure = self._failures.get(route_id) + prior_truncated = self._truncated + record['ignored'] = bool(current and current.get('ignored')) + refreshing_complete = (current and current.get('complete') and current.get('sourceSignature') == signature) + if record['complete'] or not refreshing_complete: + self._records[route_id] = record + if record['complete']: + self._failures.pop(route_id, None) + else: + self._failures[route_id] = self._failure(signature, prior_failure, record.get('reason') or 'Incomplete recording') + self._bound() + try: + self._save() + except OSError: + self._records = prior_records + self._truncated = prior_truncated + if prior_failure is None: + self._failures.pop(route_id, None) + else: + self._failures[route_id] = prior_failure + raise + except Exception as error: + with self._lock: + self._failures[route_id] = self._failure(signature, failed, type(error).__name__) + if len(self._failures) > MAX_CACHED_ROUTES: + self._failures.pop(next(iter(self._failures))) + finally: + with self._lock: + self._pending = max(0, self._pending - 1) + with self._lock: + if not self._closed.is_set(): + self._save() + except Exception: + with self._lock: + self._scan_incomplete = True + + def _failure(self, signature, previous, reason): + now = self.clock() + same = previous is not None and previous['signature'] == signature + attempts = min(MAX_RETRY_ATTEMPTS, previous['attempts'] + 1 if same else 1) + return {'signature': signature, 'attempts': attempts, + 'firstAt': previous['firstAt'] if same else now, + 'retryAt': now + min(MAX_RETRY_DELAY_SECONDS, RETRY_SECONDS * 2 ** (attempts - 1)), + 'reason': str(reason)[:200]} + + def _bound(self): + estimated = 1000 + sum(len(key) + len(json.dumps(record, separators=(',', ':'))) + 8 + for key, record in self._records.items()) + limit = max(1000, MAX_CACHE_BYTES - 1000) + if len(self._records) <= MAX_CACHED_ROUTES and estimated <= limit: + return + ordered = sorted(self._records.values(), key=lambda r: (r.get('startTime') or 0, r['routeId'])) + for record in ordered: + if len(self._records) <= MAX_CACHED_ROUTES and estimated <= limit: + break + del self._records[record['routeId']] + estimated -= len(record['routeId']) + len(json.dumps(record, separators=(',', ':'))) + 8 + self._truncated = True + + def _start(self): + with self._lock: + if self._closed.is_set() or (self._worker is not None and self._worker.is_alive()): + return + now = self.monotonic() + if now < self._next_scan or not self._allowed(): + return + self._next_scan = now + SCAN_INTERVAL_SECONDS + self._worker = threading.Thread(target=self._run, name='galaxy-drive-stats', daemon=True) + self._worker.start() + + def start(self): + """Start the device-owned analysis lifecycle independently of HTTP requests.""" + with self._lock: + if self._closed.is_set() or self._scheduler is not None: + return + self._scheduler = threading.Thread(target=self._schedule, name='galaxy-drive-history', daemon=True) + self._scheduler.start() + + def _schedule(self): + was_allowed = False + while not self._closed.is_set(): + allowed = self._allowed() + if allowed: + if not was_allowed: + with self._lock: + self._next_scan = 0.0 + self._start() + was_allowed = allowed + self._closed.wait(.1) + + def recording_dates(self, route_ids) -> dict: + """Read already analyzed capture dates; listing recordings does not start analysis.""" + with self._lock: + return {identifier: self._records[identifier]['startTime'] for identifier in route_ids + if identifier in self._records and self._records[identifier].get('startTime') is not None} + + def snapshot(self, timezone: str = "UTC") -> dict: + with self._lock: + result = self._snapshot_locked(timezone) + return result + + def _snapshot_locked(self, timezone: str = "UTC") -> dict: + from zoneinfo import ZoneInfo + if type(timezone) is not str or not 1 <= len(timezone) <= 128: + raise ValueError("Invalid timezone") + try: + zone = ZoneInfo(timezone) + except (KeyError, ValueError) as error: + raise ValueError("Invalid timezone") from error + try: + is_metric = bool(self.metric()) + except Exception: + is_metric = True + now = datetime.fromtimestamp(self.clock(), zone) + day = now.date() + monday = day - timedelta(days=day.weekday()) + dates = [(monday + timedelta(days=n)).isoformat() for n in range(7)] + complete = [r for r in self._records.values() if r['complete'] and not r['ignored']] + complete.sort(key=lambda r: (r.get('startTime') or 0, r['routeId']), reverse=True) + recent = sorted(self._records.values(), key=lambda r: (r.get('startTime') or 0, r['routeId']), reverse=True) + week = [r for r in complete if datetime.fromtimestamp(r['startTime'], zone).date().isoformat() in dates] + daily = defaultdict(float) + for record in week: + daily[datetime.fromtimestamp(record['startTime'], zone).date().isoformat()] += record['distanceMeters'] + def sums(drives): + return {'distanceMeters': sum(r['distanceMeters'] for r in drives), + 'durationSeconds': sum(r['durationSeconds'] for r in drives), + 'engagedSeconds': sum(r['engagedSeconds'] for r in drives), 'drives': len(drives)} + totals = sums(complete) + week_totals = sums(week) + week_result = {'days': [{'date': date, 'distanceMeters': daily[date]} for date in dates], + 'distanceMeters': week_totals['distanceMeters'], 'durationSeconds': week_totals['durationSeconds'], + 'engagedPercent': round(100 * week_totals['engagedSeconds'] / week_totals['durationSeconds'], 1) + if week_totals['durationSeconds'] else None, 'drives': len(week), 'timezone': timezone} + model_totals = defaultdict(lambda: {'distanceMeters': 0.0, 'durationSeconds': 0.0, 'drives': 0}) + for record in complete: + if type(record.get('model')) is str and record['model']: + model = model_totals[record['model']] + model['distanceMeters'] += record['distanceMeters'] + model['durationSeconds'] += record['durationSeconds'] + model['drives'] += 1 + models = [{'name': name, **amounts} for name, amounts in model_totals.items()] + models.sort(key=lambda m: (-m['distanceMeters'], m['name'])) + records = _personal_records(complete, zone) + running = self._worker is not None and self._worker.is_alive() + pending = max(self._pending, int(not self._scanned)) + state = ('analyzing' if running else 'waitingParked' if pending and not self._allowed() else + 'queued' if pending else 'retrying' if any( + f['attempts'] < MAX_RETRY_ATTEMPTS and self.clock() - f['firstAt'] < MAX_RETRY_AGE_SECONDS + for f in self._failures.values()) else 'ready') + def public(record): + return {key: value for key, value in record.items() if key not in ('sourceSignature', 'sourceManifest')} + included_recent = [record for record in recent if not record['ignored']] + return {'schemaVersion': SCHEMA_VERSION, 'isMetric': is_metric, + 'analysis': {'running': running, 'pending': pending, 'state': state, + 'failed': len(self._failures), + 'quarantined': sum(f['attempts'] >= MAX_RETRY_ATTEMPTS or + self.clock() - f['firstAt'] >= MAX_RETRY_AGE_SECONDS for f in self._failures.values()), + 'failures': [{'routeId': key, 'reason': f['reason'], 'attempts': f['attempts']} for key, f in self._failures.items()], + 'scanIncomplete': self._scan_incomplete}, + 'totals': totals, 'week': week_result, 'lastDrive': public(included_recent[0]) if included_recent else None, + 'recentDrives': [public(r) for r in recent[:RECENT_LIMIT]], 'records': records, 'models': models, + 'coverage': {'scope': 'analyzedLocalRoutes', 'timezone': timezone, 'partialHistory': True, + 'scanIncomplete': self._scan_incomplete, 'scanned': self._scanned, + 'cachedRoutes': len(self._records), 'completeRoutes': sum(r['complete'] for r in self._records.values()), + 'incompleteRoutes': sum(not r['complete'] for r in self._records.values()), + 'ignoredRoutes': sum(r['ignored'] for r in self._records.values()), + 'truncated': self._truncated, 'cacheCorrupt': self._corrupt}} + + def ignore(self, route_id: str, ignored: bool, authorized: Callable[[], bool]) -> dict: + if not authorized(): + raise PermissionError('Local Galaxy session expired') + if type(route_id) is not str or type(ignored) is not bool: + raise ValueError('Invalid route selection') + with self._lock: + record = self._records.get(route_id) + if record is None: + raise ValueError('Unknown route') + if record['ignored'] != ignored: + prior = record['ignored'] + record['ignored'] = ignored + try: + self._save() + except OSError: + record['ignored'] = prior + raise + return self.snapshot() + + def close(self): + self._closed.set() + scheduler = self._scheduler + if scheduler is not None and scheduler is not threading.current_thread(): + scheduler.join(timeout=2) + worker = self._worker + if worker is not None and worker is not threading.current_thread(): + worker.join(timeout=31) diff --git a/openpilot/starpilot/galaxy/tests/test_drive_analysis.py b/openpilot/starpilot/galaxy/tests/test_drive_analysis.py new file mode 100644 index 0000000000..27e21f53e0 --- /dev/null +++ b/openpilot/starpilot/galaxy/tests/test_drive_analysis.py @@ -0,0 +1,280 @@ +from pathlib import Path +import json +import tempfile +import unittest +from unittest import mock + +from openpilot.cereal import messaging +import zstandard as zstd + +from openpilot.starpilot.models.catalog import BY_ID +from openpilot.starpilot.galaxy import drive_analysis +from openpilot.starpilot.galaxy.drive_analysis import analyze_route + + +ROUTE = '2026-09-25--12-00-00' +BASE = 1_000_000_000 + + +def event(kind, stamp, **fields): + alerts = fields.pop('onroadEvents', None) + message = messaging.new_message(kind, size=len(alerts), valid=True) if alerts is not None else messaging.new_message(kind, valid=True) + message.logMonoTime = stamp + payload = getattr(message, kind) + if alerts is not None: + for item, alert in zip(payload, alerts, strict=True): + item.name = alert['name'] + for key, value in fields.items(): + setattr(payload, key, value) + return message + + +def write(root, number, events, *, lock=False, kind='rlog'): + name = f'{ROUTE}--{number}' + segment = root / name + segment.mkdir() + (segment / f'{kind}.zst').write_bytes(zstd.compress(b''.join(item.as_reader().as_builder().to_bytes() for item in events))) + if lock: + (segment / 'rlog.lock').touch() + return {'number': number, 'segmentName': name, 'files': {kind: True}} + + +def selection(*segments): + return {'routeId': ROUTE, 'segmentCount': len(segments), 'segments': list(segments)} + + +class DriveAnalysisTest(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name).resolve() + + def sample(self, stamp, speed, enabled, alerts=()): + return [event('carState', stamp, vEgo=speed), event('selfdriveState', stamp, enabled=enabled), + event('onroadEvents', stamp, onroadEvents=[{'name': name} for name in alerts])] + + def test_complete_contiguous_route_uses_observed_values_and_event_edges(self): + first = [event('clocks', BASE, wallTimeNanos=1_790_334_400_000_000_000), + *self.sample(BASE, 0, True), + *self.sample(BASE + 500_000_000, 10, True, ('driverDistracted2',)), + event('sentinel', BASE + 900_000_000, type='endOfSegment')] + second = [*self.sample(BASE + 1_000_000_000, 10, False, ('driverDistracted2', 'driverUnresponsive3')), + *self.sample(BASE + 1_500_000_000, 10, False), + event('sentinel', BASE + 1_500_000_000, type='endOfRoute')] + route = selection(write(self.root, 0, first), write(self.root, 1, second)) + result = analyze_route(self.root, route, permitted=lambda: True) + self.assertTrue(result['complete'], result) + self.assertEqual(result['distanceMeters'], 12.5) + self.assertEqual(result['durationSeconds'], 1.5) + self.assertEqual(result['engagedSeconds'], 1) + self.assertAlmostEqual(result['engagedPercent'], 66.7) + self.assertEqual(result['distractedMoments'], 1) + self.assertEqual(result['unresponsiveMoments'], 1) + self.assertEqual(result['startTime'], 1_790_334_400) + self.assertIsNone(result['model']) + + def test_missing_sentinel_gap_and_noncontiguous_segments_are_incomplete(self): + events = [*self.sample(BASE, 5, True), *self.sample(BASE + 2_000_000_000, 5, True)] + route = selection(write(self.root, 0, events)) + result = analyze_route(self.root, route, permitted=lambda: True) + self.assertFalse(result['complete']) + self.assertIsNone(result['distanceMeters']) + self.assertIn('End of route', result['reason']) + route['segments'][0]['number'] = 1 + self.assertIn('missing segments', analyze_route(self.root, route, permitted=lambda: True)['reason']) + + def test_lock_and_added_segment_invalidate_inventory(self): + first = write(self.root, 0, [*self.sample(BASE, 5, False), *self.sample(BASE + 500_000_000, 5, False), + event('sentinel', BASE + 500_000_000, type='endOfRoute')], lock=True) + result = analyze_route(self.root, selection(first), permitted=lambda: True) + self.assertFalse(result['complete']) + (self.root / first['segmentName'] / 'rlog.lock').unlink() + write(self.root, 1, [event('sentinel', BASE + 1_000_000_000, type='endOfRoute')]) + result = analyze_route(self.root, selection(first), permitted=lambda: True) + self.assertFalse(result['complete']) + self.assertIn('segments changed', result['reason']) + + def test_cancellation_is_incomplete(self): + first = write(self.root, 0, [*self.sample(BASE, 5, True), *self.sample(BASE + 500_000_000, 5, True), + event('sentinel', BASE + 500_000_000, type='endOfRoute')]) + self.assertFalse(analyze_route(self.root, selection(first), permitted=lambda: False)['complete']) + + def test_qlog_without_attention_samples_preserves_unknown_moments(self): + events = [event('sentinel', BASE, type='startOfRoute'), + event('carState', BASE + 2_000_000_000, vEgo=5), + event('selfdriveState', BASE + 2_000_000_000, enabled=True), + event('carState', BASE + 2_500_000_000, vEgo=5), + event('selfdriveState', BASE + 2_500_000_000, enabled=False), + event('sentinel', BASE + 3_000_000_000, type='endOfRoute')] + route = selection(write(self.root, 0, events, kind='qlog')) + result = analyze_route(self.root, route, permitted=lambda: True) + self.assertTrue(result['complete'], result) + self.assertEqual(result['distanceMeters'], 2.5) + self.assertEqual(result['engagedSeconds'], .5) + self.assertIsNotNone(result['startTime']) + self.assertIsNone(result['distractedMoments']) + self.assertIsNone(result['unresponsiveMoments']) + + def test_model_is_only_the_selection_recorded_in_init_data(self): + init = messaging.new_message('initData', valid=True) + init.logMonoTime = BASE + entries = init.initData.init('params').init('entries', 1) + entries[0].key = 'DrivingModelName' + entries[0].value = b'Recorded choice' + events = [init, event('carState', BASE, vEgo=5), event('selfdriveState', BASE, enabled=False), + event('carState', BASE + 500_000_000, vEgo=5), + event('selfdriveState', BASE + 500_000_000, enabled=False), + event('sentinel', BASE + 500_000_000, type='endOfRoute')] + route = selection(write(self.root, 0, events)) + result = analyze_route(self.root, route, permitted=lambda: True) + self.assertTrue(result['complete'], result) + self.assertEqual(result['model'], 'Selected: Recorded choice') + + def model_events(self, big=False): + return [*self.sample(BASE, 5, True), event('drivingModelData', BASE + 100_000_000, big=big), + *self.sample(BASE + 500_000_000, 5, True), event('sentinel', BASE + 500_000_000, type='endOfRoute')] + + def load_event(self, model_id='sc23', big=False, stamp=BASE): + message = messaging.new_message(None, valid=True) + message.logMonoTime = stamp + message.logMessage = json.dumps({'process': 123, 'msg': {'event': 'modeld.loaded', 'version': 1, + 'pid': 123, 'processStartTicks': 17, 'loadedMonoNs': stamp, 'modelId': model_id, + 'variant': 'chestnut' if big else 'small', 'artifactSha256': 'a' * 64, 'fallbackReason': None}}) + return message + + def test_qlog_reports_actual_small_fallback_without_a_recorded_identity(self): + events = self.model_events() + result = analyze_route(self.root, selection(write(self.root, 0, events, kind='qlog')), permitted=lambda: True) + self.assertTrue(result['complete'], result) + self.assertEqual(result['model'], 'Small model (identity not recorded)') + self.assertEqual(result['analysisVersion'], drive_analysis.ANALYSIS_VERSION) + + def test_retained_rlog_supplies_verified_identity_while_qlog_supplies_stats(self): + events = self.model_events() + segment = write(self.root, 0, events, kind='qlog') + rlog = self.root / segment['segmentName'] / 'rlog.zst' + rlog.write_bytes(zstd.compress(b''.join(item.as_reader().as_builder().to_bytes() for item in [self.load_event(), *events]))) + segment['files']['rlog'] = True + result = analyze_route(self.root, selection(segment), permitted=lambda: True) + self.assertTrue(result['complete'], result) + self.assertEqual(result['model'], f'{BY_ID["sc23"].name} (Small)') + rlog.write_bytes(b'corrupt optional metadata') + result = analyze_route(self.root, selection(segment), permitted=lambda: True) + self.assertTrue(result['complete'], result) + self.assertEqual(result['model'], 'Small model (identity not recorded)') + + def test_model_fallback_and_restart_do_not_keep_prior_identity(self): + events = [self.load_event('cinquev3', True), *self.sample(BASE, 5, True), + event('drivingModelData', BASE + 50_000_000, big=True), + self.load_event(stamp=BASE + 100_000_000), event('drivingModelData', BASE + 200_000_000, big=False), + *self.sample(BASE + 500_000_000, 5, True), event('sentinel', BASE + 500_000_000, type='endOfRoute')] + segment = write(self.root, 0, events) + result = analyze_route(self.root, selection(segment), permitted=lambda: True) + self.assertEqual(result['model'], f'{BY_ID["cinquev3"].name} (Chestnut big) → {BY_ID["sc23"].name} (Small)') + restart = messaging.new_message(None, valid=True) + restart.logMonoTime = BASE + 150_000_000 + restart.logMessage = json.dumps({'msg': 'modeld init', 'module': 'modeld', 'process': 124}) + events.insert(6, restart) + (self.root / segment['segmentName'] / 'rlog.zst').write_bytes(zstd.compress(b''.join(item.as_reader().as_builder().to_bytes() for item in events))) + result = analyze_route(self.root, selection(segment), permitted=lambda: True) + self.assertIn('Small model (identity not recorded)', result['model']) + self.assertNotIn(f'{BY_ID["sc23"].name} (Small)', result['model']) + + def test_failed_or_unconfirmed_load_never_overrides_actual_output(self): + events = [self.load_event('cinquev3', True), *self.model_events()] + result = analyze_route(self.root, selection(write(self.root, 0, events)), permitted=lambda: True) + self.assertEqual(result['model'], 'Small model (identity not recorded)') + with mock.patch.object(drive_analysis, 'MAX_MODEL_METADATA_SECONDS', 0): + result = analyze_route(self.root, selection({'number': 0, 'segmentName': f'{ROUTE}--0', 'files': {'rlog': True}}), + permitted=lambda: True) + self.assertTrue(result['complete'], result) + self.assertEqual(result['model'], 'Small model (identity not recorded)') + + def test_sample_gap_and_missing_wall_time_cannot_be_complete(self): + events = [event('carState', BASE, vEgo=10), event('selfdriveState', BASE, enabled=True), + event('carState', BASE + 2_000_000_000, vEgo=10), + event('selfdriveState', BASE + 2_000_000_000, enabled=True), + event('sentinel', BASE + 2_000_000_000, type='endOfRoute')] + segment = write(self.root, 0, events, kind='qlog') + gap = analyze_route(self.root, selection(segment), permitted=lambda: True) + self.assertFalse(gap['complete']) + self.assertIsNone(gap['distanceMeters']) + self.assertIsNone(gap['engagedSeconds']) + self.assertIn('gapped', gap['reason']) + + old_name = segment['segmentName'] + no_date = selection(segment) + no_date['routeId'] = '00000231--8c5f1c4f8b' + no_date['segments'][0]['segmentName'] = '00000231--8c5f1c4f8b--0' + (self.root / old_name).rename(self.root / no_date['segments'][0]['segmentName']) + (self.root / no_date['segments'][0]['segmentName'] / 'qlog.zst').write_bytes(zstd.compress(b''.join( + item.as_reader().as_builder().to_bytes() for item in [event('carState', BASE, vEgo=10), event('selfdriveState', BASE, enabled=True), + event('carState', BASE + 500_000_000, vEgo=10), + event('selfdriveState', BASE + 500_000_000, enabled=False), + event('sentinel', BASE + 500_000_000, type='endOfRoute')]))) + result = analyze_route(self.root, no_date, permitted=lambda: True) + self.assertFalse(result['complete']) + self.assertIsNone(result['startTime']) + self.assertIn('wall time', result['reason']) + + def test_129_segment_qlog_route_and_source_guards(self): + segments = [] + for number in range(129): + events = [] + for sample in range(67): + stamp = BASE + number * 60_000_000_000 + sample * 900_000_000 + events.extend([event('carState', stamp, vEgo=10), event('selfdriveState', stamp, enabled=True)]) + if number == 128: + events.append(event('sentinel', stamp, type='endOfRoute')) + segments.append(write(self.root, number, events, kind='qlog')) + route = selection(*segments) + real_open = drive_analysis.os.open + real_close = drive_analysis.os.close + opened = set() + peak = [0] + def tracking_open(*args, **kwargs): + fd = real_open(*args, **kwargs) + opened.add(fd) + peak[0] = max(peak[0], len(opened)) + return fd + def tracking_close(fd): + opened.discard(fd) + return real_close(fd) + with mock.patch.object(drive_analysis.os, 'open', side_effect=tracking_open), \ + mock.patch.object(drive_analysis.os, 'close', side_effect=tracking_close): + result = analyze_route(self.root, route, permitted=lambda: True) + self.assertTrue(result['complete'], result) + self.assertLessEqual(peak[0], 3) + self.assertFalse(opened) + self.assertEqual(result['segmentCount'], 129) + self.assertEqual(result['distanceMeters'], 77394) + self.assertEqual(result['durationSeconds'], 7739.4) + self.assertEqual(result['engagedSeconds'], 7739.4) + + middle = self.root / segments[64]['segmentName'] + original = (middle / 'qlog.zst').read_bytes() + (middle / 'qlog.zst').unlink() + middle.rmdir() + self.assertFalse(analyze_route(self.root, route, permitted=lambda: True)['complete']) + middle.mkdir() + (middle / 'qlog.zst').write_bytes(original) + (middle / 'qlog.lock').touch() + self.assertFalse(analyze_route(self.root, route, permitted=lambda: True)['complete']) + (middle / 'qlog.lock').unlink() + + first_log = self.root / segments[0]['segmentName'] / 'qlog.zst' + decode = drive_analysis.decode_segment + changed = [False] + def changing_decode(*args, **kwargs): + if not changed[0]: + changed[0] = True + first_log.write_bytes(first_log.read_bytes()) + return decode(*args, **kwargs) + with mock.patch.object(drive_analysis, 'decode_segment', side_effect=changing_decode): + result = analyze_route(self.root, route, permitted=lambda: True) + self.assertFalse(result['complete']) + self.assertIn('source changed', result['reason'].lower()) + + +if __name__ == '__main__': + unittest.main() diff --git a/openpilot/starpilot/galaxy/tests/test_drive_history.py b/openpilot/starpilot/galaxy/tests/test_drive_history.py new file mode 100644 index 0000000000..8ceb66c3e5 --- /dev/null +++ b/openpilot/starpilot/galaxy/tests/test_drive_history.py @@ -0,0 +1,179 @@ +"""Disposable local segment trees only; no log decoding or cloud access.""" + +import os +from pathlib import Path +import tempfile +import unittest +from unittest import mock + +from openpilot.starpilot.galaxy import drive_history + + +class DriveHistoryTest(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) / 'logs' + self.root.mkdir() + self.reader = drive_history.DriveHistory(self.root) + + def segment(self, name: str, *files: str) -> Path: + path = self.root / name + path.mkdir() + for filename in files: + (path / filename).write_bytes(b'x') + return path + + def test_current_local_legacy_and_downloaded_names_without_invented_stats(self): + current = '0000021e--371eaf116b' + self.segment(current + '--0', 'rlog.zst', 'qlog.zst', 'fcamera.hevc') + self.segment(current + '--2', 'qlog.bz2', 'dcamera.hevc') + self.segment('2021-02-03--04-05-06--1', 'qlog.zst') + self.segment('8bc6f0b166ab516e_0000021e--371eaf116b--3', 'rlog.bz2') + output = self.reader.snapshot() + self.assertEqual(output['source'], 'local') + self.assertTrue(output['partialHistory']) + self.assertFalse(output['scanIncomplete']) + by_id = {route['routeId']: route for route in output['routes']} + self.assertEqual(set(by_id), {current, '2021-02-03--04-05-06', + '8bc6f0b166ab516e|0000021e--371eaf116b'}) + self.assertEqual(by_id[current]['segmentCount'], 2) + self.assertEqual([s['number'] for s in by_id[current]['segments']], [0, 2]) + self.assertTrue(by_id[current]['segments'][0]['files']['rlog']) + self.assertFalse(by_id[current]['segments'][1]['files']['rlog']) + self.assertTrue(by_id[current]['segments'][1]['files']['dcamera']) + self.assertEqual(by_id[current]['segments'][0]['segmentName'], current + '--0') + self.assertEqual(by_id['8bc6f0b166ab516e|0000021e--371eaf116b']['segments'][0]['segmentName'], + '8bc6f0b166ab516e_0000021e--371eaf116b--3') + self.assertNotIn('duration', str(output).lower()) + self.assertNotIn('distance', str(output).lower()) + + def test_recording_dates_and_connect_links_keep_observed_device_identity(self): + local = '0000021e--371eaf116b' + saved = self.segment(local + '--0', 'qlog.zst') + os.utime(saved / 'qlog.zst', (1760000000, 1760000000)) + self.segment('0123456789abcdef_' + local + '--1', 'rlog.zst') + result = drive_history.recording_details(self.reader.snapshot(), {local: 1759999900}, 'fedcba9876543210') + routes = {route['routeId']: route for route in result['routes']} + self.assertEqual(routes[local]['startTime'], 1759999900) + self.assertEqual(routes[local]['fileTime'], 1760000000) + self.assertEqual(routes[local]['connectUrl'], 'https://connect.comma.ai/fedcba9876543210/' + local) + downloaded = routes['0123456789abcdef|' + local] + self.assertIsNone(downloaded['startTime']) + self.assertEqual(downloaded['connectUrl'], 'https://connect.comma.ai/0123456789abcdef/' + local) + invalid = drive_history.recording_details(self.reader.snapshot(), {}, '../not-a-device') + self.assertIsNone(next(route for route in invalid['routes'] if route['routeId'] == local)['connectUrl']) + + def test_active_empty_symlink_and_unrecognized_entries_are_excluded(self): + self.segment('00000001--aaaaaaaaaa--0', 'rlog.zst', 'rlog.lock') + self.segment('00000002--bbbbbbbbbb--0', 'unrelated.txt') + valid = self.segment('00000003--cccccccccc--0', 'qlog.zst') + (valid / 'fcamera.hevc').symlink_to(self.root / 'outside.hevc') + (self.root / '00000004--dddddddddd--0').symlink_to(valid, target_is_directory=True) + result = self.reader.snapshot() + self.assertEqual([r['routeId'] for r in result['routes']], ['00000003--cccccccccc']) + self.assertFalse(result['routes'][0]['segments'][0]['files']['fcamera']) + self.assertTrue(result['scanIncomplete']) # matching symlink directory cannot be inventoried + + def test_disappearing_file_skips_uncertain_segment(self): + segment = self.segment('00000001--aaaaaaaaaa--0', 'rlog.zst') + original = os.stat + removed = False + def disappearing(path, *args, **kwargs): + nonlocal removed + if path == 'rlog.zst' and kwargs.get('dir_fd') is not None and not removed: + removed = True + (segment / 'rlog.zst').unlink() + return original(path, *args, **kwargs) + with mock.patch.object(drive_history.os, 'stat', side_effect=disappearing): + result = self.reader.snapshot() + self.assertEqual(result['routes'], []) + self.assertTrue(result['scanIncomplete']) + + def test_root_and_segment_scan_bounds_are_explicit(self): + for n in range(4): + self.segment(f'{n:08x}--aaaaaaaaaa--0', 'rlog.zst') + with mock.patch.object(drive_history, 'MAX_ROOT_ENTRIES', 2): + result = self.reader.snapshot() + self.assertTrue(result['scanIncomplete']) + self.assertLessEqual(sum(r['segmentCount'] for r in result['routes']), 2) + too_many = self.segment('00000009--bbbbbbbbbb--0', 'qlog.zst') + for n in range(4): + (too_many / f'extra{n}').write_bytes(b'') + with mock.patch.object(drive_history, 'MAX_SEGMENT_ENTRIES', 2): + result = self.reader.snapshot() + self.assertTrue(result['scanIncomplete']) + self.assertNotIn('00000009--bbbbbbbbbb', [r['routeId'] for r in result['routes']]) + + def test_stats_only_limit_includes_long_route_without_changing_default(self): + route_id = '2026-09-25--12-00-00' + for number in range(129): + self.segment(f'{route_id}--{number}', 'qlog.zst') + ordinary = self.reader.snapshot() + self.assertTrue(ordinary['scanIncomplete']) + self.assertEqual(ordinary['routes'][0]['segmentCount'], 64) + extended = drive_history.DriveHistory(self.root, max_segments_per_route=129).snapshot() + self.assertFalse(extended['scanIncomplete']) + self.assertEqual(extended['routes'][0]['segmentCount'], 129) + with self.assertRaises(ValueError): + drive_history.DriveHistory(self.root, max_segments_per_route=513) + + def test_stats_total_limit_keeps_competing_route_visible(self): + long_id = '2026-09-25--12-00-00' + short_id = '2026-09-26--12-00-00' + for number in range(512): + self.segment(f'{long_id}--{number}', 'qlog.zst') + self.segment(f'{short_id}--0', 'qlog.zst') + capped = drive_history.DriveHistory(self.root, max_segments_per_route=512).snapshot() + self.assertTrue(capped['scanIncomplete']) + self.assertEqual(sum(route['segmentCount'] for route in capped['routes']), 512) + stats = drive_history.DriveHistory(self.root, max_segments_per_route=512, max_segments=2048).snapshot() + self.assertFalse(stats['scanIncomplete']) + self.assertEqual([(route['routeId'], route['segmentCount']) for route in stats['routes']], + [(long_id, 512), (short_id, 1)]) + for invalid in (True, 0, 2049): + with self.assertRaises(ValueError): + drive_history.DriveHistory(self.root, max_segments_per_route=512, max_segments=invalid) + + def test_missing_root_is_empty_but_root_symlink_is_unavailable(self): + self.assertEqual(drive_history.DriveHistory(self.root / 'missing').snapshot()['routes'], []) + link = Path(self.temp.name) / 'linked' + link.symlink_to(self.root, target_is_directory=True) + with self.assertRaises(drive_history.DriveHistoryUnavailable): + drive_history.DriveHistory(link).snapshot() + + def test_delete_videos_keeps_logs_and_refuses_open_or_foreign_segments(self): + name = '0000021e--371eaf116b--0' + path = self.segment(name, 'rlog.zst', 'qlog.zst', 'fcamera.hevc', 'ecamera.hevc', 'qcamera.ts') + result = self.reader.delete_videos(name) + self.assertEqual(result, {'segmentName': name, 'deleted': ['fcamera', 'ecamera', 'qcamera'], 'freedBytes': 3}) + self.assertEqual(sorted(p.name for p in path.iterdir()), ['qlog.zst', 'rlog.zst']) + self.assertEqual(self.reader.delete_videos(name)['deleted'], []) + + open_segment = self.segment('0000021e--371eaf116b--1', 'rlog.lock', 'fcamera.hevc') + with self.assertRaises(drive_history.SegmentInUse): + self.reader.delete_videos(open_segment.name) + self.assertTrue((open_segment / 'fcamera.hevc').exists()) + + with self.assertRaises(drive_history.SegmentMissing): + self.reader.delete_videos('0000021e--371eaf116b--9') + for bad in ('../logs', '0000021e--371eaf116b', 'x/0000021e--371eaf116b--0', None): + with self.assertRaises(ValueError): + self.reader.delete_videos(bad) + + outside = Path(self.temp.name) / 'outside' + outside.mkdir() + (outside / 'fcamera.hevc').write_bytes(b'x') + os.symlink(outside, self.root / '0000021e--371eaf116b--2') + with self.assertRaises(OSError): + self.reader.delete_videos('0000021e--371eaf116b--2') + self.assertTrue((outside / 'fcamera.hevc').exists()) + + linked = self.segment('0000021e--371eaf116b--3', 'rlog.zst') + os.symlink(outside / 'fcamera.hevc', linked / 'qcamera.ts') + self.assertEqual(self.reader.delete_videos(linked.name)['deleted'], []) + self.assertTrue((outside / 'fcamera.hevc').exists()) + + +if __name__ == '__main__': + unittest.main() diff --git a/openpilot/starpilot/galaxy/tests/test_drive_stats.mjs b/openpilot/starpilot/galaxy/tests/test_drive_stats.mjs new file mode 100644 index 0000000000..1f748928c8 --- /dev/null +++ b/openpilot/starpilot/galaxy/tests/test_drive_stats.mjs @@ -0,0 +1,92 @@ +import assert from "node:assert/strict" +import { DriveStatsFeed, validDriveStats } from "../web/js/drive-stats.js" + +const route = { routeId: "2026-09-27--12-00-00", startTime: 1_790_509_200, endTime: 1_790_512_800, + distanceMeters: 12000, durationSeconds: 3600, engagedSeconds: 2700, engagedPercent: 75, + model: "Chestnut", distractedMoments: null, unresponsiveMoments: 0, + complete: true, reason: null, segmentCount: 2, ignored: false } +const fixture = { schemaVersion: 1, isMetric: true, analysis: { running: false, pending: 0, failed: 0, scanIncomplete: false }, + totals: { distanceMeters: 12000, durationSeconds: 3600, engagedSeconds: 2700, drives: 1 }, + week: { days: Array.from({ length: 7 }, (_, i) => ({ date: `2026-09-${String(21 + i).padStart(2, "0")}`, + distanceMeters: i === 6 ? 12000 : 0 })), distanceMeters: 12000, durationSeconds: 3600, + engagedPercent: 75, drives: 1, timezone: "UTC" }, + lastDrive: route, recentDrives: [route], records: [{ id: "longestDrive", label: "Longest drive", value: 12000, unit: "m" }], + models: [{ name: "Chestnut", distanceMeters: 12000, durationSeconds: 3600, drives: 1 }], + coverage: { scope: "analyzedLocalRoutes", partialHistory: true } } +const response = (body) => ({ ok: true, status: 200, json: async () => body }) +const tick = async () => { for (let i = 0; i < 8; i++) await Promise.resolve() } + +assert.equal(validDriveStats(fixture), true) +assert.equal(validDriveStats({ ...fixture, isMetric: null }), false) +assert.equal(validDriveStats({ ...fixture, week: { ...fixture.week, days: [] } }), false) +assert.equal(validDriveStats({ ...fixture, recentDrives: [{ ...route, ignored: "no" }] }), false) +assert.equal(validDriveStats({ ...fixture, totals: { ...fixture.totals, distanceMeters: NaN } }), false) +assert.equal(validDriveStats({ ...fixture, lastDrive: { ...route, routeId: "