Fix offroad learner report types and hosted static checks

openpilot/starpilot/flm/offroad_control_metadata.py

@@ -69,10 +69,16 @@ def read_control_metadata(event):
     int(row.sourceReceiptMonoTime),
     int(row.nativeObservedMonoTime),
     int(row.nativeValidUntilMonoTime),
-    *(
-      bool(getattr(row, name))
-      for name in ('enabled', 'active', 'valid', 'manual', 'override', 'fault', 'saturated', 'limited', 'nativeRequired', 'nativeAllowed')
-    ),
+    bool(row.enabled),
+    bool(row.active),
+    bool(row.valid),
+    bool(row.manual),
+    bool(row.override),
+    bool(row.fault),
+    bool(row.saturated),
+    bool(row.limited),
+    bool(row.nativeRequired),
+    bool(row.nativeAllowed),
     mode,
     revision,
     baseline,

openpilot/starpilot/flm/offroad_daemon.py

@@ -7,6 +7,7 @@ import signal
 import tempfile
 import time
 import traceback
+from typing import Any
 
 from openpilot.cereal import messaging
 from openpilot.common.params import Params
@@ -19,7 +20,7 @@ class StatusReport:
     self.path = report_path() if path is None else Path(path)
     self.clock_ns = clock_ns
     self.started = self.clock_ns()
-    self.current = {'status': 'collecting', 'reason': 'waiting_for_completed_drive', 'vehicle': ''}
+    self.current: dict[str, Any] = {'status': 'collecting', 'reason': 'waiting_for_completed_drive', 'vehicle': ''}
     self.next_ns = 0
 
   def set(self, value):
@@ -42,7 +43,7 @@ class StatusReport:
       value = counts.get(key, 0)
       return value if type(value) is int and 0 <= value < 2**63 else 0
 
-    details = {'Analysis runs while parked': True, 'Next-drive trial is reversible': True, 'Physical road improvement proven': False}
+    details: dict[str, Any] = {'Analysis runs while parked': True, 'Next-drive trial is reversible': True, 'Physical road improvement proven': False}
     for key, title in (('source_sha256', 'Original completed route hash'), ('assessment_sha256', 'Offroad assessment hash')):
       if key in value:
         details[title] = value[key]

openpilot/starpilot/flm/offroad_evidence.py

@@ -71,7 +71,7 @@ class RouteWindow:
       and 20 <= len(self.points) <= MAX_SAMPLES
       and all(p.valid() for p in self.points)
       and len({(p.route, p.segment, p.domain, p.context_key) for p in self.points}) == 1
-      and all(0 < b.mono_ns - a.mono_ns <= MAX_GAP_NS for a, b in zip(self.points, self.points[1:]))
+      and all(0 < b.mono_ns - a.mono_ns <= MAX_GAP_NS for a, b in zip(self.points, self.points[1:], strict=False))
       and self.identity == window_identity(self.points)
     )
 
@@ -141,7 +141,7 @@ def eligible_points(points):
   """Preserve the original override buffers; gaps are never interpolated."""
   if not isinstance(points, tuple) or len(points) > MAX_SAMPLES or any(not p.valid() for p in points):
     raise ValueError('bounded valid original route points required')
-  if any(b.mono_ns <= a.mono_ns for a, b in zip(points, points[1:])):
+  if any(b.mono_ns <= a.mono_ns for a, b in zip(points, points[1:], strict=False)):
     raise ValueError('original route points must increase')
   if len({(p.route, p.domain, p.context_key) for p in points}) > 1:
     raise ValueError('process each original route/controller context separately')
@@ -207,7 +207,8 @@ def tracking_metrics(points, *, model_rmse=0.0):
     'mean_absolute_error': sum(abs(error) for error in errors) / len(errors),
     'worst_absolute_error': max(abs(error) for error in errors),
     'error_slew_rms': math.sqrt(
-      sum(((right - left) / ((b.mono_ns - a.mono_ns) / 1e9)) ** 2 for a, b, left, right in zip(points, points[1:], errors, errors[1:])) / (len(points) - 1)
+      sum(((right - left) / ((b.mono_ns - a.mono_ns) / 1e9)) ** 2 for a, b, left, right in zip(points, points[1:], errors, errors[1:], strict=False))
+      / (len(points) - 1)
     )
     if len(points) > 1
     else 0.0,

openpilot/starpilot/flm/offroad_processor.py

@@ -3,20 +3,19 @@
 from dataclasses import asdict, dataclass, field, replace
 from bisect import bisect_left
 import hashlib
-import json
 
 from openpilot.starpilot.flm.offroad_analysis import analyze, MAX_WINDOWS, cohort, _spread, MAX_COHORT_WINDOWS
 from openpilot.starpilot.flm.offroad_baseline_windows import baseline_windows
 from openpilot.starpilot.flm.canonical_surface_adapter import NAMES as AMPLITUDE_NAMES
 from openpilot.starpilot.flm.offroad_residual import analyze_stable_windows
 from openpilot.starpilot.flm.offroad_evidence import read_point, whole_windows, eligible_points, tracking_metrics
-from openpilot.starpilot.flm.offroad_route import latest_route, walk_completed_route
+from openpilot.starpilot.flm.offroad_route import latest_route, walk_completed_route, RouteReceipt
 from openpilot.starpilot.flm.runtime_adapter_registry import car_params_digest
 
 
 @dataclass(frozen=True)
 class CompletedAssessment:
-  receipt: object
+  receipt: RouteReceipt
   groups: tuple
   counts: dict
   cp: object = field(repr=False, compare=False)
@@ -172,9 +171,9 @@ def analyze_latest(params, root, previous, *, permitted):
           raise ValueError('Original segment tracking bound exceeded')
       if len(controls) > 12000 or applied_count > 12000:
         raise ValueError('Original segment controller input bound exceeded')
-    if any(b.control_ns <= a.control_ns for a, b in zip(controls, controls[1:])):
+    if any(b.control_ns <= a.control_ns for a, b in zip(controls, controls[1:], strict=False)):
       raise ValueError('Original controller metadata order changed')
-    if any(b[0] <= a[0] for a, b in zip(timeline, timeline[1:])):
+    if any(b[0] <= a[0] for a, b in zip(timeline, timeline[1:], strict=False)):
       raise ValueError('Original controller envelope order changed')
     times = tuple(header for header, _ in timeline)
     for key, groups in rows.items():

openpilot/starpilot/flm/offroad_residual.py

@@ -111,7 +111,7 @@ def estimate_residual(windows, *, cohort, permitted=lambda: True):
     normalizer = max(reference, floor)
     normalized.append(metric['rmse'] / normalizer)
     duration = (points[-1].mono_ns - points[0].mono_ns) / 1e9
-    variation = math.sqrt(sum((b.desired - a.desired) ** 2 / ((b.mono_ns - a.mono_ns) / 1e9) for a, b in zip(points, points[1:])) / duration)
+    variation = math.sqrt(sum((b.desired - a.desired) ** 2 / ((b.mono_ns - a.mono_ns) / 1e9) for a, b in zip(points, points[1:], strict=False)) / duration)
     variations.append(variation / normalizer)
     means.append(sum(p.desired for p in points) / len(points))
   source = hashlib.sha256(repr((cohort, tuple(w.identity for w in windows))).encode()).hexdigest()

openpilot/starpilot/flm/offroad_route.py

@@ -7,7 +7,6 @@ from pathlib import Path
 import re
 import time
 
-from openpilot.starpilot.flm.local_logs import read_closed_rlog
 from openpilot.starpilot.flm.log_decode import decode_full_rlog_segment
 from openpilot.starpilot.galaxy.drive_history import DriveHistory
 from openpilot.starpilot.galaxy.drive_analysis import _selected, _route_names, _closed_source

openpilot/starpilot/flm/offroad_runtime.py

@@ -58,8 +58,9 @@ def configuration_data(cp, ci, controller, params, *, manual_status):
   if selection.source == 'invalid' or descriptor.controller_mode not in (str(selection.mode), 'native'):
     raise ValueError('Actual controller does not match saved selection')
   fixed_manual = ()
-  if raw[DOCUMENT_KEY] is not None:
-    profile = parse_document(raw[DOCUMENT_KEY]).get(str(cp.carFingerprint))
+  document = raw[DOCUMENT_KEY]
+  if document is not None:
+    profile = parse_document(document).get(str(cp.carFingerprint))
     if profile is not None:
       if profile.flm is not None or profile.manual_coordinates is not None or profile.geometry is not None and profile.geometry.learning == 'force_off':
         raise ValueError('Manual or explicitly disabled tuning selected')

openpilot/starpilot/flm/offroad_source.py

@@ -35,7 +35,7 @@ def record_drive(params, *, stopped, on_status=None):
       super().__init__(['extrinsicsCalibration', 'controlsState', 'carState', 'carControl', 'starpilotLateralState'])
 
     def update_msgs(self, cur_time, msgs):
-      packets = tuple(msgs)
+      packets = list(msgs)
       for packet in packets:
         if packet is not None:
           self.original[packet.which()] = packet

openpilot/starpilot/flm/offroad_stable_bridge.py

@@ -7,12 +7,12 @@ atomic source-bound save and next-drive comparison/rollback.
 
 import hashlib
 from pathlib import Path
+from typing import Any
 
 from openpilot.starpilot.flm.offroad_stable_policy import (
   FLMSample,
   StablePolicy,
   cadenced_runs,
-  classify_torque_samples,
   current_catalog,
   recommended_route_profile,
   ORIGINAL_COMMIT,
@@ -49,7 +49,7 @@ class StableRouteSelector:
     factors = [1.0 + previous[name] for name in CURVE_NAMES]
     # No old platform preset or invented baseline/friction curve is installed.
     self.policy = StablePolicy(self.catalog, {self.family: factors})
-    self.rows = None
+    self.rows: dict[str, list[FLMSample] | None] | None = None
     self.spans = {}
     self.contexts = {}
     for group in assessment.stable_groups:
@@ -160,10 +160,10 @@ class StableRouteSelector:
         or not RouteWindow(key, tuple(original_points[key])).valid()
         or abs(values[0].t * 1e9 - span['start_ns']) > MAX_GAP_NS
         or abs(values[-1].t * 1e9 - span['end_ns']) > MAX_GAP_NS
-        or any(not 0 < b.t - a.t <= MAX_GAP_NS / 1e9 + 1e-9 for a, b in zip(values, values[1:]))
+        or any(not 0 < b.t - a.t <= MAX_GAP_NS / 1e9 + 1e-9 for a, b in zip(values, values[1:], strict=False))
       ):
-        rows[key] = None
-    self.rows = rows
+        invalid.add(key)
+    self.rows = {key: None if key in invalid else values for key, values in rows.items()}
 
   def propose(self, group, cohort):
     ids = {role: tuple(cohort[role]['windows']) for role in ('training', 'heldout')}
@@ -175,11 +175,13 @@ class StableRouteSelector:
       raise ValueError('Disjoint original training and held-out windows required')
     if self.rows is None:
       self._load()
+    rows = self.rows
+    assert rows is not None
     all_ids = ids['training'] + ids['heldout']
-    if any(key not in self.rows or self.rows[key] is None for key in all_ids):
+    if any(key not in rows or rows[key] is None for key in all_ids):
       return None, {'reason': 'original_stable_input_fields_missing', 'original_windows': all_ids}
     mapped = {}
-    reports = {}
+    reports: dict[str, dict[str, Any]] = {}
     brand = str(self.assessment.cp.brand)
     family = 'gm' if brand == 'gm' else 'standard'
     if brand == 'hyundai':
@@ -193,7 +195,9 @@ class StableRouteSelector:
       selected = []
       unavailable = []
       for key in windows:
-        for run in cadenced_runs(self.rows[key]):
+        values = rows[key]
+        assert values is not None
+        for run in cadenced_runs(values):
           summaries, _ = classify_current_stable(list(run))
           for summary in summaries:
             # Capability-based exclusions affect proposals only. No scored

openpilot/starpilot/flm/offroad_stable_policy.py

@@ -223,13 +223,15 @@ def _build_plot_svg(plot_data: dict[str, Any]) -> str:
       coords.append(f"{x:.1f},{y:.1f}")
     return " ".join(coords)
 
-  return (
-    "<svg viewBox='0 0 380 140' class='flm-plot' preserveAspectRatio='none'>"
-    "<rect x='0' y='0' width='380' height='140' rx='8' ry='8' fill='#0f172a'/>"
-    "<line x1='0' y1='120' x2='380' y2='120' stroke='#334155' stroke-width='1'/>"
-    f"<polyline fill='none' stroke='#ef4444' stroke-width='2' points='{_points(desired)}'/>"
-    f"<polyline fill='none' stroke='#38bdf8' stroke-width='2' points='{_points(actual)}'/>"
-    "</svg>"
+  return ''.join(
+    (
+      "<svg viewBox='0 0 380 140' class='flm-plot' preserveAspectRatio='none'>",
+      "<rect x='0' y='0' width='380' height='140' rx='8' ry='8' fill='#0f172a'/>",
+      "<line x1='0' y1='120' x2='380' y2='120' stroke='#334155' stroke-width='1'/>",
+      f"<polyline fill='none' stroke='#ef4444' stroke-width='2' points='{_points(desired)}'/>",
+      f"<polyline fill='none' stroke='#38bdf8' stroke-width='2' points='{_points(actual)}'/>",
+      "</svg>",
+    )
   )
 
 
@@ -885,7 +887,7 @@ class StablePolicy:
         continue
       avg_delta_curve = [value / bucket['weight'] for value in bucket['weightedDelta']]
       values = [round(max(0.05, float(bucket['current'][idx]) + avg_delta_curve[idx] * multiplier), 4) for idx in range(len(bucket['current']))]
-      if any((not math.isclose(float(bucket['current'][idx]), values[idx], abs_tol=1e-06) for idx in range(len(values)))):
+      if any(not math.isclose(float(bucket['current'][idx]), values[idx], abs_tol=1e-06) for idx in range(len(values))):
         overrides['baseFrictionThresholds'][family] = {'speedKnots': list(FLM_FRICTION_SPEED_KNOTS), 'values': values}
     overrides = self._normalize(overrides)
     return (params_delta, overrides, requires_force_auto_tune_off)
@@ -899,8 +901,8 @@ class StablePolicy:
     path_key: str = 'cleanup_pass',
     path_label: str = 'Cleanup Pass',
   ) -> list[dict[str, Any]]:
-    ignored = set((str(item) for item in feedback.get('ignoredDimensions', [])))
-    accepted = set((str(item) for item in feedback.get('acceptedDimensions', [])))
+    ignored = {str(item) for item in feedback.get('ignoredDimensions', [])}
+    accepted = {str(item) for item in feedback.get('acceptedDimensions', [])}
     has_feedback_decisions = bool(ignored or accepted)
     considered = [
       suggestion for suggestion in suggestions if suggestion.get('dimensionId') not in ignored and (not accepted or suggestion.get('dimensionId') in accepted)
@@ -1245,7 +1247,16 @@ def _current_stable_build_plot_svg(plot_data: dict[str, Any]) -> str:
       coords.append(f'{x:.1f},{y:.1f}')
     return ' '.join(coords)
 
-  return f"<svg viewBox='0 0 380 140' class='flm-plot' preserveAspectRatio='none'><rect x='0' y='0' width='380' height='140' rx='8' ry='8' fill='#0f172a'/><line x1='0' y1='120' x2='380' y2='120' stroke='#334155' stroke-width='1'/><polyline fill='none' stroke='#ef4444' stroke-width='2' points='{_points(desired)}'/><polyline fill='none' stroke='#38bdf8' stroke-width='2' points='{_points(actual)}'/></svg>"
+  return ''.join(
+    (
+      "<svg viewBox='0 0 380 140' class='flm-plot' preserveAspectRatio='none'>",
+      "<rect x='0' y='0' width='380' height='140' rx='8' ry='8' fill='#0f172a'/>",
+      "<line x1='0' y1='120' x2='380' y2='120' stroke='#334155' stroke-width='1'/>",
+      f"<polyline fill='none' stroke='#ef4444' stroke-width='2' points='{_points(desired)}'/>",
+      f"<polyline fill='none' stroke='#38bdf8' stroke-width='2' points='{_points(actual)}'/>",
+      "</svg>",
+    )
+  )
 
 
 def _current_stable_direction_reversal_count(values: np.ndarray, min_step: float) -> int:
@@ -1383,7 +1394,7 @@ def _current_stable_build_event_summaries(samples: list[FLMSample]) -> tuple[lis
     desired_sign = float(np.mean([sample.desired_la for sample in window]))
     if abs(desired_sign) < 0.35:
       continue
-    if any((sample.desired_la * desired_sign < 0.0 for sample in window)):
+    if any(sample.desired_la * desired_sign < 0.0 for sample in window):
       continue
     error_series = np.array([sample.actual_la - sample.desired_la for sample in window])
     sign_changes = int(np.sum(np.sign(error_series[1:]) != np.sign(error_series[:-1])))
@@ -1424,7 +1435,7 @@ def _current_stable_build_event_summaries(samples: list[FLMSample]) -> tuple[lis
     'highwayStraightAngleP2P': round(float(np.percentile(np.abs(angle), 95) - np.percentile(np.abs(angle), 5)), 4) if angle else 0.0,
     'meanOutputAbs': round(float(np.mean(np.abs(output))), 4) if output else 0.0,
   }
-  if summary_stats['meanErrorAbs'] < 0.08 and (not any((summary['severity'] > 0.65 for summary in summaries))):
+  if summary_stats['meanErrorAbs'] < 0.08 and (not any(summary['severity'] > 0.65 for summary in summaries)):
     summaries.append(
       {
         'bucket': 'model_limited',
@@ -1540,7 +1551,7 @@ def recommended_route_profile(profile, previous, baseline_revision, *, domain):
     curve = curves[family]
     if curve.get('speedKnots') != list(CURVE_SPEEDS) or len(curve.get('values', ())) != 5:
       raise ValueError('Actual five-point threshold curve required')
-    for name, target in zip(CURVE_NAMES, curve['values']):
+    for name, target in zip(CURVE_NAMES, curve['values'], strict=True):
       if type(target) not in (int, float) or not math.isfinite(target) or target <= 0:
         raise ValueError('Finite positive threshold factor required')
       deltas[name] = target - 1.0 - current[name]

openpilot/starpilot/flm/offroad_store.py

@@ -10,6 +10,7 @@ import json
 import math
 import os
 from pathlib import Path
+from typing import Any
 
 from openpilot.starpilot.flm.canonical_surface_adapter import (
   NAMES as CANONICAL_NAMES,
@@ -143,7 +144,7 @@ def remember(raw, row):
   from openpilot.starpilot.flm.learned_lifecycle import decode, encode
 
   scope = validate_row(row)
-  doc = decode(raw)
+  doc: dict[str, Any] = decode(raw)
   if doc['version'] == 1:
     doc = {'version': 3, 'profiles': doc['profiles'], 'fullSurfaces': {}, NAMESPACE: {}}
   elif doc['version'] == 2:

openpilot/starpilot/flm/offroad_surface.py

@@ -7,7 +7,7 @@ per-coordinate bounds, sign and 25% stage envelopes remain authoritative.
 
 import hashlib
 import math
-from openpilot.starpilot.flm.canonical_surface_adapter import CanonicalSurfaceEffects, MANIFEST_SHA256, NAMES, WINDOWS
+from openpilot.starpilot.flm.canonical_surface_adapter import CanonicalSurfaceEffects, MANIFEST_SHA256, NAMES
 
 LEGACY_REVISION = 'route-reference-relative-v2'
 REVISION = 'route-reference-relative-v3'

openpilot/starpilot/flm/offroad_tracking.py

@@ -138,7 +138,7 @@ def estimate(windows, *, cohort, permitted=lambda: True):
   residual_interval = (max(0.0, residual_interval[0]), residual_interval[1])
   variability = (
     math.sqrt(
-      sum(sum((b.desired - a.desired) ** 2 / ((b.mono_ns - a.mono_ns) / 1e9) for a, b in zip(w.points, w.points[1:])) for w in windows)
+      sum(sum((b.desired - a.desired) ** 2 / ((b.mono_ns - a.mono_ns) / 1e9) for a, b in zip(w.points, w.points[1:], strict=False)) for w in windows)
       / sum((w.points[-1].mono_ns - w.points[0].mono_ns) / 1e9 for w in windows)
     )
     / reference_rms

openpilot/starpilot/flm/tests/test_offroad_backend.py

@@ -81,7 +81,7 @@ class TestOffroadCache(unittest.TestCase):
         real_fsync = os.fsync
         calls = 0
 
-        def interrupted(fd):
+        def interrupted(fd, failure_at=failure_at, real_fsync=real_fsync):
           nonlocal calls
           calls += 1
           if calls == failure_at: