move live fingerprint debugging code to debug/car (#33308)

* remove cereal from fw_versions

* fix it

* do ecu_addrs and vin

* do last one

* move executable-ness

* notexe

* nice test
old-commit-hash: 2ed567b0f5

.importlinter

@@ -32,10 +32,6 @@ ignore_imports =
     openpilot.selfdrive.car.gm.interface -> openpilot.common.basedir
     openpilot.selfdrive.car.interfaces -> openpilot.common.basedir
 
-    # params will need to move to new openpilot files that just call car files
-    openpilot.selfdrive.car.fw_versions -> openpilot.common.params
-    openpilot.selfdrive.car.ecu_addrs -> openpilot.common.params
-
     # these are okay
     openpilot.selfdrive.car.card -> openpilot.common.swaglog
     openpilot.selfdrive.car.card -> openpilot.common.realtime

selfdrive/car/disable_ecu.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python3
 from openpilot.selfdrive.car import carlog
 from openpilot.selfdrive.car.isotp_parallel_query import IsoTpParallelQuery
 
@@ -35,17 +34,3 @@ def disable_ecu(can_recv, can_send, bus=0, addr=0x7d0, sub_addr=None, com_cont_r
     carlog.error(f"ecu disable retry ({i + 1}) ...")
   carlog.error("ecu disable failed")
   return False
-
-
-if __name__ == "__main__":
-  import time
-  import cereal.messaging as messaging
-  from openpilot.selfdrive.car.card import can_comm_callbacks
-  sendcan = messaging.pub_sock('sendcan')
-  logcan = messaging.sub_sock('can')
-  can_callbacks = can_comm_callbacks(logcan, sendcan)
-  time.sleep(1)
-
-  # honda bosch radar disable
-  disabled = disable_ecu(*can_callbacks, bus=1, addr=0x18DAB0F1, com_cont_req=b'\x28\x83\x03', timeout=0.5, debug=False)
-  print(f"disabled: {disabled}")

selfdrive/car/ecu_addrs.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python3
 import capnp
 import time
 
@@ -22,7 +21,7 @@ def _is_tester_present_response(msg: capnp.lib.capnp._DynamicStructReader, subad
   return False
 
 
-def _get_all_ecu_addrs(can_recv: CanRecvCallable, can_send: CanSendCallable, bus: int, timeout: float = 1, debug: bool = True) -> set[EcuAddrBusType]:
+def get_all_ecu_addrs(can_recv: CanRecvCallable, can_send: CanSendCallable, bus: int, timeout: float = 1, debug: bool = True) -> set[EcuAddrBusType]:
   addr_list = [0x700 + i for i in range(256)] + [0x18da00f1 + (i << 8) for i in range(256)]
   queries: set[EcuAddrBusType] = {(addr, None, bus) for addr in addr_list}
   responses = queries
@@ -56,41 +55,3 @@ def get_ecu_addrs(can_recv: CanRecvCallable, can_send: CanSendCallable, queries:
   except Exception:
     carlog.exception("ECU addr scan exception")
   return ecu_responses
-
-
-if __name__ == "__main__":
-  import argparse
-  import cereal.messaging as messaging
-  from openpilot.common.params import Params
-  from openpilot.selfdrive.car.card import can_comm_callbacks, obd_callback
-
-  parser = argparse.ArgumentParser(description='Get addresses of all ECUs')
-  parser.add_argument('--debug', action='store_true')
-  parser.add_argument('--bus', type=int, default=1)
-  parser.add_argument('--no-obd', action='store_true')
-  parser.add_argument('--timeout', type=float, default=1.0)
-  args = parser.parse_args()
-
-  logcan = messaging.sub_sock('can')
-  sendcan = messaging.pub_sock('sendcan')
-  can_callbacks = can_comm_callbacks(logcan, sendcan)
-
-  # Set up params for pandad
-  params = Params()
-  params.remove("FirmwareQueryDone")
-  params.put_bool("IsOnroad", False)
-  time.sleep(0.2)  # thread is 10 Hz
-  params.put_bool("IsOnroad", True)
-
-  obd_callback(params)(not args.no_obd)
-
-  print("Getting ECU addresses ...")
-  ecu_addrs = _get_all_ecu_addrs(*can_callbacks, args.bus, args.timeout, debug=args.debug)
-
-  print()
-  print("Found ECUs on rx addresses:")
-  for addr, subaddr, _ in ecu_addrs:
-    msg = f"  {hex(addr)}"
-    if subaddr is not None:
-      msg += f" (sub-address: {hex(subaddr)})"
-    print(msg)

selfdrive/car/fw_query_definitions.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python3
 import capnp
 import copy
 from dataclasses import dataclass, field

selfdrive/car/fw_versions.py

@@ -1,7 +1,6 @@
-#!/usr/bin/env python3
 from collections import defaultdict
 from collections.abc import Callable, Iterator
-from typing import Any, Protocol, TypeVar
+from typing import Protocol, TypeVar
 
 from tqdm import tqdm
 import capnp
@@ -327,69 +326,3 @@ def get_fw_versions(can_recv: CanRecvCallable, can_send: CanSendCallable, set_ob
           carlog.exception("FW query exception")
 
   return car_fw
-
-
-if __name__ == "__main__":
-  import time
-  import argparse
-  import cereal.messaging as messaging
-  from openpilot.common.params import Params
-  from openpilot.selfdrive.car.vin import get_vin
-  from openpilot.selfdrive.car.card import can_comm_callbacks, obd_callback
-
-  parser = argparse.ArgumentParser(description='Get firmware version of ECUs')
-  parser.add_argument('--scan', action='store_true')
-  parser.add_argument('--debug', action='store_true')
-  parser.add_argument('--brand', help='Only query addresses/with requests for this brand')
-  args = parser.parse_args()
-
-  logcan = messaging.sub_sock('can')
-  pandaStates_sock = messaging.sub_sock('pandaStates')
-  sendcan = messaging.pub_sock('sendcan')
-  can_callbacks = can_comm_callbacks(logcan, sendcan)
-
-  # Set up params for pandad
-  params = Params()
-  params.remove("FirmwareQueryDone")
-  params.put_bool("IsOnroad", False)
-  time.sleep(0.2)  # thread is 10 Hz
-  params.put_bool("IsOnroad", True)
-  set_obd_multiplexing = obd_callback(params)
-
-  extra: Any = None
-  if args.scan:
-    extra = {}
-    # Honda
-    for i in range(256):
-      extra[(Ecu.unknown, 0x18da00f1 + (i << 8), None)] = []
-      extra[(Ecu.unknown, 0x700 + i, None)] = []
-      extra[(Ecu.unknown, 0x750, i)] = []
-    extra = {"any": {"debug": extra}}
-
-  num_pandas = len(messaging.recv_one_retry(pandaStates_sock).pandaStates)
-
-  t = time.time()
-  print("Getting vin...")
-  set_obd_multiplexing(True)
-  vin_rx_addr, vin_rx_bus, vin = get_vin(*can_callbacks, (0, 1), debug=args.debug)
-  print(f'RX: {hex(vin_rx_addr)}, BUS: {vin_rx_bus}, VIN: {vin}')
-  print(f"Getting VIN took {time.time() - t:.3f} s")
-  print()
-
-  t = time.time()
-  fw_vers = get_fw_versions(*can_callbacks, set_obd_multiplexing, query_brand=args.brand, extra=extra, num_pandas=num_pandas, debug=args.debug, progress=True)
-  _, candidates = match_fw_to_car(fw_vers, vin)
-
-  print()
-  print("Found FW versions")
-  print("{")
-  padding = max([len(fw.brand) for fw in fw_vers] or [0])
-  for version in fw_vers:
-    subaddr = None if version.subAddress == 0 else hex(version.subAddress)
-    print(f"  Brand: {version.brand:{padding}}, bus: {version.bus}, OBD: {version.obdMultiplexing} - " +
-          f"(Ecu.{version.ecu}, {hex(version.address)}, {subaddr}): [{version.fwVersion}]")
-  print("}")
-
-  print()
-  print("Possible matches:", candidates)
-  print(f"Getting fw took {time.time() - t:.3f} s")

selfdrive/car/vin.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python3
 import re
 
 from panda.python.uds import get_rx_addr_for_tx_addr, FUNCTIONAL_ADDRS
@@ -57,25 +56,3 @@ def get_vin(can_recv, can_send, buses, timeout=0.1, retry=2, debug=False):
     carlog.error(f"vin query retry ({i+1}) ...")
 
   return -1, -1, VIN_UNKNOWN
-
-
-if __name__ == "__main__":
-  import argparse
-  import time
-  import cereal.messaging as messaging
-  from openpilot.selfdrive.car.card import can_comm_callbacks
-
-  parser = argparse.ArgumentParser(description='Get VIN of the car')
-  parser.add_argument('--debug', action='store_true')
-  parser.add_argument('--bus', type=int, default=1)
-  parser.add_argument('--timeout', type=float, default=0.1)
-  parser.add_argument('--retry', type=int, default=5)
-  args = parser.parse_args()
-
-  sendcan = messaging.pub_sock('sendcan')
-  logcan = messaging.sub_sock('can')
-  can_callbacks = can_comm_callbacks(logcan, sendcan)
-  time.sleep(1)
-
-  vin_rx_addr, vin_rx_bus, vin = get_vin(*can_callbacks, (args.bus,), args.timeout, args.retry, debug=args.debug)
-  print(f'RX: {hex(vin_rx_addr)}, BUS: {vin_rx_bus}, VIN: {vin}')

selfdrive/debug/car/disable_ecu.py

@@ -0,0 +1,15 @@
+#!/usr/bin/env python3
+import time
+import cereal.messaging as messaging
+from openpilot.selfdrive.car.card import can_comm_callbacks
+from openpilot.selfdrive.car.disable_ecu import disable_ecu
+
+if __name__ == "__main__":
+  sendcan = messaging.pub_sock('sendcan')
+  logcan = messaging.sub_sock('can')
+  can_callbacks = can_comm_callbacks(logcan, sendcan)
+  time.sleep(1)
+
+  # honda bosch radar disable
+  disabled = disable_ecu(*can_callbacks, bus=1, addr=0x18DAB0F1, com_cont_req=b'\x28\x83\x03', timeout=0.5, debug=False)
+  print(f"disabled: {disabled}")

selfdrive/debug/car/ecu_addrs.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+import argparse
+import time
+import cereal.messaging as messaging
+from openpilot.common.params import Params
+from openpilot.selfdrive.car.card import can_comm_callbacks, obd_callback
+from openpilot.selfdrive.car.ecu_addrs import get_all_ecu_addrs
+
+
+if __name__ == "__main__":
+  parser = argparse.ArgumentParser(description='Get addresses of all ECUs')
+  parser.add_argument('--debug', action='store_true')
+  parser.add_argument('--bus', type=int, default=1)
+  parser.add_argument('--no-obd', action='store_true')
+  parser.add_argument('--timeout', type=float, default=1.0)
+  args = parser.parse_args()
+
+  logcan = messaging.sub_sock('can')
+  sendcan = messaging.pub_sock('sendcan')
+  can_callbacks = can_comm_callbacks(logcan, sendcan)
+
+  # Set up params for pandad
+  params = Params()
+  params.remove("FirmwareQueryDone")
+  params.put_bool("IsOnroad", False)
+  time.sleep(0.2)  # thread is 10 Hz
+  params.put_bool("IsOnroad", True)
+
+  obd_callback(params)(not args.no_obd)
+
+  print("Getting ECU addresses ...")
+  ecu_addrs = get_all_ecu_addrs(*can_callbacks, args.bus, args.timeout, debug=args.debug)
+
+  print()
+  print("Found ECUs on rx addresses:")
+  for addr, subaddr, _ in ecu_addrs:
+    msg = f"  {hex(addr)}"
+    if subaddr is not None:
+      msg += f" (sub-address: {hex(subaddr)})"
+    print(msg)

selfdrive/debug/car/fw_versions.py

@@ -0,0 +1,70 @@
+#!/usr/bin/env python3
+import time
+import argparse
+import cereal.messaging as messaging
+from cereal import car
+from openpilot.common.params import Params
+from openpilot.selfdrive.car.card import can_comm_callbacks, obd_callback
+from openpilot.selfdrive.car.fw_versions import get_fw_versions, match_fw_to_car
+from openpilot.selfdrive.car.vin import get_vin
+from typing import Any
+
+Ecu = car.CarParams.Ecu
+
+if __name__ == "__main__":
+  parser = argparse.ArgumentParser(description='Get firmware version of ECUs')
+  parser.add_argument('--scan', action='store_true')
+  parser.add_argument('--debug', action='store_true')
+  parser.add_argument('--brand', help='Only query addresses/with requests for this brand')
+  args = parser.parse_args()
+
+  logcan = messaging.sub_sock('can')
+  pandaStates_sock = messaging.sub_sock('pandaStates')
+  sendcan = messaging.pub_sock('sendcan')
+  can_callbacks = can_comm_callbacks(logcan, sendcan)
+
+  # Set up params for pandad
+  params = Params()
+  params.remove("FirmwareQueryDone")
+  params.put_bool("IsOnroad", False)
+  time.sleep(0.2)  # thread is 10 Hz
+  params.put_bool("IsOnroad", True)
+  set_obd_multiplexing = obd_callback(params)
+
+  extra: Any = None
+  if args.scan:
+    extra = {}
+    # Honda
+    for i in range(256):
+      extra[(Ecu.unknown, 0x18da00f1 + (i << 8), None)] = []
+      extra[(Ecu.unknown, 0x700 + i, None)] = []
+      extra[(Ecu.unknown, 0x750, i)] = []
+    extra = {"any": {"debug": extra}}
+
+  num_pandas = len(messaging.recv_one_retry(pandaStates_sock).pandaStates)
+
+  t = time.time()
+  print("Getting vin...")
+  set_obd_multiplexing(True)
+  vin_rx_addr, vin_rx_bus, vin = get_vin(*can_callbacks, (0, 1), debug=args.debug)
+  print(f'RX: {hex(vin_rx_addr)}, BUS: {vin_rx_bus}, VIN: {vin}')
+  print(f"Getting VIN took {time.time() - t:.3f} s")
+  print()
+
+  t = time.time()
+  fw_vers = get_fw_versions(*can_callbacks, set_obd_multiplexing, query_brand=args.brand, extra=extra, num_pandas=num_pandas, debug=args.debug, progress=True)
+  _, candidates = match_fw_to_car(fw_vers, vin)
+
+  print()
+  print("Found FW versions")
+  print("{")
+  padding = max([len(fw.brand) for fw in fw_vers] or [0])
+  for version in fw_vers:
+    subaddr = None if version.subAddress == 0 else hex(version.subAddress)
+    print(f"  Brand: {version.brand:{padding}}, bus: {version.bus}, OBD: {version.obdMultiplexing} - " +
+          f"(Ecu.{version.ecu}, {hex(version.address)}, {subaddr}): [{version.fwVersion}]")
+  print("}")
+
+  print()
+  print("Possible matches:", candidates)
+  print(f"Getting fw took {time.time() - t:.3f} s")

selfdrive/debug/car/vin.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python3
+import argparse
+import time
+import cereal.messaging as messaging
+from openpilot.selfdrive.car.card import can_comm_callbacks
+from openpilot.selfdrive.car.vin import get_vin
+
+if __name__ == "__main__":
+  parser = argparse.ArgumentParser(description='Get VIN of the car')
+  parser.add_argument('--debug', action='store_true')
+  parser.add_argument('--bus', type=int, default=1)
+  parser.add_argument('--timeout', type=float, default=0.1)
+  parser.add_argument('--retry', type=int, default=5)
+  args = parser.parse_args()
+
+  sendcan = messaging.pub_sock('sendcan')
+  logcan = messaging.sub_sock('can')
+  can_callbacks = can_comm_callbacks(logcan, sendcan)
+  time.sleep(1)
+
+  vin_rx_addr, vin_rx_bus, vin = get_vin(*can_callbacks, (args.bus,), args.timeout, args.retry, debug=args.debug)
+  print(f'RX: {hex(vin_rx_addr)}, BUS: {vin_rx_bus}, VIN: {vin}')