Add calibration reset beside driver monitoring in Galaxy

openpilot/starpilot/galaxy/developer_settings.py

@@ -4,6 +4,7 @@ from dataclasses import replace
 
 from openpilot.starpilot.longitudinal.voacc import KEY, DESCRIPTION
 from openpilot.starpilot.lateral.chubbs_lane import KEY as CHUBBS_KEY, DESCRIPTION as CHUBBS_DESCRIPTION
+from openpilot.starpilot.ui.calibration_reset import calibration_reset
 from openpilot.starpilot.saved_document import commit_exact
 from openpilot.starpilot.saved_source import read_saved
 from openpilot.starpilot.ui.feature_settings_state import FeatureRow, FeatureSettingsState
@@ -12,6 +13,8 @@ from openpilot.starpilot.ui.feature_settings_state import FeatureRow, FeatureSet
 STARPORT_WIFI_KEY = "StarPortDedicatedWifi"
 STARPORT_WIFI_DESCRIPTION = ("Experimental. Join the car's Wi-Fi on a second interface so the comma stays on home Wi-Fi or your " +
                        "phone hotspot during wireless StarPort. Applies from the next StarPort connection.")
+CALIBRATION_RESET = "reset_calibration"
+CALIBRATION_RESET_DESCRIPTION = "Clear camera and steering calibration. The device recalibrates on the next drive."
 DRIVER_MONITORING_RESET = "reset_driver_monitoring"
 DRIVER_MONITORING_RESET_DESCRIPTION = "Clear saved wheel-side detection. Automatic detection restarts on the next drive."
 
@@ -34,6 +37,9 @@ class DeveloperSettingsOwner:
   def _driver_monitoring_reset_allowed(self):
     return self.authorized() and not self.params.get_bool("IsDriverViewEnabled")
 
+  def _calibration_reset_allowed(self):
+    return self.authorized() and not calibration_reset.pending and not self.params.get_bool("OnroadCycleRequested")
+
   def snapshot(self):
     raw, readable = read_saved(self.params, KEY, 8)
     valid = readable and raw in (None, b"0", b"1")
@@ -48,6 +54,12 @@ class DeveloperSettingsOwner:
                     "Park the vehicle and close driver camera preview before resetting driver monitoring.")
     reset_row = FeatureRow(DRIVER_MONITORING_RESET, "Reset Driver Monitoring", "Ready" if reset_available else "Unavailable",
                            rhd_raw if rhd_readable else None, available=reset_available, reason=reset_reason)
+    calibration_raw, calibration_readable = read_saved(self.params, "CalibrationParams", 1024 * 1024)
+    calibration_available = self._calibration_reset_allowed() and calibration_readable
+    calibration_row = FeatureRow(CALIBRATION_RESET, "Reset Calibration", "Ready" if calibration_available else "Unavailable",
+                                 calibration_raw if calibration_readable else None, available=calibration_available,
+                                 reason=CALIBRATION_RESET_DESCRIPTION if calibration_available else
+                                        "Park the vehicle and wait for any pending reset to finish.")
     firmware_rows = (() if self.panda_wake is None else
                      tuple(replace(item, vehicle_fingerprint=item.capability[0])
                            for item in self.panda_wake.rows(parked) if item.key == "IgnoreIgnitionLine"))
@@ -64,9 +76,17 @@ class DeveloperSettingsOwner:
                         reason=CHUBBS_DESCRIPTION if parked else "Park the vehicle to change this setting.", default_value="Off")
     summary = "Changes apply next drive. Panda firmware changes require a device restart." if firmware_rows else "Changes apply next drive."
     return FeatureSettingsState("developer", "Developer driving options", summary,
-                                (row, chubbs, reset_row, *firmware_rows, *starport_rows), parked)
+                                (row, chubbs, reset_row, calibration_row, *firmware_rows, *starport_rows), parked)
 
   def apply(self, request):
+    if request.key == CALIBRATION_RESET:
+      if request.value != "confirm" or not self._calibration_reset_allowed():
+        return False
+      current, readable = read_saved(self.params, "CalibrationParams", 1024 * 1024)
+      if not readable or current != request.expected:
+        return False
+      return calibration_reset.request(self.params)
+
     if request.key == DRIVER_MONITORING_RESET:
       if request.value != "confirm" or not self._driver_monitoring_reset_allowed():
         return False

openpilot/starpilot/galaxy/settings.py

@@ -30,7 +30,7 @@ from openpilot.starpilot.ui.feature_settings_state import (
   FeatureRow, FeatureSettingsRequest, FEATURE_CONFIRM_ACTIONS, is_long_confirm_action, row_change, row_default,
 )
 from openpilot.starpilot.saved_source import read_saved
-from openpilot.starpilot.galaxy.developer_settings import DRIVER_MONITORING_RESET, DeveloperSettingsOwner
+from openpilot.starpilot.galaxy.developer_settings import CALIBRATION_RESET, DRIVER_MONITORING_RESET, DeveloperSettingsOwner
 from openpilot.starpilot.speed_limits.vision_gate import diagnostic_choice_enabled
 from openpilot.starpilot.parked_evidence import ParkedEvidence, RESUME_SKEW_NS, fresh_offroad, host_parked
 from openpilot.starpilot.ui.pip_owner import EDITOR as PIP_EDITOR, FORMAT_PREFIX as PIP_FORMAT_PREFIX, RESET as PIP_RESET, PiPOwner, validated_editor_draft
@@ -45,7 +45,7 @@ PAGES = frozenset(("hub", "localization", "developer", "vehicle", "slc", "lane",
                    "conditional", "conditional/cem", "conditional/ccm",
                    *(f"{name}/{category}" for name in ("traffic", "aggressive", "standard", "relaxed")
                      for category in ("acceleration", "braking", "following"))))
-CONFIRM_ACTIONS = FEATURE_CONFIRM_ACTIONS | {DRIVER_MONITORING_RESET}
+CONFIRM_ACTIONS = FEATURE_CONFIRM_ACTIONS | {DRIVER_MONITORING_RESET, CALIBRATION_RESET}
 MAX_VIEWS = 64
 MAX_INTENTS = 64
 TTL = 60.0
@@ -431,6 +431,8 @@ def _question(row: FeatureRow, request: FeatureSettingsRequest) -> str:
     return f"Save {row.label.lower()} as {value}?"
   if key == SENTRY_RESET:
     return "Replace invalid saved motion settings with Off and the default sensitivity and warning time?"
+  if key == CALIBRATION_RESET:
+    return "Reset camera and steering calibration? Automatic steering learning will be invalidated and the device will recalibrate on the next drive."
   if key == DRIVER_MONITORING_RESET:
     return "Clear saved driver monitoring wheel-side detection? Automatic detection will run again on the next drive."
   if key.startswith("sentry:"):

openpilot/starpilot/galaxy/tests/test_developer_settings.py

@@ -6,8 +6,9 @@ from types import SimpleNamespace
 from typing import cast
 import tempfile
 import unittest
+from unittest.mock import patch
 
-from openpilot.starpilot.galaxy.developer_settings import DRIVER_MONITORING_RESET, STARPORT_WIFI_KEY, DeveloperSettingsOwner
+from openpilot.starpilot.galaxy.developer_settings import CALIBRATION_RESET, DRIVER_MONITORING_RESET, STARPORT_WIFI_KEY, DeveloperSettingsOwner
 from openpilot.starpilot.galaxy.settings import AuthorityContext, ContextSource, SettingsChanged, SettingsGateway
 from openpilot.starpilot.longitudinal.voacc import KEY, selected
 from openpilot.starpilot.ui.feature_settings_state import FeatureSettingsRequest
@@ -170,3 +171,60 @@ class DeveloperSettingsTest(unittest.TestCase):
     self.assertFalse(row.available)
     self.assertFalse(self.owner.apply(FeatureSettingsRequest(DRIVER_MONITORING_RESET, b"1", "confirm", confirmation=True)))
     self.assertTrue(Path(self.params.get_param_path("IsRhdDetected")).exists())
+
+
+class CalibrationResetSettingsTest(unittest.TestCase):
+  setUp = DeveloperSettingsTest.setUp
+  def test_calibration_reset_adjacent_and_uses_existing_reset_service(self):
+    with patch("openpilot.starpilot.galaxy.developer_settings.calibration_reset") as reset:
+      reset.pending = False
+      rows = self.owner.snapshot().rows
+      keys = [row.key for row in rows]
+      self.assertEqual(keys[keys.index(DRIVER_MONITORING_RESET) + 1], CALIBRATION_RESET)
+      self.assertTrue(rows[keys.index(CALIBRATION_RESET)].available)
+      reset.request.return_value = True
+      self.assertTrue(self.owner.apply(FeatureSettingsRequest(CALIBRATION_RESET, None, "confirm", confirmation=True)))
+      reset.request.assert_called_once_with(self.params)
+
+  def test_calibration_reset_refuses_onroad_pending_stale_and_wrong_value(self):
+    with patch("openpilot.starpilot.galaxy.developer_settings.calibration_reset") as reset:
+      reset.pending = False
+      request = FeatureSettingsRequest(CALIBRATION_RESET, None, "confirm", confirmation=True)
+      self.assertFalse(self.owner.apply(replace(request, value="On")))
+      self.parked = False
+      self.assertFalse(self.owner.apply(request))
+      self.assertFalse(next(row for row in self.owner.snapshot().rows if row.key == CALIBRATION_RESET).available)
+      self.parked = True
+      reset.pending = True
+      self.assertFalse(self.owner.apply(request))
+      reset.pending = False
+      self.params.put_bool("OnroadCycleRequested", True)
+      self.assertFalse(self.owner.apply(request))
+      self.params.remove("OnroadCycleRequested")
+      Path(self.params.get_param_path("CalibrationParams")).write_bytes(b"changed")
+      self.assertFalse(self.owner.apply(request))
+      reset.request.assert_not_called()
+
+  def test_calibration_reset_gateway_requires_confirmation_and_fresh_parked_state(self):
+    with patch("openpilot.starpilot.galaxy.developer_settings.calibration_reset") as reset:
+      reset.pending = False
+      reset.request.return_value = True
+      context = SimpleNamespace(value=AuthorityContext(True, None, None))
+      context.sample = lambda: context.value
+      gateway = SettingsGateway(self.params, cast(ContextSource, context), clock=lambda: 10.)
+      def preview():
+        page = gateway.page("developer", "session", b"generation")
+        i = next(i for i,row in enumerate(page["rows"]) if row["label"] == "Reset Calibration")
+        self.assertTrue(page["rows"][i]["confirm"])
+        return gateway.preview(page["view"], i, 0, "session", b"generation")
+      pending = preview()
+      self.assertIn("recalibrate", pending["question"])
+      reset.request.assert_not_called()
+      context.value = replace(context.value, parked=False)
+      with self.assertRaises(SettingsChanged):
+        gateway.confirm(pending["intent"], "session", b"generation")
+      reset.request.assert_not_called()
+      context.value = replace(context.value, parked=True)
+      pending = preview()
+      self.assertTrue(gateway.confirm(pending["intent"], "session", b"generation"))
+      reset.request.assert_called_once()