Skip to content

Transport

Transport adapters enable real ECU I/O over CAN and DoIP.

Components

  • Transport base class
  • CANTransport (python-can)
  • DoIPTransport (doipy)
  • IsoTpSegmenter / IsoTpReassembler
  • UDSTransportClient (request/response session)
  • MockTransport (tests/sim)

Example

from pyudskit.transport import CANTransport

tr = CANTransport(channel="can0", bustype="socketcan", rx_id=0x7E8, tx_id=0x7E0)
tr.send(bytes.fromhex("22 F1 90"))
resp = tr.recv(2000)
print(resp)
tr.close()

from pyudskit.transport import UDSTransportClient, MockTransport

mock = MockTransport(responses=[bytes.fromhex("50 03 00 19 01 F4")])
client = UDSTransportClient(mock)
print(client.request(bytes.fromhex("10 03")))

API Docs

pyudskit.transport

CANTransport

Bases: Transport

CAN transport using python-can. Requires python-can installed.

Source code in pyudskit/transport/can.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class CANTransport(Transport):
    """CAN transport using python-can. Requires python-can installed."""

    def __init__(self, channel: str, bustype: str, rx_id: int, tx_id: int, config: Optional[TransportConfig] = None) -> None:
        super().__init__(config=config)
        try:
            import can  # type: ignore
        except Exception as exc:  # pragma: no cover
            raise RuntimeError("python-can is required for CANTransport") from exc
        self._can = can
        self.bus = can.Bus(channel=channel, bustype=bustype)
        self.rx_id = rx_id
        self.tx_id = tx_id
        self._reassembler = IsoTpReassembler()
        self._segmenter = IsoTpSegmenter()

    def send(self, request_bytes: bytes) -> None:
        for frame in self._segmenter.segment(request_bytes):
            msg = self._can.Message(arbitration_id=self.tx_id, data=frame, is_extended_id=False)
            self.bus.send(msg)

    def recv(self, timeout_ms: Optional[int] = None) -> Optional[bytes]:
        timeout = (timeout_ms or self.config.timeout_ms) / 1000.0
        msg = self.bus.recv(timeout=timeout)
        if msg is None:
            return None
        payload = self._reassembler.feed(bytes(msg.data))
        return payload

    def close(self) -> None:
        try:
            self.bus.shutdown()
        except Exception:
            pass

DoIPTransport

Bases: Transport

DoIP transport using doipy. Requires doipy installed.

Source code in pyudskit/transport/doip.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class DoIPTransport(Transport):
    """DoIP transport using doipy. Requires doipy installed."""

    def __init__(self, ip: str, logical_address: int, tx_id: int, rx_id: int, config: Optional[TransportConfig] = None) -> None:
        super().__init__(config=config)
        try:
            from doipy import DoIPClient  # type: ignore
        except Exception as exc:  # pragma: no cover
            raise RuntimeError("doipy is required for DoIPTransport") from exc
        self.client = DoIPClient(ip, logical_address)
        self.tx_id = tx_id
        self.rx_id = rx_id

    def send(self, request_bytes: bytes) -> None:
        self.client.send_diagnostic_message(self.tx_id, request_bytes)

    def recv(self, timeout_ms: Optional[int] = None) -> Optional[bytes]:
        timeout = (timeout_ms or self.config.timeout_ms) / 1000.0
        msg = self.client.receive_diagnostic_message(timeout)
        if msg is None:
            return None
        return msg.payload

    def close(self) -> None:
        try:
            self.client.close()
        except Exception:
            pass

IsoTpReassembler

Minimal ISO-TP reassembler for incoming frames.

Source code in pyudskit/transport/isotp.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class IsoTpReassembler:
    """Minimal ISO-TP reassembler for incoming frames."""

    def __init__(self) -> None:
        self._buf = bytearray()
        self._expected_len: Optional[int] = None
        self._next_seq: int = 1

    def feed(self, frame: bytes) -> Optional[bytes]:
        if not frame:
            return None
        pci = frame[0]
        frame_type = (pci & 0xF0) >> 4
        if frame_type == 0x0:  # Single Frame
            length = pci & 0x0F
            return bytes(frame[1 : 1 + length])
        if frame_type == 0x1:  # First Frame
            length = ((pci & 0x0F) << 8) | frame[1]
            self._expected_len = length
            self._buf = bytearray(frame[2:])
            self._next_seq = 1
            return self._maybe_complete()
        if frame_type == 0x2:  # Consecutive Frame
            seq = pci & 0x0F
            if seq != (self._next_seq & 0x0F):
                raise IsoTpError("sequence mismatch")
            self._next_seq += 1
            self._buf.extend(frame[1:])
            return self._maybe_complete()
        if frame_type == 0x3:  # Flow Control (ignored by reassembler)
            return None
        raise IsoTpError("unknown PCI")

    def _maybe_complete(self) -> Optional[bytes]:
        if self._expected_len is None:
            return None
        if len(self._buf) >= self._expected_len:
            payload = bytes(self._buf[: self._expected_len])
            self._reset()
            return payload
        return None

    def _reset(self) -> None:
        self._buf = bytearray()
        self._expected_len = None
        self._next_seq = 1

IsoTpSegmenter

Segment outgoing payloads into ISO-TP frames (8-byte CAN).

Source code in pyudskit/transport/isotp.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class IsoTpSegmenter:
    """Segment outgoing payloads into ISO-TP frames (8-byte CAN)."""

    def segment(self, payload: bytes) -> list[bytes]:
        if len(payload) <= 7:
            return [bytes([len(payload) & 0x0F]) + payload.ljust(7, b"\x00")]
        frames: list[bytes] = []
        length = len(payload)
        first = bytes([0x10 | ((length >> 8) & 0x0F), length & 0xFF]) + payload[:6]
        frames.append(first.ljust(8, b"\x00"))
        seq = 1
        idx = 6
        while idx < length:
            chunk = payload[idx : idx + 7]
            frame = bytes([0x20 | (seq & 0x0F)]) + chunk
            frames.append(frame.ljust(8, b"\x00"))
            seq = (seq + 1) & 0x0F
            idx += 7
        return frames

MockTransport

Bases: Transport

Mock transport for tests and offline simulations.

Source code in pyudskit/transport/mock.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class MockTransport(Transport):
    """Mock transport for tests and offline simulations."""

    def __init__(
        self,
        responses: Optional[list[bytes]] = None,
        handler: Optional[Callable[[bytes], Optional[bytes]]] = None,
        config: Optional[TransportConfig] = None,
    ) -> None:
        super().__init__(config=config)
        self.sent: list[bytes] = []
        self._queue: Deque[bytes] = deque(responses or [])
        self._handler = handler

    def send(self, request_bytes: bytes) -> None:
        self.sent.append(request_bytes)
        if self._handler:
            resp = self._handler(request_bytes)
            if resp:
                self._queue.append(resp)

    def recv(self, timeout_ms: Optional[int] = None) -> Optional[bytes]:
        if self._queue:
            return self._queue.popleft()
        return None

    def close(self) -> None:
        self._queue.clear()

Transport

Bases: ABC

Abstract transport interface for UDS over CAN/DoIP/etc.

Source code in pyudskit/transport/base.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class Transport(ABC):
    """Abstract transport interface for UDS over CAN/DoIP/etc."""

    def __init__(self, config: Optional[TransportConfig] = None) -> None:
        self.config = config or TransportConfig()

    @abstractmethod
    def send(self, request_bytes: bytes) -> None:
        """Send raw UDS payload bytes."""
        ...

    @abstractmethod
    def recv(self, timeout_ms: Optional[int] = None) -> Optional[bytes]:
        """Receive raw UDS payload bytes, or None on timeout."""
        ...

    @abstractmethod
    def close(self) -> None:
        """Close underlying transport resources."""
        ...

close() abstractmethod

Close underlying transport resources.

Source code in pyudskit/transport/base.py 
29
30
31
32
@abstractmethod
def close(self) -> None:
    """Close underlying transport resources."""
    ...

recv(timeout_ms=None) abstractmethod

Receive raw UDS payload bytes, or None on timeout.

Source code in pyudskit/transport/base.py 
24
25
26
27
@abstractmethod
def recv(self, timeout_ms: Optional[int] = None) -> Optional[bytes]:
    """Receive raw UDS payload bytes, or None on timeout."""
    ...

send(request_bytes) abstractmethod

Send raw UDS payload bytes.

Source code in pyudskit/transport/base.py 
19
20
21
22
@abstractmethod
def send(self, request_bytes: bytes) -> None:
    """Send raw UDS payload bytes."""
    ...

UDSTransportClient

End-to-end UDS request/response over a Transport.

Source code in pyudskit/transport/session.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class UDSTransportClient:
    """End-to-end UDS request/response over a Transport."""

    def __init__(self, transport: Transport, timing: Optional[UDSTiming] = None) -> None:
        self.transport = transport
        self.timing = timing or UDSTiming()

    def request(self, request_bytes: bytes, timeout_ms: Optional[int] = None) -> Optional[bytes]:
        self.transport.send(request_bytes)
        deadline = time.time() + ((timeout_ms or self.timing.overall_timeout_ms) / 1000.0)
        wait_ms = self.timing.p2_ms

        while True:
            remaining = deadline - time.time()
            if remaining <= 0:
                return None
            resp = self.transport.recv(int(min(wait_ms, remaining * 1000)))
            if resp is None:
                continue
            msg = UDSMessage.from_bytes(resp)
            if msg.is_negative_response and msg.nrc == 0x78:
                wait_ms = self.timing.p2_star_ms
                continue
            return resp

    def request_message(self, request_bytes: bytes, timeout_ms: Optional[int] = None) -> Optional[UDSMessage]:
        resp = self.request(request_bytes, timeout_ms=timeout_ms)
        if resp is None:
            return None
        return UDSMessage.from_bytes(resp)

pyudskit.transport.base.Transport

Bases: ABC

Abstract transport interface for UDS over CAN/DoIP/etc.

Source code in pyudskit/transport/base.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class Transport(ABC):
    """Abstract transport interface for UDS over CAN/DoIP/etc."""

    def __init__(self, config: Optional[TransportConfig] = None) -> None:
        self.config = config or TransportConfig()

    @abstractmethod
    def send(self, request_bytes: bytes) -> None:
        """Send raw UDS payload bytes."""
        ...

    @abstractmethod
    def recv(self, timeout_ms: Optional[int] = None) -> Optional[bytes]:
        """Receive raw UDS payload bytes, or None on timeout."""
        ...

    @abstractmethod
    def close(self) -> None:
        """Close underlying transport resources."""
        ...

close() abstractmethod

Close underlying transport resources.

Source code in pyudskit/transport/base.py 
29
30
31
32
@abstractmethod
def close(self) -> None:
    """Close underlying transport resources."""
    ...

recv(timeout_ms=None) abstractmethod

Receive raw UDS payload bytes, or None on timeout.

Source code in pyudskit/transport/base.py 
24
25
26
27
@abstractmethod
def recv(self, timeout_ms: Optional[int] = None) -> Optional[bytes]:
    """Receive raw UDS payload bytes, or None on timeout."""
    ...

send(request_bytes) abstractmethod

Send raw UDS payload bytes.

Source code in pyudskit/transport/base.py 
19
20
21
22
@abstractmethod
def send(self, request_bytes: bytes) -> None:
    """Send raw UDS payload bytes."""
    ...

pyudskit.transport.can.CANTransport

Bases: Transport

CAN transport using python-can. Requires python-can installed.

Source code in pyudskit/transport/can.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class CANTransport(Transport):
    """CAN transport using python-can. Requires python-can installed."""

    def __init__(self, channel: str, bustype: str, rx_id: int, tx_id: int, config: Optional[TransportConfig] = None) -> None:
        super().__init__(config=config)
        try:
            import can  # type: ignore
        except Exception as exc:  # pragma: no cover
            raise RuntimeError("python-can is required for CANTransport") from exc
        self._can = can
        self.bus = can.Bus(channel=channel, bustype=bustype)
        self.rx_id = rx_id
        self.tx_id = tx_id
        self._reassembler = IsoTpReassembler()
        self._segmenter = IsoTpSegmenter()

    def send(self, request_bytes: bytes) -> None:
        for frame in self._segmenter.segment(request_bytes):
            msg = self._can.Message(arbitration_id=self.tx_id, data=frame, is_extended_id=False)
            self.bus.send(msg)

    def recv(self, timeout_ms: Optional[int] = None) -> Optional[bytes]:
        timeout = (timeout_ms or self.config.timeout_ms) / 1000.0
        msg = self.bus.recv(timeout=timeout)
        if msg is None:
            return None
        payload = self._reassembler.feed(bytes(msg.data))
        return payload

    def close(self) -> None:
        try:
            self.bus.shutdown()
        except Exception:
            pass

pyudskit.transport.doip.DoIPTransport

Bases: Transport

DoIP transport using doipy. Requires doipy installed.

Source code in pyudskit/transport/doip.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class DoIPTransport(Transport):
    """DoIP transport using doipy. Requires doipy installed."""

    def __init__(self, ip: str, logical_address: int, tx_id: int, rx_id: int, config: Optional[TransportConfig] = None) -> None:
        super().__init__(config=config)
        try:
            from doipy import DoIPClient  # type: ignore
        except Exception as exc:  # pragma: no cover
            raise RuntimeError("doipy is required for DoIPTransport") from exc
        self.client = DoIPClient(ip, logical_address)
        self.tx_id = tx_id
        self.rx_id = rx_id

    def send(self, request_bytes: bytes) -> None:
        self.client.send_diagnostic_message(self.tx_id, request_bytes)

    def recv(self, timeout_ms: Optional[int] = None) -> Optional[bytes]:
        timeout = (timeout_ms or self.config.timeout_ms) / 1000.0
        msg = self.client.receive_diagnostic_message(timeout)
        if msg is None:
            return None
        return msg.payload

    def close(self) -> None:
        try:
            self.client.close()
        except Exception:
            pass

pyudskit.transport.isotp.IsoTpSegmenter

Segment outgoing payloads into ISO-TP frames (8-byte CAN).

Source code in pyudskit/transport/isotp.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class IsoTpSegmenter:
    """Segment outgoing payloads into ISO-TP frames (8-byte CAN)."""

    def segment(self, payload: bytes) -> list[bytes]:
        if len(payload) <= 7:
            return [bytes([len(payload) & 0x0F]) + payload.ljust(7, b"\x00")]
        frames: list[bytes] = []
        length = len(payload)
        first = bytes([0x10 | ((length >> 8) & 0x0F), length & 0xFF]) + payload[:6]
        frames.append(first.ljust(8, b"\x00"))
        seq = 1
        idx = 6
        while idx < length:
            chunk = payload[idx : idx + 7]
            frame = bytes([0x20 | (seq & 0x0F)]) + chunk
            frames.append(frame.ljust(8, b"\x00"))
            seq = (seq + 1) & 0x0F
            idx += 7
        return frames

pyudskit.transport.isotp.IsoTpReassembler

Minimal ISO-TP reassembler for incoming frames.

Source code in pyudskit/transport/isotp.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class IsoTpReassembler:
    """Minimal ISO-TP reassembler for incoming frames."""

    def __init__(self) -> None:
        self._buf = bytearray()
        self._expected_len: Optional[int] = None
        self._next_seq: int = 1

    def feed(self, frame: bytes) -> Optional[bytes]:
        if not frame:
            return None
        pci = frame[0]
        frame_type = (pci & 0xF0) >> 4
        if frame_type == 0x0:  # Single Frame
            length = pci & 0x0F
            return bytes(frame[1 : 1 + length])
        if frame_type == 0x1:  # First Frame
            length = ((pci & 0x0F) << 8) | frame[1]
            self._expected_len = length
            self._buf = bytearray(frame[2:])
            self._next_seq = 1
            return self._maybe_complete()
        if frame_type == 0x2:  # Consecutive Frame
            seq = pci & 0x0F
            if seq != (self._next_seq & 0x0F):
                raise IsoTpError("sequence mismatch")
            self._next_seq += 1
            self._buf.extend(frame[1:])
            return self._maybe_complete()
        if frame_type == 0x3:  # Flow Control (ignored by reassembler)
            return None
        raise IsoTpError("unknown PCI")

    def _maybe_complete(self) -> Optional[bytes]:
        if self._expected_len is None:
            return None
        if len(self._buf) >= self._expected_len:
            payload = bytes(self._buf[: self._expected_len])
            self._reset()
            return payload
        return None

    def _reset(self) -> None:
        self._buf = bytearray()
        self._expected_len = None
        self._next_seq = 1

pyudskit.transport.session.UDSTransportClient

End-to-end UDS request/response over a Transport.

Source code in pyudskit/transport/session.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class UDSTransportClient:
    """End-to-end UDS request/response over a Transport."""

    def __init__(self, transport: Transport, timing: Optional[UDSTiming] = None) -> None:
        self.transport = transport
        self.timing = timing or UDSTiming()

    def request(self, request_bytes: bytes, timeout_ms: Optional[int] = None) -> Optional[bytes]:
        self.transport.send(request_bytes)
        deadline = time.time() + ((timeout_ms or self.timing.overall_timeout_ms) / 1000.0)
        wait_ms = self.timing.p2_ms

        while True:
            remaining = deadline - time.time()
            if remaining <= 0:
                return None
            resp = self.transport.recv(int(min(wait_ms, remaining * 1000)))
            if resp is None:
                continue
            msg = UDSMessage.from_bytes(resp)
            if msg.is_negative_response and msg.nrc == 0x78:
                wait_ms = self.timing.p2_star_ms
                continue
            return resp

    def request_message(self, request_bytes: bytes, timeout_ms: Optional[int] = None) -> Optional[UDSMessage]:
        resp = self.request(request_bytes, timeout_ms=timeout_ms)
        if resp is None:
            return None
        return UDSMessage.from_bytes(resp)

pyudskit.transport.session.UDSTiming dataclass

Source code in pyudskit/transport/session.py 
11
12
13
14
15
@dataclass
class UDSTiming:
    p2_ms: int = 50
    p2_star_ms: int = 5000
    overall_timeout_ms: int = 10000

pyudskit.transport.mock.MockTransport

Bases: Transport

Mock transport for tests and offline simulations.

Source code in pyudskit/transport/mock.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class MockTransport(Transport):
    """Mock transport for tests and offline simulations."""

    def __init__(
        self,
        responses: Optional[list[bytes]] = None,
        handler: Optional[Callable[[bytes], Optional[bytes]]] = None,
        config: Optional[TransportConfig] = None,
    ) -> None:
        super().__init__(config=config)
        self.sent: list[bytes] = []
        self._queue: Deque[bytes] = deque(responses or [])
        self._handler = handler

    def send(self, request_bytes: bytes) -> None:
        self.sent.append(request_bytes)
        if self._handler:
            resp = self._handler(request_bytes)
            if resp:
                self._queue.append(resp)

    def recv(self, timeout_ms: Optional[int] = None) -> Optional[bytes]:
        if self._queue:
            return self._queue.popleft()
        return None

    def close(self) -> None:
        self._queue.clear()