HKG: use platform codes to fuzzy fingerprint

selfdrive/car/fw_query_definitions.py

@@ -3,7 +3,7 @@
 import copy
 from dataclasses import dataclass, field
 import struct
-from typing import Dict, List, Optional, Tuple
+from typing import Callable, Dict, List, Optional, Set, Tuple
 
 import panda.python.uds as uds
 
@@ -74,6 +74,9 @@ class FwQueryConfig:
   non_essential_ecus: Dict[capnp.lib.capnp._EnumModule, List[str]] = field(default_factory=dict)
   # Ecus added for data collection, not to be fingerprinted on
   extra_ecus: List[Tuple[capnp.lib.capnp._EnumModule, int, Optional[int]]] = field(default_factory=list)
+  # Function a brand can implement to provide better fuzzy matching. Takes in FW versions,
+  # returns set of candidates. Only will match if one candidate is returned
+  match_fw_to_car_fuzzy: Optional[Callable[[Dict[Tuple[int, Optional[int]], Set[bytes]]], Set[str]]] = None
 
   def __post_init__(self):
     for i in range(len(self.requests)):

selfdrive/car/fw_versions.py

@@ -151,6 +151,11 @@ def match_fw_to_car(fw_versions, allow_exact=True, allow_fuzzy=True, log=True):
       fw_versions_dict = build_fw_dict(fw_versions, filter_brand=brand)
       matches |= match_func(fw_versions_dict, log=log)
 
+      # If specified and no matches so far, fall back to brand's fuzzy fingerprinting function
+      config = FW_QUERY_CONFIGS[brand]
+      if not exact_match and not len(matches) and config.match_fw_to_car_fuzzy is not None:
+        matches |= config.match_fw_to_car_fuzzy(fw_versions_dict)
+
     if len(matches):
       return exact_match, matches
 

selfdrive/car/hyundai/tests/print_platform_codes.py

@@ -13,14 +13,9 @@
       if ecu[0] not in PLATFORM_CODE_ECUS:
         continue
 
-      codes = set()
-      dates = set()
-      for fw in ecus[ecu]:
-        code = list(get_platform_codes([fw]))[0]
-        codes.add(code.split(b"-")[0])
-        if b"-" in code:
-          dates.add(code.split(b"-")[1])
-
+      platform_codes = get_platform_codes(ecus[ecu])
+      codes = {code for code, _ in platform_codes}
+      dates = {date for _, date in platform_codes if date is not None}
       print(f'  (Ecu.{ECU_NAME[ecu[0]]}, {hex(ecu[1])}, {ecu[2]}):')
       print(f'    Codes: {codes}')
       print(f'    Dates: {dates}')

selfdrive/car/hyundai/tests/test_hyundai.py

@@ -2,8 +2,9 @@
 import unittest
 
 from cereal import car
-from selfdrive.car.hyundai.values import CAMERA_SCC_CAR, CANFD_CAR, CAN_GEARS, CAR, CHECKSUM, FW_QUERY_CONFIG, \
-                                         FW_VERSIONS, LEGACY_SAFETY_MODE_CAR, PART_NUMBER_FW_PATTERN, PLATFORM_CODE_ECUS, \
+from selfdrive.car.fw_versions import build_fw_dict
+from selfdrive.car.hyundai.values import CAMERA_SCC_CAR, CANFD_CAR, CAN_GEARS, CAR, CHECKSUM, EV_CAR, \
+                                         FW_QUERY_CONFIG, FW_VERSIONS, LEGACY_SAFETY_MODE_CAR, PLATFORM_CODE_ECUS, \
                                          get_platform_codes
 
 Ecu = car.CarParams.Ecu
@@ -50,6 +51,30 @@ def test_fw_format(self):
     # - every supported ECU FW version has a part number
     # - expected parsing of ECU FW dates
 
+    # Some platforms have date codes in a different format we don't yet parse (or are missing).
+    # For now, assert list of expected missing date cars
+    no_dates_platforms = {
+      # CAN FD
+      CAR.TUCSON_4TH_GEN,
+      CAR.TUCSON_HYBRID_4TH_GEN,
+      CAR.KIA_SPORTAGE_HYBRID_5TH_GEN,
+      CAR.SANTA_CRUZ_1ST_GEN,
+      CAR.KIA_SPORTAGE_5TH_GEN,
+      # CAN
+      CAR.KONA,
+      CAR.SONATA_LF,
+      CAR.VELOSTER,
+      CAR.KIA_CEED,
+      CAR.KIA_FORTE,
+      CAR.KONA_EV,
+      CAR.KONA_EV_2022,
+      CAR.KIA_OPTIMA_G4,
+      CAR.KIA_OPTIMA_G4_FL,
+      CAR.ELANTRA,
+      CAR.KONA_HEV,
+      CAR.KIA_SORENTO,
+    }
+
     for car_model, ecus in FW_VERSIONS.items():
       with self.subTest(car_model=car_model):
         for ecu, fws in ecus.items():
@@ -62,48 +87,75 @@ def test_fw_format(self):
             self.assertEqual(1, len(result), f"Unable to parse FW: {fw}")
             codes |= result
 
-          # Some newer platforms have date codes in a different format we don't yet parse,
-          # for now assert we can parse all FW or none
-          self.assertEqual(len({b"-" in code for code in codes}), 1,
-                           "Not all FW dates are parsable")
+          # So far we've only seen dates in fwdCamera
+          if car_model in no_dates_platforms or ecu[0] != Ecu.fwdCamera:
+            self.assertTrue(all({date is None for _, date in codes}))
+          else:
+            self.assertTrue(all({date is not None for _, date in codes}))
 
           if car_model == CAR.HYUNDAI_GENESIS:
             raise unittest.SkipTest("No part numbers for car model")
 
           # Hyundai places the ECU part number in their FW versions, assert all parsable
           # Some examples of valid formats: b"56310-L0010", b"56310L0010", b"56310/M6300"
-          for fw in fws:
-            match = PART_NUMBER_FW_PATTERN.search(fw)
-            self.assertIsNotNone(match, fw)
+          self.assertTrue(all({b"-" in code for code, _ in codes}),
+                          f"FW does not have part number: {fw}")
 
   def test_platform_codes_spot_check(self):
     # Asserts basic platform code parsing behavior for a few cases
-    codes = get_platform_codes([b"\xf1\x00DH LKAS 1.1 -150210"])
-    self.assertEqual(codes, {b"DH-1502"})
+    results = get_platform_codes([b"\xf1\x00DH LKAS 1.1 -150210"])
+    self.assertEqual(results, {(b"DH", b"150210")})
 
     # Some cameras and all radars do not have dates
-    codes = get_platform_codes([b"\xf1\x00AEhe SCC H-CUP      1.01 1.01 96400-G2000         "])
-    self.assertEqual(codes, {b"AEhe"})
+    results = get_platform_codes([b"\xf1\x00AEhe SCC H-CUP      1.01 1.01 96400-G2000         "])
+    self.assertEqual(results, {(b"AEhe-G2000", None)})
 
-    codes = get_platform_codes([b"\xf1\x00CV1_ RDR -----      1.00 1.01 99110-CV000         "])
-    self.assertEqual(codes, {b"CV1"})
+    results = get_platform_codes([b"\xf1\x00CV1_ RDR -----      1.00 1.01 99110-CV000         "])
+    self.assertEqual(results, {(b"CV1-CV000", None)})
 
-    codes = get_platform_codes([
+    results = get_platform_codes([
       b"\xf1\x00DH LKAS 1.1 -150210",
       b"\xf1\x00AEhe SCC H-CUP      1.01 1.01 96400-G2000         ",
       b"\xf1\x00CV1_ RDR -----      1.00 1.01 99110-CV000         ",
     ])
-    self.assertEqual(codes, {b"DH-1502", b"AEhe", b"CV1"})
+    self.assertEqual(results, {(b"DH", b"150210"), (b"AEhe-G2000", None), (b"CV1-CV000", None)})
 
-    # Returned platform codes must inclusively contain start/end dates
-    codes = get_platform_codes([
+    results = get_platform_codes([
       b"\xf1\x00LX2 MFC  AT USA LHD 1.00 1.07 99211-S8100 220222",
       b"\xf1\x00LX2 MFC  AT USA LHD 1.00 1.08 99211-S8100 211103",
       b"\xf1\x00ON  MFC  AT USA LHD 1.00 1.01 99211-S9100 190405",
       b"\xf1\x00ON  MFC  AT USA LHD 1.00 1.03 99211-S9100 190720",
     ])
-    self.assertEqual(codes, {b"LX2-2111", b"LX2-2112", b"LX2-2201", b"LX2-2202",
-                             b"ON-1904", b"ON-1905", b"ON-1906", b"ON-1907"})
+    self.assertEqual(results, {(b"LX2-S8100", b"220222"), (b"LX2-S8100", b"211103"),
+                               (b"ON-S9100", b"190405"), (b"ON-S9100", b"190720")})
+
+  def test_excluded_platforms(self):
+    # Asserts a list of platforms that will not fuzzy fingerprint with platform codes due to them being shared.
+    # This list can be shrunk as we combine platforms and detect features
+    excluded_platforms = {
+      CAR.GENESIS_G70,            # shared platform code, part number, and date
+      CAR.GENESIS_G70_2020,
+      CAR.TUCSON_4TH_GEN,         # shared platform code and part number
+      CAR.TUCSON_HYBRID_4TH_GEN,
+    } | set(CANFD_CAR - EV_CAR)   # shared platform codes
+
+    platforms_with_shared_codes = set()
+    for platform, fw_by_addr in FW_VERSIONS.items():
+      car_fw = []
+      for ecu, fw_versions in fw_by_addr.items():
+        ecu_name, addr, sub_addr = ecu
+        for fw in fw_versions:
+          car_fw.append({"ecu": ecu_name, "fwVersion": fw, "address": addr,
+                         "subAddress": 0 if sub_addr is None else sub_addr})
+
+      CP = car.CarParams.new_message(carFw=car_fw)
+      matches = FW_QUERY_CONFIG.match_fw_to_car_fuzzy(build_fw_dict(CP.carFw))
+      if len(matches) == 1:
+        self.assertEqual(list(matches)[0], platform)
+      else:
+        platforms_with_shared_codes.add(platform)
+
+    self.assertEqual(platforms_with_shared_codes, excluded_platforms)
 
 
 if __name__ == "__main__":

selfdrive/car/hyundai/values.py

@@ -1,18 +1,14 @@
 import re
-from datetime import datetime
-from dateutil import rrule
-from collections import defaultdict
 from dataclasses import dataclass
 from enum import Enum, IntFlag
-from typing import DefaultDict, Dict, List, Optional, Set, Union, cast
+from typing import Dict, List, Optional, Set, Tuple, Union
 
 from cereal import car
 from panda.python import uds
 from common.conversions import Conversions as CV
 from selfdrive.car import dbc_dict
 from selfdrive.car.docs_definitions import CarFootnote, CarHarness, CarInfo, CarParts, Column
 from selfdrive.car.fw_query_definitions import FwQueryConfig, Request, p16
-from system.swaglog import cloudlog
 
 Ecu = car.CarParams.Ecu
 
@@ -348,35 +344,72 @@ class Buttons:
 }
 
 
-def get_platform_codes(fw_versions: List[bytes]) -> Set[bytes]:
-  codes: DefaultDict[bytes, Set[Optional[bytes]]] = defaultdict(set)
+def get_platform_codes(fw_versions: List[bytes]) -> Set[Tuple[bytes, Optional[bytes]]]:
+  # Returns unique, platform-specific identification codes for a set of versions
+  codes = set()  # (code-Optional[part], date)
   for fw in fw_versions:
-    code_match, date_match = (PLATFORM_CODE_FW_PATTERN.search(fw),
-                              DATE_FW_PATTERN.search(fw))
+    code_match = PLATFORM_CODE_FW_PATTERN.search(fw)
+    part_match = PART_NUMBER_FW_PATTERN.search(fw)
+    date_match = DATE_FW_PATTERN.search(fw)
     if code_match is not None:
-      code = code_match.group()
+      code: bytes = code_match.group()
+      part = part_match.group() if part_match else None
       date = date_match.group() if date_match else None
-      codes[code].add(date)
+      if part is not None:
+        # part number starts with generic ECU part type, add what is specific to platform
+        code += b"-" + part[-5:]
 
-  # Create platform codes for all dates inclusive if ECU has FW dates
-  final_codes = set()
-  for code, dates in codes.items():
-    # Radar and some cameras don't have FW dates
-    if None in dates:
-      final_codes.add(code)
-      continue
+      codes.add((code, date))
+  return codes
 
-    try:
-      parsed = {datetime.strptime(cast(bytes, date).decode()[:4], '%y%m') for date in dates}
-    except ValueError:
-      cloudlog.exception(f'Error parsing date in FW versions: {code!r}, {dates}')
-      final_codes.add(code)
-      continue
 
-    for monthly in rrule.rrule(rrule.MONTHLY, dtstart=min(parsed), until=max(parsed)):
-      final_codes.add(code + b'-' + monthly.strftime('%y%m').encode())
+def match_fw_to_car_fuzzy(live_fw_versions, log=True) -> Set[str]:
+  # Non-electric CAN FD platforms often do not have platform code specifiers needed
+  # to distinguish between hybrid and ICE. All EVs so far are either exclusively
+  # electric or specify electric in the platform code.
+  fuzzy_platform_blacklist = set(CANFD_CAR - EV_CAR)
+  candidates = set()
 
-  return final_codes
+  for candidate, fws in FW_VERSIONS.items():
+    # Keep track of ECUs which pass all checks (platform codes, within date range)
+    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 codes
+      if ecu[0] not in PLATFORM_CODE_ECUS:
+        continue
+
+      # Expected platform codes & dates
+      codes = get_platform_codes(expected_versions)
+      expected_platform_codes = {code for code, _ in codes}
+      expected_dates = {date for _, date in codes if date is not None}
+
+      # Found platform codes & dates
+      codes = get_platform_codes(live_fw_versions.get(addr, set()))
+      found_platform_codes = {code for code, _ in codes}
+      found_dates = {date for _, date in codes if date is not None}
+
+      # Check platform code + 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
+
+      # Don't check dates if none in database
+      if len(expected_dates):
+        if not len(found_dates):
+          break
+
+        # Check all dates within range in the database, format is %y%m%d
+        if not all(min(expected_dates) <= found_date <= max(expected_dates) for found_date in found_dates):
+          break
+
+      valid_found_ecus.add(addr)
+
+    # If all live ECUs pass all checks for candidate, add it as a match
+    if len(valid_expected_ecus - valid_found_ecus) == 0:
+      candidates.add(candidate)
+
+  return candidates - fuzzy_platform_blacklist
 
 
 HYUNDAI_VERSION_REQUEST_LONG = bytes([uds.SERVICE_TYPE.READ_DATA_BY_IDENTIFIER]) + \
@@ -458,6 +491,8 @@ def get_platform_codes(fw_versions: List[bytes]) -> Set[bytes]:
     (Ecu.hvac, 0x7b3, None),         # HVAC Control Assembly
     (Ecu.cornerRadar, 0x7b7, None),
   ],
+  # Custom fuzzy fingerprinting function using platform codes, part numbers + FW dates:
+  match_fw_to_car_fuzzy=match_fw_to_car_fuzzy,
 )
 
 FW_VERSIONS = {