Chrysler: reliable fuzzy fingerprinting

opendbc/car/chrysler/tests/print_platform_codes.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+from collections import defaultdict
+
+from opendbc.car.structs import CarParams
+from opendbc.car.chrysler.values import get_platform_codes
+from opendbc.car.chrysler.fingerprints import FW_VERSIONS
+
+Ecu = CarParams.Ecu
+
+if __name__ == "__main__":
+  cars_for_code: defaultdict = defaultdict(lambda: defaultdict(set))
+
+  for car_model, ecus in FW_VERSIONS.items():
+    print(car_model)
+    for ecu in sorted(ecus):
+      platform_codes = get_platform_codes(ecus[ecu])
+      for code in platform_codes:
+        cars_for_code[ecu[0]][code].add(car_model)
+
+      print(f'  (Ecu.{ecu[0]}, {hex(ecu[1])}, {ecu[2]}):')
+      print(f'    Codes: {sorted(platform_codes)}')
+    print()
+
+  print('\nCar models vs. platform codes:')
+  for ecu, codes in cars_for_code.items():
+    print(f'  Ecu.{ecu}:')
+    for code, cars in codes.items():
+      if len(cars) > 1:
+        print(f'    {code!r}: {sorted(map(str, cars))}')

opendbc/car/chrysler/tests/test_chrysler.py

@@ -0,0 +1,107 @@
+import random
+import unittest
+
+from hypothesis import settings, given, strategies as st
+
+from opendbc.car.structs import CarParams
+from opendbc.car.fw_versions import build_fw_dict
+from opendbc.car.chrysler.values import FW_QUERY_CONFIG, FW_PATTERN, PLATFORM_CODE_ECUS, get_platform_codes
+from opendbc.car.chrysler.fingerprints import FW_VERSIONS
+from opendbc.testing import parameterized
+
+Ecu = CarParams.Ecu
+
+
+class TestChryslerFW(unittest.TestCase):
+  @parameterized("car_model, fw_versions", FW_VERSIONS.items())
+  def test_fw_versions(self, car_model, fw_versions):
+    for (_ecu, _addr, _subaddr), fws in fw_versions.items():
+      for fw in fws:
+        match = FW_PATTERN.match(fw)
+        assert match is not None, f"Unable to parse FW: {fw!r}"
+
+        codes = get_platform_codes([fw])
+        assert 1 == len(codes), f"Unable to parse FW: {fw!r}"
+
+  @parameterized("car_model, fw_versions", FW_VERSIONS.items())
+  def test_platform_code_ecus_available(self, car_model, fw_versions):
+    # Asserts ECU keys essential for fuzzy fingerprinting are available on all platforms
+    present_ecus = {ecu[0] for ecu in fw_versions if ecu[0] in PLATFORM_CODE_ECUS}
+    assert len(present_ecus) >= 3, "Platform has too few ECUs to fuzzy fingerprint"
+
+  @settings(max_examples=100)
+  @given(data=st.data())
+  def test_platform_codes_fuzzy_fw(self, data):
+    """Ensure function doesn't raise an exception"""
+    fw_strategy = st.lists(st.binary())
+    fws = data.draw(fw_strategy)
+    get_platform_codes(fws)
+
+  def test_platform_codes_spot_check(self):
+    # Asserts basic platform code parsing behavior for a few cases
+    results = get_platform_codes([
+      b"68227902AF",
+      b"68227902AG",
+      b"68360252AC",
+      b"68267018AO ",
+      b"22DTRHD_AA",
+      b"M2370131MB",
+    ])
+    assert results == {b"68227902", b"68360252", b"68267018", b"22DTRHD_", b"M2370131"}
+
+  def test_fuzzy_match(self):
+    # Ensure that unique part number combinations map to one platform
+    for platform, fw_by_addr in FW_VERSIONS.items():
+      for _ in range(20):
+        car_fw = []
+        for ecu, fw_versions in fw_by_addr.items():
+          ecu_name, addr, sub_addr = ecu
+          fw = random.choice(fw_versions)
+          car_fw.append(CarParams.CarFw(ecu=ecu_name, fwVersion=fw, address=addr,
+                                        subAddress=0 if sub_addr is None else sub_addr))
+
+        CP = CarParams(carFw=car_fw)
+        matches = FW_QUERY_CONFIG.match_fw_to_car_fuzzy(build_fw_dict(CP.carFw), CP.carVin, FW_VERSIONS)
+        assert matches == {platform}
+
+  def test_fuzzy_match_new_revision(self):
+    # Ensure fuzzy matching is robust to unseen software revisions of known part numbers
+    for platform, fw_by_addr in FW_VERSIONS.items():
+      for _ in range(20):
+        live_fw = {}
+        for (_ecu, addr, sub_addr), fw_versions in fw_by_addr.items():
+          fw = random.choice(fw_versions)
+          part_number = FW_PATTERN.match(fw).group('part_number')
+          live_fw[(addr, sub_addr)] = {part_number + b'ZZ'}
+
+        matches = FW_QUERY_CONFIG.match_fw_to_car_fuzzy(live_fw, '', FW_VERSIONS)
+        assert matches == {platform}
+
+  def test_fuzzy_match_unknown_part_number(self):
+    # Ensure fuzzy matching rejects platforms on an unseen part number for a platform code ECU
+    for _platform, fw_by_addr in FW_VERSIONS.items():
+      live_fw = {}
+      for (ecu, addr, sub_addr), fw_versions in fw_by_addr.items():
+        fw = random.choice(fw_versions)
+        if ecu == Ecu.combinationMeter:
+          fw = b'99999999AA'
+        live_fw[(addr, sub_addr)] = {fw}
+
+      matches = FW_QUERY_CONFIG.match_fw_to_car_fuzzy(live_fw, '', FW_VERSIONS)
+      assert matches == set()
+
+  def test_fuzzy_match_missing_ecu(self):
+    # Ensure fuzzy matching rejects platforms when an expected platform code ECU is missing
+    for _platform, fw_by_addr in FW_VERSIONS.items():
+      live_fw = {}
+      for (ecu, addr, sub_addr), fw_versions in fw_by_addr.items():
+        if ecu == Ecu.combinationMeter:
+          continue
+        live_fw[(addr, sub_addr)] = {random.choice(fw_versions)}
+
+      matches = FW_QUERY_CONFIG.match_fw_to_car_fuzzy(live_fw, '', FW_VERSIONS)
+      assert matches == set()
+
+  def test_fuzzy_match_empty(self):
+    matches = FW_QUERY_CONFIG.match_fw_to_car_fuzzy({}, '', FW_VERSIONS)
+    assert matches == set()

opendbc/car/chrysler/values.py

@@ -1,10 +1,11 @@
+import re
 from enum import IntFlag
 from dataclasses import dataclass, field
 
 from opendbc.car import Bus, CarSpecs, DbcDict, PlatformConfig, Platforms, uds
 from opendbc.car.structs import CarParams
 from opendbc.car.docs_definitions import CarHarness, CarDocs, CarParts
-from opendbc.car.fw_query_definitions import FwQueryConfig, Request, p16
+from opendbc.car.fw_query_definitions import FwQueryConfig, LiveFwVersions, OfflineFwVersions, Request, p16
 
 Ecu = CarParams.Ecu
 
@@ -129,6 +130,62 @@ def __init__(self, CP):
 CUSW_CARS = {CAR.JEEP_CHEROKEE_5TH_GEN, }
 
 
+# FW response contains an FCA part number and a software revision,
+# e.g. b'68227902AF' or b'68267018AO ' (engine responses have a trailing space)
+#      1111111122
+# 1 = Part number, unique to a component on a platform and its model year range
+# 2 = Software revision, updates alphabetically with new software versions
+FW_PATTERN = re.compile(b'^(?P<part_number>[0-9A-Z_]{8})(?P<revision>[A-Z]{2,3})[ ]*$')
+
+
+def get_platform_codes(fw_versions: list[bytes] | set[bytes]) -> set[bytes]:
+  codes = set()
+  for fw in fw_versions:
+    match = FW_PATTERN.match(fw)
+    if match is not None:
+      codes.add(match.group('part_number'))
+
+  return codes
+
+
+def match_fw_to_car_fuzzy(live_fw_versions: LiveFwVersions, vin: str, offline_fw_versions: OfflineFwVersions) -> set[str]:
+  candidates: set[str] = set()
+
+  for candidate, fws in offline_fw_versions.items():
+    # Keep track of ECUs which pass all checks (part number matches)
+    valid_found_ecus = set()
+    valid_expected_ecus = {ecu[1:] for ecu in fws if ecu[0] in PLATFORM_CODE_ECUS}
+    for ecu, expected_versions in fws.items():
+      addr = ecu[1:]
+      # Only check ECUs expected to have platform-specific part numbers
+      if ecu[0] not in PLATFORM_CODE_ECUS:
+        continue
+
+      # Expected part numbers
+      expected_platform_codes = get_platform_codes(expected_versions)
+
+      # Found part numbers
+      found_platform_codes = get_platform_codes(live_fw_versions.get(addr, set()))
+
+      # Check any part number matches for any found versions
+      if not any(found_platform_code in expected_platform_codes for found_platform_code in found_platform_codes):
+        break
+
+      valid_found_ecus.add(addr)
+
+    # If all live ECUs pass all checks for candidate, add it as a match
+    if valid_expected_ecus.issubset(valid_found_ecus):
+      candidates.add(candidate)
+
+  return candidates
+
+
+# All of these ECUs must be present and their part numbers are expected to be platform-specific.
+# Part numbers are shared between some sibling platforms (e.g. ICE and hybrid variants, trucks and SUVs
+# on the same platform), but no two platforms share all of them
+PLATFORM_CODE_ECUS = (Ecu.combinationMeter, Ecu.srs, Ecu.abs, Ecu.eps)
+
+
 CHRYSLER_VERSION_REQUEST = bytes([uds.SERVICE_TYPE.READ_DATA_BY_IDENTIFIER]) + \
   p16(0xf132)
 CHRYSLER_VERSION_RESPONSE = bytes([uds.SERVICE_TYPE.READ_DATA_BY_IDENTIFIER + 0x40]) + \
@@ -166,6 +223,8 @@ def __init__(self, CP):
   extra_ecus=[
     (Ecu.abs, 0x7e4, None),  # alt address for abs on hybrids, NOTE: not on all hybrid platforms
   ],
+  # Custom fuzzy fingerprinting function using platform-specific part numbers
+  match_fw_to_car_fuzzy=match_fw_to_car_fuzzy,
 )
 
 DBC = CAR.create_dbc_map()