diff --git a/bindings/python/CMakeLists.txt b/bindings/python/CMakeLists.txt index 56778228..2a7102bb 100644 --- a/bindings/python/CMakeLists.txt +++ b/bindings/python/CMakeLists.txt @@ -26,6 +26,7 @@ pybind11_add_module(icsneopy icsneopy/communication/message/message.cpp icsneopy/communication/message/canmessage.cpp icsneopy/communication/message/canerrormessage.cpp + icsneopy/communication/message/iso15765message.cpp icsneopy/communication/message/ethernetmessage.cpp icsneopy/communication/message/linmessage.cpp icsneopy/communication/message/tc10statusmessage.cpp diff --git a/bindings/python/icsneopy/communication/message/iso15765message.cpp b/bindings/python/icsneopy/communication/message/iso15765message.cpp new file mode 100644 index 00000000..0c57bc0c --- /dev/null +++ b/bindings/python/icsneopy/communication/message/iso15765message.cpp @@ -0,0 +1,48 @@ +#include +#include + +#include "icsneo/communication/message/iso15765message.h" + +namespace icsneo { + +void init_iso15765message(pybind11::module_& m) { + auto args = pybind11::classh(m, "Iso15765MessageArgs"); + + pybind11::classh(args, "ArbId") + .def(pybind11::init(), pybind11::arg("id") = 0, pybind11::arg("is_29bit") = false) + .def_readwrite("id", &Iso15765MessageArgs::ArbId::Id) + .def_readwrite("is_29bit", &Iso15765MessageArgs::ArbId::Is29Bit); + + args.def(pybind11::init<>()) + .def_property("tx_index", &Iso15765MessageArgs::getTxIndex, &Iso15765MessageArgs::setTxIndex) + .def_property("arb_id", + [](Iso15765MessageArgs& self) -> Iso15765MessageArgs::ArbId& { + return const_cast(self.getArbId()); + }, + &Iso15765MessageArgs::setArbId, + pybind11::return_value_policy::reference_internal) + .def_property("flow_control_arb_id", + [](Iso15765MessageArgs& self) -> Iso15765MessageArgs::ArbId& { + return const_cast(self.getFlowControlArbId()); + }, + &Iso15765MessageArgs::setFlowControlArbId, + pybind11::return_value_policy::reference_internal) + .def_property("flow_control_arb_id_mask", &Iso15765MessageArgs::getFlowControlArbIdMask, &Iso15765MessageArgs::setFlowControlArbIdMask) + .def_property("is_can_fd", &Iso15765MessageArgs::getIsCanFd, &Iso15765MessageArgs::setIsCanFd) + .def_property("is_brs_enabled", &Iso15765MessageArgs::getIsBrsEnabled, &Iso15765MessageArgs::setIsBrsEnabled) + .def_property("tx_dl", &Iso15765MessageArgs::getTxDl, &Iso15765MessageArgs::setTxDl) + .def_property("st_min", &Iso15765MessageArgs::getStMin, &Iso15765MessageArgs::setStMin) + .def_property("block_size", &Iso15765MessageArgs::getBlockSize, &Iso15765MessageArgs::setBlockSize) + .def_property("extended_address", &Iso15765MessageArgs::getExtendedAddress, &Iso15765MessageArgs::setExtendedAddress) + .def_property("flow_control_extended_address", &Iso15765MessageArgs::getFlowControlExtendedAddress, &Iso15765MessageArgs::setFlowControlExtendedAddress) + .def_property("padding_value", &Iso15765MessageArgs::getPaddingValue, &Iso15765MessageArgs::setPaddingValue) + .def_property("fs_timeout", &Iso15765MessageArgs::getFsTimeout, &Iso15765MessageArgs::setFsTimeout) + .def_property("fs_wait_timeout", &Iso15765MessageArgs::getFsWaitTimeout, &Iso15765MessageArgs::setFsWaitTimeout) + .def_property("cf_timeout", &Iso15765MessageArgs::getCfTimeout, &Iso15765MessageArgs::setCfTimeout) + .def_property("data", + &Iso15765MessageArgs::getData, + [](Iso15765MessageArgs& self, std::vector data) { self.setData(std::move(data)); }) + .def_property("is_flow_control_enabled", &Iso15765MessageArgs::getIsFlowControlEnabled, &Iso15765MessageArgs::setIsFlowControlEnabled); +} + +} // namespace icsneo diff --git a/bindings/python/icsneopy/device/device.cpp b/bindings/python/icsneopy/device/device.cpp index 3176784c..07244012 100644 --- a/bindings/python/icsneopy/device/device.cpp +++ b/bindings/python/icsneopy/device/device.cpp @@ -45,6 +45,10 @@ void init_device(pybind11::module_& m) { .def("is_online_supported", &Device::isOnlineSupported) .def("is_online", &Device::isOnline) .def("is_open", &Device::isOpen) + .def("iso15765_enable", &Device::iso15765Enable, pybind11::arg("network"), pybind11::call_guard()) + .def("iso15765_disable_all", &Device::iso15765DisableAll, pybind11::call_guard()) + .def("iso15765_transmit_message", &Device::iso15765TransmitMessage, pybind11::arg("network"), pybind11::arg("message"), pybind11::arg("timeout"), pybind11::call_guard()) + .def("iso15765_setup_rx_flow_control", &Device::iso15765SetupRxFlowControl, pybind11::arg("network"), pybind11::arg("message"), pybind11::call_guard()) .def("open", [](Device& device) { return device.open(); }, pybind11::call_guard()) .def("prepare_script_load", &Device::prepareScriptLoad, pybind11::call_guard()) .def("remove_message_callback", &Device::removeMessageCallback, pybind11::call_guard()) diff --git a/bindings/python/icsneopy/icsneocpp.cpp b/bindings/python/icsneopy/icsneocpp.cpp index 24eddfe2..4a513431 100644 --- a/bindings/python/icsneopy/icsneocpp.cpp +++ b/bindings/python/icsneopy/icsneocpp.cpp @@ -15,6 +15,7 @@ void init_devicetype(pybind11::module_&); void init_message(pybind11::module_&); void init_canmessage(pybind11::module_&); void init_canerrormessage(pybind11::module_&); +void init_iso15765message(pybind11::module_&); void init_ethernetmessage(pybind11::module_&); void init_linmessage(pybind11::module_&); void init_tc10statusmessage(pybind11::module_&); @@ -60,6 +61,7 @@ PYBIND11_MODULE(icsneopy, m) { init_message(m); init_canmessage(m); init_canerrormessage(m); + init_iso15765message(m); init_ethernetmessage(m); init_linmessage(m); init_tc10statusmessage(m); diff --git a/docs/icsneopy/examples.rst b/docs/icsneopy/examples.rst index 0c3034d2..595f9154 100644 --- a/docs/icsneopy/examples.rst +++ b/docs/icsneopy/examples.rst @@ -107,3 +107,14 @@ Analog Output Control .. literalinclude:: ../../examples/python/analog_out/analog_out_basic.py :language: python + +ISO-TP CAN1 Loopback +==================== + +Two devices, CAN1 wired together. ``--raw`` checks the cable, ``--sf`` sends one +firmware single frame, ``--both`` sends a segmented payload with firmware flow control. + +:download:`Download example <../../examples/python/iso15765/iso15765_loopback.py>` + +.. literalinclude:: ../../examples/python/iso15765/iso15765_loopback.py + :language: python diff --git a/examples/python/iso15765/iso15765_loopback.py b/examples/python/iso15765/iso15765_loopback.py new file mode 100644 index 00000000..f9ec4384 --- /dev/null +++ b/examples/python/iso15765/iso15765_loopback.py @@ -0,0 +1,264 @@ +""" +Two-device ISO-TP loopback on CAN1. + +Firmware segments the payload. This script sends it once and prints CAN1 +frames with the ISO-TP PCI nibble decoded (SF/FF/CF/FC). + +Wire CAN1 of two devices together (CANH/CANL, terminated). + + python iso15765_loopback.py --raw + python iso15765_loopback.py --sf + python iso15765_loopback.py --both +""" + +import argparse +import threading +import time +from datetime import timedelta + +import icsneopy + + +CAN1 = icsneopy.Network.NetID.DWCAN_01 + + +def pci_describe(data): + if not data: + return "" + pci_type = data[0] >> 4 + pci_low = data[0] & 0x0F + if pci_type == 0x0: + if pci_low > 7 or len(data) < 1 + pci_low: + return "" + return f"SF len={pci_low}" + if pci_type == 0x1: + if len(data) < 8: + return "" + length = (pci_low << 8) | data[1] + if length < 8: + return "" + return f"FF len={length}" + if pci_type == 0x2: + if len(data) < 2: + return "" + return f"CF SN={pci_low}" + if pci_type == 0x3: + if len(data) < 3 or pci_low > 2: + return "" + fs_name = ("CTS", "WAIT", "OVFLW")[pci_low] + return f"FC FS={fs_name} BS={data[1]} STmin=0x{data[2]:02X}" + return "" + + +def print_api_result(what, ok): + line = f" {what}: {'OK' if ok else 'FAIL'}" + if not ok: + err = icsneopy.get_last_error() + if err.get_type() != icsneopy.APIEvent.Type.NoErrorFound: + line += f" [{err.describe()}]" + print(line) + + +def is_extended(arb_id): + return arb_id > 0x7FF + + +def make_args(tx_id, fc_id, payload, rx): + msg = icsneopy.Iso15765MessageArgs() + msg.tx_index = 0 + msg.arb_id = icsneopy.Iso15765MessageArgs.ArbId(tx_id, is_extended(tx_id)) + msg.flow_control_arb_id = icsneopy.Iso15765MessageArgs.ArbId(fc_id, is_extended(fc_id)) + msg.flow_control_arb_id_mask = 0x1FFFFFFF if is_extended(tx_id) else 0x7FF + msg.is_can_fd = False + msg.is_brs_enabled = False + msg.tx_dl = 8 + msg.padding_value = 0x00 + msg.fs_timeout = 75 + msg.fs_wait_timeout = 150 + msg.cf_timeout = 150 + msg.is_flow_control_enabled = rx + if payload: + msg.data = payload + return msg + + +def transmit_can(device, arb_id, data): + frame = icsneopy.CANMessage() + frame.network = icsneopy.Network(CAN1) + frame.arbid = arb_id + # RawMessage.data is vector: a list of ints, not bytes. + frame.data = list(data) + frame.isExtended = is_extended(arb_id) + frame.isCANFD = False + ok = device.transmit(frame) + if not ok: + print_api_result("transmit CAN", ok) + return ok + + +class Sniffer: + def __init__(self, tag): + self.tag = tag + self.t0 = None + self.lock = threading.Lock() + + def on_message(self, message): + if not isinstance(message, icsneopy.CANMessage): + return + if message.network.get_type() != icsneopy.Network.Type.CAN: + return + with self.lock: + if self.t0 is None: + self.t0 = message.timestamp + rel_ms = 0.0 + if message.timestamp >= self.t0: + rel_ms = (message.timestamp - self.t0) / 1e6 + direction = "TX" if message.transmitted else "RX" + width = 8 if is_extended(message.arbid) else 3 + data = " ".join(f"{b:02X}" for b in message.data) + line = ( + f" [{self.tag} {direction}] {rel_ms:8.3f} ms " + f"{message.network} 0x{message.arbid:0{width}X} " + f"[{len(message.data)}] {data}" + ) + pci = pci_describe(message.data) + if pci: + line += f" {pci}" + print(line) + + +def open_and_online(device, baud, keep_baud): + print(f"Opening {device.describe()} ...") + if not device.open(): + print_api_result("open", False) + return False + if not keep_baud and device.settings is not None: + net = icsneopy.Network(CAN1) + if not device.settings.set_baudrate(net, baud): + print_api_result("set_baudrate", False) + elif not device.settings.apply(True): + print_api_result("settings.apply(temporary)", False) + if not device.go_online(): + print_api_result("go_online", False) + device.close() + return False + return True + + +def close_device(device): + if device is None: + return + device.iso15765_disable_all() + if device.is_online(): + device.go_offline() + if device.is_open(): + device.close() + + +def find_by_serial(devices, serial): + for device in devices: + if device.get_serial().lower() == serial.lower(): + return device + return None + + +def parse_args(): + parser = argparse.ArgumentParser( + description="Two Intrepid devices with CAN1 wired together (CANH/CANL, terminated)." + ) + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument("--raw", action="store_true", help="Raw CAN ping-pong (cable check)") + mode.add_argument("--sf", action="store_true", help="Firmware ISO-TP single-frame TX from tester") + mode.add_argument("--both", action="store_true", help="Firmware ISO-TP TX on tester, firmware flow-control on ECU") + parser.add_argument("--tester", default="", help="Tester (sender) serial. Optional if exactly two devices are found.") + parser.add_argument("--ecu", default="", help="ECU / peer serial. Optional if exactly two devices are found.") + parser.add_argument("--baud", type=int, default=500000, help="CAN1 baud, applied temporarily (default 500000)") + parser.add_argument("--keep-baud", action="store_true", help="Do not change device baud") + parser.add_argument("--tx-id", type=lambda s: int(s, 0), default=0x7E0, help="Tester ISO-TP CAN ID (default 0x7E0)") + parser.add_argument("--fc-id", type=lambda s: int(s, 0), default=0x7E8, help="Flow-control CAN ID (default 0x7E8)") + parser.add_argument("--bytes", type=int, default=None, help="ISO-TP payload size (sf default 5, both default 20)") + parser.add_argument("--listen-ms", type=int, default=2000, help="Time to listen after the action (default 2000)") + return parser.parse_args() + + +def main(): + args = parse_args() + print(f"libicsneo {icsneopy.get_version()}") + + devices = icsneopy.find_all_devices() + print(f"Found {len(devices)} device(s)") + for device in devices: + print(f" {device.describe()} serial={device.get_serial()}") + if len(devices) < 2: + raise SystemExit("Need two devices connected.") + + if not args.tester and not args.ecu: + if len(devices) != 2: + raise SystemExit("More than two devices found; pass --tester and --ecu serials.") + tester, ecu = devices[0], devices[1] + print("No serials given; using first as tester, second as ECU.") + else: + if not args.tester or not args.ecu: + raise SystemExit("Pass both --tester and --ecu, or neither.") + tester = find_by_serial(devices, args.tester) + ecu = find_by_serial(devices, args.ecu) + if tester is None or ecu is None: + raise SystemExit("Could not match tester/ECU serials.") + + payload_size = args.bytes if args.bytes is not None else (5 if args.sf else 20) + payload = [i & 0xFF for i in range(payload_size)] + + print(f"Tester: {tester.describe()}") + print(f"ECU: {ecu.describe()}") + if not args.raw: + print(f"CAN1 TX 0x{args.tx_id:X} FC 0x{args.fc_id:X} payload {payload_size} bytes") + print() + + network = icsneopy.Network(CAN1) + try: + if not open_and_online(tester, args.baud, args.keep_baud): + return 1 + if not open_and_online(ecu, args.baud, args.keep_baud): + return 1 + + for device, sniffer in ((tester, Sniffer("TESTER")), (ecu, Sniffer("ECU "))): + callback = icsneopy.MessageCallback(sniffer.on_message, icsneopy.MessageFilter(CAN1)) + device.add_message_callback(callback) + + if args.raw: + print("Mode --raw: tester 0x123, then ECU 0x456") + transmit_can(tester, 0x123, [0x11, 0x22, 0x33, 0x44]) + time.sleep(0.2) + transmit_can(ecu, 0x456, [0x55, 0x66, 0x77, 0x88]) + elif args.sf: + print("Mode --sf: firmware ISO-TP single frame from tester") + if len(payload) > 7: + print(" warning: payload > 7 bytes will be a First Frame (no flow control in this mode)") + ok = tester.iso15765_enable(network) + print_api_result("tester iso15765_enable", ok) + msg = make_args(args.tx_id, args.fc_id, payload, rx=False) + ok = tester.iso15765_transmit_message(network, msg, timedelta(0)) + print_api_result("tester iso15765_transmit_message", ok) + else: + print("Mode --both: firmware TX on tester, firmware flow-control on ECU") + ok = tester.iso15765_enable(network) + print_api_result("tester iso15765_enable", ok) + ok = ecu.iso15765_enable(network) + print_api_result("ECU iso15765_enable", ok) + msg = make_args(args.tx_id, args.fc_id, [], rx=True) + ok = ecu.iso15765_setup_rx_flow_control(network, msg) + print_api_result("ECU iso15765_setup_rx_flow_control", ok) + time.sleep(0.1) + msg = make_args(args.tx_id, args.fc_id, payload, rx=False) + ok = tester.iso15765_transmit_message(network, msg, timedelta(0)) + print_api_result("tester iso15765_transmit_message", ok) + + time.sleep(args.listen_ms / 1000.0) + finally: + close_device(tester) + close_device(ecu) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())