python-can-isotp

The isotp.TransportLayer includes a rate limiter to throttle output data rates and prevent overwhelming CAN driver buffers. It uses a sliding window mechanism where the maximum allowed bits per window is the product of rate_limit_max_bitrate and rate_limit_window_size.

Key considerations:

• Burst Size: The product of bitrate and window size also defines the maximum burst size. For example, a bitrate of 80,000 bps and a window of 0.1s allows a burst of 8,000 bits (1,000 bytes) every 0.1s.
• Payload vs. Hardware Bitrate: The rate_limit_max_bitrate applies only to the CAN payload. It does not include CAN layer overhead (ID, CRC, etc.). Consequently, the actual hardware bitrate measured may be 1x to 1.5x higher than requested.
• Accuracy: Bitrate is achieved using OS Sleep() calls. Accuracy may be poor if the bitrate is very low or the window size is very small due to OS scheduler resolution.

ISO-15765 defines several addressing modes supported by this module. The way source and target addresses are defined depends on the selected isotp.AddressingMode. Addressing is primarily handled via the isotp.Address object.

Key addressing modes include:

• Normal addressing: Uses standard CAN arbitration IDs (rxid and txid). Works with both 11-bit and 29-bit identifiers.
• Normal fixed addressing: Encodes target_address and source_address directly into a 29-bit CAN arbitration ID. Only available for 29-bit identifiers.
• Extended addressing: Uses standard rxid/txid but adds source_address and target_address as the first byte of the payload.
• Mixed addressing (11-bit): Uses rxid/txid and an address_extension byte in the payload.
• Mixed addressing (29-bit): Combines normal fixed addressing (in the ID) with an address_extension byte in the payload.

Mixed addressing uses a payload prefix called address_extension.

Mixed addressing - 11 bits

Combines normal addressing with extended addressing. The address_extension byte in the payload acts as both the source and target address.

Reception Condition:

• Message arbitration ID must match receiver rxid.
• Payload first byte must match receiver address_extension.

Example:

• rxid: 0x123, txid: 0x456, address_extension: 0x99

0x123    [8]   99 10 0A 00 01 02 03    // First frame
0x456    [5]   99 30 00 08 00             // Flow control
0x123    [8]   99 21 04 05 06 07 08 09 // consecutive frame

Mixed addressing - 29 bits

Combines normal fixed addressing with extended addressing.

ID Encoding Patterns:

• 1-to-1 communication (target_address_type = Physical): 0x18CE<TA><SA>
• 1-to-n communication (target_address_type = Functional): 0x18CD<TA><SA>

Reception Condition:

• Message Target Address must match receiver source_address.
• Message Source Address must match the receiver target_address.
• Payload first byte must match receiver address_extension.

Example:

• source_address: 0x55, target_address: 0xAA, address_extension: 0x99

0x18CE55AA    [8]   99 10 0A 00 01 02 03    // First frame
0x18CEAA55    [5]   99 30 00 08 00             // Flow control
0x18CE55AA    [8]   99 21 04 05 06 07 08 09 // consecutive frame

Since isotp.TransportLayer calls rxfn and txfn without additional arguments, you cannot pass hardware handles (like file descriptors or API handles) directly to them. To solve this, use functools.partial to wrap your functions with the required handles before passing them to the TransportLayer constructor.

import isotp
import functools
from typing import Optional

def my_rxfn(hardware_handle, timeout:float) -> Optional[isotp.CanMesage]:
    msg = my_hardware_api_recv(hardware_handle, timeout)
    if msg is None:
        return None
    return isotp.CanMesage(arbitration_id=msg.get_id(), data=msg.get_data(), dlc=msg.get_dlc(), extended_id=msg.is_extended_id())

def my_txfn(hardware_handle, isotp_msg:isotp.CanMesage):
    msg = my_hardware_api_make_msg()
    msg.set_id(isotp_msg.arbitration_id)
    msg.set_data(isotp_msg.data)
    msg.set_dlc(isotp_msg.dlc)
    msg.set_extended_id(isotp_msg.is_extended_id)
    my_hardware_api_send(hardware_handle, msg)

hardware_handle = my_hardware_open()
addr = isotp.Address(isotp.AddressingMode.Normal_29bits, txid=0x123456, rxid = 0x123457)

# Wrap functions with the handle
partial_rxfn = functools.partial(my_rxfn, hardware_handle)
partial_txfn = functools.partial(my_txfn, hardware_handle)

layer = isotp.TransportLayer(rxfn=partial_rxfn, txfn=partial_txfn, address=addr)
layer.start()
# ... rest of program
layer.stop()
my_hardware_close()

Errors occurring during transmission or reception are reported via a user-provided error handler. An error handler must be a callable that accepts an isotp.IsoTpError as its first parameter.

Important: The error handler is called from the internal thread. Any interaction with your main application from the handler must use thread-safe mechanisms.

All errors inherit from isotp.IsoTpError (which inherits from Exception). Common error types include:

• FlowControlTimeoutError
• ConsecutiveFrameTimeoutError
• InvalidCanDataError
• UnexpectedFlowControlError
• UnexpectedConsecutiveFrameError
• ReceptionInterruptedWithSingleFrameError
• ReceptionInterruptedWithFirstFrameError
• WrongSequenceNumberError
• UnsupportedWaitFrameError
• MaximumWaitFrameReachedError
• FrameTooLongError
• ChangingInvalidRXDLError
• MissingEscapeSequenceError
• InvalidCanFdFirstFrameRXDL
• OverflowError
• BadGeneratorError

def my_error_handler(error):
    # error is an instance of isotp.IsoTpError
    print(f"An error occurred: {error}")

# When initializing your TransportLayer, pass this handler