diff --git a/opendbc/car/chrysler/tests/__init__.py b/opendbc/car/chrysler/tests/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/opendbc/car/chrysler/tests/test_chrysler.py b/opendbc/car/chrysler/tests/test_chrysler.py new file mode 100644 index 00000000000..712a0ae9b57 --- /dev/null +++ b/opendbc/car/chrysler/tests/test_chrysler.py @@ -0,0 +1,166 @@ +import itertools +import random +import unittest +from unittest.mock import patch + +from opendbc.car.chrysler.fingerprints import FW_VERSIONS +from opendbc.car.chrysler.values import CAR, FW_QUERY_CONFIG, OPTIONAL_CODE_ECUS, PLATFORM_CODE_ECUS, get_platform_codes, match_fw_to_car_fuzzy +from opendbc.car.fw_versions import match_fw_to_car +from opendbc.car.structs import CarParams +from opendbc.testing import fuzzy_test + +Ecu = CarParams.Ecu + + +def unseen_revision(fw): + codes = get_platform_codes([fw]) + return next(iter(codes)) + b'ZZ' if codes else fw + + +class TestChryslerFW(unittest.TestCase): + def test_platform_codes(self): + self.assertEqual(get_platform_codes([b'68227902AF', b'68227902AG', b'68267018AO ']), {b'68227902', b'68267018'}) + for fw in (b'', b'68227902A', b'68227902ABC', b'68227902AF\n', b'68227902AF\x00', + b' 68227902AF', b'68227902AF ', b'68227902af', b'22DTRHD_AA'): + with self.subTest(fw=fw): + self.assertEqual(get_platform_codes([fw]), set()) + + @fuzzy_test(max_examples=100) + def test_platform_codes_arbitrary_bytes(self, fuzzy): + get_platform_codes(fuzzy.list(fuzzy.binary)) + + def test_all_database_versions(self): + unparsed = set() + for fws in FW_VERSIONS.values(): + for ecu, versions in fws.items(): + if ecu[0] not in PLATFORM_CODE_ECUS: + continue + for fw in versions: + if not get_platform_codes([fw]): + unparsed.add((ecu[0], fw)) + self.assertEqual(unparsed, {(Ecu.fwdRadar, b'22DTRHD_AA')}) + + def test_known_and_unseen_revisions(self): + rng = random.Random(0) + for platform, fws in FW_VERSIONS.items(): + for revise in (False, True): + for _ in range(100): + live = {ecu[1:]: {rng.choice(versions)} for ecu, versions in fws.items()} + if revise: + live = {addr: {unseen_revision(fw) for fw in versions} for addr, versions in live.items()} + with self.subTest(platform=platform, revise=revise): + self.assertEqual(match_fw_to_car_fuzzy(live, '', FW_VERSIONS), {platform}) + + def test_each_unseen_revision(self): + # Exercise every checked-in version, including the non-part-number radar + # response, without relying on random draws to select rare firmware. + for platform, fws in FW_VERSIONS.items(): + baseline = {ecu[1:]: {unseen_revision(versions[0])} for ecu, versions in fws.items()} + for ecu, versions in fws.items(): + for fw in versions: + with self.subTest(platform=platform, ecu=ecu, fw=fw): + live = baseline | {ecu[1:]: {unseen_revision(fw)}} + self.assertEqual(match_fw_to_car_fuzzy(live, '', FW_VERSIONS), {platform}) + + def test_reject_changed_or_missing_required_ecu(self): + for platform, fws in FW_VERSIONS.items(): + baseline = {ecu[1:]: {unseen_revision(versions[0])} for ecu, versions in fws.items()} + for ecu in fws: + if ecu[0] not in PLATFORM_CODE_ECUS or ecu[0] in OPTIONAL_CODE_ECUS: + continue + for replacements in (set(), {b'99999999AA'}, {b'68227902AF\n'}): + with self.subTest(platform=platform, ecu=ecu, replacements=replacements): + self.assertEqual(match_fw_to_car_fuzzy(baseline | {ecu[1:]: replacements}, '', FW_VERSIONS), set()) + + def test_one_unseen_cluster(self): + for platform, fws in FW_VERSIONS.items(): + baseline = {ecu[1:]: {unseen_revision(versions[0])} for ecu, versions in fws.items()} + body_addrs = {ecu[1:] for ecu in fws if ecu[0] in OPTIONAL_CODE_ECUS} + for addr in body_addrs: + for versions in (set(), {b'99999999AA'}): + with self.subTest(platform=platform, addr=addr, versions=versions): + self.assertEqual(match_fw_to_car_fuzzy(baseline | {addr: versions}, '', FW_VERSIONS), {platform}) + self.assertEqual(match_fw_to_car_fuzzy(baseline | {addr: {b'corrupt'}}, '', FW_VERSIONS), set()) + srs_addr = next(ecu[1:] for ecu in fws if ecu[0] == Ecu.srs) + unknown = {addr: {b'99999999AA'} for addr in body_addrs | {srs_addr}} + self.assertEqual(match_fw_to_car_fuzzy(baseline | unknown, '', FW_VERSIONS), set()) + + def test_recorded_ram_hd_cluster_parts(self): + # Public commaCarSegments devices 64dbf4f1e713bfa8 and 77d9432d15a85e17: + # cluster parts are absent from the database, while the control ECUs match. + fws = FW_VERSIONS[CAR.RAM_HD_5TH_GEN] + live = {ecu[1:]: {unseen_revision(versions[0])} for ecu, versions in fws.items()} + for cluster_fw in (b'68525438AB', b'68620923AB'): + self.assertEqual(match_fw_to_car_fuzzy(live | {(0x742, None): {cluster_fw}}, '', FW_VERSIONS), {CAR.RAM_HD_5TH_GEN}) + + def test_known_conflicting_cluster(self): + for platform, other in itertools.permutations(FW_VERSIONS, 2): + fws = FW_VERSIONS[platform] + live = {ecu[1:]: {unseen_revision(versions[0])} for ecu, versions in fws.items()} + for ecu, versions in fws.items(): + if ecu[0] not in OPTIONAL_CODE_ECUS or ecu not in FW_VERSIONS[other]: + continue + foreign = get_platform_codes(FW_VERSIONS[other][ecu]) - get_platform_codes(versions) + for code in foreign: + with self.subTest(platform=platform, other=other, ecu=ecu): + self.assertEqual(match_fw_to_car_fuzzy(live | {ecu[1:]: {code + b'ZZ'}}, '', FW_VERSIONS), set()) + + def test_non_platform_ecus_do_not_prevent_matching(self): + for platform, fws in FW_VERSIONS.items(): + live = {ecu[1:]: {unseen_revision(versions[0])} for ecu, versions in fws.items() if ecu[0] in PLATFORM_CODE_ECUS} + self.assertEqual(match_fw_to_car_fuzzy(live, '', FW_VERSIONS), {platform}) + for ecu in fws: + if ecu[0] not in PLATFORM_CODE_ECUS: + live[ecu[1:]] = {b'99999999AA'} + self.assertEqual(match_fw_to_car_fuzzy(live, '', FW_VERSIONS), {platform}) + + def test_shared_modules_cannot_identify_sibling(self): + for platform, other in itertools.permutations(FW_VERSIONS, 2): + other_fws = FW_VERSIONS[other] + live = {ecu[1:]: {unseen_revision(fw) for fw in versions} for ecu, versions in other_fws.items()} + self.assertEqual(match_fw_to_car_fuzzy(live, '', {platform: FW_VERSIONS[platform]}), set()) + + def test_empty_or_incomplete_database(self): + eps = (Ecu.eps, 0x730, None) + meter = (Ecu.combinationMeter, 0x742, None) + live = {eps[1:]: {b'68421036ZZ'}, meter[1:]: {b'68227902ZZ'}} + for offline in ({}, {'car': {}}, {'car': {eps: [b'68421036AC']}}, {'car': {meter: [b'68227902AF']}}): + self.assertEqual(match_fw_to_car_fuzzy(live, '', offline), set()) + complete = {'car': {eps: [b'68421036AC'], meter: [b'68227902AF']}} + self.assertEqual(match_fw_to_car_fuzzy(live, '', complete), {'car'}) + self.assertEqual(match_fw_to_car_fuzzy({}, '', complete), set()) + wrong_subaddress = {(address, 1): versions for (address, _), versions in live.items()} + self.assertEqual(match_fw_to_car_fuzzy(wrong_subaddress, '', complete), set()) + + def test_ambiguous_platforms_are_not_guessed(self): + fws = next(iter(FW_VERSIONS.values())) + live = {ecu[1:]: {unseen_revision(versions[0])} for ecu, versions in fws.items()} + self.assertEqual(match_fw_to_car_fuzzy(live, '', {'first': fws, 'second': fws}), {'first', 'second'}) + + def test_public_matcher_uses_chrysler_fallback(self): + self.assertIs(FW_QUERY_CONFIG.match_fw_to_car_fuzzy, match_fw_to_car_fuzzy) + for platform, fws in FW_VERSIONS.items(): + car_fw = [CarParams.CarFw(ecu=ecu, address=address, subAddress=subaddress or 0, brand='chrysler', + fwVersion=unseen_revision(versions[0])) for (ecu, address, subaddress), versions in fws.items()] + with self.subTest(platform=platform): + exact, matches = match_fw_to_car(car_fw, '', log=False) + self.assertFalse(exact) + self.assertEqual(matches, {platform}) + + def test_generic_matching_cannot_bypass_control_ecus(self): + # Keep the other firmware exact: revising every ECU would prevent the generic + # matcher from finding the body/powertrain matches which trigger this bug. + for platform, fws in FW_VERSIONS.items(): + for changed_ecu in {ecu[0] for ecu in fws if ecu[0] in (Ecu.eps, Ecu.abs, Ecu.fwdRadar)}: + for missing in (False, True): + car_fw = [CarParams.CarFw(ecu=ecu, address=address, subAddress=subaddress or 0, brand='chrysler', + fwVersion=b'99999999AA' if ecu == changed_ecu else versions[0]) + for (ecu, address, subaddress), versions in fws.items() if not (missing and ecu == changed_ecu)] + with self.subTest(platform=platform, ecu=changed_ecu, missing=missing): + _, matches = match_fw_to_car(car_fw, '', log=False) + self.assertEqual(matches, set()) + # Demonstrate a reachable generic-matcher bypass; opt-out is needed + # even when a brand hook verifies the control modules. + with patch.object(FW_QUERY_CONFIG, 'use_generic_fuzzy', True): + _, matches = match_fw_to_car(car_fw, '', log=False) + self.assertEqual(matches, {platform}) diff --git a/opendbc/car/chrysler/values.py b/opendbc/car/chrysler/values.py index d5e0fca3ae5..d68c539010f 100644 --- a/opendbc/car/chrysler/values.py +++ b/opendbc/car/chrysler/values.py @@ -1,10 +1,12 @@ +import re +from collections import defaultdict 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 @@ -141,6 +143,59 @@ def __init__(self, CP): CHRYSLER_RX_OFFSET = -0x280 +# Mopar part numbers end with a two-letter revision. Keep the whole part number, +# including any hardware/model-year distinction, when matching unseen revisions. +FW_PATTERN = re.compile(br'(?P[0-9A-Z]{8})[A-Z]{2} ?') +PLATFORM_CODE_ECUS = (Ecu.abs, Ecu.combinationMeter, Ecu.srs, Ecu.eps, Ecu.fwdRadar) +OPTIONAL_CODE_ECUS = (Ecu.combinationMeter,) + + +def get_platform_codes(fw_versions: list[bytes] | set[bytes]) -> set[bytes]: + return {match.group('part_number') for fw in fw_versions if (match := FW_PATTERN.fullmatch(fw)) is not None} + + +def match_fw_to_car_fuzzy(live_fw_versions: LiveFwVersions, vin: str, offline_fw_versions: OfflineFwVersions) -> set[str]: + known_body_codes: dict[tuple[int, int | None], set[bytes]] = defaultdict(set) + for fws in offline_fw_versions.values(): + for ecu, versions in fws.items(): + if ecu[0] in OPTIONAL_CODE_ECUS: + known_body_codes[ecu[1:]].update(get_platform_codes(versions)) + + candidates = set() + for candidate, fws in offline_fw_versions.items(): + expected_ecus = {ecu: versions for ecu, versions in fws.items() if ecu[0] in PLATFORM_CODE_ECUS} + # Require the steering ECU and independent platform evidence. A powertrain or + # body-module match alone cannot establish a compatible steering interface. + if not any(ecu[0] == Ecu.eps for ecu in expected_ecus) or len(expected_ecus) < 2: + continue + + matched = 0 + unseen_body = 0 + for ecu, expected_versions in expected_ecus.items(): + found_versions = live_fw_versions.get(ecu[1:], set()) + # Some radar responses (e.g. 22DTRHD_AA) are not Mopar part numbers. Such + # responses must match exactly; do not guess how to normalize them. + exact = set(expected_versions) & found_versions + found_codes = get_platform_codes(found_versions) + codes = get_platform_codes(expected_versions) & found_codes + if exact or codes: + matched += 1 + continue + + # Recorded RAM HDs have new cluster part numbers with known control ECUs. + # Tolerate one missing/unseen cluster, never an unknown control module + # or a cluster part number belonging to another known platform. + unknown_body = not found_versions or (found_codes and not (found_codes & known_body_codes[ecu[1:]])) + if ecu[0] not in OPTIONAL_CODE_ECUS or not unknown_body or unseen_body: + break + unseen_body += 1 + else: + if matched >= 2: + candidates.add(candidate) + + return candidates + + FW_QUERY_CONFIG = FwQueryConfig( fw_version_regex=br"[A-Z0-9_]{10} ?", requests=[ @@ -167,6 +222,8 @@ def __init__(self, CP): extra_ecus=[ (Ecu.abs, 0x7e4, None), # alt address for abs on hybrids, NOTE: not on all hybrid platforms ], + match_fw_to_car_fuzzy=match_fw_to_car_fuzzy, + use_generic_fuzzy=False, ) DBC = CAR.create_dbc_map() diff --git a/opendbc/car/fw_query_definitions.py b/opendbc/car/fw_query_definitions.py index fdcb0e990c5..f08b9d83520 100644 --- a/opendbc/car/fw_query_definitions.py +++ b/opendbc/car/fw_query_definitions.py @@ -108,6 +108,8 @@ class FwQueryConfig: # Function a brand can implement to provide better fuzzy matching. Takes in FW versions and VIN, # returns set of candidates. Only will match if one candidate is returned match_fw_to_car_fuzzy: Callable[[LiveFwVersions, str, OfflineFwVersions], set[str]] | None = None + # Disable the generic fallback when a brand requires ECU evidence it ignores. + use_generic_fuzzy: bool = True def __post_init__(self): # Asserts that a request exists if extra ecus are used diff --git a/opendbc/car/fw_versions.py b/opendbc/car/fw_versions.py index d672b039984..a8b658f4219 100644 --- a/opendbc/car/fw_versions.py +++ b/opendbc/car/fw_versions.py @@ -156,11 +156,12 @@ def match_fw_to_car(fw_versions: list[CarParams.CarFw], vin: str, allow_exact: b # For each brand, attempt to fingerprint using all FW returned from its queries matches: set[str] = set() for brand in VERSIONS.keys(): + config = FW_QUERY_CONFIGS[brand] fw_versions_dict = build_fw_dict(fw_versions, filter_brand=brand) - matches |= match_func(fw_versions_dict, match_brand=brand, log=log) + if exact_match or config.use_generic_fuzzy: + matches |= match_func(fw_versions_dict, match_brand=brand, 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, vin, VERSIONS[brand])