sadmen
/
openpilot_comma
mirror of https://github.com/commaai/openpilot.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix test

Browse Source
pull/28423/head
Shane Smiskol 2 years ago
parent e9b3175264
commit 1d34269fe2
1 changed files with 147 additions and 147 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 294
      selfdrive/car/tests/test_fw_fingerprint.py

294
selfdrive/car/tests/test_fw_fingerprint.py
Unescape View File

@ -32,26 +32,26 @@ class TestFwFingerprint(unittest.TestCase):
self.assertEqual(len(candidates), 1, f"got more than one candidate: {candidates}")
self.assertEqual(candidates[0], expected)
@parameterized.expand([(b, c, e[c]) for b, e in VERSIONS.items() for c in e])
def test_exact_match(self, brand, car_model, ecus):
CP = car.CarParams.new_message()
for _ in range(200):
fw = []
for ecu, fw_versions in ecus.items():
if not len(fw_versions):
raise unittest.SkipTest("Car model has no FW versions")
ecu_name, addr, sub_addr = ecu
fw.append({"ecu": ecu_name, "fwVersion": random.choice(fw_versions), 'brand': brand,
"address": addr, "subAddress": 0 if sub_addr is None else sub_addr})
CP.carFw = fw
_, matches = match_fw_to_car(CP.carFw, allow_fuzzy=False)
self.assertFingerprints(matches, car_model)
# @parameterized.expand([(b, c, e[c]) for b, e in VERSIONS.items() for c in e])
# def test_exact_match(self, brand, car_model, ecus):
# CP = car.CarParams.new_message()
# for _ in range(200):
# fw = []
# for ecu, fw_versions in ecus.items():
# if not len(fw_versions):
# raise unittest.SkipTest("Car model has no FW versions")
# ecu_name, addr, sub_addr = ecu
# fw.append({"ecu": ecu_name, "fwVersion": random.choice(fw_versions), 'brand': brand,
# "address": addr, "subAddress": 0 if sub_addr is None else sub_addr})
# CP.carFw = fw
# _, matches = match_fw_to_car(CP.carFw, allow_fuzzy=False)
# self.assertFingerprints(matches, car_model)
@parameterized.expand([(b, c, e[c]) for b, e in VERSIONS.items() for c in e])
def test_fuzzy_match(self, brand, car_model, ecus):
# TODO: speed up fuzzy matching and test more
CP = car.CarParams.new_message()
for _ in range(5):
for _ in range(1):
fw = []
for ecu, fw_versions in ecus.items():
if not len(fw_versions):
@ -63,142 +63,142 @@ class TestFwFingerprint(unittest.TestCase):
_, matches = match_fw_to_car(CP.carFw, allow_exact=False, log=False)
# Assert no match if there are not enough valid ECUs
valid_ecus = [f['ecu'] for f in fw if f['ecu'] not in FUZZY_EXCLUDE_ECUS]
if len(valid_ecus) < 2:
self.assertEqual(len(matches), 0, valid_ecus)
valid_ecus = [(f['address'], f['subAddress']) for f in fw if f['ecu'] not in FUZZY_EXCLUDE_ECUS]
if len(set(valid_ecus)) < 2:
self.assertEqual(len(matches), 0)
# There won't always be a match due to shared FW, but if there is it should be correct
elif len(matches):
self.assertFingerprints(matches, car_model)
def test_no_duplicate_fw_versions(self):
for car_model, ecus in FW_VERSIONS.items():
with self.subTest(car_model=car_model):
for ecu, ecu_fw in ecus.items():
with self.subTest(ecu):
duplicates = {fw for fw in ecu_fw if ecu_fw.count(fw) > 1}
self.assertFalse(len(duplicates), f"{car_model}: Duplicate FW versions: Ecu.{ECU_NAME[ecu[0]]}, {duplicates}")
def test_all_addrs_map_to_one_ecu(self):
for brand, cars in VERSIONS.items():
addr_to_ecu = defaultdict(set)
for ecus in cars.values():
for ecu_type, addr, sub_addr in ecus.keys():
addr_to_ecu[(addr, sub_addr)].add(ecu_type)
ecus_for_addr = addr_to_ecu[(addr, sub_addr)]
ecu_strings = ", ".join([f'Ecu.{ECU_NAME[ecu]}' for ecu in ecus_for_addr])
self.assertLessEqual(len(ecus_for_addr), 1, f"{brand} has multiple ECUs that map to one address: {ecu_strings} -> ({hex(addr)}, {sub_addr})")
def test_data_collection_ecus(self):
# Asserts no extra ECUs are in the fingerprinting database
for brand, config in FW_QUERY_CONFIGS.items():
for car_model, ecus in VERSIONS[brand].items():
bad_ecus = set(ecus).intersection(config.extra_ecus)
with self.subTest(car_model=car_model):
self.assertFalse(len(bad_ecus), f'{car_model}: Fingerprints contain ECUs added for data collection: {bad_ecus}')
def test_blacklisted_ecus(self):
blacklisted_addrs = (0x7c4, 0x7d0) # includes A/C ecu and an unknown ecu
for car_model, ecus in FW_VERSIONS.items():
with self.subTest(car_model=car_model):
CP = interfaces[car_model][0].get_non_essential_params(car_model)
if CP.carName == 'subaru':
for ecu in ecus.keys():
self.assertNotIn(ecu[1], blacklisted_addrs, f'{car_model}: Blacklisted ecu: (Ecu.{ECU_NAME[ecu[0]]}, {hex(ecu[1])})')
elif CP.carName == "chrysler":
# Some HD trucks have a combined TCM and ECM
if CP.carFingerprint.startswith("RAM HD"):
for ecu in ecus.keys():
self.assertNotEqual(ecu[0], Ecu.transmission, f"{car_model}: Blacklisted ecu: (Ecu.{ECU_NAME[ecu[0]]}, {hex(ecu[1])})")
def test_missing_versions_and_configs(self):
brand_versions = set(VERSIONS.keys())
brand_configs = set(FW_QUERY_CONFIGS.keys())
if len(brand_configs - brand_versions):
with self.subTest():
self.fail(f"Brands do not implement FW_VERSIONS: {brand_configs - brand_versions}")
if len(brand_versions - brand_configs):
with self.subTest():
self.fail(f"Brands do not implement FW_QUERY_CONFIG: {brand_versions - brand_configs}")
def test_fw_request_ecu_whitelist(self):
for brand, config in FW_QUERY_CONFIGS.items():
with self.subTest(brand=brand):
whitelisted_ecus = {ecu for r in config.requests for ecu in r.whitelist_ecus}
brand_ecus = {fw[0] for car_fw in VERSIONS[brand].values() for fw in car_fw}
brand_ecus |= {ecu[0] for ecu in config.extra_ecus}
# each ecu in brand's fw versions + extra ecus needs to be whitelisted at least once
ecus_not_whitelisted = brand_ecus - whitelisted_ecus
ecu_strings = ", ".join([f'Ecu.{ECU_NAME[ecu]}' for ecu in ecus_not_whitelisted])
self.assertFalse(len(whitelisted_ecus) and len(ecus_not_whitelisted),
f'{brand.title()}: FW query whitelist missing ecus: {ecu_strings}')
class TestFwFingerprintTiming(unittest.TestCase):
@staticmethod
def _benchmark(brand, num_pandas, n):
params = Params()
fake_socket = FakeSocket()
times = []
for _ in range(n):
params.put_bool("ObdMultiplexingEnabled", True)
thread = threading.Thread(target=get_fw_versions, args=(fake_socket, fake_socket, brand), kwargs=dict(num_pandas=num_pandas))
thread.start()
t = time.perf_counter()
while thread.is_alive():
time.sleep(0.02)
if not params.get_bool("ObdMultiplexingChanged"):
params.put_bool("ObdMultiplexingChanged", True)
times.append(time.perf_counter() - t)
return round(sum(times) / len(times), 2)
def _assert_timing(self, avg_time, ref_time, tol):
self.assertLess(avg_time, ref_time + tol)
self.assertGreater(avg_time, ref_time - tol, "Performance seems to have improved, update test refs.")
def test_fw_query_timing(self):
tol = 0.1
total_ref_time = 4.6
brand_ref_times = {
1: {
'body': 0.1,
'chrysler': 0.3,
'ford': 0.2,
'honda': 0.5,
'hyundai': 0.7,
'mazda': 0.1,
'nissan': 0.3,
'subaru': 0.1,
'tesla': 0.2,
'toyota': 0.7,
'volkswagen': 0.2,
},
2: {
'hyundai': 1.1,
}
}
total_time = 0
for num_pandas in (1, 2):
for brand, config in FW_QUERY_CONFIGS.items():
with self.subTest(brand=brand, num_pandas=num_pandas):
multi_panda_requests = [r for r in config.requests if r.bus > 3]
if not len(multi_panda_requests) and num_pandas > 1:
raise unittest.SkipTest("No multi-panda FW queries")
avg_time = self._benchmark(brand, num_pandas, 10)
total_time += avg_time
self._assert_timing(avg_time, brand_ref_times[num_pandas][brand], tol)
print(f'{brand=}, {num_pandas=}, {len(config.requests)=}, avg FW query time={avg_time} seconds')
with self.subTest(brand='all_brands'):
self._assert_timing(total_time, total_ref_time, tol)
print(f'all brands, total FW query time={total_time} seconds')
# def test_no_duplicate_fw_versions(self):
# for car_model, ecus in FW_VERSIONS.items():
# with self.subTest(car_model=car_model):
# for ecu, ecu_fw in ecus.items():
# with self.subTest(ecu):
# duplicates = {fw for fw in ecu_fw if ecu_fw.count(fw) > 1}
# self.assertFalse(len(duplicates), f"{car_model}: Duplicate FW versions: Ecu.{ECU_NAME[ecu[0]]}, {duplicates}")
#
# def test_all_addrs_map_to_one_ecu(self):
# for brand, cars in VERSIONS.items():
# addr_to_ecu = defaultdict(set)
# for ecus in cars.values():
# for ecu_type, addr, sub_addr in ecus.keys():
# addr_to_ecu[(addr, sub_addr)].add(ecu_type)
# ecus_for_addr = addr_to_ecu[(addr, sub_addr)]
# ecu_strings = ", ".join([f'Ecu.{ECU_NAME[ecu]}' for ecu in ecus_for_addr])
# self.assertLessEqual(len(ecus_for_addr), 1, f"{brand} has multiple ECUs that map to one address: {ecu_strings} -> ({hex(addr)}, {sub_addr})")
#
# def test_data_collection_ecus(self):
# # Asserts no extra ECUs are in the fingerprinting database
# for brand, config in FW_QUERY_CONFIGS.items():
# for car_model, ecus in VERSIONS[brand].items():
# bad_ecus = set(ecus).intersection(config.extra_ecus)
# with self.subTest(car_model=car_model):
# self.assertFalse(len(bad_ecus), f'{car_model}: Fingerprints contain ECUs added for data collection: {bad_ecus}')
#
# def test_blacklisted_ecus(self):
# blacklisted_addrs = (0x7c4, 0x7d0) # includes A/C ecu and an unknown ecu
# for car_model, ecus in FW_VERSIONS.items():
# with self.subTest(car_model=car_model):
# CP = interfaces[car_model][0].get_non_essential_params(car_model)
# if CP.carName == 'subaru':
# for ecu in ecus.keys():
# self.assertNotIn(ecu[1], blacklisted_addrs, f'{car_model}: Blacklisted ecu: (Ecu.{ECU_NAME[ecu[0]]}, {hex(ecu[1])})')
#
# elif CP.carName == "chrysler":
# # Some HD trucks have a combined TCM and ECM
# if CP.carFingerprint.startswith("RAM HD"):
# for ecu in ecus.keys():
# self.assertNotEqual(ecu[0], Ecu.transmission, f"{car_model}: Blacklisted ecu: (Ecu.{ECU_NAME[ecu[0]]}, {hex(ecu[1])})")
#
# def test_missing_versions_and_configs(self):
# brand_versions = set(VERSIONS.keys())
# brand_configs = set(FW_QUERY_CONFIGS.keys())
# if len(brand_configs - brand_versions):
# with self.subTest():
# self.fail(f"Brands do not implement FW_VERSIONS: {brand_configs - brand_versions}")
#
# if len(brand_versions - brand_configs):
# with self.subTest():
# self.fail(f"Brands do not implement FW_QUERY_CONFIG: {brand_versions - brand_configs}")
#
# def test_fw_request_ecu_whitelist(self):
# for brand, config in FW_QUERY_CONFIGS.items():
# with self.subTest(brand=brand):
# whitelisted_ecus = {ecu for r in config.requests for ecu in r.whitelist_ecus}
# brand_ecus = {fw[0] for car_fw in VERSIONS[brand].values() for fw in car_fw}
# brand_ecus |= {ecu[0] for ecu in config.extra_ecus}
#
# # each ecu in brand's fw versions + extra ecus needs to be whitelisted at least once
# ecus_not_whitelisted = brand_ecus - whitelisted_ecus
#
# ecu_strings = ", ".join([f'Ecu.{ECU_NAME[ecu]}' for ecu in ecus_not_whitelisted])
# self.assertFalse(len(whitelisted_ecus) and len(ecus_not_whitelisted),
# f'{brand.title()}: FW query whitelist missing ecus: {ecu_strings}')
#
#
# class TestFwFingerprintTiming(unittest.TestCase):
# @staticmethod
# def _benchmark(brand, num_pandas, n):
# params = Params()
# fake_socket = FakeSocket()
#
# times = []
# for _ in range(n):
# params.put_bool("ObdMultiplexingEnabled", True)
# thread = threading.Thread(target=get_fw_versions, args=(fake_socket, fake_socket, brand), kwargs=dict(num_pandas=num_pandas))
# thread.start()
# t = time.perf_counter()
# while thread.is_alive():
# time.sleep(0.02)
# if not params.get_bool("ObdMultiplexingChanged"):
# params.put_bool("ObdMultiplexingChanged", True)
# times.append(time.perf_counter() - t)
#
# return round(sum(times) / len(times), 2)
#
# def _assert_timing(self, avg_time, ref_time, tol):
# self.assertLess(avg_time, ref_time + tol)
# self.assertGreater(avg_time, ref_time - tol, "Performance seems to have improved, update test refs.")
#
# def test_fw_query_timing(self):
# tol = 0.1
# total_ref_time = 4.6
# brand_ref_times = {
# 1: {
# 'body': 0.1,
# 'chrysler': 0.3,
# 'ford': 0.2,
# 'honda': 0.5,
# 'hyundai': 0.7,
# 'mazda': 0.1,
# 'nissan': 0.3,
# 'subaru': 0.1,
# 'tesla': 0.2,
# 'toyota': 0.7,
# 'volkswagen': 0.2,
# },
# 2: {
# 'hyundai': 1.1,
# }
# }
#
# total_time = 0
# for num_pandas in (1, 2):
# for brand, config in FW_QUERY_CONFIGS.items():
# with self.subTest(brand=brand, num_pandas=num_pandas):
# multi_panda_requests = [r for r in config.requests if r.bus > 3]
# if not len(multi_panda_requests) and num_pandas > 1:
# raise unittest.SkipTest("No multi-panda FW queries")
#
# avg_time = self._benchmark(brand, num_pandas, 10)
# total_time += avg_time
# self._assert_timing(avg_time, brand_ref_times[num_pandas][brand], tol)
# print(f'{brand=}, {num_pandas=}, {len(config.requests)=}, avg FW query time={avg_time} seconds')
#
# with self.subTest(brand='all_brands'):
# self._assert_timing(total_time, total_ref_time, tol)
# print(f'all brands, total FW query time={total_time} seconds')
if __name__ == "__main__":

Write Preview
Loading…
Cancel
Save