Keep automatic Tesla AP1 discovery on normal wiring

opendbc_repo/opendbc/car/car_helpers.py

@@ -15,6 +15,7 @@ from opendbc.car.ford.discovery import (normal_bus0_selection as ford_normal_bus
                                        normal_bus0_queries as ford_normal_bus0_queries, normal_bus_query_brand)
 from opendbc.car.volvo.discovery import normal_bus_topology as volvo_normal_bus_topology
 from opendbc.car.hyundai.discovery import normal_bus_topology
+from opendbc.car.tesla.discovery import hw1_normal_bus0_topology
 from opendbc.car.values import BRANDS
 from opendbc.car.vin import get_vin, is_valid_vin, VIN_UNKNOWN
 
@@ -169,13 +170,14 @@ def fingerprint(can_recv: CanRecvCallable, can_send: CanSendCallable, set_obd_mu
       elif (normal_brand := normal_bus_query_brand(
           can_recv, normal_bus_topology,
           gm_probe=gm_normal_bus0_topology if _gm_fw_uses_normal_bus0() else None,
-          volvo_probe=volvo_normal_bus_topology if not fw_versions.FW_QUERY_CONFIGS['volvo'].requests else None)) is not None:
+          volvo_probe=volvo_normal_bus_topology if not fw_versions.FW_QUERY_CONFIGS['volvo'].requests else None,
+          tesla_probe=hw1_normal_bus0_topology if _tesla_fw_uses_normal_bus0() else None)) is not None:
 
         set_obd_multiplexing(False)
         warmup_remaining = 0.1 - (time.monotonic() - start_time)
         if warmup_remaining > 0:
           time.sleep(warmup_remaining)
-        vin_buses = (0,) if normal_brand in ('ford', 'gm', 'volvo') else (0, 1)
+        vin_buses = (0,) if normal_brand in ('ford', 'gm', 'volvo', 'tesla') else (0, 1)
         vin_rx_addr, vin_rx_bus, vin = get_vin(can_recv, can_send, vin_buses)
         car_fw = get_fw_versions(can_recv, can_send, set_obd_multiplexing,
                                  query_brand=normal_brand, normal_bus_only=True)

opendbc_repo/opendbc/car/ford/discovery.py

@@ -45,7 +45,7 @@ NORMAL_BUS0_LAYOUT = {(0, 0x82): 8, (0, 0x165): 8, (0, 0x204): 8,
                       (0, 0x217): 8, (2, 0x3CA): 8, (2, 0x3D6): 8}
 
 
-def normal_bus_query_brand(can_recv, hyundai_probe, *, gm_probe=None, volvo_probe=None, clock=time.monotonic):
+def normal_bus_query_brand(can_recv, hyundai_probe, *, gm_probe=None, volvo_probe=None, tesla_probe=None, clock=time.monotonic):
   # Share the existing bounded passive HKG read, rather than consuming a second
   # startup window or enabling the OBD mux to find out whether this is a Ford.
   received = {}
@@ -62,7 +62,7 @@ def normal_bus_query_brand(can_recv, hyundai_probe, *, gm_probe=None, volvo_prob
     for packet in packets:
       for frame in packet:
         key = (frame.src, frame.address)
-        if gm_probe is not None or volvo_probe is not None:
+        if gm_probe is not None or volvo_probe is not None or tesla_probe is not None:
           bus = gm_received.setdefault(frame.src, {})
           length = len(frame.dat)
           gm_malformed |= frame.address in bus and bus[frame.address] != length
@@ -78,7 +78,7 @@ def normal_bus_query_brand(can_recv, hyundai_probe, *, gm_probe=None, volvo_prob
   # Rejecting HKG or Ford does not reject GM. Finish the shared passive window
   # within its original budget even when an early frame excludes Ford.
   while not (not malformed and received == NORMAL_BUS0_LAYOUT) and reads < 24 and clock() < deadline:
-    if malformed and ((gm_probe is None and volvo_probe is None) or gm_malformed):
+    if malformed and ((gm_probe is None and volvo_probe is None and tesla_probe is None) or gm_malformed):
       break
     observe(wait_for_one=True)
   if not malformed and received == NORMAL_BUS0_LAYOUT and normal_bus0_queries():
@@ -87,4 +87,6 @@ def normal_bus_query_brand(can_recv, hyundai_probe, *, gm_probe=None, volvo_prob
     return 'gm'
   if volvo_probe is not None and not gm_malformed and volvo_probe(gm_received):
     return 'volvo'
+  if tesla_probe is not None and not gm_malformed and tesla_probe(gm_received):
+    return 'tesla'
   return None

opendbc_repo/opendbc/car/tesla/discovery.py

@@ -0,0 +1,12 @@
+"""Passive AP1 query-wiring selection; live firmware still owns vehicle identity."""
+
+HW1_NORMAL_BUS0_LAYOUT = {(0, 0x108): 8, (0, 0x155): 8, (0, 0x20A): 8,
+                         (0, 0x368): 8, (0, 0x370): 8, (2, 0x488): 4, (2, 0x2B9): 8}
+
+
+def hw1_normal_bus0_topology(fingerprint):
+  try:
+    return all(fingerprint.get(bus, {}).get(address) == length
+               for (bus, address), length in HW1_NORMAL_BUS0_LAYOUT.items())
+  except (AttributeError, TypeError, ValueError):
+    return False

opendbc_repo/opendbc/car/tesla/tests/test_automatic_hw1_discovery.py

@@ -0,0 +1,95 @@
+"""AP1 discovery must collect live identity without probing unrelated OBD wiring."""
+from unittest import mock
+
+import pytest
+from opendbc.car import car_helpers
+from opendbc.car.can_definitions import CanData
+from opendbc.car.tesla.discovery import HW1_NORMAL_BUS0_LAYOUT, hw1_normal_bus0_topology
+from opendbc.car.tesla.tests.test_selected_discovery import TeslaEcu
+from opendbc.car.tesla.values import CAR
+
+HW1_FIRMWARE = b'1016704-00-HAA' + b'\x00' * 10
+
+
+class AP1Ecu(TeslaEcu):
+  def __init__(self, firmware=HW1_FIRMWARE):
+    super().__init__()
+    self.firmware = firmware
+
+  def send(self, messages):
+    for message in messages:
+      payload = message.dat
+      if (message.src == 0 and message.address == 0x730 and payload[0] >> 4 == 0 and
+          payload[1:1 + payload[0]] == b'\x22\xf1\x95'):
+        self.sent.append(message)
+        self.reply(0x730, 0x738, b'\x62\xf1\x95' + self.firmware)
+      else:
+        super().send([message])
+
+  def recv(self, wait_for_one=False):
+    self.clock += .001
+    values, self.frames = self.frames, []
+    return [values + [CanData(address, bytes(length), bus)
+                      for (bus, address), length in HW1_NORMAL_BUS0_LAYOUT.items()]]
+
+
+def automatic(peer, *, alpha=False, release=False):
+  callbacks = []
+  with mock.patch.dict(car_helpers.os.environ, {'FINGERPRINT': '', 'SKIP_FW_QUERY': '', 'DISABLE_FW_CACHE': ''}), \
+       mock.patch.object(car_helpers.time, 'monotonic', side_effect=lambda: peer.clock), \
+       mock.patch.object(car_helpers.time, 'sleep', side_effect=lambda seconds: setattr(peer, 'clock', peer.clock + seconds)):
+    result = car_helpers.get_car(peer.recv, peer.send, callbacks.append, alpha, release)
+  return result, callbacks
+
+
+@pytest.mark.parametrize('alpha,release', [(False, False), (True, False), (True, True)])
+def test_actual_automatic_ap1_vin_fw_can_and_factory_use_normal_wiring(alpha, release):
+  peer = AP1Ecu()
+  ci, callbacks = automatic(peer, alpha=alpha, release=release)
+  assert callbacks and set(callbacks) == {False}
+  assert peer.sent and all(frame.src == 0 for frame in peer.sent)
+  assert ci.CP.carFingerprint == CAR.TESLA_MODEL_S_HW1
+  assert ci.CP.fingerprintSource == car_helpers.CarParams.FingerprintSource.fw
+  assert ci.CP.carFw and all(fw.brand == 'tesla' and fw.bus == 0 for fw in ci.CP.carFw)
+  assert any(fw.fwVersion == HW1_FIRMWARE for fw in ci.CP.carFw)
+  assert ci.CP.safetyConfigs[0].safetyParam == 16 + int(alpha)
+  assert ci.CP.openpilotLongitudinalControl == alpha and ci.CP.pcmCruise != alpha
+
+
+@pytest.mark.parametrize('source', list(HW1_NORMAL_BUS0_LAYOUT))
+@pytest.mark.parametrize('defect', ['missing', 'length', 'bus', 'echo'])
+def test_ap1_wiring_scope_requires_each_real_source(source, defect):
+  fp = {}
+  for (bus, address), length in HW1_NORMAL_BUS0_LAYOUT.items():
+    if (bus, address) == source:
+      if defect == 'missing':
+        continue
+      if defect == 'length':
+        length -= 1
+      else:
+        bus += 128 if defect == 'echo' else 1
+    fp.setdefault(bus, {})[address] = length
+  assert not hw1_normal_bus0_topology(fp)
+
+
+def test_conflicting_lengths_cannot_be_hidden_by_later_valid_ap1_frames():
+  from opendbc.car.ford.discovery import normal_bus_query_brand
+  values = [CanData(address, bytes(length), bus) for (bus, address), length in HW1_NORMAL_BUS0_LAYOUT.items()]
+  packets = [CanData(0x108, bytes(7), 0), *values]
+  result = normal_bus_query_brand(lambda **kwargs: [packets], lambda recv: bool(recv() and False),
+                                  tesla_probe=hw1_normal_bus0_topology, clock=lambda: 0.)
+  assert result is None
+
+
+@pytest.mark.parametrize('firmware', [b'', b'1016704-00-HAA', b'\x10\x00B'])
+def test_ap1_layout_never_substitutes_for_live_firmware_identity(firmware):
+  ci, _ = automatic(AP1Ecu(firmware))
+  assert ci.CP.carFingerprint != CAR.TESLA_MODEL_S_HW1
+
+
+def test_passive_ap1_layout_keeps_existing_shared_read_budget():
+  from opendbc.car.ford.discovery import normal_bus_query_brand
+  calls = []
+  assert normal_bus_query_brand(lambda **kwargs: calls.append(kwargs) or [], lambda recv: bool(recv() and False),
+                                tesla_probe=hw1_normal_bus0_topology, clock=lambda: 0.) is None
+  assert len(calls) == 24

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "76719de784909665231700c43e3485008e4c0e0f8361b273ffac4e9c09d262bd", "schema_version": 1}
+{"protected_inventory_sha256": "e1e0e932477e66be40286fa12f9318bf0767a6ea192554d19e05418d811aef3a", "schema_version": 1}