Examples

These are the Python programs from the repository’s python/examples/ directory. They run against real adapters and ECUs. Select a suitable adapter and ECU address for your setup before running them.

Print CAN frames

Discover an adapter and print received CAN frames.

"""Print frames from the first available adapter. Requires CAN hardware or vcan0."""

import asyncio

from automotive.adapters import CanAdapter


async def main() -> None:
    """Discover an adapter and print frames until interrupted."""
    adapter = CanAdapter()
    stream = adapter.recv()

    async for frame in stream:
        print(frame)


if __name__ == "__main__":
    asyncio.run(main())

Send a CAN frame

Discover an adapter and send one Classic CAN frame.

"""Send one Classic CAN frame through the first available adapter."""

import asyncio

from automotive.adapters import CanAdapter
from automotive.can import Frame


async def main() -> None:
    """Discover an adapter and send a frame."""
    adapter = CanAdapter()
    frame = Frame(id=0x123, data=b"\x01\x02\x03\x04")
    await adapter.send(frame)
    print("Sent", frame)


if __name__ == "__main__":
    asyncio.run(main())

Send ISO-TP messages

Follow the Rust example’s four explicit exchanges: two short requests, a 33-byte segmented request, and another DID request that can produce a long reply.

"""Exchange short and segmented PDUs with an ECU at 0x7A1/0x7A9."""

import asyncio

from automotive.adapters import CanAdapter
from automotive.transports.isotp import IsoTP, IsoTPConfig


async def main() -> None:
    """Send short and segmented requests, printing each response as hex."""
    adapter = CanAdapter()
    config = IsoTPConfig(tx_id=0x7A1)
    isotp = IsoTP(adapter, config)

    stream = isotp.recv()

    await isotp.send(b"\x3e\x00")
    response = await anext(stream)
    print("RX:", response.hex())

    await isotp.send(b"\x22\xf1\x81")
    response = await anext(stream)
    print("RX:", response.hex())

    long_request = b"\x10" + b"\x00" * 32
    await isotp.send(long_request)
    response = await anext(stream)
    print("RX:", response.hex())

    long_response = b"\x22\xf1\x81"
    await isotp.send(long_response)
    response = await anext(stream)
    print("RX:", response.hex())


if __name__ == "__main__":
    asyncio.run(main())

Query an ECU with UDS

Read diagnostic data through a persistent UDS client.

"""Read software identification from a diagnostic ECU at transmit ID 0x7A1."""

import asyncio

from automotive.adapters import CanAdapter
from automotive.protocols.uds import DataIdentifier, SessionType, UDSClient
from automotive.transports.isotp import IsoTP


async def main() -> None:
    """Use defaults, select a session, and read an ECU data identifier."""
    adapter = CanAdapter()
    isotp = IsoTP.from_id(adapter, 0x7A1)
    uds = UDSClient(isotp)
    await uds.tester_present()
    await uds.diagnostic_session_control(SessionType.EXTENDED_DIAGNOSTIC)

    did = DataIdentifier.APPLICATION_SOFTWARE_IDENTIFICATION
    resp = await uds.read_data_by_identifier(did)

    print(f"{did.name}: {resp!r}")


if __name__ == "__main__":
    asyncio.run(main())

Configure bitrate and sample points

Configure a PEAK PCAN-USB FD adapter with 500 kbit/s nominal and 2 Mbit/s data timing, both with an 80% sample point.

"""Configure a PEAK PCAN-USB FD adapter for 500 kbit/s CAN and 2 Mbit/s CAN FD."""

import asyncio

from automotive.adapters import PeakAdapter, PeakConfig
from automotive.can.bitrate import BitrateSettings


async def main() -> None:
    """Configure nominal and data-phase timing, then print received frames."""
    config = PeakConfig(
        bitrate=BitrateSettings(
            bitrate=500_000,
            sample_point=0.8,
            data_bitrate=2_000_000,
            data_sample_point=0.8,
        ),
    )
    print("Resolved timing:", config.resolved_bitrate)
    adapter = PeakAdapter(config)
    stream = adapter.recv()

    async for frame in stream:
        print(frame)


if __name__ == "__main__":
    asyncio.run(main())

Query firmware versions concurrently

Use independent ECU transports for concurrent diagnostic requests.

"""Query named DIDs across standard and extended ECU addresses."""

import asyncio

from automotive.adapters import CanAdapter
from automotive.errors import AutomotiveError
from automotive.protocols.uds import DataIdentifier, UDSClient, UDSConfig
from automotive.transports.isotp import IsoTP, IsoTPConfig


async def get_version(adapter: CanAdapter, identifier: int) -> None:
    """Query each named DID for one ECU, retaining its configured transport."""
    # Increased timeout for adapters not supporting real ACKs.
    config = IsoTPConfig(tx_id=identifier, timeout=1.0)
    isotp = IsoTP(adapter, config)
    uds = UDSClient(isotp, UDSConfig(p2_timeout=1.0))

    await uds.tester_present()

    for did in DataIdentifier:
        try:
            resp = await uds.read_data_by_identifier(did)
        except (AutomotiveError, OSError):
            continue
        print(f"{identifier:x} 0x{did.value:x} {did.name}: {resp!r}")


async def main() -> None:
    """Scan the Rust example's standard/extended address ranges concurrently."""
    adapter = CanAdapter()

    standard_ids = range(0x700, 0x7F8)
    extended_ids = (0x18DA0000 + (i << 8) + 0xF1 for i in range(0x100))
    ids = [*standard_ids, *extended_ids]

    await asyncio.gather(*(get_version(adapter, identifier) for identifier in ids), return_exceptions=True)


if __name__ == "__main__":
    asyncio.run(main())

Synchronous examples

These programs use the matching modules under automotive.sync, including automotive.sync.adapters, automotive.sync.transports.isotp, and automotive.sync.protocols.uds, with the same configuration and value types. They use ordinary constructors, blocking calls, and receive iterators. Both the async and sync examples let ownership release resources when they go out of scope. Their setup, requests, output, and flow follow the corresponding Rust programs.

Send a CAN frame

"""Send one Classic CAN frame through the first available adapter."""

from automotive.sync.adapters import CanAdapter
from automotive.sync.can import Frame


def main() -> None:
    """Discover an adapter and send a frame."""
    adapter = CanAdapter()
    frame = Frame(id=0x123, data=b"\x01\x02\x03\x04")
    adapter.send(frame)
    print("Sent", frame)


if __name__ == "__main__":
    main()

Print CAN frames

"""Print frames from the first available adapter. Requires CAN hardware or vcan0."""

from automotive.sync.adapters import CanAdapter


def main() -> None:
    """Discover an adapter and print frames until interrupted."""
    adapter = CanAdapter()
    stream = adapter.recv()

    for frame in stream:
        print(frame)


if __name__ == "__main__":
    main()

Send ISO-TP messages

"""Exchange short and segmented PDUs with an ECU at 0x7A1/0x7A9."""

from automotive.sync.adapters import CanAdapter
from automotive.sync.transports.isotp import IsoTP, IsoTPConfig


def main() -> None:
    """Send short and segmented requests, printing each response as hex."""
    adapter = CanAdapter()
    config = IsoTPConfig(tx_id=0x7A1)
    isotp = IsoTP(adapter, config)

    stream = isotp.recv()

    isotp.send(b"\x3e\x00")
    response = next(stream)
    print("RX:", response.hex())

    isotp.send(b"\x22\xf1\x81")
    response = next(stream)
    print("RX:", response.hex())

    long_request = b"\x10" + b"\x00" * 32
    isotp.send(long_request)
    response = next(stream)
    print("RX:", response.hex())

    long_response = b"\x22\xf1\x81"
    isotp.send(long_response)
    response = next(stream)
    print("RX:", response.hex())


if __name__ == "__main__":
    main()

Query an ECU with UDS

"""Read software identification from a diagnostic ECU at transmit ID 0x7A1."""

from automotive.sync.adapters import CanAdapter
from automotive.sync.protocols.uds import DataIdentifier, SessionType, UDSClient
from automotive.sync.transports.isotp import IsoTP


def main() -> None:
    """Use defaults, select a session, and read an ECU data identifier."""
    adapter = CanAdapter()
    isotp = IsoTP.from_id(adapter, 0x7A1)
    uds = UDSClient(isotp)
    uds.tester_present()
    uds.diagnostic_session_control(SessionType.EXTENDED_DIAGNOSTIC)

    did = DataIdentifier.APPLICATION_SOFTWARE_IDENTIFICATION
    resp = uds.read_data_by_identifier(did)

    print(f"{did.name}: {resp!r}")


if __name__ == "__main__":
    main()

Configure bitrate and sample points

Configure the same PEAK timing through the synchronous API.

"""Configure a PEAK PCAN-USB FD adapter for 500 kbit/s CAN and 2 Mbit/s CAN FD."""

from automotive.sync.adapters import PeakAdapter, PeakConfig
from automotive.sync.can.bitrate import BitrateSettings


def main() -> None:
    """Configure nominal and data-phase timing, then print received frames."""
    config = PeakConfig(
        bitrate=BitrateSettings(
            bitrate=500_000,
            sample_point=0.8,
            data_bitrate=2_000_000,
            data_sample_point=0.8,
        ),
    )
    print("Resolved timing:", config.resolved_bitrate)
    adapter = PeakAdapter(config)
    stream = adapter.recv()

    for frame in stream:
        print(frame)


if __name__ == "__main__":
    main()

Query firmware versions

Query the same DIDs and ECU address ranges as the Rust and async Python examples. The synchronous version visits ECUs sequentially; the async version queries them concurrently.

"""Query named DIDs across standard and extended ECU addresses."""

from automotive.sync.adapters import CanAdapter
from automotive.sync.errors import AutomotiveError
from automotive.sync.protocols.uds import DataIdentifier, UDSClient, UDSConfig
from automotive.sync.transports.isotp import IsoTP, IsoTPConfig


def get_version(adapter: CanAdapter, identifier: int) -> None:
    """Query each named DID for one ECU, retaining its configured transport."""
    # Lower timeout for the sync version since we scan each ECU individually and bus load is low
    config = IsoTPConfig(tx_id=identifier, timeout=0.05)
    isotp = IsoTP(adapter, config)
    uds = UDSClient(isotp, UDSConfig(p2_timeout=0.05))

    uds.tester_present()

    for did in DataIdentifier:
        try:
            resp = uds.read_data_by_identifier(did)
        except (AutomotiveError, OSError):
            continue
        print(f"{identifier:x} 0x{did.value:x} {did.name}: {resp!r}")


def main() -> None:
    """Scan the Rust example's standard/extended address ranges sequentially."""
    adapter = CanAdapter()

    standard_ids = range(0x700, 0x7F8)
    extended_ids = (0x18DA0000 + (i << 8) + 0xF1 for i in range(0x100))
    ids = [*standard_ids, *extended_ids]

    for identifier in ids:
        try:
            get_version(adapter, identifier)
        except (AutomotiveError, OSError):
            continue


if __name__ == "__main__":
    main()