diff --git a/selfdrive/car/ecu_addrs.py b/selfdrive/car/ecu_addrs.py index 756cd7f963..0fd4bbd0be 100755 --- a/selfdrive/car/ecu_addrs.py +++ b/selfdrive/car/ecu_addrs.py @@ -10,7 +10,7 @@ from openpilot.selfdrive.pandad import can_list_to_can_capnp from openpilot.common.swaglog import cloudlog -def make_tester_present_msg(addr, bus, subaddr=None): +def _make_tester_present_msg(addr, bus, subaddr=None): dat = [0x02, SERVICE_TYPE.TESTER_PRESENT, 0x0] if subaddr is not None: dat.insert(0, subaddr) @@ -19,7 +19,7 @@ def make_tester_present_msg(addr, bus, subaddr=None): return make_can_msg(addr, bytes(dat), bus) -def is_tester_present_response(msg: capnp.lib.capnp._DynamicStructReader, subaddr: int = None) -> bool: +def _is_tester_present_response(msg: capnp.lib.capnp._DynamicStructReader, subaddr: int = None) -> bool: # ISO-TP messages are always padded to 8 bytes # tester present response is always a single frame dat_offset = 1 if subaddr is not None else 0 @@ -33,7 +33,7 @@ def is_tester_present_response(msg: capnp.lib.capnp._DynamicStructReader, subadd return False -def get_all_ecu_addrs(logcan: messaging.SubSocket, sendcan: messaging.PubSocket, bus: int, timeout: float = 1, debug: bool = True) -> set[EcuAddrBusType]: +def _get_all_ecu_addrs(logcan: messaging.SubSocket, sendcan: messaging.PubSocket, 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 @@ -44,7 +44,7 @@ def get_ecu_addrs(logcan: messaging.SubSocket, sendcan: messaging.PubSocket, que responses: set[EcuAddrBusType], timeout: float = 1, debug: bool = False) -> set[EcuAddrBusType]: ecu_responses: set[EcuAddrBusType] = set() # set((addr, subaddr, bus),) try: - msgs = [make_tester_present_msg(addr, bus, subaddr) for addr, subaddr, bus in queries] + msgs = [_make_tester_present_msg(addr, bus, subaddr) for addr, subaddr, bus in queries] messaging.drain_sock_raw(logcan) sendcan.send(can_list_to_can_capnp(msgs, msgtype='sendcan')) @@ -58,7 +58,7 @@ def get_ecu_addrs(logcan: messaging.SubSocket, sendcan: messaging.PubSocket, que continue subaddr = None if (msg.address, None, msg.src) in responses else msg.dat[0] - if (msg.address, subaddr, msg.src) in responses and is_tester_present_response(msg, subaddr): + if (msg.address, subaddr, msg.src) in responses and _is_tester_present_response(msg, subaddr): if debug: print(f"CAN-RX: {hex(msg.address)} - 0x{bytes.hex(msg.dat)}") if (msg.address, subaddr, msg.src) in ecu_responses: @@ -94,7 +94,7 @@ if __name__ == "__main__": set_obd_multiplexing(params, not args.no_obd) print("Getting ECU addresses ...") - ecu_addrs = get_all_ecu_addrs(logcan, sendcan, args.bus, args.timeout, debug=args.debug) + ecu_addrs = _get_all_ecu_addrs(logcan, sendcan, args.bus, args.timeout, debug=args.debug) print() print("Found ECUs on rx addresses:")