sunnypilot v2026.003.000

version: sunnypilot v2026.003.000 (staging)
date: 2026-09-03T14:24:40
master commit: 132b31f4cf

tools/scripts/car/can_print_changes.py

@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+import argparse
+import binascii
+import time
+from collections import defaultdict
+
+import openpilot.cereal.messaging as messaging
+from tools.scripts.car.can_table import can_table
+from openpilot.tools.lib.logreader import LogIterable, LogReader
+
+RED = '\033[91m'
+CLEAR = '\033[0m'
+
+def update(msgs, bus, dat, low_to_high, high_to_low, quiet=False):
+  for x in msgs:
+    if x.which() != 'can':
+      continue
+
+    for y in x.can:
+      if y.src == bus:
+        dat[y.address] = y.dat
+
+        i = int.from_bytes(y.dat, byteorder='big')
+        l_h = low_to_high[y.address]
+        h_l = high_to_low[y.address]
+
+        change = None
+        if (i | l_h) != l_h:
+          low_to_high[y.address] = i | l_h
+          change = "+"
+
+        if (~i | h_l) != h_l:
+          high_to_low[y.address] = ~i | h_l
+          change = "-"
+
+        if change and not quiet:
+          print(f"{time.monotonic():.2f}\t{hex(y.address)} ({y.address})\t{change}{binascii.hexlify(y.dat)}")
+
+
+def can_printer(bus=0, init_msgs=None, new_msgs=None, table=False):
+  logcan = messaging.sub_sock('can', timeout=10)
+
+  dat = defaultdict(int)
+  low_to_high = defaultdict(int)
+  high_to_low = defaultdict(int)
+
+  if init_msgs is not None:
+    update(init_msgs, bus, dat, low_to_high, high_to_low, quiet=True)
+
+  low_to_high_init = low_to_high.copy()
+  high_to_low_init = high_to_low.copy()
+
+  if new_msgs is not None:
+    update(new_msgs, bus, dat, low_to_high, high_to_low)
+  else:
+    # Live mode
+    print(f"Waiting for messages on bus {bus}")
+    try:
+      while 1:
+        can_recv = messaging.drain_sock(logcan)
+        update(can_recv, bus, dat, low_to_high, high_to_low)
+        time.sleep(0.02)
+    except KeyboardInterrupt:
+      pass
+
+  print("\n\n")
+  tables = ""
+  for addr in sorted(dat.keys()):
+    init = low_to_high_init[addr] & high_to_low_init[addr]
+    now = low_to_high[addr] & high_to_low[addr]
+    d = now & ~init
+    if d == 0:
+      continue
+    b = d.to_bytes(len(dat[addr]), byteorder='big')
+
+    byts = ''.join([(c if c == '0' else f'{RED}{c}{CLEAR}') for c in str(binascii.hexlify(b))[2:-1]])
+    header = f"{hex(addr).ljust(6)}({str(addr).ljust(4)})"
+    print(header, byts)
+    tables += f"{header}\n"
+    tables += can_table(b) + "\n\n"
+
+  if table:
+    print(tables)
+
+if __name__ == "__main__":
+  desc = """Collects messages and prints when a new bit transition is observed.
+  This is very useful to find signals based on user triggered actions, such as blinkers and seatbelt.
+  Leave the script running until no new transitions are seen, then perform the action."""
+  parser = argparse.ArgumentParser(description=desc,
+                                   formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+  parser.add_argument("--bus", type=int, help="CAN bus to print out", default=0)
+  parser.add_argument("--table", action="store_true", help="Print a cabana-like table")
+  parser.add_argument("init", type=str, nargs='?', help="Route or segment to initialize with. Use empty quotes to compare against all zeros.")
+  parser.add_argument("comp", type=str, nargs='?', help="Route or segment to compare against init")
+
+  args = parser.parse_args()
+
+  init_lr: LogIterable | None = None
+  new_lr: LogIterable | None = None
+
+  if args.init:
+    if args.init == '':
+      init_lr = []
+    else:
+      init_lr = LogReader(args.init)
+  if args.comp:
+    new_lr = LogReader(args.comp)
+
+  can_printer(args.bus, init_msgs=init_lr, new_msgs=new_lr, table=args.table)

tools/scripts/car/can_printer.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+import argparse
+import binascii
+import time
+from collections import defaultdict
+
+import openpilot.cereal.messaging as messaging
+
+
+def can_printer(bus, max_msg, addr, ascii_decode):
+  logcan = messaging.sub_sock('can', addr=addr)
+
+  start = time.monotonic()
+  lp = time.monotonic()
+  msgs = defaultdict(list)
+  while 1:
+    can_recv = messaging.drain_sock(logcan, wait_for_one=True)
+    for x in can_recv:
+      for y in x.can:
+        if y.src == bus:
+          msgs[y.address].append(y.dat)
+
+    if time.monotonic() - lp > 0.1:
+      dd = chr(27) + "[2J"
+      dd += f"{time.monotonic() - start:5.2f}\n"
+      for _addr in sorted(msgs.keys()):
+        a = f"\"{msgs[_addr][-1].decode('ascii', 'backslashreplace')}\"" if ascii_decode else ""
+        x = binascii.hexlify(msgs[_addr][-1]).decode('ascii')
+        freq = len(msgs[_addr]) / (time.monotonic() - start)
+        if max_msg is None or _addr < max_msg:
+          dd += f"{_addr:04X}({_addr:4d})({len(msgs[_addr]):6d})({freq:3}dHz) {x.ljust(20)} {a}\n"
+      print(dd)
+      lp = time.monotonic()
+
+if __name__ == "__main__":
+  parser = argparse.ArgumentParser(description="simple CAN data viewer",
+                                   formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+
+  parser.add_argument("--bus", type=int, help="CAN bus to print out", default=0)
+  parser.add_argument("--max_msg", type=int, help="max addr")
+  parser.add_argument("--ascii", action='store_true', help="decode as ascii")
+  parser.add_argument("--addr", default="127.0.0.1")
+
+  args = parser.parse_args()
+  can_printer(args.bus, args.max_msg, args.addr, args.ascii)

tools/scripts/car/can_table.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+import argparse
+import pandas as pd
+
+import openpilot.cereal.messaging as messaging
+
+
+def can_table(dat):
+  rows = []
+  for b in dat:
+    r = list(bin(b).lstrip('0b').zfill(8))
+    r += [hex(b)]
+    rows.append(r)
+
+  df = pd.DataFrame(data=rows)
+  df.columns = [str(n) for n in range(7, -1, -1)] + [' ']
+  table = df.to_markdown(tablefmt='grid')
+  return table
+
+
+if __name__ == "__main__":
+  parser = argparse.ArgumentParser(description="Cabana-like table of bits for your terminal",
+                                   formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+  parser.add_argument("addr", type=str, nargs=1)
+  parser.add_argument("bus", type=int, default=0, nargs='?')
+
+  args = parser.parse_args()
+
+  addr = int(args.addr[0], 0)
+  can = messaging.sub_sock('can', conflate=False, timeout=None)
+
+  print(f"waiting for {hex(addr)} ({addr}) on bus {args.bus}...")
+
+  latest = None
+  while True:
+    for msg in messaging.drain_sock(can, wait_for_one=True):
+      for m in msg.can:
+        if m.address == addr and m.src == args.bus:
+          latest = m
+
+    if latest is None:
+      continue
+
+    table = can_table(latest.dat)
+    print(f"\n\n{hex(addr)} ({addr}) on bus {args.bus}\n{table}")

tools/scripts/car/clear_dtc.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+import sys
+import argparse
+from subprocess import check_output, CalledProcessError
+from opendbc.car.carlog import carlog
+from opendbc.car.uds import UdsClient, MessageTimeoutError, SESSION_TYPE, DTC_GROUP_TYPE
+from opendbc.car.structs import CarParams
+from panda import Panda
+
+parser = argparse.ArgumentParser(description="clear DTC status")
+parser.add_argument("addr", type=lambda x: int(x,0), nargs="?", default=0x7DF) # default is functional (broadcast) address
+parser.add_argument("--bus", type=int, default=0)
+parser.add_argument('--debug', action='store_true')
+args = parser.parse_args()
+
+if args.debug:
+  carlog.setLevel('DEBUG')
+
+try:
+  check_output(["pidof", "pandad"])
+  print("pandad is running, please kill openpilot before running this script! (aborted)")
+  sys.exit(1)
+except CalledProcessError as e:
+  if e.returncode != 1: # 1 == no process found (pandad not running)
+    raise e
+
+panda = Panda()
+panda.set_safety_mode(CarParams.SafetyModel.elm327)
+uds_client = UdsClient(panda, args.addr, bus=args.bus)
+print("extended diagnostic session ...")
+try:
+  uds_client.diagnostic_session_control(SESSION_TYPE.EXTENDED_DIAGNOSTIC)
+except MessageTimeoutError:
+  # functional address isn't properly handled so a timeout occurs
+  if args.addr != 0x7DF:
+    raise
+print("clear diagnostic info ...")
+try:
+  uds_client.clear_diagnostic_information(DTC_GROUP_TYPE.ALL)
+except MessageTimeoutError:
+  # functional address isn't properly handled so a timeout occurs
+  if args.addr != 0x7DF:
+    pass
+print("")
+print("you may need to power cycle your vehicle now")

tools/scripts/car/disable_ecu.py

@@ -0,0 +1,15 @@
+#!/usr/bin/env python3
+import time
+import openpilot.cereal.messaging as messaging
+from opendbc.car.disable_ecu import disable_ecu
+from openpilot.selfdrive.car.card import can_comm_callbacks
+
+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)
+  print(f"disabled: {disabled}")

tools/scripts/car/ecu_addrs.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+import argparse
+import time
+import openpilot.cereal.messaging as messaging
+from opendbc.car.carlog import carlog
+from opendbc.car.ecu_addrs import get_all_ecu_addrs
+from openpilot.common.params import Params
+from openpilot.selfdrive.car.card import can_comm_callbacks, obd_callback
+
+
+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()
+
+  if args.debug:
+    carlog.setLevel('DEBUG')
+
+  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("IsOffroad", True, block=True)
+  time.sleep(0.2)  # thread is 10 Hz
+  params.put_bool("IsOffroad", False, block=True)
+
+  obd_callback(params)(not args.no_obd)
+
+  print("Getting ECU addresses ...")
+  ecu_addrs = get_all_ecu_addrs(*can_callbacks, args.bus, args.timeout)
+
+  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)

tools/scripts/car/fw_versions.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+import time
+import argparse
+import openpilot.cereal.messaging as messaging
+from opendbc.car.structs import car
+from opendbc.car.carlog import carlog
+from opendbc.car.fw_versions import get_fw_versions, match_fw_to_car
+from opendbc.car.vin import get_vin
+from openpilot.common.params import Params
+from openpilot.selfdrive.car.card import can_comm_callbacks, obd_callback
+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()
+
+  if args.debug:
+    carlog.setLevel('DEBUG')
+
+  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("IsOffroad", True, block=True)
+  time.sleep(0.2)  # thread is 10 Hz
+  params.put_bool("IsOffroad", False, block=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}}
+
+  t = time.monotonic()
+  print("Getting vin...")
+  set_obd_multiplexing(True)
+  vin_rx_addr, vin_rx_bus, vin = get_vin(*can_callbacks, (0, 1))
+  print(f'RX: {hex(vin_rx_addr)}, BUS: {vin_rx_bus}, VIN: {vin}')
+  print(f"Getting VIN took {time.monotonic() - t:.3f} s")
+  print()
+
+  t = time.monotonic()
+  fw_vers = get_fw_versions(*can_callbacks, set_obd_multiplexing, query_brand=args.brand, extra=extra, 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!r}]")
+  print("}")
+
+  print()
+  print("Possible matches:", candidates)
+  print(f"Getting fw took {time.monotonic() - t:.3f} s")

tools/scripts/car/hyundai_enable_radar_points.py

@@ -0,0 +1,142 @@
+#!/usr/bin/env python3
+"""Some Hyundai radars can be reconfigured to output (debug) radar points on bus 1.
+Reconfiguration is done over UDS by reading/writing to 0x0142 using the Read/Write Data By Identifier
+endpoints (0x22 & 0x2E). This script checks your radar firmware version against a list of known
+firmware versions. If you want to try on a new radar make sure to note the default config value
+in case it's different from the other radars and you need to revert the changes.
+
+After changing the config the car should not show any faults when openpilot is not running.
+These config changes are persistent across car reboots. You need to run this script again
+to go back to the default values.
+
+USE AT YOUR OWN RISK! Safety features, like AEB and FCW, might be affected by these changes."""
+
+import sys
+import argparse
+from typing import NamedTuple
+from subprocess import check_output, CalledProcessError
+from opendbc.car.carlog import carlog
+from opendbc.car.uds import UdsClient, SESSION_TYPE, DATA_IDENTIFIER_TYPE
+from opendbc.car.structs import CarParams
+from panda.python import Panda
+
+class ConfigValues(NamedTuple):
+  default_config: bytes
+  tracks_enabled: bytes
+
+# If your radar supports changing data identifier 0x0142 as well make a PR to
+# this file to add your firmware version. Make sure to post a drive as proof!
+# NOTE: these firmware versions do not match what openpilot uses
+#       because this script uses a different diagnostic session type
+SUPPORTED_FW_VERSIONS = {
+  # 2020 SONATA
+  b"DN8_ SCC FHCUP      1.00 1.00 99110-L0000\x19\x08)\x15T    ": ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+  b"DN8_ SCC F-CUP      1.00 1.00 99110-L0000\x19\x08)\x15T    ": ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+  # 2021 SONATA HYBRID
+  b"DNhe SCC FHCUP      1.00 1.00 99110-L5000\x19\x04&\x13'    ": ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+  b"DNhe SCC FHCUP      1.00 1.02 99110-L5000 \x01#\x15#    ": ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+  # 2020 PALISADE
+  b"LX2_ SCC FHCUP      1.00 1.04 99110-S8100\x19\x05\x02\x16V    ": ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+  # 2022 PALISADE
+  b"LX2_ SCC FHCUP      1.00 1.00 99110-S8110!\x04\x05\x17\x01    ": ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+  # 2020 SANTA FE
+  b"TM__ SCC F-CUP      1.00 1.03 99110-S2000\x19\x050\x13'    ": ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+  # 2020 GENESIS G70
+  b'IK__ SCC F-CUP      1.00 1.02 96400-G9100\x18\x07\x06\x17\x12    ': ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+  # 2019 SANTA FE
+  b"TM__ SCC F-CUP      1.00 1.00 99110-S1210\x19\x01%\x168    ": ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+  b"TM__ SCC F-CUP      1.00 1.02 99110-S2000\x18\x07\x08\x18W    ": ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+  # 2021 K5 HEV
+  b"DLhe SCC FHCUP      1.00 1.02 99110-L7000 \x01 \x102    ": ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+  # 2022 Niro EV
+  b"DEev SCC F-CUP      1.00 1.00 99110-Q4600\x01\x42      ": ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+  b"DEev SCC F-CUP      1.00 1.00 99110-Q4600 \x07\x03\t%      ": ConfigValues(
+    default_config=b"\x00\x00\x00\x01\x00\x00",
+    tracks_enabled=b"\x00\x00\x00\x01\x00\x01"),
+}
+
+if __name__ == "__main__":
+  parser = argparse.ArgumentParser(description='configure radar to output points (or reset to default)')
+  parser.add_argument('--default', action="store_true", default=False, help='reset to default configuration (default: false)')
+  parser.add_argument('--debug', action="store_true", default=False, help='enable debug output (default: false)')
+  parser.add_argument('--bus', type=int, default=0, help='can bus to use (default: 0)')
+  args = parser.parse_args()
+
+  if args.debug:
+    carlog.setLevel('DEBUG')
+
+  try:
+    check_output(["pidof", "pandad"])
+    print("pandad is running, please kill openpilot before running this script! (aborted)")
+    sys.exit(1)
+  except CalledProcessError as e:
+    if e.returncode != 1: # 1 == no process found (pandad not running)
+      raise e
+
+  confirm = input("power on the vehicle keeping the engine off (press start button twice) then type OK to continue: ").upper().strip()
+  if confirm != "OK":
+    print("\nyou didn't type 'OK! (aborted)")
+    sys.exit(0)
+
+  panda = Panda()
+  panda.set_safety_mode(CarParams.SafetyModel.elm327)
+  uds_client = UdsClient(panda, 0x7D0, bus=args.bus)
+
+  print("\n[START DIAGNOSTIC SESSION]")
+  session_type : SESSION_TYPE = 0x07
+  uds_client.diagnostic_session_control(session_type)
+
+  print("[HARDWARE/SOFTWARE VERSION]")
+  fw_version_data_id : DATA_IDENTIFIER_TYPE = 0xf100
+  fw_version = uds_client.read_data_by_identifier(fw_version_data_id)
+  print(fw_version)
+  if fw_version not in SUPPORTED_FW_VERSIONS.keys():
+    print("radar not supported! (aborted)")
+    sys.exit(1)
+
+  print("[GET CONFIGURATION]")
+  config_data_id : DATA_IDENTIFIER_TYPE = 0x0142
+  current_config = uds_client.read_data_by_identifier(config_data_id)
+  config_values = SUPPORTED_FW_VERSIONS[fw_version]
+  new_config = config_values.default_config if args.default else config_values.tracks_enabled
+  print(f"current config: 0x{current_config.hex()}")
+  if current_config != new_config:
+    print("[CHANGE CONFIGURATION]")
+    print(f"new config:     0x{new_config.hex()}")
+    uds_client.write_data_by_identifier(config_data_id, new_config)
+    if not args.default and current_config != SUPPORTED_FW_VERSIONS[fw_version].default_config:
+      print("\ncurrent config does not match expected default! (aborted)")
+      sys.exit(1)
+
+    print("[DONE]")
+    print("\nrestart your vehicle and ensure there are no faults")
+    if not args.default:
+      print("you can run this script again with --default to go back to the original (factory) settings")
+  else:
+    print("[DONE]")
+    print("\ncurrent config is already the desired configuration")
+    sys.exit(0)

tools/scripts/car/max_lat_accel.py

@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+import argparse
+import numpy as np
+import matplotlib.pyplot as plt
+from functools import partial
+from tqdm import tqdm
+from typing import NamedTuple
+from openpilot.tools.lib.logreader import LogReader
+from openpilot.selfdrive.locationd.models.pose_kf import EARTH_G
+
+RLOG_MIN_LAT_ACTIVE = 50
+RLOG_MIN_STEERING_UNPRESSED = 50
+RLOG_MIN_REQUESTING_MAX = 25  # sample many times after reaching max torque
+
+QLOG_DECIMATION = 10
+
+
+class Event(NamedTuple):
+  lateral_accel: float
+  speed: float
+  roll: float
+  timestamp: float  # relative to start of route (s)
+
+
+def find_events(lr: LogReader, extrapolate: bool = False, qlog: bool = False) -> list[Event]:
+  min_lat_active = RLOG_MIN_LAT_ACTIVE // QLOG_DECIMATION if qlog else RLOG_MIN_LAT_ACTIVE
+  min_steering_unpressed = RLOG_MIN_STEERING_UNPRESSED // QLOG_DECIMATION if qlog else RLOG_MIN_STEERING_UNPRESSED
+  min_requesting_max = RLOG_MIN_REQUESTING_MAX // QLOG_DECIMATION if qlog else RLOG_MIN_REQUESTING_MAX
+
+  # if we test with driver torque safety, max torque can be slightly noisy
+  steer_threshold = 0.7 if extrapolate else 0.95
+
+  events = []
+
+  # state tracking
+  steering_unpressed = 0  # frames
+  requesting_max = 0  # frames
+  lat_active = 0  # frames
+
+  # current state
+  curvature = 0
+  v_ego = 0
+  roll = 0
+  out_torque = 0
+
+  start_ts = 0
+  for msg in lr:
+    if msg.which() == 'carControl':
+      if start_ts == 0:
+        start_ts = msg.logMonoTime
+
+      lat_active = lat_active + 1 if msg.carControl.latActive else 0
+
+    elif msg.which() == 'carOutput':
+      out_torque = msg.carOutput.actuatorsOutput.torque
+      requesting_max = requesting_max + 1 if abs(out_torque) > steer_threshold else 0
+
+    elif msg.which() == 'carState':
+      steering_unpressed = steering_unpressed + 1 if not msg.carState.steeringPressed else 0
+      v_ego = msg.carState.vEgo
+
+    elif msg.which() == 'controlsState':
+      curvature = msg.controlsState.curvature
+
+    elif msg.which() == 'vehicleParameters':
+      roll = msg.vehicleParameters.roll
+
+    if lat_active > min_lat_active and steering_unpressed > min_steering_unpressed and requesting_max > min_requesting_max:
+      # TODO: record max lat accel at the end of the event, need to use the past lat accel as overriding can happen before we detect it
+      requesting_max = 0
+
+      factor = 1 / abs(out_torque)
+      current_lateral_accel = (curvature * v_ego ** 2 * factor) - roll * EARTH_G
+      events.append(Event(current_lateral_accel, v_ego, roll, round((msg.logMonoTime - start_ts) * 1e-9, 2)))
+      print(events[-1])
+
+  return events
+
+
+if __name__ == '__main__':
+  parser = argparse.ArgumentParser(description="Find max lateral acceleration events",
+                                   formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+
+  parser.add_argument("route", nargs='+')
+  parser.add_argument("-e", "--extrapolate", action="store_true", help="Extrapolates max lateral acceleration events linearly. " +
+                                                                       "This option can be far less accurate.")
+  args = parser.parse_args()
+
+  events = []
+  for route in tqdm(args.route):
+    try:
+      lr = LogReader(route, sort_by_time=True)
+    except Exception:
+      print(f'Skipping {route}')
+      continue
+
+    qlog = route.endswith('/q')
+    if qlog:
+      print('WARNING: Treating route as qlog!')
+
+    print('Finding events...')
+    events += lr.run_across_segments(8, partial(find_events, extrapolate=args.extrapolate, qlog=qlog), disable_tqdm=True)
+
+  print()
+  print(f'Found {len(events)} events')
+
+  perc_left_accel = -np.percentile([-ev.lateral_accel for ev in events if ev.lateral_accel < 0] or [0], 90)
+  perc_right_accel = np.percentile([ev.lateral_accel for ev in events if ev.lateral_accel > 0] or [0], 90)
+
+  CP = lr.first('carParams')
+
+  plt.ion()
+  plt.clf()
+  plt.suptitle(f'{CP.carFingerprint} - Max lateral acceleration events')
+  plt.title(', '.join(args.route))
+  plt.scatter([ev.speed for ev in events], [ev.lateral_accel for ev in events], label='max lateral accel events')
+
+  plt.plot([0, 35], [3, 3], c='r', label='ISO 11270 - 3 m/s^2')
+  plt.plot([0, 35], [-3, -3], c='r')
+
+  plt.plot([0, 35], [perc_left_accel, perc_left_accel], c='g', linestyle='--', label='90th percentile left lateral accel')
+  plt.plot([0, 35], [perc_right_accel, perc_right_accel], c='#ff7f0e', linestyle='--', label='90th percentile right lateral accel')
+  plt.text(0.4, float(perc_left_accel + 0.4), f'{perc_left_accel:.2f} m/s^2', verticalalignment='center', fontsize=12)
+  plt.text(0.4, float(perc_right_accel - 0.4), f'{perc_right_accel:.2f} m/s^2', verticalalignment='center', fontsize=12)
+
+  plt.xlim(0, 35)
+  plt.ylim(-5, 5)
+  plt.xlabel('speed (m/s)')
+  plt.ylabel('lateral acceleration (m/s^2)')
+  plt.legend()
+  plt.show(block=True)

tools/scripts/car/measure_torque_time_to_max.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+
+import os
+import argparse
+import struct
+from collections import deque
+from statistics import mean
+
+from openpilot.cereal import log
+import openpilot.cereal.messaging as messaging
+
+if __name__ == "__main__":
+
+  parser = argparse.ArgumentParser(description='Sniff a communication socket')
+  parser.add_argument('--addr', default='127.0.0.1')
+  args = parser.parse_args()
+
+  if args.addr != "127.0.0.1":
+    os.environ["ZMQ"] = "1"
+    messaging.reset_context()
+
+  poller = messaging.Poller()
+  messaging.sub_sock('can', poller, addr=args.addr)
+
+  active = 0
+  start_t = 0
+  start_v = 0
+  max_v = 0
+  max_t = 0
+  window = deque(maxlen=10)
+  avg = 0
+  while 1:
+    polld = poller.poll(1000)
+    for sock in polld:
+      msg = sock.receive()
+      with log.Event.from_bytes(msg) as log_evt:
+        evt = log_evt
+
+      for item in evt.can:
+        if item.address == 0xe4 and item.src == 128:
+          torque_req = struct.unpack('!h', item.dat[0:2])[0]
+          # print(torque_req)
+          active = abs(torque_req) > 0
+          if abs(torque_req) < 100:
+            if max_v > 5:
+              print(f'{start_v} -> {max_v} = {round(max_v - start_v, 2)} over {round(max_t - start_t, 2)}s')
+            start_t = evt.logMonoTime / 1e9
+            start_v = avg
+            max_t = 0
+            max_v = 0
+        if item.address == 0x1ab and item.src == 0:
+          motor_torque = ((item.dat[0] & 0x3) << 8) + item.dat[1]
+          window.append(motor_torque)
+          avg = mean(window)
+          #print(f'{evt.logMonoTime}: {avg}')
+          if active and avg > max_v + 0.5:
+            max_v = avg
+            max_t = evt.logMonoTime / 1e9

tools/scripts/car/read_dtc_status.py

@@ -0,0 +1,39 @@
+#!/usr/bin/env python3
+import sys
+import argparse
+from subprocess import check_output, CalledProcessError
+from opendbc.car.carlog import carlog
+from opendbc.car.uds import UdsClient, SESSION_TYPE, DTC_REPORT_TYPE, DTC_STATUS_MASK_TYPE, get_dtc_num_as_str, get_dtc_status_names
+from opendbc.car.structs import CarParams
+from panda import Panda
+
+parser = argparse.ArgumentParser(description="read DTC status")
+parser.add_argument("addr", type=lambda x: int(x,0))
+parser.add_argument("--bus", type=int, default=0)
+parser.add_argument('--debug', action='store_true')
+args = parser.parse_args()
+
+if args.debug:
+  carlog.setLevel('DEBUG')
+
+try:
+  check_output(["pidof", "pandad"])
+  print("pandad is running, please kill openpilot before running this script! (aborted)")
+  sys.exit(1)
+except CalledProcessError as e:
+  if e.returncode != 1: # 1 == no process found (pandad not running)
+    raise e
+
+panda = Panda()
+panda.set_safety_mode(CarParams.SafetyModel.elm327)
+uds_client = UdsClient(panda, args.addr, bus=args.bus)
+print("extended diagnostic session ...")
+uds_client.diagnostic_session_control(SESSION_TYPE.EXTENDED_DIAGNOSTIC)
+print("read diagnostic codes ...")
+data = uds_client.read_dtc_information(DTC_REPORT_TYPE.DTC_BY_STATUS_MASK, DTC_STATUS_MASK_TYPE.ALL)
+print("status availability:", " ".join(get_dtc_status_names(data[0])))
+print("DTC status:")
+for i in range(1, len(data), 4):
+  dtc_num = get_dtc_num_as_str(data[i:i+3])
+  dtc_status = " ".join(get_dtc_status_names(data[i+3]))
+  print(dtc_num, dtc_status)

tools/scripts/car/toyota_eps_factor.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+import sys
+import numpy as np
+import matplotlib.pyplot as plt
+from sklearn import linear_model
+from opendbc.car.toyota.values import STEER_THRESHOLD
+
+from openpilot.tools.lib.logreader import LogReader
+
+MIN_SAMPLES = 30 * 100
+
+
+def to_signed(n, bits):
+  if n >= (1 << max((bits - 1), 0)):
+    n = n - (1 << max(bits, 0))
+  return n
+
+
+def get_eps_factor(lr, plot=False):
+  engaged = False
+  steering_pressed = False
+  torque_cmd, eps_torque = None, None
+  cmds, eps = [], []
+
+  for msg in lr:
+    if msg.which() != 'can':
+      continue
+
+    for m in msg.can:
+      if m.address == 0x2e4 and m.src == 128:
+        engaged = bool(m.dat[0] & 1)
+        torque_cmd = to_signed((m.dat[1] << 8) | m.dat[2], 16)
+      elif m.address == 0x260 and m.src == 0:
+        eps_torque = to_signed((m.dat[5] << 8) | m.dat[6], 16)
+        steering_pressed = abs(to_signed((m.dat[1] << 8) | m.dat[2], 16)) > STEER_THRESHOLD
+
+    if engaged and torque_cmd is not None and eps_torque is not None and not steering_pressed:
+      cmds.append(torque_cmd)
+      eps.append(eps_torque)
+    else:
+      if len(cmds) > MIN_SAMPLES:
+        break
+      cmds, eps = [], []
+
+  if len(cmds) < MIN_SAMPLES:
+    raise Exception("too few samples found in route")
+
+  lm = linear_model.LinearRegression(fit_intercept=False)
+  lm.fit(np.array(cmds).reshape(-1, 1), eps)
+  scale_factor = 1. / lm.coef_[0]
+
+  if plot:
+    plt.plot(np.array(eps) * scale_factor)
+    plt.plot(cmds)
+    plt.show()
+  return scale_factor
+
+
+if __name__ == "__main__":
+  lr = LogReader(sys.argv[1])
+  n = get_eps_factor(lr, plot="--plot" in sys.argv)
+  print("EPS torque factor: ", n)

tools/scripts/car/vin.py

@@ -0,0 +1,26 @@
+#!/usr/bin/env python3
+import argparse
+import time
+import openpilot.cereal.messaging as messaging
+from opendbc.car.carlog import carlog
+from opendbc.car.vin import get_vin
+from openpilot.selfdrive.car.card import can_comm_callbacks
+
+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()
+
+  if args.debug:
+    carlog.setLevel('DEBUG')
+
+  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)
+  print(f'RX: {hex(vin_rx_addr)}, BUS: {vin_rx_bus}, VIN: {vin}')

tools/scripts/car/vw_mqb_config.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+
+import argparse
+import struct
+from enum import IntEnum
+from opendbc.car.carlog import carlog
+from opendbc.car.uds import UdsClient, MessageTimeoutError, NegativeResponseError, SESSION_TYPE,\
+  DATA_IDENTIFIER_TYPE, ACCESS_TYPE
+from opendbc.car.structs import CarParams
+from panda import Panda
+from datetime import date
+
+# TODO: extend UDS library to allow custom/vendor-defined data identifiers without ignoring type checks
+class VOLKSWAGEN_DATA_IDENTIFIER_TYPE(IntEnum):
+  CODING = 0x0600
+
+# TODO: extend UDS library security_access() to take an access level offset per ISO 14229-1:2020 10.4 and remove this
+class ACCESS_TYPE_LEVEL_1(IntEnum):
+  REQUEST_SEED = ACCESS_TYPE.REQUEST_SEED + 2
+  SEND_KEY = ACCESS_TYPE.SEND_KEY + 2
+
+MQB_EPS_CAN_ADDR = 0x712
+RX_OFFSET = 0x6a
+
+if __name__ == "__main__":
+  desc_text =   "Shows Volkswagen EPS software and coding info, and enables or disables Heading Control Assist " + \
+                "(Lane Assist). Useful for enabling HCA on cars without factory Lane Assist that want to use " + \
+                "openpilot integrated at the CAN gateway (J533)."
+  epilog_text = "This tool is meant to run directly on a vehicle-installed comma three, with the " + \
+                "openpilot/tmux processes stopped. It should also work on a separate PC with a USB-attached comma " + \
+                "panda. Vehicle ignition must be on. Recommend engine not be running when making changes. Must " + \
+                "turn ignition off and on again for any changes to take effect."
+  parser = argparse.ArgumentParser(description=desc_text, epilog=epilog_text)
+  parser.add_argument("--debug", action="store_true", help="enable ISO-TP/UDS stack debugging output")
+  parser.add_argument("action", choices={"show", "enable", "disable"}, help="show or modify current EPS HCA config")
+  args = parser.parse_args()
+
+  if args.debug:
+    carlog.setLevel('DEBUG')
+
+  panda = Panda()
+  panda.set_safety_mode(CarParams.SafetyModel.elm327)
+  uds_client = UdsClient(panda, MQB_EPS_CAN_ADDR, MQB_EPS_CAN_ADDR + RX_OFFSET, 1, timeout=0.2)
+
+  try:
+    uds_client.diagnostic_session_control(SESSION_TYPE.EXTENDED_DIAGNOSTIC)
+  except MessageTimeoutError:
+    print("Timeout opening session with EPS")
+    quit()
+
+  odx_file, current_coding = None, None
+  try:
+    hw_pn = uds_client.read_data_by_identifier(DATA_IDENTIFIER_TYPE.VEHICLE_MANUFACTURER_ECU_HARDWARE_NUMBER).decode("utf-8")
+    sw_pn = uds_client.read_data_by_identifier(DATA_IDENTIFIER_TYPE.VEHICLE_MANUFACTURER_SPARE_PART_NUMBER).decode("utf-8")
+    sw_ver = uds_client.read_data_by_identifier(DATA_IDENTIFIER_TYPE.VEHICLE_MANUFACTURER_ECU_SOFTWARE_VERSION_NUMBER).decode("utf-8")
+    component = uds_client.read_data_by_identifier(DATA_IDENTIFIER_TYPE.SYSTEM_NAME_OR_ENGINE_TYPE).decode("utf-8")
+    odx_file = uds_client.read_data_by_identifier(DATA_IDENTIFIER_TYPE.ODX_FILE).decode("utf-8").rstrip('\x00')
+    current_coding = uds_client.read_data_by_identifier(VOLKSWAGEN_DATA_IDENTIFIER_TYPE.CODING)
+    coding_text = current_coding.hex()
+
+    print("\nEPS diagnostic data\n")
+    print(f"   Part No HW:   {hw_pn}")
+    print(f"   Part No SW:   {sw_pn}")
+    print(f"   SW Version:   {sw_ver}")
+    print(f"   Component:    {component}")
+    print(f"   Coding:       {coding_text}")
+    print(f"   ASAM Dataset: {odx_file}")
+  except NegativeResponseError:
+    print("Error fetching data from EPS")
+    quit()
+  except MessageTimeoutError:
+    print("Timeout fetching data from EPS")
+    quit()
+
+  coding_variant, current_coding_array, coding_byte, coding_bit = None, None, 0, 0
+  coding_length = len(current_coding)
+
+  # EPS_MQB_ZFLS
+  if odx_file in ("EV_SteerAssisMQB", "EV_SteerAssisMNB"):
+    coding_variant = "ZFLS"
+    coding_byte = 0
+    coding_bit = 4
+
+  # MQB_PP_APA, MQB_VWBS_GEN2
+  elif odx_file in ("EV_SteerAssisVWBSMQBA", "EV_SteerAssisVWBSMQBGen2"):
+    coding_variant = "APA"
+    coding_byte = 3
+    coding_bit = 0
+
+  else:
+    print("Configuration changes not yet supported on this EPS!")
+    quit()
+
+  current_coding_array = struct.unpack(f"!{coding_length}B", current_coding)
+  hca_enabled = (current_coding_array[coding_byte] & (1 << coding_bit) != 0)
+  hca_text = ("DISABLED", "ENABLED")[hca_enabled]
+  print(f"   Lane Assist:  {hca_text}")
+
+  try:
+    params = uds_client.read_data_by_identifier(DATA_IDENTIFIER_TYPE.APPLICATION_DATA_IDENTIFICATION).decode("utf-8")
+    param_version_system_params = params[1:3]
+    param_vehicle_type = params[3:5]
+    param_index_char_curve = params[5:7]
+    param_version_char_values = params[7:9]
+    param_version_memory_map = params[9:11]
+    print("\nEPS parameterization (per-vehicle calibration) data\n")
+    print(f"   Version of system parameters:     {param_version_system_params}")
+    print(f"   Vehicle type:                     {param_vehicle_type}")
+    print(f"   Index of characteristic curve:    {param_index_char_curve}")
+    print(f"   Version of characteristic values: {param_version_char_values}")
+    print(f"   Version of memory map:            {param_version_memory_map}")
+  except (NegativeResponseError, MessageTimeoutError):
+    print("Error fetching parameterization data from EPS!")
+    quit()
+
+  if args.action in ["enable", "disable"]:
+    print("\nAttempting configuration update")
+
+    assert(coding_variant in ("ZFLS", "APA"))
+    # ZFLS EPS config coding length can be anywhere from 1 to 4 bytes, but the
+    # bit we care about is always in the same place in the first byte
+    if args.action == "enable":
+      new_byte = current_coding_array[coding_byte] | (1 << coding_bit)
+    else:
+      new_byte = current_coding_array[coding_byte] & ~(1 << coding_bit)
+    new_coding = current_coding[0:coding_byte] + new_byte.to_bytes(1, "little") + current_coding[coding_byte+1:]
+
+    try:
+      seed = uds_client.security_access(ACCESS_TYPE_LEVEL_1.REQUEST_SEED)
+      key = struct.unpack("!I", seed)[0] + 28183  # yeah, it's like that
+      uds_client.security_access(ACCESS_TYPE_LEVEL_1.SEND_KEY, struct.pack("!I", key))
+    except (NegativeResponseError, MessageTimeoutError):
+      print("Security access failed!")
+      print("Open the hood and retry (disables the \"diagnostic firewall\" on newer vehicles)")
+      quit()
+
+    try:
+      # Programming date and tester number must be written before making
+      # a change, or write to CODING will fail with request sequence error
+      # Encoding on tester is unclear, it contains the workshop code in the
+      # last two bytes, but not the VZ/importer or tester serial number
+      # Can't seem to read it back, but we can read the calibration tester,
+      # so fib a little and say that same tester did the programming
+      current_date = date.today()
+      formatted_date = current_date.strftime('%y-%m-%d')
+      year, month, day = (int(part) for part in formatted_date.split('-'))
+      prog_date = bytes([year, month, day])
+      uds_client.write_data_by_identifier(DATA_IDENTIFIER_TYPE.PROGRAMMING_DATE, prog_date)
+      tester_num = uds_client.read_data_by_identifier(DATA_IDENTIFIER_TYPE.CALIBRATION_REPAIR_SHOP_CODE_OR_CALIBRATION_EQUIPMENT_SERIAL_NUMBER)
+      uds_client.write_data_by_identifier(DATA_IDENTIFIER_TYPE.REPAIR_SHOP_CODE_OR_TESTER_SERIAL_NUMBER, tester_num)
+      uds_client.write_data_by_identifier(VOLKSWAGEN_DATA_IDENTIFIER_TYPE.CODING, new_coding)
+    except (NegativeResponseError, MessageTimeoutError):
+      print("Writing new configuration failed!")
+      print("Make sure the comma processes are stopped: tmux kill-session -t comma")
+      quit()
+
+    try:
+      # Read back result just to make 100% sure everything worked
+      current_coding_text = uds_client.read_data_by_identifier(VOLKSWAGEN_DATA_IDENTIFIER_TYPE.CODING).hex()
+      print(f"   New coding:   {current_coding_text}")
+    except (NegativeResponseError, MessageTimeoutError):
+      print("Reading back updated coding failed!")
+      quit()
+    print("EPS configuration successfully updated")