Drive Statistics

Implement background drive analysis and statistics with explicit worker and persistence ownership.

Co-authored-by: Danny <dannyirwin31@yahoo.com>
Co-authored-by: Prabhaav Pillai <prabhaav@msu.edu>
Co-authored-by: firestarsdog <229254897+firestarsdog@users.noreply.github.com>
This commit is contained in:
firestar5683
2026-09-25 14:53:21 -05:00
parent 1a67eedd00
commit 8cea672429
12 changed files with 3005 additions and 0 deletions
@@ -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
+546
View File
@@ -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)
@@ -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()
@@ -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()
@@ -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: "<script>" } }), false)
const updates = [], requests = [], timers = new Map()
let serial = 0
let ignored = false
const later = (fn, ms) => { const id = ++serial; timers.set(id, { fn, ms }); return id }
const cancelTimer = (id) => timers.delete(id)
const feed = new DriveStatsFeed({ publish: (state) => updates.push(state),
fetcher: async (url, options) => {
requests.push({ url, options })
if (url.endsWith("ignore")) ignored = true
return response(ignored ? { ...fixture, totals: { ...fixture.totals, drives: 0 },
recentDrives: [{ ...route, ignored: true }] } : fixture)
}, later, cancelTimer })
await feed.start()
assert.equal(feed.status, "ready")
assert.equal(feed.data.lastDrive.distractedMoments, null)
assert.deepEqual(requests.map((request) => request.url), ["./api/drives/stats?timezone=" + encodeURIComponent(Intl.DateTimeFormat().resolvedOptions().timeZone || "UTC")])
assert.equal(requests[0].options.credentials, "same-origin")
assert.equal([...timers.values()].some((timer) => timer.ms === 60000), true)
await feed.ignore(route.routeId, true)
assert.deepEqual(JSON.parse(requests[1].options.body), { routeId: route.routeId, ignored: true })
assert.equal(feed.data.totals.drives, 0)
assert.equal(feed.data.recentDrives[0].ignored, true)
await feed.ignore(route.routeId, true)
assert.equal(requests.length, 3, "duplicate selection should not be sent")
assert.equal(requests[2].url, requests[0].url, "mutation refresh retains the viewer timezone")
feed.stop()
assert.equal(timers.size, 0)
assert.equal(feed.data, null)
const pending = []
let revoked = 0
const slow = new DriveStatsFeed({ publish: () => {}, unauthorized: () => { revoked++ },
fetcher: (url, options) => new Promise((resolve) => pending.push({ url, options, resolve })), later, cancelTimer })
const loading = slow.start()
assert.equal(slow.status, "loading")
slow.stop()
assert.equal(pending[0].options.signal.aborted, true)
pending[0].resolve(response(fixture))
await loading
assert.equal(slow.data, null)
const waiting = slow.start()
pending[1].resolve({ ok: false, status: 401 })
await waiting
assert.equal(revoked, 1)
assert.equal(slow.active, false)
let failedRequest
const uncertain = new DriveStatsFeed({ publish: () => {},
fetcher: (url, options) => url.includes("stats?") ? Promise.resolve(response(fixture)) :
new Promise((resolve) => { failedRequest = { resolve, options } }), later, cancelTimer })
await uncertain.start()
const changing = uncertain.ignore(route.routeId, true)
const [deadlineId, deadline] = [...timers.entries()].find(([, timer]) => timer.ms === 5000)
assert.ok(deadline)
timers.delete(deadlineId)
deadline.fn()
assert.equal(failedRequest.options.signal.aborted, true)
assert.equal(uncertain.data.recentDrives[0].ignored, false, "timeout must not invent a saved change")
assert.match(uncertain.error, /timed out/)
failedRequest.resolve(response({ ...fixture, recentDrives: [{ ...route, ignored: true }] }))
await changing
assert.equal(uncertain.data.recentDrives[0].ignored, false, "late reply must not refill state")
uncertain.stop()
assert.equal(timers.size, 0)
assert.ok(updates.some((state) => state.status === "ready"))
await tick()
console.log("Galaxy drive statistics feed checks passed")
@@ -0,0 +1,538 @@
from unittest.mock import Mock
"""Durable local-drive statistics and background analysis contract."""
from datetime import UTC, datetime
from pathlib import Path
import copy
import json
import tempfile
import threading
import time
import unittest
from unittest import mock
from openpilot.starpilot.galaxy import drive_stats
from openpilot.starpilot.galaxy.drive_stats import DriveStatsOwner, MAX_CACHED_ROUTES
NOW = datetime(2026, 9, 27, 12, tzinfo=UTC).timestamp()
class History:
def __init__(self, root, routes):
self.root = root
self.routes = routes
self.scans = 0
self.incomplete = False
def snapshot(self):
self.scans += 1
return {'routes': list(self.routes), 'scanIncomplete': self.incomplete}
def route(identifier, number=0):
return {'routeId': identifier, 'segmentCount': 1,
'segments': [{'number': number, 'segmentName': f'{identifier}--{number}', 'files': {'rlog': True}}]}
def analyzed(root, selected, permitted):
return {'routeId': selected['routeId'], 'startTime': NOW - 60, 'endTime': NOW,
'distanceMeters': 1000.0, 'durationSeconds': 60.0, 'engagedSeconds': 30.0,
'engagedPercent': 50.0, 'model': 'Test Model', 'distractedMoments': 1,
'unresponsiveMoments': 0, 'complete': True, 'reason': None, 'segmentCount': 1}
class DriveStatsTest(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.store = Path(self.temp.name) / 'state' / 'drive-stats.json'
self.history = History(self.root, [route('abc')])
self.now = [NOW]
self.owners = []
def owner(self, analyzer=analyzed, permitted=lambda: True, monotonic=None, metric=None):
owner = DriveStatsOwner(root=self.root, store=self.store, history=Mock(wraps=self.history),
analyzer=analyzer, permitted=permitted, clock=lambda: self.now[0],
monotonic=monotonic if monotonic is not None else lambda: self.now[0], metric=metric)
self.owners.append(owner)
self.addCleanup(owner.close)
return owner
def completed(self, owner, scans=1):
deadline = time.monotonic() + 3
while time.monotonic() < deadline:
with owner._lock:
worker = owner._worker
if self.history.scans >= scans and worker and not worker.is_alive():
return owner.snapshot()
time.sleep(.01)
self.fail('background scan did not finish')
def test_recording_dates_reads_cache_without_starting_analysis(self):
owner = self.owner()
record = analyzed(self.root, route('abc'), lambda: True)
owner._records = {'abc': record}
self.assertEqual(owner.recording_dates(['abc', 'missing']), {'abc': NOW - 60})
self.assertIsNone(owner._worker)
self.assertEqual(self.history.scans, 0)
def test_complete_aggregation_week_models_and_exclusion_persist(self):
owner = self.owner()
owner._start()
initial = owner.snapshot()
self.assertEqual(initial['totals']['drives'], 0)
result = self.completed(owner)
self.assertEqual(result['totals']['distanceMeters'], 1000)
self.assertEqual(result['totals']['engagedSeconds'], 30)
self.assertEqual(result['week']['engagedPercent'], 50)
self.assertEqual(result['week']['days'][-1]['distanceMeters'], 1000)
self.assertEqual(result['models'][0]['name'], 'Test Model')
self.assertEqual(result['records'][0]['id'], 'longestDrive')
ignored = owner.ignore('abc', True, lambda: True)
self.assertEqual(ignored['totals']['drives'], 0)
self.assertTrue(ignored['recentDrives'][0]['ignored'])
self.assertIsNone(ignored['lastDrive'])
second = self.owner()
loaded = second.snapshot()
self.assertEqual(loaded['totals']['drives'], 0)
self.assertTrue(loaded['recentDrives'][0]['ignored'])
with self.assertRaises(PermissionError):
second.ignore('abc', False, lambda: False)
self.assertEqual(second.ignore('abc', False, lambda: True)['totals']['drives'], 1)
def test_week_uses_viewer_calendar_at_midnight_and_dst(self):
owner = self.owner(permitted=lambda: False)
self.now[0] = datetime(2026, 9, 28, 2, tzinfo=UTC).timestamp()
record = analyzed(self.root, route('abc'), lambda: True)
record.update(startTime=self.now[0] - 60, ignored=False)
owner._records = {'abc': record}
local = owner.snapshot('America/Chicago')
self.assertEqual(local['week']['days'][-1]['date'], '2026-09-27')
self.assertEqual(local['week']['days'][-1]['distanceMeters'], 1000)
self.assertEqual(owner.snapshot()['week']['days'][0]['date'], '2026-09-28')
for stamp in ('2026-11-01T06:30:00+00:00', '2026-11-01T07:30:00+00:00'):
self.now[0] = datetime.fromisoformat(stamp).timestamp()
record['startTime'] = self.now[0]
self.assertEqual(owner.snapshot('America/Chicago')['week']['days'][-1]['distanceMeters'], 1000)
for zone in ('../UTC', '/etc/passwd', 'Unknown/Zone', '', 'x' * 129):
with self.subTest(zone=zone), self.assertRaises(ValueError):
owner.snapshot(zone)
def test_metric_preference_changes_presentation_flag_only(self):
owner = self.owner(metric=lambda: False)
owner._start()
owner.snapshot()
output = self.completed(owner)
self.assertFalse(output['isMetric'])
self.assertEqual(output['totals']['distanceMeters'], 1000)
def test_personal_records_share_the_viewer_calendar_with_week_graph(self):
owner = self.owner(permitted=lambda: False)
self.now[0] = datetime(2026, 9, 28, 6, tzinfo=UTC).timestamp()
for identifier, stamp, distance, engaged in (
('sunday', '2026-09-28T04:30:00+00:00', 1000, 60),
('monday', '2026-09-28T05:30:00+00:00', 2000, 0),
):
record = analyzed(self.root, route(identifier), lambda: True)
start = datetime.fromisoformat(stamp).timestamp()
record.update(startTime=start, endTime=start + 60, ignored=False,
distanceMeters=distance, engagedSeconds=engaged, engagedPercent=100 * engaged / 60)
owner._records[identifier] = record
saved = copy.deepcopy(owner._records)
local = owner.snapshot('America/Chicago')
records = {record['id']: record['value'] for record in local['records']}
self.assertEqual(local['week']['distanceMeters'], 2000)
self.assertEqual(local['week']['days'][0]['distanceMeters'], 2000)
self.assertEqual(records['bestWeek'], 2000)
self.assertEqual(records['mostEngagedDay'], 100)
self.assertEqual(records['highestStreak'], 2)
self.assertEqual(local['coverage']['timezone'], local['week']['timezone'])
utc = owner.snapshot()
records = {record['id']: record['value'] for record in utc['records']}
self.assertEqual(utc['week']['distanceMeters'], 3000)
self.assertEqual(records['bestWeek'], 3000)
self.assertEqual(records['mostEngagedDay'], 50)
self.assertEqual(records['highestStreak'], 1)
self.assertEqual(utc['coverage']['timezone'], 'UTC')
self.assertEqual(local['totals'], utc['totals'])
self.assertEqual(owner._records, saved)
self.assertFalse(self.store.exists(), 'viewer timezone must not rewrite stored driving history')
def test_personal_record_streak_counts_calendar_days_across_dst(self):
owner = self.owner(permitted=lambda: False)
for stamps in (
('2026-03-07T18:00:00+00:00', '2026-03-08T17:00:00+00:00', '2026-03-09T17:00:00+00:00'),
('2026-10-31T17:00:00+00:00', '2026-11-01T18:00:00+00:00', '2026-11-02T18:00:00+00:00'),
):
with self.subTest(stamps=stamps):
owner._records.clear()
for index, stamp in enumerate(stamps):
record = analyzed(self.root, route(str(index)), lambda: True)
start = datetime.fromisoformat(stamp).timestamp()
record.update(startTime=start, endTime=start + 60, ignored=False)
owner._records[str(index)] = record
records = {record['id']: record['value'] for record in owner.snapshot('America/Chicago')['records']}
self.assertEqual(records['highestStreak'], 3)
def test_repeated_dst_hour_is_one_record_day(self):
owner = self.owner(permitted=lambda: False)
self.now[0] = datetime(2026, 11, 1, 8, tzinfo=UTC).timestamp()
for index, stamp in enumerate(('2026-11-01T06:30:00+00:00', '2026-11-01T07:30:00+00:00')):
record = analyzed(self.root, route(str(index)), lambda: True)
start = datetime.fromisoformat(stamp).timestamp()
record.update(startTime=start, endTime=start + 60, ignored=False, engagedSeconds=60 * index,
engagedPercent=100 * index)
owner._records[str(index)] = record
local = owner.snapshot('America/Chicago')
records = {record['id']: record['value'] for record in local['records']}
self.assertEqual(local['week']['days'][-1]['distanceMeters'], 2000)
self.assertEqual(records['highestStreak'], 1)
self.assertEqual(records['mostEngagedDay'], 50)
def test_default_store_inventory_uses_bounded_long_route_limit(self):
owner = DriveStatsOwner(root=self.root, store=self.store, analyzer=analyzed, permitted=lambda: False)
self.addCleanup(owner.close)
self.assertEqual(owner.history.max_segments_per_route, 512)
self.assertEqual(owner.history.max_segments, 2048)
def test_cleaned_logs_retain_complete_records_and_changed_logs_invalidate(self):
segment = self.root / 'abc--0'
segment.mkdir()
log = segment / 'rlog.zst'
log.write_bytes(b'first')
owner = self.owner()
owner._start()
owner.snapshot()
self.completed(owner)
self.history.routes = []
self.now[0] += 61
owner._start()
owner.snapshot()
self.assertEqual(self.completed(owner, scans=2)['totals']['drives'], 1)
self.history.routes = [route('abc')]
log.write_bytes(b'changed')
analyzing = threading.Event()
release = threading.Event()
def delayed(root, selected, permitted):
analyzing.set()
release.wait(2)
return analyzed(root, selected, permitted)
owner.analyzer = delayed
self.now[0] += 61
owner._start()
owner.snapshot()
self.assertTrue(analyzing.wait(2))
self.assertEqual(owner.snapshot()['totals']['drives'], 0)
release.set()
self.assertEqual(self.completed(owner, scans=3)['totals']['drives'], 1)
def test_incomplete_and_failed_routes_never_contribute(self):
def incomplete(root, selected, permitted):
result = analyzed(root, selected, permitted)
result.update(complete=False, reason='gap', distanceMeters=None, engagedSeconds=None)
return result
owner = self.owner(analyzer=incomplete)
owner._start()
owner.snapshot()
result = self.completed(owner)
self.assertEqual(result['totals']['drives'], 0)
self.assertEqual(result['coverage']['incompleteRoutes'], 1)
self.assertFalse(result['recentDrives'][0]['complete'])
self.assertEqual(result['records'], [])
self.assertTrue(result['coverage']['partialHistory'])
def test_log_cleanup_keeps_completed_drive_and_qlog_change_retries(self):
segment = self.root / 'abc--0'
segment.mkdir()
rlog = segment / 'rlog.zst'
qlog = segment / 'qlog.zst'
rlog.write_bytes(b'rlog')
qlog.write_bytes(b'qlog')
calls = []
def count(root, selected, permitted):
calls.append(1)
return analyzed(root, selected, permitted)
owner = self.owner(analyzer=count)
owner._start()
owner.snapshot()
self.completed(owner)
rlog.unlink()
self.now[0] += 61
owner._start()
owner.snapshot()
self.assertEqual(self.completed(owner, scans=2)['totals']['drives'], 1)
self.assertEqual(len(calls), 1)
qlog.write_bytes(b'changed qlog')
self.now[0] += 61
owner._start()
owner.snapshot()
self.assertEqual(self.completed(owner, scans=3)['totals']['drives'], 1)
self.assertEqual(len(calls), 2)
def test_attention_records_use_independent_observations(self):
def attention(root, selected, permitted):
result = analyzed(root, selected, permitted)
result.update(distractedMoments=2, unresponsiveMoments=0)
return result
owner = self.owner(analyzer=attention)
owner._start()
owner.snapshot()
records = {record['id']: record for record in self.completed(owner)['records']}
self.assertEqual(records['cleanDriveStreak']['value'], 1)
self.assertIsNone(records['longestUndistractedDrive']['value'])
def test_last_drive_can_be_incomplete_without_inflating_totals(self):
self.history.routes = [route('old'), route('new')]
def mixed(root, selected, permitted):
result = analyzed(root, selected, permitted)
if selected['routeId'] == 'old':
result.update(startTime=NOW - 3600, endTime=NOW - 3540)
else:
result.update(complete=False, reason='gap', distanceMeters=None, engagedSeconds=None)
return result
owner = self.owner(analyzer=mixed)
owner._start()
owner.snapshot()
output = self.completed(owner)
self.assertEqual(output['totals']['drives'], 1)
self.assertEqual(output['lastDrive']['routeId'], 'new')
self.assertFalse(output['lastDrive']['complete'])
def test_parked_check_is_throttled_and_close_is_immediate(self):
ticks = [100.0]
calls = []
def parked():
calls.append(1)
return True
owner = self.owner(permitted=parked, monotonic=lambda: ticks[0])
self.assertTrue(owner._allowed())
for _ in range(100):
self.assertTrue(owner._allowed())
self.assertEqual(len(calls), 1)
ticks[0] += .11
self.assertTrue(owner._allowed())
self.assertEqual(len(calls), 2)
owner.close()
self.assertFalse(owner._allowed())
self.assertEqual(len(calls), 2)
def test_backward_wall_clock_does_not_freeze_scan(self):
ticks = [100.0]
owner = self.owner(monotonic=lambda: ticks[0])
owner._start()
owner.snapshot()
self.completed(owner)
self.now[0] -= 86400
ticks[0] += 61
owner._start()
owner.snapshot()
self.completed(owner, scans=2)
def test_failed_cache_write_rolls_back_ignore(self):
owner = self.owner()
owner._start()
owner.snapshot()
self.completed(owner)
with mock.patch.object(owner, '_save', side_effect=OSError('full disk')):
with self.assertRaises(OSError):
owner.ignore('abc', True, lambda: True)
self.assertFalse(owner.snapshot()['recentDrives'][0]['ignored'])
def test_background_work_is_bounded_and_resumes_other_routes(self):
self.history.routes = [route(f'route-{index}') for index in range(5)]
owner = self.owner()
owner._start()
owner.snapshot()
first = self.completed(owner)
self.assertEqual(first['totals']['drives'], 2)
self.assertEqual(first['analysis']['pending'], 3)
self.now[0] += 61
owner._start()
owner.snapshot()
self.assertEqual(self.completed(owner, scans=2)['totals']['drives'], 4)
self.now[0] += 61
owner._start()
owner.snapshot()
self.assertEqual(self.completed(owner, scans=3)['totals']['drives'], 5)
def test_incomplete_source_waits_for_retry_or_revision(self):
calls = []
def incomplete(root, selected, permitted):
calls.append(1)
result = analyzed(root, selected, permitted)
result.update(complete=False, reason='gap')
return result
owner = self.owner(analyzer=incomplete)
owner._start()
owner.snapshot()
result = self.completed(owner)
self.assertEqual(result['analysis']['pending'], 0)
self.assertEqual(result['analysis']['state'], 'retrying')
self.assertFalse(result['analysis']['running'])
self.now[0] += 61
owner._start()
owner.snapshot()
self.completed(owner, scans=2)
self.assertEqual(len(calls), 1)
self.now[0] += 300
owner.analyzer = analyzed
owner._start()
owner.snapshot()
self.assertEqual(self.completed(owner, scans=3)['totals']['drives'], 1)
def test_failed_analysis_leaves_queue_and_retries_after_cooldown(self):
owner = self.owner(analyzer=mock.Mock(side_effect=TimeoutError('bounded route timeout')))
owner._start()
owner.snapshot()
result = self.completed(owner)
self.assertEqual(result['analysis']['pending'], 0)
self.assertEqual(result['analysis']['failed'], 1)
self.assertEqual(result['analysis']['state'], 'retrying')
self.now[0] += 61
owner._start()
owner.snapshot()
self.completed(owner, scans=2)
self.assertEqual(owner.analyzer.call_count, 1)
owner.analyzer = analyzed
self.now[0] += 300
owner._start()
owner.snapshot()
result = self.completed(owner, scans=3)
self.assertEqual(result['analysis']['state'], 'ready')
self.assertEqual(result['analysis']['failed'], 0)
def test_analysis_upgrade_refreshes_retained_logs_but_preserves_deleted_history(self):
segment = self.root / 'abc--0'
segment.mkdir()
log = segment / 'rlog.zst'
log.write_bytes(b'original source')
owner = self.owner()
owner._start()
owner.snapshot()
self.completed(owner)
owner._records['abc'].update(analysisVersion=1, model=None)
owner._save()
self.now[0] += 61
owner._start()
owner.snapshot()
result = self.completed(owner, scans=2)
self.assertEqual(result['lastDrive']['model'], 'Test Model')
self.assertEqual(result['lastDrive']['analysisVersion'], drive_stats.ANALYSIS_VERSION)
owner._records['abc']['analysisVersion'] = 1
log.unlink()
self.now[0] += 61
owner.analyzer = mock.Mock(side_effect=AssertionError('deleted source must retain prior measurements'))
owner._start()
owner.snapshot()
result = self.completed(owner, scans=3)
self.assertEqual(result['totals']['distanceMeters'], 1000)
owner.analyzer.assert_not_called()
def test_metadata_upgrade_failure_preserves_previous_complete_totals(self):
owner = self.owner()
owner._start()
owner.snapshot()
self.completed(owner)
owner._records['abc']['analysisVersion'] = 1
owner.analyzer = lambda *args, **kwargs: {**analyzed(*args, **kwargs), 'complete': False, 'reason': 'timeout'}
self.now[0] += 61
owner._start()
owner.snapshot()
result = self.completed(owner, scans=2)
self.assertEqual(result['totals']['distanceMeters'], 1000)
self.assertEqual(result['analysis']['state'], 'retrying')
def test_offroad_wait_is_distinct_from_running_analysis(self):
result = self.owner(permitted=lambda: False).snapshot()
self.assertEqual(result['analysis']['state'], 'waitingParked')
self.assertFalse(result['analysis']['running'])
def test_byte_limit_evicts_oldest_with_partial_coverage(self):
owner = self.owner()
with owner._lock, mock.patch('openpilot.starpilot.galaxy.drive_stats.MAX_CACHE_BYTES', 2200):
for index in range(10):
record = analyzed(None, {'routeId': f'route-{index}'}, lambda: True)
record.update(ignored=False, sourceSignature='x', sourceManifest={}, model='x' * 120)
owner._records[record['routeId']] = record
owner._bound()
owner._save()
self.assertLess(len(owner._records), 10)
self.assertTrue(owner.snapshot()['coverage']['truncated'])
def test_corrupt_cache_not_replaced_until_valid_scan(self):
self.store.parent.mkdir(parents=True)
self.store.write_text('broken json')
permitted = [False]
owner = self.owner(permitted=lambda: permitted[0])
self.assertTrue(owner.snapshot()['coverage']['cacheCorrupt'])
self.assertEqual(self.store.read_text(), 'broken json')
permitted[0] = True
self.now[0] += .11
owner._start()
owner.snapshot()
result = self.completed(owner)
self.assertFalse(result['coverage']['cacheCorrupt'])
self.assertEqual(json.loads(self.store.read_text())['schemaVersion'], 1)
def test_oversized_number_in_cache_is_treated_as_corrupt(self):
self.store.parent.mkdir(parents=True)
record = analyzed(None, {'routeId': 'abc'}, lambda: True)
record.update(ignored=False, sourceSignature='x', sourceManifest={}, distanceMeters=10**400)
self.store.write_text(json.dumps({'schemaVersion': 1, 'routes': {'abc': record}}))
owner = self.owner(permitted=lambda: False)
result = owner.snapshot()
self.assertTrue(result['coverage']['cacheCorrupt'])
self.assertEqual(result['totals']['drives'], 0)
self.assertIn('abc', self.store.read_text())
def test_changed_source_during_analysis_cannot_commit_stale_result(self):
segment = self.root / 'abc--0'
segment.mkdir()
log = segment / 'qlog.zst'
log.write_bytes(b'initial')
def changes_log(root, selected, permitted):
log.write_bytes(b'replaced while decoding')
return analyzed(root, selected, permitted)
owner = self.owner(analyzer=changes_log)
owner._start()
owner.snapshot()
first = self.completed(owner)
self.assertEqual(first['totals']['drives'], 0)
self.assertEqual(first['analysis']['failed'], 1)
owner.analyzer = analyzed
self.now[0] += 61
owner._start()
owner.snapshot()
self.assertEqual(self.completed(owner, scans=2)['totals']['drives'], 1)
def test_cancel_prevents_late_commit_and_storage_is_bounded(self):
entered = threading.Event()
release = threading.Event()
def blocked(root, selected, permitted):
entered.set()
release.wait(2)
return analyzed(root, selected, permitted)
owner = self.owner(analyzer=blocked)
owner._start()
owner.snapshot()
self.assertTrue(entered.wait(2))
owner.close()
release.set()
self.assertFalse(self.store.exists())
bounded = self.owner()
with bounded._lock:
for index in range(MAX_CACHED_ROUTES + 1):
record = analyzed(None, {'routeId': f'route-{index}'}, lambda: True)
record.update(ignored=False, sourceSignature='x')
bounded._records[record['routeId']] = record
bounded._bound()
self.assertEqual(len(bounded._records), MAX_CACHED_ROUTES)
self.assertTrue(bounded._truncated)
if __name__ == '__main__':
unittest.main()
@@ -0,0 +1,91 @@
from unittest.mock import Mock
"""Device lifecycle analysis continues with no HTTP clients."""
import tempfile
import threading
import time
from pathlib import Path
from unittest import TestCase
from openpilot.starpilot.galaxy.drive_stats import DriveStatsOwner
class BackgroundHistoryTest(TestCase):
def test_no_browser_pause_resume_and_restart_reuses_completion(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
allowed = threading.Event()
entered = threading.Event()
cancelled = threading.Event()
complete = threading.Event()
calls = []
record = {
'routeId': 'example',
'startTime': 1700000000.0,
'endTime': 1700000060.0,
'distanceMeters': 100.0,
'durationSeconds': 60.0,
'engagedSeconds': 20.0,
'engagedPercent': 100 / 3,
'model': 'test',
'distractedMoments': 0,
'unresponsiveMoments': 0,
'complete': True,
'reason': None,
'segmentCount': 1,
}
class History:
def __init__(self):
self.root = root
def snapshot(self):
return {'routes': [{'routeId': 'example', 'segments': []}]}
def analyze(_root, route, *, permitted):
calls.append(route['routeId'])
entered.set()
if len(calls) == 1:
deadline = time.monotonic() + 2
while permitted() and time.monotonic() < deadline:
time.sleep(0.005)
cancelled.set()
return dict(record, complete=False, reason='Analysis cancelled')
complete.set()
return record
owner = DriveStatsOwner(root=root, store=root / 'cache.json', history=Mock(wraps=History()), analyzer=analyze, permitted=allowed.is_set)
try:
owner.snapshot()
self.assertIsNone(owner._worker, 'dashboard must only read cached status')
owner.start()
owner.start()
self.assertFalse(entered.wait(0.15))
allowed.set()
self.assertTrue(entered.wait(2), 'offroad must start without requests')
allowed.clear()
self.assertTrue(cancelled.wait(0.4), 'onroad must revoke analysis promptly')
time.sleep(0.15)
self.assertFalse(owner._failures, 'interruption is not a corrupt log')
self.assertFalse((root / 'cache.json').exists() and owner._records)
allowed.set()
self.assertTrue(complete.wait(2), 'offroad must resume without requests')
deadline = time.monotonic() + 2
while owner._worker.is_alive() and time.monotonic() < deadline:
time.sleep(0.01)
self.assertEqual(owner.snapshot()['totals']['drives'], 1)
finally:
owner.close()
restored = DriveStatsOwner(root=root, store=root / 'cache.json', history=Mock(wraps=History()), analyzer=analyze, permitted=allowed.is_set)
try:
restored.start()
deadline = time.monotonic() + 2
while restored._worker is None and time.monotonic() < deadline:
time.sleep(0.01)
restored._worker.join(2)
self.assertEqual(calls, ['example', 'example'])
self.assertEqual(restored.snapshot()['totals']['drives'], 1)
finally:
restored.close()
self.assertFalse(owner._scheduler.is_alive())
self.assertFalse(restored._scheduler.is_alive())
@@ -0,0 +1,93 @@
from unittest.mock import Mock
import tempfile
import unittest
from pathlib import Path
from openpilot.starpilot.galaxy.drive_stats import DriveStatsOwner, MAX_RETRY_ATTEMPTS
from openpilot.starpilot.galaxy.tests.test_drive_stats import History, route, analyzed, NOW
class HistoryRecoveryTest(unittest.TestCase):
def test_retry_budget_survives_restart_and_changed_sources_recover(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory) / 'logs'
root.mkdir()
store = Path(directory) / 'cache.json'
selected = route('abc')
segment = root / 'abc--0'
segment.mkdir()
source = segment / 'rlog.zst'
source.write_bytes(b'bad')
now = [NOW]
calls = []
def analyzer(root, selected, permitted):
calls.append(selected['routeId'])
result = analyzed(root, selected, permitted)
if source.read_bytes() == b'bad':
result.update(complete=False, reason='Invalid compressed recording')
return result
def owner():
return DriveStatsOwner(
root=root, store=store, history=Mock(wraps=History(root, [selected])), analyzer=analyzer, permitted=lambda: True,
clock=lambda: now[0], monotonic=lambda: now[0]
)
for _ in range(MAX_RETRY_ATTEMPTS):
current = owner()
current._run()
current.close()
now[0] += 3601
current = owner()
current._run()
self.assertEqual(len(calls), MAX_RETRY_ATTEMPTS)
self.assertEqual(current._snapshot_locked()['analysis']['state'], 'ready')
self.assertEqual(current._snapshot_locked()['analysis']['quarantined'], 1)
source.write_bytes(b'repaired')
current._run()
self.assertEqual(len(calls), MAX_RETRY_ATTEMPTS + 1)
self.assertEqual(current._snapshot_locked()['totals']['drives'], 1)
self.assertFalse(current._failures)
current.close()
def test_interruption_does_not_consume_failure_budget(self):
with tempfile.TemporaryDirectory() as directory:
active = [True]
def analyzer(root, selected, permitted):
active[0] = False
return {'routeId': selected['routeId'], 'complete': False, 'reason': 'Analysis cancelled'}
current = DriveStatsOwner(
root=Path(directory),
store=Path(directory) / 'cache.json',
history=Mock(wraps=History(Path(directory), [route('abc')])),
analyzer=analyzer,
permitted=lambda: active[0],
monotonic=lambda: NOW if active[0] else NOW + 1,
)
current._run()
self.assertFalse(current._failures)
current.close()
class RlogRecoveryTest(unittest.TestCase):
def test_complete_rlog_recovers_gapped_qlog_without_source_mismatch(self):
from openpilot.starpilot.galaxy.drive_analysis import analyze_route
from openpilot.starpilot.galaxy.tests.test_drive_analysis import BASE, DriveAnalysisTest, event, selection, write
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
def sample(stamp):
return DriveAnalysisTest.sample(Mock(wraps=self), stamp, 5, True)
qlog = [*sample(BASE), *sample(BASE + 2_000_000_000), event('sentinel', BASE + 2_000_000_000, type='endOfRoute')]
segment = write(root, 0, qlog, kind='qlog')
rlog = [*sample(BASE), *sample(BASE + 500_000_000), *sample(BASE + 1_000_000_000),
*sample(BASE + 1_500_000_000), *sample(BASE + 2_000_000_000),
event('sentinel', BASE + 2_000_000_000, type='endOfRoute')]
import zstandard as zstd
(root / segment['segmentName'] / 'rlog.zst').write_bytes(zstd.compress(b''.join(item.to_bytes() for item in rlog)))
segment['files']['rlog'] = True
result = analyze_route(root, selection(segment), permitted=lambda: True)
self.assertTrue(result['complete'], result)
self.assertEqual(result['durationSeconds'], 2)
self.assertEqual(result['distanceMeters'], 10)
@@ -0,0 +1,66 @@
import unittest
import time
from pathlib import Path
from tempfile import TemporaryDirectory
from openpilot.cereal import messaging
from openpilot.cereal.services import SERVICE_LIST
from openpilot.starpilot.models.receipt import ModelReceiptOwner, recorded_model_message, logged_model_load
from openpilot.starpilot.models.status import ModelLoad, ModelVariant
from openpilot.starpilot.galaxy.drive_analysis import analyze_route
from openpilot.starpilot.galaxy.tests.test_drive_analysis import event, write, selection, BASE
class ModelRecordingTest(unittest.TestCase):
def test_actual_model_heartbeat_transport_and_late_route_qlog(self):
load = ModelLoad(123, 17, BASE - 500_000_000, 'sc23', ModelVariant.SMALL, 'a' * 64)
owner = ModelReceiptOwner()
owner.load = load
class Publisher:
def send(self, service, message):
self.service = service
self.raw = message.to_bytes()
publisher = Publisher()
self.assertTrue(owner.publish(publisher, BASE + 200_000_000))
self.assertEqual(publisher.service, 'modelIdentity')
self.assertTrue(SERVICE_LIST['modelIdentity'].should_log)
self.assertEqual(SERVICE_LIST['modelIdentity'].decimation, 1)
heartbeat = messaging.log_from_bytes(publisher.raw)
self.assertEqual(heartbeat.which(), 'logMessage')
self.assertEqual(logged_model_load(str(heartbeat.logMessage)), load)
self.assertFalse(owner.publish(publisher, BASE + 1_000_000_000))
message = messaging.new_message(None, valid=True, logMonoTime=BASE + 200_000_000)
message.logMessage = recorded_model_message(load)
with TemporaryDirectory() as directory:
events = [
event('carState', BASE, vEgo=5),
event('selfdriveState', BASE, enabled=True),
event('drivingModelData', BASE + 100_000_000, big=False),
message,
event('carState', BASE + 500_000_000, vEgo=5),
event('selfdriveState', BASE + 500_000_000, enabled=True),
event('sentinel', BASE + 500_000_000, type='endOfRoute'),
]
result = analyze_route(Path(directory), selection(write(Path(directory), 0, events, kind='qlog')), permitted=lambda: True)
self.assertTrue(result['complete'], result)
self.assertIn('(', result['model'])
self.assertNotIn('identity not recorded', result['model'])
def test_recorder_socket_accepts_existing_event_union(self):
from openpilot.common.prefix import OpenpilotPrefix
prefix = OpenpilotPrefix()
prefix.__enter__()
self.addCleanup(prefix.__exit__, None, None, None)
publisher = messaging.PubMaster(['modelIdentity'])
subscriber = messaging.sub_sock('modelIdentity', timeout=1000)
load = ModelLoad(123, 17, BASE, 'sc23', ModelVariant.SMALL, 'a' * 64)
owner = ModelReceiptOwner()
owner.load = load
time.sleep(0.05)
self.assertTrue(owner.publish(publisher, BASE + 5_000_000_000))
received = messaging.recv_one(subscriber)
self.assertIsNotNone(received)
self.assertEqual(received.which(), 'logMessage')
self.assertEqual(logged_model_load(str(received.logMessage)), load)
@@ -0,0 +1,216 @@
import assert from 'node:assert/strict'
import { PlotsFeed, PlotsPage, PlotGraph, MAX_POINTS, appendPlotSample, chartGeometry, matchQuality, seriesPoints, validatePlotPayload } from '../web/js/plots.js'
const values = () => ({ desiredLateralAccel: .2, actualLateralAccel: .18, desiredLongitudinalAccel: .3,
actualLongitudinalAccel: .28, lateralP: .1, lateralI: .02, lateralD: null, lateralF: .08,
longitudinalP: .1, longitudinalI: .1, longitudinalF: .1, speedMps: 12,
controlsActive: true, lateralControlActive: true, longitudinalControlActive: true, controlsFresh: true, poseFresh: true,
speedSource: 'deviceMotion',
lateralSource: 'torqueState', longitudinalSource: 'aTarget', lateralTermsSource: 'torqueState',
longitudinalTermsSource: 'controlsState' })
const payload = (index, extra = {}) => ({ schemaVersion: 1, sessionId: 'ab'.repeat(8), state: 'current', sampleIndex: index,
sampleAgeSeconds: .1, values: values(), error: '', bootStabilizing: false, ...extra })
const flush = async () => { for (let i = 0; i < 8; i++) await Promise.resolve() }
function fixture() {
const requests = [], states = [], timers = new Map()
let now = 1000, timerIndex = 0, unauthorized = 0
const feed = new PlotsFeed({ publish: (state) => states.push(state), unauthorized: () => unauthorized++,
fetcher: (url, options) => new Promise((resolve) => requests.push({ url, signal: options.signal, resolve })),
now: () => now,
later: (fn, delay) => { timers.set(++timerIndex, { fn, due: now + delay }); return timerIndex },
cancel: (id) => timers.delete(id) })
const reply = async (index, body, status = 200) => {
requests[index].resolve({ ok: status === 200, status, json: async () => body })
await flush()
}
const advance = (milliseconds) => {
const end = now + milliseconds
while (true) {
const next = [...timers.entries()].filter(([, timer]) => timer.due <= end).sort((a, b) => a[1].due - b[1].due)[0]
if (!next) break
now = next[1].due; timers.delete(next[0]); next[1].fn()
}
now = end
}
return { feed, states, requests, reply, advance, timers, get unauthorized() { return unauthorized } }
}
assert.equal(validatePlotPayload(payload(1)).values.lateralSource, 'torqueState')
assert.throws(() => validatePlotPayload(payload(1, { values: { ...values(), actualLateralAccel: Infinity } })))
assert.throws(() => validatePlotPayload(payload(1, { values: { ...values(), poseFresh: false } })))
assert.throws(() => validatePlotPayload(payload(1, { state: 'stale' })))
let history = []
for (let i = 1; i <= 260; i++) history = appendPlotSample(history, payload(i), i * 750)
assert.equal(history.length, MAX_POINTS)
assert.equal(history[0].index, 21)
assert.equal(appendPlotSample(history, payload(260), 10000).length, MAX_POINTS)
const restarted = appendPlotSample(history, payload(260, { sessionId: 'cd'.repeat(8) }), 10000)
assert.equal(restarted.length, 1)
assert.equal(restarted[0].sessionId, 'cd'.repeat(8))
assert.equal(matchQuality(history, 'desiredLateralAccel', 'actualLateralAccel', { minSpeed: .5,
minDemand: .008, great: .15, good: .30, fair: .50 }).label, 'Great')
assert.equal(matchQuality([{ ...values(), at: 1, controlsActive: false }], 'desiredLateralAccel',
'actualLateralAccel', { great: .15, good: .30, fair: .50 }).label, 'N/A')
const aolHistory = Array.from({ length: 8 }, (_, i) => ({ ...values(), at: i * 750, controlsActive: false,
lateralControlActive: true, longitudinalControlActive: false }))
assert.equal(matchQuality(aolHistory, 'desiredLateralAccel', 'actualLateralAccel', { minSpeed: .5,
minDemand: .008, great: .15, good: .30, fair: .50, activeKey: 'lateralControlActive' }).label, 'Great')
assert.equal(matchQuality(aolHistory, 'desiredLongitudinalAccel', 'actualLongitudinalAccel', { minDemand: .05,
great: .32, good: .52, fair: .78, activeKey: 'longitudinalControlActive' }).label, 'N/A')
assert.equal(seriesPoints(history, ['desiredLateralAccel', 'actualLateralAccel']).length, 2)
assert.deepEqual(seriesPoints([{ at: 1, desiredLateralAccel: null }, { at: 2, desiredLateralAccel: .1 }],
['desiredLateralAccel']), [[]]) // Null is a gap, never a zero line.
assert.equal(seriesPoints([{ at: 1, desiredLateralAccel: .1 }, { at: 2, desiredLateralAccel: null },
{ at: 3, desiredLateralAccel: .2 }], ['desiredLateralAccel'])[0].length, 0) // Do not bridge missing evidence.
const shortChart = chartGeometry([{ at: 1000, desiredLateralAccel: -.25 },
{ at: 2500, desiredLateralAccel: .5 }], ['desiredLateralAccel'])
assert.equal(shortChart.top, '0.50')
assert.equal(shortChart.bottom, '-0.25')
assert.equal(shortChart.from, '−1.5 s')
assert.ok(shortChart.zeroY > 18 && shortChart.zeroY < 132)
assert.equal(chartGeometry([], ['desiredLateralAccel']).valid, false)
const live = fixture()
live.feed.start()
assert.equal(live.requests[0].url, './api/plots/live')
await live.reply(0, payload(1))
assert.equal(live.states.at(-1).status, 'current')
assert.equal(live.states.at(-1).history.length, 1)
live.advance(750)
assert.equal(live.requests.length, 2)
await live.reply(1, payload(2))
assert.equal(live.states.at(-1).history.length, 2)
live.advance(750)
await live.reply(2, payload(2)) // same producer sample cannot grow history
assert.equal(live.states.at(-1).history.length, 2)
live.feed.setPaused(true)
assert.equal(live.states.at(-1).status, 'paused')
assert.equal(live.timers.size, 0)
live.feed.setPaused(false)
assert.equal(live.requests.length, 4)
live.feed.stop()
await live.reply(3, payload(3))
assert.equal(live.states.at(-1).status, 'idle')
assert.equal(live.states.at(-1).history.length, 0)
const stale = fixture()
stale.feed.start()
await stale.reply(0, payload(1))
stale.advance(1400)
assert.equal(stale.states.at(-1).status, 'stale')
assert.equal(stale.states.at(-1).data.values.desiredLateralAccel, .2)
assert.equal(stale.states.at(-1).history.length, 1)
await stale.reply(1, payload(1, { state: 'unavailable', sampleAgeSeconds: null, values: null, error: 'No fresh controls' }))
assert.equal(stale.states.at(-1).status, 'stale')
assert.equal(stale.states.at(-1).data.sampleIndex, 1)
stale.advance(750)
await stale.reply(2, payload(2))
assert.equal(stale.states.at(-1).status, 'current')
assert.equal(stale.states.at(-1).history.length, 2)
stale.feed.stop()
assert.equal(stale.states.at(-1).data, null)
const interrupted = fixture()
interrupted.feed.start()
await interrupted.reply(0, payload(1))
interrupted.advance(2250)
assert.equal(interrupted.states.at(-1).status, 'stale')
assert.equal(interrupted.states.at(-1).data.sampleIndex, 1)
assert.equal(interrupted.states.at(-1).history.length, 1)
interrupted.feed.stop()
const gap = chartGeometry([{ at: 0, value: 1 }, { at: 750, value: 2 },
{ at: 4000, value: 3 }, { at: 4750, value: 4 }], ['value'])
assert.equal(gap.paths[0].length, 2, 'reconnection must not draw across a missing interval')
const aged = fixture()
aged.feed.start()
aged.advance(90)
await aged.reply(0, payload(1, { sampleAgeSeconds: 1.4 }))
assert.equal(aged.states.at(-1).status, 'current')
assert.equal(aged.states.at(-1).history[0].at, -400)
aged.advance(10)
assert.equal(aged.states.at(-1).status, 'stale')
aged.feed.stop()
const expiredInTransit = fixture()
expiredInTransit.feed.start()
expiredInTransit.advance(200)
await expiredInTransit.reply(0, payload(1, { sampleAgeSeconds: 1.4 }))
assert.equal(expiredInTransit.states.at(-1).status, 'stale')
assert.equal(expiredInTransit.states.at(-1).history.length, 0)
expiredInTransit.feed.stop()
const denied = fixture()
denied.feed.start()
await denied.reply(0, {}, 401)
assert.equal(denied.unauthorized, 1)
assert.equal(denied.timers.size, 0)
const timeout = fixture()
timeout.feed.start()
timeout.advance(1500)
assert.equal(timeout.states.at(-1).status, 'unavailable')
await timeout.reply(0, payload(1))
assert.equal(timeout.states.at(-1).status, 'unavailable')
timeout.feed.stop()
const prior = globalThis.document, listeners = new Map()
globalThis.document = { hidden: false, addEventListener: (name, fn) => listeners.set(name, fn),
removeEventListener: (name, fn) => { if (listeners.get(name) === fn) listeners.delete(name) } }
const page = { mode: 'local', paused: false, feed: { starts: 0, pauses: [], stops: 0,
active: false, start() { this.starts++; this.active = true }, setPaused(value) { this.pauses.push(value) }, stop() { this.stops++; this.active = false } } }
PlotsPage.mounted.call(page)
assert.equal(page.feed.starts, 1)
document.hidden = true; listeners.get('visibilitychange')()
assert.deepEqual(page.feed.pauses, [true])
document.hidden = false; listeners.get('visibilitychange')()
assert.deepEqual(page.feed.pauses, [true, false])
PlotsPage.beforeUnmount.call(page)
assert.equal(page.feed.stops, 1)
assert.equal(listeners.size, 0)
const hidden = { mode: 'local', paused: false, feed: { active: false, starts: 0, pauses: [], stops: 0,
start() { this.active = true; this.starts++ }, setPaused(value) { this.pauses.push(value) }, stop() { this.stops++ } } }
document.hidden = true
PlotsPage.mounted.call(hidden)
assert.equal(hidden.feed.starts, 0)
document.hidden = false
listeners.get('visibilitychange')()
assert.equal(hidden.feed.starts, 1)
listeners.get('visibilitychange')()
assert.equal(hidden.feed.starts, 1)
PlotsPage.beforeUnmount.call(hidden)
assert.equal(listeners.size, 0)
globalThis.document = prior
assert.match(PlotsPage.template, /Lateral acceleration/)
assert.match(PlotsPage.template, /Longitudinal acceleration/)
assert.doesNotMatch(PlotsPage.template, /Download|Apply tune|Set control/)
// Compile and render the actual Vue page template with live-looking values.
globalThis.document = { createElement: () => ({
set innerHTML(value) {
this.textContent = value.replaceAll('&amp;', '&')
this.children = [{ getAttribute: () => this.textContent.match(/foo="([\s\S]*?)"/)?.[1] ?? '' }]
}, textContent: '', children: [],
}) }
const { compile } = await import('../web/vendor/vue/vue.esm-browser.js')
const render = compile(PlotsPage.template)
const chart = chartGeometry([{ ...values(), at: 1000 }, { ...values(), desiredLateralAccel: .3, at: 2500 }],
['desiredLateralAccel', 'actualLateralAccel'])
const context = { mode: 'local', status: 'current', data: payload(1), paused: false, advanced: false, error: '',
lateralChart: chart, longChart: chart, lateralTermsChart: chart, longTermsChart: chart,
lateralQuality: { label: 'Good', detail: '8 samples' }, longQuality: { label: 'Fair', detail: '8 samples' },
source: (source) => source, label: (number) => String(number), togglePause: () => {}, }
const rendered = JSON.stringify(render(context, []))
assert.match(rendered, /Lateral acceleration/)
assert.match(rendered, /Longitudinal acceleration/)
assert.match(rendered, /Good/)
assert.match(rendered, /−1.5 s/)
const graphRender = compile(PlotGraph.template)
const graph = JSON.stringify(graphRender({ chart, title: 'Lateral acceleration history', terms: false }, []))
assert.match(graph, /0.30/)
assert.match(graph, /0.00/)
assert.match(graph, /−1.5 s/)
assert.match(graph, /Latest/)
assert.match(graph, /48.0,/) // Plot geometry stays inside the labeled axes.
globalThis.document = prior
console.log('Galaxy Plots browser lifecycle and rendering contracts passed')
@@ -0,0 +1,341 @@
import assert from 'node:assert/strict'
import { readFileSync } from 'node:fs'
import { BottomNav } from '../web/js/bottom-nav.js'
import { LocalHistoryFeed, LocalRecordingsPage, availableFiles, hasVideo, validLocalHistory, quickRoadUrl,
segmentSummaryUrl, validSegmentSummary, routeFiles, firstQuickVideo, routeDate, connectRouteUrl, routeMatchesSearch } from '../web/js/record-history.js'
const flush = async () => { for (let i = 0; i < 8; i++) await Promise.resolve() }
const current = '0000021e--371eaf116b'
const history = { schemaVersion: 1, source: 'local', partialHistory: true, scanIncomplete: true,
routes: [{ routeId: current, segmentCount: 2, segments: [
{ number: 0, segmentName: `${current}--0`, files: { rlog: true, qlog: true, fcamera: true, dcamera: false, ecamera: false, qcamera: true } },
{ number: 2, segmentName: `${current}--2`, files: { rlog: false, qlog: true, fcamera: false, dcamera: true, ecamera: false, qcamera: true } },
] }] }
assert.equal(validLocalHistory(history), true)
assert.equal(validLocalHistory({ ...history, source: 'cloud' }), false)
assert.equal(validLocalHistory({ ...history, routes: [{ ...history.routes[0], routeId: '<script>' }] }), false)
assert.equal(validLocalHistory({ ...history, routes: [{ ...history.routes[0], segmentCount: 3 }] }), false)
assert.deepEqual(availableFiles(history.routes[0].segments[0].files), ['Full log', 'Quick log', 'Road video', 'Quick video'])
assert.equal(quickRoadUrl(`${current}--0`), `./api/recordings/media/${current}--0`)
assert.equal(quickRoadUrl('../qcamera.ts'), null)
assert.equal(segmentSummaryUrl(`${current}--0`), `./api/recordings/segment-summary?segmentName=${current}--0`)
assert.equal(segmentSummaryUrl('../rlog.zst'), null)
const summary = { schemaVersion: 1, source: 'closed_local_rlog', segmentName: `${current}--0`, sourceSha256: 'a'.repeat(64),
observedCarSpanSeconds: 2, estimatedDistanceMeters: 15, observedLatActiveSeconds: 2,
observedLongActiveSeconds: 1, gaps: { carState: 0, carControl: 0 }, sampleCoverageComplete: true }
assert.equal(validSegmentSummary(summary, `${current}--0`), true)
assert.equal(validSegmentSummary({ ...summary, observedLongActiveSeconds: -1 }, `${current}--0`), false)
assert.equal(validSegmentSummary(summary, `${current}--2`), false)
assert.match(LocalRecordingsPage.template, /Local recordings/)
assert.equal(hasVideo(history.routes[0].segments[0].files), true)
assert.equal(hasVideo({ rlog: true, qlog: true, fcamera: false, dcamera: false, ecamera: false, qcamera: false }), false)
assert.match(LocalRecordingsPage.template, /v-if="hasVideo\(segment\.files\)"[^>]*@click="confirmDeleteVideos\(segment\)"/)
assert.match(LocalRecordingsPage.template, /scan was incomplete/)
assert.doesNotMatch(LocalRecordingsPage.template, /v-html|deleteRoute/)
assert.match(LocalRecordingsPage.template, /aria-label="Video segment"/)
assert.doesNotMatch(LocalRecordingsPage.template, /<details[^>]*\sopen[\s=>]/)
assert.deepEqual(routeFiles(history.routes[0]), ['Full log', 'Quick log', 'Road video', 'Driver video', 'Quick video'])
assert.equal(firstQuickVideo(history.routes[0]).number, 0)
assert.equal(firstQuickVideo({ segments: [] }), null)
const dated = { ...history.routes[0], startTime: Date.parse('2026-09-28T01:30:00Z') / 1000, fileTime: 1 }
assert.match(routeDate(dated, 'en-US', 'America/Chicago').label, /Sep 27, 2026/)
assert.equal(routeDate(dated).source, '')
assert.equal(routeDate({ ...dated, startTime: null }, 'en-US', 'UTC').source, 'File date')
assert.equal(routeDate({ ...dated, startTime: '2026-09-28', fileTime: Infinity }).label, 'Date unavailable')
assert.equal(routeDate(history.routes[0]).datetime, null, 'legacy server data remains usable without an invented date')
const connected = { ...dated, connectUrl: `https://connect.comma.ai/abcdef0123456789/${current}` }
assert.equal(connectRouteUrl(connected), connected.connectUrl)
assert.equal(connectRouteUrl({ ...connected, routeId: `abcdef0123456789|${current}` }), connected.connectUrl)
for (const connectUrl of [`https://example.com/abcdef0123456789/${current}`, `${connected.connectUrl}?redirect=evil`,
`https://connect.comma.ai/abcdef0123456789/wrong-route`, 'javascript:alert(1)', null]) {
assert.equal(connectRouteUrl({ ...connected, connectUrl }), null)
}
assert.equal(connectRouteUrl({ ...connected, routeId: `0000000000000000|${current}` }), null)
const cards = LocalRecordingsPage.computed.routeCards.call({ query: "", data: { routes: [connected] } })
assert.equal(cards[0].connect, connected.connectUrl)
assert.equal(cards[0].firstQuick.number, 0)
assert.equal(cards[0].fileLabels.length, 5)
const app = readFileSync(new URL('../web/js/app.js', import.meta.url), 'utf8')
assert.match(app, /<LocalRecordingsPage v-else-if="route.path === '\/recordings' && authState.localAccess"/)
assert.match(app, /<div v-if="authState.localAccess" class="gx-nav-section"><div class="gx-nav-section__title">Routes<\/div>/)
assert.match(app, /<GxState v-else-if="route.path === '\/recordings'">Routes require a local Galaxy connection/)
assert.match(app, /<BottomNav[^>]*:local-access="authState.localAccess"/)
assert.match(BottomNav.template, /:disabled="item.path === '\/recordings' && !localAccess"/)
function fixture() {
const requests = [], states = [], timers = new Map()
let next = 0, revoked = 0
const feed = new LocalHistoryFeed({ publish: (state) => states.push(state), unauthorized: () => revoked++,
fetcher: (url, options) => new Promise((resolve) => requests.push({ url, options, resolve })),
later: (fn, ms) => { timers.set(++next, { fn, ms }); return next }, cancelTimer: (id) => timers.delete(id) })
async function reply(index, body, code = 200) {
requests[index].resolve({ status: code, ok: code === 200, json: async () => body })
await flush()
}
return { feed, requests, states, timers, reply, get revoked() { return revoked } }
}
const normal = fixture()
normal.feed.start()
assert.equal(normal.requests[0].url, './api/recordings/local')
assert.equal(normal.requests[0].options.credentials, 'same-origin')
await normal.reply(0, history)
assert.equal(normal.states.at(-1).status, 'ready')
assert.equal(normal.states.at(-1).data.routes[0].routeId, current)
assert.equal(normal.timers.size, 1)
await flush()
assert.equal(normal.requests.length, 1, 'no repeated scans without explicit refresh')
normal.feed.load()
assert.equal(normal.requests.length, 2)
await normal.reply(1, { ...history, scanIncomplete: false })
normal.feed.stop()
assert.equal(normal.states.at(-1).data, null)
const stale = fixture()
stale.feed.start()
stale.feed.stop()
stale.feed.start()
await stale.reply(1, history)
const count = stale.states.length
await stale.reply(0, { ...history, routes: [] })
assert.equal(stale.states.length, count)
assert.equal(stale.states.at(-1).data.routes.length, 1)
const revoked = fixture()
revoked.feed.start()
await revoked.reply(0, { code: 'setup_required' }, 503)
assert.equal(revoked.revoked, 1)
assert.equal(revoked.feed.data, null)
assert.equal(revoked.timers.size, 0)
const loggedOut = fixture()
loggedOut.feed.start()
await loggedOut.reply(0, {}, 401)
assert.equal(loggedOut.revoked, 1)
const unavailable = fixture()
unavailable.feed.start()
await unavailable.reply(0, { code: 'backend_unavailable' }, 503)
assert.equal(unavailable.revoked, 0)
assert.equal(unavailable.feed.status, 'unavailable')
const late = fixture()
late.feed.start()
let resolveBody
late.requests[0].resolve({ status: 200, ok: true, json: () => new Promise((resolve) => { resolveBody = resolve }) })
await flush()
late.feed.stop()
resolveBody(history)
await flush()
assert.equal(late.feed.data, null)
// A transport that ignores AbortSignal cannot strand the UI in loading.
const timedOut = fixture()
timedOut.feed.start()
assert.deepEqual([...timedOut.timers.values()].map((timer) => timer.ms), [10000, 4000])
const [timeoutId, timeout] = [...timedOut.timers.entries()].find(([, timer]) => timer.ms === 4000)
timedOut.timers.delete(timeoutId)
timeout.fn()
assert.equal(timedOut.feed.status, 'unavailable')
assert.match(timedOut.feed.error, /timed out/)
timedOut.feed.load()
assert.equal(timedOut.requests.length, 2, 'manual refresh works after timeout')
await timedOut.reply(1, history)
await timedOut.reply(0, { ...history, routes: [] })
assert.equal(timedOut.feed.status, 'ready')
assert.equal(timedOut.feed.data.routes.length, 1)
const slowBody = fixture()
slowBody.feed.start()
let finishSlowBody
slowBody.requests[0].resolve({ status: 200, ok: true, json: () => new Promise((resolve) => { finishSlowBody = resolve }) })
await flush()
const [slowId, slowTimer] = [...slowBody.timers.entries()].find(([, timer]) => timer.ms === 4000)
slowBody.timers.delete(slowId)
slowTimer.fn()
assert.equal(slowBody.feed.status, 'unavailable')
slowBody.feed.load()
await slowBody.reply(1, history)
finishSlowBody({ ...history, routes: [] })
await flush()
assert.equal(slowBody.feed.data.routes.length, 1, 'old body cannot replace refreshed inventory')
const priorDocument = globalThis.document
const listeners = new Map()
globalThis.document = { hidden: true, addEventListener: (name, cb) => listeners.set(name, cb),
removeEventListener: (name, cb) => { if (listeners.get(name) === cb) listeners.delete(name) } }
const page = { mode: 'local', $data: { status: 'idle', data: null, error: '' }, $refs: {}, playerGeneration: 0,
unauthorized() { throw new Error('revoked') } }
Object.assign(page, LocalRecordingsPage.methods)
LocalRecordingsPage.created.call(page)
const pending = []
page.feed.fetcher = (url, options) => new Promise((resolve) => pending.push({ url, options, resolve }))
LocalRecordingsPage.mounted.call(page)
assert.equal(pending.length, 0, 'hidden mount does not scan')
document.hidden = false
listeners.get('visibilitychange')()
assert.equal(pending.length, 1)
document.hidden = true
listeners.get('visibilitychange')()
assert.equal(page.$data.data, null)
LocalRecordingsPage.beforeUnmount.call(page)
assert.equal(listeners.size, 0)
globalThis.document = priorDocument
// A selected segment is immediately visible, and retired media errors cannot act on a new video.
const controls = LocalRecordingsPage.methods
const player = { mode: 'local', status: 'ready', replayRoute: null, playing: null, playerError: '', playerGeneration: 0,
sessionProbe: null, $refs: {}, $nextTick(callback) { callback() }, unauthorized() { this.revoked = true } }
for (const [name, method] of Object.entries(controls)) player[name] = method
player.openPlayer(history.routes[0], history.routes[0].segments[0])
assert.equal(player.playing.segments.length, 2)
assert.equal(player.playing.index, 0)
player.chooseSegment(1)
assert.equal(player.playing.index, 1)
const oldVideo = {}
player.$refs.quickVideo = { dataset: { playerGeneration: String(player.playerGeneration) },
pause() {}, removeAttribute() {}, load() {} }
const oldFetch = globalThis.fetch
let sessionRequest
globalThis.fetch = (_url, options) => new Promise((resolve) => { sessionRequest = { options, resolve } })
await player.videoError({ currentTarget: oldVideo })
assert.equal(sessionRequest, undefined, 'late error from retired element is ignored')
const active = player.$refs.quickVideo
active.dataset.playerGeneration = String(player.playerGeneration - 1)
await player.videoError({ currentTarget: active })
assert.equal(sessionRequest, undefined, 'retired generation on an old element is ignored')
active.dataset.playerGeneration = String(player.playerGeneration)
const checking = player.videoError({ currentTarget: active })
await flush()
assert.equal(sessionRequest.options.credentials, 'same-origin')
player.closePlayer()
assert.equal(sessionRequest.options.signal.aborted, true)
sessionRequest.resolve({ status: 401, json: async () => ({ authenticated: false }) })
await checking
assert.equal(player.revoked, undefined, 'retired session check cannot revoke reopened player')
player.openPlayer(history.routes[0], history.routes[0].segments[0])
player.$refs.quickVideo = { dataset: { playerGeneration: String(player.playerGeneration) },
pause() {}, removeAttribute() {}, load() {} }
sessionRequest = undefined
const revokedVideo = player.videoError({ currentTarget: player.$refs.quickVideo })
await flush()
sessionRequest.resolve({ status: 401, json: async () => ({ authenticated: false }) })
await revokedVideo
assert.equal(player.playing, null)
assert.equal(player.revoked, true, 'current media auth loss clears playback and session')
player.mode = 'offline-preview'
sessionRequest = undefined
player.openPlayer(history.routes[0], history.routes[0].segments[0])
await player.videoError({ currentTarget: player.$refs.quickVideo })
assert.equal(player.playing, null, 'offline preview never opens a video')
assert.equal(sessionRequest, undefined, 'offline preview does not request media or session')
globalThis.fetch = oldFetch
assert.match(LocalRecordingsPage.template, /cameraLabel\(playing.camera\)/)
assert.match(LocalRecordingsPage.template, /@error="videoError\(\$event\)"/)
assert.match(LocalRecordingsPage.template, /Distance is estimated from recorded speed/)
assert.match(LocalRecordingsPage.template, /Steering active/)
assert.match(LocalRecordingsPage.template, /Longitudinal active/)
// The mounted page requests details only on selection; hide, timeout and late responses retire them.
const savedDocument = globalThis.document
const savedSetTimeout = globalThis.setTimeout
const savedClearTimeout = globalThis.clearTimeout
globalThis.document = { hidden: false }
const detailTimers = new Map()
let timerId = 0
globalThis.setTimeout = (callback, delay) => { detailTimers.set(++timerId, { callback, delay }); return timerId }
globalThis.clearTimeout = (id) => detailTimers.delete(id)
const detailsPage = { mode: 'local', status: 'ready', details: null, detailsName: null, detailsStatus: 'idle',
detailsError: '', detailsGeneration: 0, $refs: {}, unauthorized() { this.revoked = true } }
for (const [name, method] of Object.entries(controls)) detailsPage[name] = method
const detailRequests = []
globalThis.fetch = (url, options) => new Promise((resolve) => detailRequests.push({ url, options, resolve }))
const segment = history.routes[0].segments[0]
const firstDetails = detailsPage.openDetails(segment)
assert.equal(detailRequests.length, 1)
assert.equal(detailRequests[0].options.credentials, 'same-origin')
assert.deepEqual([...detailTimers.values()].map((item) => item.delay), [10000])
detailsPage.closeDetails()
assert.equal(detailRequests[0].options.signal.aborted, true)
detailRequests[0].resolve({ status: 200, ok: true, json: async () => summary })
await firstDetails
assert.equal(detailsPage.details, null)
const timedDetails = detailsPage.openDetails(segment)
const detailTimeout = [...detailTimers.values()][0]
detailTimeout.callback()
assert.equal(detailsPage.detailsStatus, 'unavailable')
const freshDetails = detailsPage.openDetails(segment)
detailRequests[2].resolve({ status: 200, ok: true, json: async () => summary })
await freshDetails
detailRequests[1].resolve({ status: 200, ok: true, json: async () => ({ ...summary, estimatedDistanceMeters: 999 }) })
await timedDetails
assert.equal(detailsPage.details.estimatedDistanceMeters, 15)
const revokeDetails = detailsPage.openDetails(segment)
detailRequests[3].resolve({ status: 503, ok: false, json: async () => ({ code: 'setup_required' }) })
await revokeDetails
assert.equal(detailsPage.revoked, true)
assert.equal(detailsPage.details, null)
detailsPage.mode = 'offline-preview'
await detailsPage.openDetails(segment)
assert.equal(detailRequests.length, 4)
globalThis.fetch = oldFetch
globalThis.document = savedDocument
globalThis.setTimeout = savedSetTimeout
globalThis.clearTimeout = savedClearTimeout
// Dom-style library filtering, duration ordering, and fallback camera selection.
assert.equal(routeMatchesSearch({ ...history.routes[0], displayName: "Café trip" }, "cafe TRIP"), true)
assert.equal(routeMatchesSearch(history.routes[0], "missing drive"), false)
const short = { ...history.routes[0], routeId: "0000021f--371eaf116b", segmentCount: 1, segments: history.routes[0].segments.slice(0, 1) }
const library = { data: { routes: [short, history.routes[0]] }, sortOrder: "longest", preservedOnly: false, query: "" }
assert.equal(LocalRecordingsPage.computed.routeCards.call(library)[0].segmentCount, 2)
library.sortOrder = "shortest"
assert.equal(LocalRecordingsPage.computed.routeCards.call(library)[0].segmentCount, 1)
library.preservedOnly = true
assert.equal(LocalRecordingsPage.computed.routeCards.call(library).length, 0)
player.mode = "local"
player.watchRoute(history.routes[0])
assert.equal(player.playing.camera, "fcamera")
player.selectCamera("dcamera")
assert.equal(player.playing.camera, "dcamera")
assert.equal(player.playing.segments[0].number, 2)
player.closePlayer()
for (const segment of history.routes[0].segments) {
player.watchRoute(history.routes[0], segment)
assert.equal(player.playing.segments[player.playing.index].number, segment.number, 'Play video starts at the selected segment')
assert.equal(player.replayRoute, null, 'plain video does not launch UI replay')
player.closePlayer()
}
assert.match(LocalRecordingsPage.template, /@click="watchRoute\(logsRoute, segment\)">Play video/)
assert.match(LocalRecordingsPage.template, /<h2>Routes<\/h2>/)
// Confirmation has no side effects; errors stay in the dialog, and success refreshes inventory.
const deletePage = { ...LocalRecordingsPage.data(), mode: 'local', status: 'ready', routeCards: [],
unauthorized: () => assert.fail('unexpected unauthorized'), feed: { async load() { deletePage.loads++ } }, loads: 0 }
for (const [key, method] of Object.entries(LocalRecordingsPage.methods)) deletePage[key] = method.bind(deletePage)
const deleteSegment = { segmentName: `${current}--0`, number: 0, files: { qcamera: true } }
const realFetch = globalThis.fetch
const deleteCalls = []
try {
globalThis.fetch = async (path, options) => {
deleteCalls.push([path, JSON.parse(options.body)])
return { ok: false, status: 409, json: async () => ({ error: 'Turn off the vehicle before managing recordings' }) }
}
deletePage.confirmDeleteVideos(deleteSegment)
assert.equal(deleteCalls.length, 0)
await deletePage.deleteVideos()
assert.equal(deletePage.deleteSelection, deleteSegment)
assert.match(deletePage.deleteError, /Turn off the vehicle/)
assert.equal(deletePage.loads, 0)
globalThis.fetch = async () => ({ ok: true, status: 200, json: async () => ({ deleted: ['qcamera'] }) })
await deletePage.deleteVideos()
assert.equal(deletePage.deleteSelection, null)
assert.equal(deletePage.loads, 1)
assert.equal(deletePage.deleting, null)
} finally { globalThis.fetch = realFetch }
const replayPage = { closePlayer() { this.playing = null }, playing: {} }
LocalRecordingsPage.methods.openReplay.call(replayPage, history.routes[0], 0)
assert.deepEqual(replayPage.replayRoute, { routeId: current, segment: 0 })
assert.equal(replayPage.playing, null)
LocalRecordingsPage.methods.openReplay.call(replayPage, history.routes[0])
assert.deepEqual(replayPage.replayRoute, { routeId: current, segment: null }, 'full route does not retain a prior selected segment')
assert.match(LocalRecordingsPage.template, /@click="openLogs\(route\)">\{\{ route.segmentCount \}\}/)
assert.match(LocalRecordingsPage.template, /:disabled="!replayableSegment\(segment\)" @click="openReplay\(logsRoute, segment.number\)"/)
assert.match(LocalRecordingsPage.template, /:start-segment="replayRoute.segment"/)
+120
View File
@@ -0,0 +1,120 @@
"""StarPilot writable state shares one namespace without touching /persist."""
import os
from pathlib import Path
import tempfile
import time
import unittest
from unittest.mock import patch
from openpilot.common.params import Params
from openpilot.starpilot.galaxy.access import AccessStatus, default_owner
from openpilot.starpilot.galaxy.drive_stats import DriveStatsOwner
from openpilot.starpilot.galaxy.remote import default_remote_pairing
from openpilot.starpilot.state_migration import prepare_manager_start
from openpilot.starpilot import storage
class TestStarPilotStorage(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.base = Path(temporary.name)
for context in (patch('openpilot.common.hardware.PC', False),
patch.dict(os.environ, {'OPENPILOT_PREFIX': 'storage-test'})):
context.start()
self.addCleanup(context.stop)
def test_desktop_keeps_existing_persist_location(self):
with patch.object(storage, 'PC', True), patch.object(storage.Paths, 'persist_root', return_value=str(self.base / 'persist')):
self.assertEqual(storage.starpilot_storage_root(), self.base / 'persist' / 'starpilot')
def test_device_default_and_isolated_prefix(self):
with patch.object(storage, 'PC', False), patch.object(storage, '_DEVICE_ROOT', self.base / 'data' / 'starpilot'):
with patch.dict(os.environ, {'OPENPILOT_PREFIX': ''}):
self.assertEqual(storage.starpilot_storage_root(), self.base / 'data' / 'starpilot')
with patch.dict(os.environ, {'OPENPILOT_PREFIX': 'desk-123'}):
self.assertEqual(storage.starpilot_storage_root(), self.base / 'data' / 'starpilot-desk-123')
for unsafe in ('../other', '/tmp/other', 'a.b', 'x' * 65):
with self.subTest(prefix=unsafe), patch.dict(os.environ, {'OPENPILOT_PREFIX': unsafe}):
with self.assertRaises(ValueError):
storage.starpilot_storage_root()
def test_device_migration_and_galaxy_use_same_private_root(self):
params = Params(str(self.base / 'params'))
device_root = self.base / 'data' / 'starpilot'
device_root.parent.mkdir()
with patch.object(storage, 'PC', False), patch('openpilot.common.hardware.PC', False), \
patch.object(storage, '_DEVICE_ROOT', device_root), \
patch.dict(os.environ, {'OPENPILOT_PREFIX': 'desk'}):
root = storage.starpilot_storage_root()
prepare_manager_start(params, root)
self.assertTrue((root / 'profiles').is_dir())
owner = default_owner()
self.assertEqual(owner.root, root / 'galaxy')
self.assertIsNone(owner.legacy_root)
self.assertTrue(owner.configure('fixture-password', lambda: True))
self.assertEqual(owner.status().status, AccessStatus.CONFIGURED_LOCAL)
self.assertEqual(owner.root.stat().st_mode & 0o777, 0o700)
self.assertFalse(device_root.exists())
def test_home_history_before_first_pairing_keeps_shared_storage_private(self):
logs = self.base / 'logs'
logs.mkdir()
with patch.object(storage, 'starpilot_storage_root', return_value=self.base):
history = DriveStatsOwner(root=logs, permitted=lambda: True)
self.addCleanup(history.close)
history.start()
history.snapshot()
deadline = time.monotonic() + 3
while not history.store.exists() and time.monotonic() < deadline:
time.sleep(.01)
self.assertTrue(history.store.is_file())
saved_history = history.store.read_bytes()
owner = default_owner()
pairing = default_remote_pairing()
self.assertEqual(owner.status().status, AccessStatus.UNCONFIGURED)
self.assertTrue(owner.configure('fixture-password', lambda: True))
self.assertIsNotNone(pairing.pair('a' * 64))
self.assertEqual(history.store.read_bytes(), saved_history)
self.assertEqual(owner.root.stat().st_mode & 0o777, 0o700)
def test_old_history_directory_permissions_are_tightened_without_changing_files(self):
root = self.base / 'galaxy'
root.mkdir(mode=0o700)
with patch.object(storage, 'starpilot_storage_root', return_value=self.base):
owner = default_owner()
self.assertTrue(owner.configure('fixture-password', lambda: True))
pairing = default_remote_pairing()
self.assertIsNotNone(pairing.pair('a' * 64))
(root / 'drive-stats.json').write_text('{"schemaVersion":1}')
saved = {path.name: path.read_bytes() for path in root.iterdir()}
root.chmod(0o755)
self.assertEqual(default_owner().status().status, AccessStatus.CONFIGURED_LOCAL)
self.assertTrue(owner.verify('fixture-password'))
self.assertIsNotNone(default_remote_pairing().read())
self.assertEqual(root.stat().st_mode & 0o777, 0o700)
self.assertEqual({path.name: path.read_bytes() for path in root.iterdir()}, saved)
def test_shared_storage_does_not_repair_symlink_or_writable_directory(self):
outside = self.base / 'outside'
outside.mkdir(mode=0o755)
root = self.base / 'galaxy'
root.symlink_to(outside, target_is_directory=True)
with patch.object(storage, 'starpilot_storage_root', return_value=self.base):
self.assertEqual(default_owner().status().status, AccessStatus.UNAVAILABLE)
self.assertEqual(outside.stat().st_mode & 0o777, 0o755)
root.unlink()
root.mkdir()
root.chmod(0o777)
self.assertEqual(default_owner().status().status, AccessStatus.UNAVAILABLE)
self.assertEqual(root.stat().st_mode & 0o777, 0o777)
def test_private_directory_migration_does_not_admit_unsafe_credential_file(self):
root = self.base / 'galaxy'
root.mkdir(mode=0o755)
(root / 'access-v1.json').write_text('{}')
(root / 'access-v1.json').chmod(0o644)
with patch.object(storage, 'starpilot_storage_root', return_value=self.base):
self.assertEqual(default_owner().status().status, AccessStatus.UNAVAILABLE)
self.assertEqual((root / 'access-v1.json').read_text(), '{}')