__all__ = [
'PDU',
'RateLimiter',
'TransportLayerLogic',
'TransportLayer',
'CanStack',
'NotifierBasedCanStack'
]
from isotp.can_message import CanMessage
from isotp.tools import Timer, FiniteByteGenerator
import isotp.address
import isotp.errors
import queue
import logging
from copy import copy
import binascii
import time
import math
import enum
from dataclasses import dataclass
import threading
import inspect
import functools
from collections.abc import Iterable
from typing import Optional, Any, List, Callable, Dict, Tuple, Union, Generator, cast
try:
import can
_can_available = True
except ImportError:
_can_available = False
def is_documented_by(original: Callable[[Any], Any]) -> Callable[[Any], Any]:
def wrapper(target: Callable[[Any], Any]) -> Callable[[Any], Any]:
target.__doc__ = original.__doc__
return target
return wrapper
class PDU:
"""
Converts a CAN Message into a meaningful PDU such as SingleFrame, FirstFrame, ConsecutiveFrame, FlowControl
:param msg: The CAN message
:type msg: `isotp.protocol.CanMessage`
"""
__slots__ = 'type', 'length', 'data', 'blocksize', 'stmin', 'stmin_sec', 'seqnum', 'flow_status', 'rx_dl', 'escape_sequence', 'can_dl'
class Type:
SINGLE_FRAME = 0
FIRST_FRAME = 1
CONSECUTIVE_FRAME = 2
FLOW_CONTROL = 3
class FlowStatus:
ContinueToSend = 0
Wait = 1
Overflow = 2
type: int
length: Optional[int]
data: bytes
blocksize: Optional[int]
stmin: Optional[int]
stmin_sec: Optional[float]
seqnum: Optional[int]
flow_status: Optional[int]
rx_dl: int
escape_sequence: bool
can_dl: int
def __init__(self, msg: CanMessage, start_of_data: int = 0) -> None:
self.data = bytes()
self.length = None
self.blocksize = None
self.stmin = None
self.stmin_sec = None
self.seqnum = None
self.flow_status = None
self.escape_sequence = False
if len(msg.data) < start_of_data:
raise ValueError("Received message is missing data according to prefix size")
self.can_dl = len(msg.data)
self.rx_dl = max(8, self.can_dl)
msg_data = msg.data[start_of_data:]
datalen = len(msg_data)
# Guarantee at least presence of byte #1
if datalen > 0:
hnb = (msg_data[0] >> 4) & 0xF
if hnb > 3:
raise ValueError('Received message with unknown frame type %d' % hnb)
self.type = int(hnb)
else:
raise ValueError('Empty CAN frame')
if self.type == self.Type.SINGLE_FRAME:
length_placeholder = int(msg_data[0]) & 0xF
if length_placeholder != 0:
self.length = length_placeholder
if self.length > datalen - 1:
raise ValueError("Received Single Frame with length of %d while there is room for %d bytes of data with this configuration" % (
self.length, datalen - 1))
self.data = msg_data[1:][:self.length]
else: # Escape sequence
if datalen < 2:
raise ValueError('Single frame with escape sequence must be at least %d bytes long with this configuration' % (2 + start_of_data))
self.escape_sequence = True
self.length = int(msg_data[1])
if self.length == 0:
raise ValueError("Received Single Frame with length of 0 bytes")
if self.length > datalen - 2:
raise ValueError("Received Single Frame with length of %d while there is room for %d bytes of data with this configuration" % (
self.length, datalen - 2))
self.data = msg_data[2:][:self.length]
elif self.type == self.Type.FIRST_FRAME:
if datalen < 2:
raise ValueError('First frame without escape sequence must be at least %d bytes long with this configuration' % (2 + start_of_data))
length_placeholder = ((int(msg_data[0]) & 0xF) << 8) | int(msg_data[1])
if length_placeholder != 0: # Frame is maximum 4095 bytes
self.length = length_placeholder
self.data = msg_data[2:][:min(self.length, datalen - 2)]
else: # Frame is larger than 4095 bytes
if datalen < 6:
raise ValueError('First frame with escape sequence must be at least %d bytes long with this configuration' % (6 + start_of_data))
self.escape_sequence = True
self.length = (msg_data[2] << 24) | (msg_data[3] << 16) | (msg_data[4] << 8) | (msg_data[5] << 0)
self.data = msg_data[6:][:min(self.length, datalen - 6)]
elif self.type == self.Type.CONSECUTIVE_FRAME:
self.seqnum = int(msg_data[0]) & 0xF
self.data = msg_data[1:] # No need to check size as this will return empty data if overflow.
elif self.type == self.Type.FLOW_CONTROL:
if datalen < 3:
raise ValueError('Flow Control frame must be at least %d bytes with the actual configuration' % (3 + start_of_data))
self.flow_status = int(msg_data[0]) & 0xF
if self.flow_status >= 3:
raise ValueError('Unknown flow status')
self.blocksize = int(msg_data[1])
stmin_temp = int(msg_data[2])
if stmin_temp >= 0 and stmin_temp <= 0x7F:
self.stmin_sec = stmin_temp / 1000
elif stmin_temp >= 0xf1 and stmin_temp <= 0xF9:
self.stmin_sec = (stmin_temp - 0xF0) / 10000
if self.stmin_sec is None:
raise ValueError('Invalid StMin received in Flow Control')
else:
self.stmin = stmin_temp
else:
raise ValueError("Unsupported PDU type: %s" % self.type)
@classmethod
def craft_flow_control_data(cls, flow_status: int, blocksize: int, stmin: int) -> bytes:
return bytes([(0x30 | (flow_status) & 0xF), blocksize & 0xFF, stmin & 0xFF])
def name(self) -> str:
if self.type is None:
return "[None]"
if self.type == self.Type.SINGLE_FRAME:
return "SINGLE_FRAME"
elif self.type == self.Type.FIRST_FRAME:
return "FIRST_FRAME"
elif self.type == self.Type.CONSECUTIVE_FRAME:
return "CONSECUTIVE_FRAME"
elif self.type == self.Type.FLOW_CONTROL:
return "FLOW_CONTROL"
else:
return "Reserved"
class RateLimiter:
TIME_SLOT_LENGTH = 0.005
enabled: bool
mean_bitrate: float
window_size_sec: float
error_reason: str
burst_bitcount: List[int]
burst_time: List[float]
bit_total: int
window_bit_max: float
def __init__(self, mean_bitrate: float = 10000000, window_size_sec: float = 0.1) -> None:
self.enabled = False
self.mean_bitrate = mean_bitrate
self.window_size_sec = window_size_sec
self.error_reason = ''
self.reset()
if self.can_be_enabled():
self.enable()
def can_be_enabled(self) -> bool:
try:
float(self.mean_bitrate)
except:
self.error_reason = 'mean_bitrate is not numerical'
return False
if float(self.mean_bitrate) <= 0:
self.error_reason = 'mean_bitrate must be greater than 0'
return False
try:
float(self.window_size_sec)
except:
self.error_reason = 'window_size_sec is not numerical'
return False
if float(self.window_size_sec) <= 0:
self.error_reason = 'window_size_sec must be greater than 0'
return False
return True
def set_bitrate(self, mean_bitrate: float) -> None:
self.mean_bitrate = mean_bitrate
def enable(self) -> None:
if self.can_be_enabled():
self.mean_bitrate = float(self.mean_bitrate)
self.window_size_sec = float(self.window_size_sec)
self.enabled = True
self.reset()
else:
raise ValueError('Cannot enable Rate Limiter. \n %s' % self.error_reason)
def disable(self) -> None:
self.enabled = False
def reset(self) -> None:
self.burst_bitcount = []
self.burst_time = []
self.bit_total = 0
self.window_bit_max = self.mean_bitrate * self.window_size_sec
def update(self) -> None:
if not self.enabled:
self.reset()
return
t = time.perf_counter()
while len(self.burst_time) > 0:
t2 = self.burst_time[0]
if t - t2 > self.window_size_sec:
self.burst_time.pop(0)
n_to_remove = self.burst_bitcount.pop(0)
self.bit_total -= n_to_remove
else:
break
def allowed_bytes(self) -> int:
no_limit = 0xFFFFFFFF
if not self.enabled:
return no_limit
allowed_bits = max(self.window_bit_max - self.bit_total, 0)
return math.floor(allowed_bits / 8)
def inform_byte_sent(self, datalen: int) -> None:
if self.enabled:
bytelen = datalen * 8
t = time.perf_counter()
self.bit_total += bytelen
if len(self.burst_time) == 0:
self.burst_time.append(t)
self.burst_bitcount.append(bytelen)
else:
last_time = self.burst_time[-1]
if t - last_time > self.TIME_SLOT_LENGTH:
self.burst_time.append(t)
self.burst_bitcount.append(bytelen)
else:
self.burst_bitcount[-1] += bytelen
class TransportLayerLogic:
LOGGER_NAME = 'isotp'
@dataclass(init=False)
class Params:
__slots__ = (
'stmin',
'blocksize',
'override_receiver_stmin',
'rx_flowcontrol_timeout',
'rx_consecutive_frame_timeout',
'tx_padding',
'wftmax',
'tx_data_length',
'tx_data_min_length',
'max_frame_size',
'can_fd',
'bitrate_switch',
'default_target_address_type',
'rate_limit_max_bitrate',
'rate_limit_window_size',
'rate_limit_enable',
'listen_mode',
'blocking_send',
'logger_name',
'wait_func'
)
stmin: int
blocksize: int
override_receiver_stmin: Optional[float]
rx_flowcontrol_timeout: float
rx_consecutive_frame_timeout: float
tx_padding: Optional[int]
wftmax: int
tx_data_length: int
tx_data_min_length: Optional[int]
max_frame_size: int
can_fd: bool
bitrate_switch: bool
default_target_address_type: isotp.TargetAddressType
rate_limit_max_bitrate: int
rate_limit_window_size: float
rate_limit_enable: bool
listen_mode: bool
blocking_send: bool
logger_name: str
wait_func: Callable[[float], None]
def __init__(self) -> None:
self.stmin = 0
self.blocksize = 8
self.override_receiver_stmin = None
self.rx_flowcontrol_timeout = 1000
self.rx_consecutive_frame_timeout = 1000
self.tx_padding = None
self.wftmax = 0
self.tx_data_length = 8
self.tx_data_min_length = None
self.max_frame_size = 4095
self.can_fd = False
self.bitrate_switch = False
self.default_target_address_type = isotp.address.TargetAddressType.Physical
self.rate_limit_max_bitrate = 100000000
self.rate_limit_window_size = 0.2
self.rate_limit_enable = False
self.listen_mode = False
self.blocking_send = False
self.logger_name = TransportLayer.LOGGER_NAME
self.wait_func = time.sleep
def set(self, key: str, val: Any, validate: bool = True) -> None:
param_alias: Dict[str, str] = {
}
if key in param_alias:
key = param_alias[key]
setattr(self, key, val)
if validate:
self.validate()
def validate(self) -> None:
if not isinstance(self.rx_flowcontrol_timeout, int):
raise ValueError('rx_flowcontrol_timeout must be an integer')
if self.rx_flowcontrol_timeout < 0:
raise ValueError('rx_flowcontrol_timeout must be positive integer')
if not isinstance(self.rx_consecutive_frame_timeout, int):
raise ValueError('rx_consecutive_frame_timeout must be an integer')
if self.rx_consecutive_frame_timeout < 0:
raise ValueError('rx_consecutive_frame_timeout must be positive integer')
if self.tx_padding is not None:
if not isinstance(self.tx_padding, int):
raise ValueError('tx_padding must be an integer')
if self.tx_padding < 0 or self.tx_padding > 0xFF:
raise ValueError('tx_padding must be an integer between 0x00 and 0xFF')
if not isinstance(self.stmin, int):
raise ValueError('stmin must be an integer')
if self.stmin < 0 or self.stmin > 0xFF:
raise ValueError('stmin must be positive integer between 0x00 and 0xFF')
if not isinstance(self.blocksize, int):
raise ValueError('blocksize must be an integer')
if self.blocksize < 0 or self.blocksize > 0xFF:
raise ValueError('blocksize must be and integer between 0x00 and 0xFF')
if self.override_receiver_stmin is not None:
if not isinstance(self.override_receiver_stmin, (int, float)) or isinstance(self.override_receiver_stmin, bool):
raise ValueError('override_receiver_stmin must be a float')
self.override_receiver_stmin = float(self.override_receiver_stmin)
if self.override_receiver_stmin < 0 or not math.isfinite(self.override_receiver_stmin):
raise ValueError('Invalid override_receiver_stmin')
if not isinstance(self.wftmax, int):
raise ValueError('wftmax must be an integer')
if self.wftmax < 0:
raise ValueError('wftmax must be and integer equal or greater than 0')
if not isinstance(self.tx_data_length, int):
raise ValueError('tx_data_length must be an integer')
if self.tx_data_length not in [8, 12, 16, 20, 24, 32, 48, 64]:
raise ValueError('tx_data_length must be one of these value : 8, 12, 16, 20, 24, 32, 48, 64 ')
if self.tx_data_min_length is not None:
if not isinstance(self.tx_data_min_length, int):
raise ValueError('tx_data_min_length must be an integer')
if self.tx_data_min_length not in [1, 2, 3, 4, 5, 6, 7, 8, 12, 16, 20, 24, 32, 48, 64]:
raise ValueError('tx_data_min_length must be one of these value : 1, 2, 3, 4, 5, 6, 7, 8, 12, 16, 20, 24, 32, 48, 64 ')
if self.tx_data_min_length > self.tx_data_length:
raise ValueError('tx_data_min_length cannot be greater than tx_data_length')
if not isinstance(self.max_frame_size, int):
raise ValueError('max_frame_size must be an integer')
if self.max_frame_size < 0:
raise ValueError('max_frame_size must be a positive integer')
if not isinstance(self.can_fd, bool):
raise ValueError('can_fd must be a boolean value')
if not isinstance(self.bitrate_switch, bool):
raise ValueError('bitrate_switch must be a boolean value')
if isinstance(self.default_target_address_type, int):
self.default_target_address_type = isotp.TargetAddressType(self.default_target_address_type)
if not isinstance(self.default_target_address_type, isotp.TargetAddressType):
raise ValueError('default_target_address_type must be an integer or a TargetAddressType instance')
if self.default_target_address_type not in [isotp.address.TargetAddressType.Physical, isotp.address.TargetAddressType.Functional]:
raise ValueError('default_target_address_type must be either be Physical (%d) or Functional (%d)' %
(isotp.address.TargetAddressType.Physical.value, isotp.address.TargetAddressType.Functional.value))
if not isinstance(self.rate_limit_max_bitrate, int):
raise ValueError('rate_limit_max_bitrate must be an integer')
if self.rate_limit_max_bitrate <= 0:
raise ValueError('rate_limit_max_bitrate must be greater than 0')
if not (isinstance(self.rate_limit_window_size, float) or isinstance(self.rate_limit_window_size, int)):
raise ValueError('rate_limit_window_size must be a float ')
if self.rate_limit_window_size <= 0:
raise ValueError('rate_limit_window_size must be greater than 0')
if not isinstance(self.rate_limit_enable, bool):
raise ValueError('rate_limit_enable must be a boolean value')
if self.rate_limit_max_bitrate * self.rate_limit_window_size < self.tx_data_length * 8:
raise ValueError(
'Rate limiter is so restrictive that a SingleFrame cannot be sent. Please, allow a higher bitrate or increase the window size. (tx_data_length = %d)' % self.tx_data_length)
if not isinstance(self.listen_mode, bool):
raise ValueError('listen_mode must be a boolean value')
if not isinstance(self.blocking_send, bool):
raise ValueError('blocking_send must be a boolean value')
if not isinstance(self.logger_name, str):
raise ValueError('logger_name must be a string')
if not callable(self.wait_func):
raise ValueError('wait_func should be a callable')
try:
self.wait_func(0.001)
except Exception as e:
raise ValueError("Given wait_func raised an exception %s" % e)
class RxState(enum.Enum):
IDLE = 0
WAIT_CF = 1
class TxState(enum.Enum):
IDLE = 0
WAIT_FC = 1
TRANSMIT_CF = 2
TRANSMIT_SF_STANDBY = 3
TRANSMIT_FF_STANDBY = 4
SendGenerator = Tuple[Generator[int, None, None], int]
@dataclass
class SendRequest:
"""An object representing a call to `TransportLayer.send() by the user. Wraps the given parameter and associate with a completion event and a success flag`"""
generator: FiniteByteGenerator
target_address_type: isotp.address.TargetAddressType
complete_event: threading.Event
success: bool
def __init__(self,
data: Union[bytearray, bytes, "TransportLayerLogic.SendGenerator"],
target_address_type: isotp.address.TargetAddressType
):
if isinstance(data, tuple):
if len(data) != 2:
raise ValueError("Given tuple must have 2 items. A generator and a length")
gen, size = data
self.generator = FiniteByteGenerator(gen, size)
elif isinstance(data, Iterable):
data = cast(Union[bytes, bytearray], data) # type:ignore
self.generator = FiniteByteGenerator((x for x in data), len(data))
else:
raise ValueError("data must be an iterable element (bytes or bytearray) or a tuple of generator,size")
self.consumed_size = 0
self.target_address_type = target_address_type
self.complete_event = threading.Event()
self.success = False
def complete(self, success: bool) -> None:
self.success = success
self.complete_event.set()
@dataclass(frozen=True)
class ProcessStats:
"""Some statistics produced by every ``process`` called indicating how much has been accomplish during that iteration."""
__slots__ = ('received', 'received_processed', 'sent', 'frame_received')
received: int
received_processed: int
sent: int
frame_received: int
def __repr__(self) -> str:
return f'<{self.__class__.__name__} received:{self.received} (processed: {self.received_processed}, sent: {self.sent})>'
@dataclass(frozen=True)
class ProcessRxReport:
immediate_tx_required: bool
frame_received: bool
__slots__ = ('immediate_tx_required', 'frame_received')
@dataclass(frozen=True)
class ProcessTxReport:
msg: Optional[CanMessage]
immediate_rx_required: bool
__slots__ = 'msg', 'immediate_rx_required'
RxFn = Callable[[float], Optional[CanMessage]]
TxFn = Callable[[CanMessage], None]
PostSendCallback = Callable[[SendRequest], None]
ErrorHandler = Callable[[Exception], None]
params: Params
logger: logging.Logger
remote_blocksize: Optional[int]
rxfn: RxFn
txfn: TxFn
tx_queue: "queue.Queue[SendRequest]"
rx_queue: "queue.Queue[bytearray]"
tx_standby_msg: Optional[CanMessage]
rx_state: RxState
tx_state: TxState
rx_block_counter: int
last_seqnum: int
rx_frame_length: int
tx_frame_length: int
last_flow_control_frame: Optional[PDU]
tx_block_counter: int
tx_seqnum: int
wft_counter: int
pending_flow_control_tx: bool
timer_tx_stmin: Timer
error_handler: Optional[ErrorHandler]
actual_rxdl: Optional[int]
timings: Dict[Tuple[RxState, TxState], float]
active_send_request: Optional[SendRequest]
rx_buffer: bytearray
address: isotp.address.AbstractAddress
timer_rx_fc: Timer
timer_rx_cf: Timer
rate_limiter: RateLimiter
blocking_rxfn: bool
post_send_callback: Optional[PostSendCallback]
def __init__(self,
rxfn: RxFn,
txfn: TxFn,
address: isotp.Address,
error_handler: Optional[ErrorHandler] = None,
params: Optional[Dict[str, Any]] = None,
post_send_callback: Optional[PostSendCallback] = None
):
self.post_send_callback = post_send_callback
self.params = self.Params()
self.logger = logging.getLogger(self.LOGGER_NAME)
if params is not None:
for k in params:
self.params.set(k, params[k], validate=False)
self.params.validate()
self.logger = logging.getLogger(self.params.logger_name)
self.remote_blocksize = None # Block size received in Flow Control message
# Backward compatibility. Handle rxfn with no params as non-blocking
if len(inspect.signature(rxfn).parameters) < 1:
self.rxfn = lambda x: rxfn() # type: ignore
self.blocking_rxfn = False
self.logger.debug("Given rxfn is considered non-blocking")
else:
self.rxfn = rxfn # Function to call to receive a CAN message
self.blocking_rxfn = True
self.logger.debug("Given rxfn is considered blocking")
self.txfn = txfn # Function to call to receive a CAN message
self.set_address(address)
self.tx_queue = queue.Queue() # Layer Input queue for IsoTP frame
self.rx_queue = queue.Queue() # Layer Output queue for IsoTP frame
self.tx_standby_msg = None # Pending message when throttling is active
self.active_send_request = None # The user request for sending. Contains a synchronizing event for blocking send
self.rx_state = self.RxState.IDLE # State of the reception FSM
self.tx_state = self.TxState.IDLE # State of the transmission FSM
self.last_rx_state = self.rx_state # Used to log changes in states
self.last_tx_state = self.tx_state # Used to log changes in states
self.rx_block_counter = 0 # Keeps track of how many block we've received. Used to determine when to send a flow control message
self.last_seqnum = 0 # Consecutive frame Sequence number of previous message
self.rx_frame_length = 0 # Length of IsoTP frame being received at the moment
self.tx_frame_length = 0 # Length of the data that we are sending
self.last_flow_control_frame = None # When a FlowControl is received. Put here
self.tx_block_counter = 0 # Keeps track of how many block we've sent. USed to determine when to wait for a flow control message
self.tx_seqnum = 0 # Keeps track of the actual sequence number while sending
self.wft_counter = 0 # Keeps track of how many wait frame we've received
self.pending_flow_control_tx = False # Flag indicating that we need to transmit a flow control message. Set by Rx Process, Cleared by Tx Process
self._empty_rx_buffer()
self.timer_tx_stmin = Timer(timeout=0)
self.error_handler = error_handler
self.actual_rxdl = None # Length of the CAN messages during a reception. Set by the first frame. All consecutive frames must be identical
# Legacy (v1.x) timing recommendation
self.timings = {
(self.RxState.IDLE, self.TxState.IDLE): 0.02,
(self.RxState.IDLE, self.TxState.WAIT_FC): 0.005,
}
self.load_params()
def load_params(self) -> None:
self.params.validate()
self.timer_rx_fc = Timer(timeout=float(self.params.rx_flowcontrol_timeout) / 1000)
self.timer_rx_cf = Timer(timeout=float(self.params.rx_consecutive_frame_timeout) / 1000)
self.rate_limiter = RateLimiter(mean_bitrate=self.params.rate_limit_max_bitrate, window_size_sec=self.params.rate_limit_window_size)
if self.params.rate_limit_enable:
self.rate_limiter.enable()
def send(self,
data: Union[bytes, bytearray, SendGenerator],
target_address_type: Optional[Union[isotp.address.TargetAddressType, int]] = None,
send_timeout: Optional[float] = None
) -> None:
"""
Enqueue an IsoTP frame to be sent over CAN network.
When performing a blocking send, this method returns only when the transmission is complete or raise an exception when a failure or a timeout occurs.
See :ref:`blocking_send<param_blocking_send>`
:param data: The data to be sent. Can either be a bytearray or a tuple containing a generator and a size. The generator should return integer
:type data: bytearray | (Generator, int)
:param target_address_type: Optional parameter that can be Physical (0) for 1-to-1 communication or Functional (1) for 1-to-n.
See :class:`isotp.TargetAddressType<isotp.TargetAddressType>`.
If not provided, parameter :ref:`default_target_address_type<param_default_target_address_type>` will be used (default to ``Physical``)
:type target_address_type: int
:param send_timeout: Timeout value for blocking send. Unused if :ref:`blocking_send<param_blocking_send>` is ``False``
:type send_timeout: float or None
:raises ValueError: Given data is not a bytearray, a tuple (generator,size) or the size is too big
:raises RuntimeError: Transmit queue is full
:raises BlockingSendTimeout: When :ref:`blocking_send<param_blocking_send>` is set to ``True`` and the send operation does not complete in the given timeout.
:raises BlockingSendFailure: When :ref:`blocking_send<param_blocking_send>` is set to ``True`` and the transmission failed for any reason (e.g. unexpected frame or bad timings), including a timeout. Note that
:class:`BlockingSendTimeout<BlockingSendTimeout>` inherits :class:`BlockingSendFailure<BlockingSendFailure>`.
"""
if target_address_type is None:
target_address_type = self.params.default_target_address_type
else:
target_address_type = isotp.address.TargetAddressType(target_address_type)
send_request = self.SendRequest(data=data, target_address_type=target_address_type)
if self.tx_queue.full():
raise RuntimeError('Transmit queue is full')
if target_address_type == isotp.address.TargetAddressType.Functional:
length_bytes = 1 if self.params.tx_data_length == 8 else 2
maxlen = self.params.tx_data_length - length_bytes - len(self.address.get_tx_payload_prefix())
if send_request.generator.total_length() > maxlen:
raise ValueError('Cannot send multi packet frame with Functional TargetAddressType')
if self.params.blocking_send:
send_request.complete_event.clear()
self.logger.debug("Enqueuing a SendRequest for %d bytes and TAT=%s" % (send_request.generator.total_length(), target_address_type.name))
self.tx_queue.put(send_request)
if self.post_send_callback is not None:
self.post_send_callback(send_request)
if self.params.blocking_send:
send_request.complete_event.wait(send_timeout)
if not send_request.complete_event.is_set():
raise isotp.errors.BlockingSendTimeout("Failed to send IsoTP frame in time")
else:
if not send_request.success:
raise isotp.errors.BlockingSendFailure("Error while sending IsoTP frame")
# Receive an IsoTP frame. Output of the layer
def recv(self, block: bool = False, timeout: Optional[float] = None) -> Optional[bytearray]:
"""
Dequeue an IsoTP frame from the reception queue if available.
:param block: Tells if the read should be blocking or not
:type block: bool
:param timeout: Timeout value used for blocking read only
:type timeout: float
:return: The next available IsoTP frame
:rtype: bytearray or None
"""
try:
return self.rx_queue.get(block=block, timeout=timeout)
except queue.Empty:
return None
def available(self) -> bool:
"""
Returns ``True`` if an IsoTP frame is awaiting in the reception queue. ``False`` otherwise
"""
return not self.rx_queue.empty()
def transmitting(self) -> bool:
"""
Returns ``True`` if an IsoTP frame is being transmitted. ``False`` otherwise
"""
return not self.tx_queue.empty() or self.tx_state != self.TxState.IDLE
def process(self, rx_timeout: float = 0.0, do_rx: bool = True, do_tx: bool = True) -> ProcessStats:
"""
Function to be called periodically, as fast as possible.
This function is expected to block only if the given rxfn performs a blocking read.
:param rx_timeout: Timeout for any read operation
:type rx_timeout: float
:param do_rx: Process reception when ``True``
:type do_rx: bool
:param do_tx: Process transmission when ``True``
:type do_tx: bool
:return: Statistics about what have been accomplished during the call
:rtype: :class:`ProcessStats<isotp.ProcessStats>`
"""
run_process = True
msg_received = 0
msg_received_processed = 0
msg_sent = 0
nb_frame_received = 0
# Run as long as RxStateMachine or Tx StateMachine request the other state machine to immediatly do a pass.
# Prevent useless latency in FSM processing
while run_process:
msg: Optional[CanMessage] = None
run_process = False
# if we have data to send and nothing else in process. start by sending that data. Avoid blocking in rxfn for nothing.
start_with_tx = do_tx \
and not self.tx_queue.empty() \
and self.rx_state == self.RxState.IDLE \
and self.tx_state == self.TxState.IDLE
if start_with_tx:
run_process = True
if do_rx and not start_with_tx:
first_loop = True
while msg is not None or first_loop:
first_loop = False
msg = self.rxfn(rx_timeout)
self._check_timeouts_rx() # Check for every message because rxfn may be blocking since v2.x. Always execute, even if msg=None (issue #41)
if msg is not None:
msg_received += 1
for_me = self.address.is_for_me(msg)
if self.logger.isEnabledFor(logging.DEBUG):
addr = "%08X" % msg.arbitration_id if msg.is_extended_id else "%03X" % msg.arbitration_id
processed = 'p' if for_me else 'i' # processed/ignored
self.logger.debug("Rx: <%s> (%02d) [%s]\t %s" % (addr,
len(msg.data), processed, binascii.hexlify(msg.data).decode('ascii')))
if for_me:
msg_received_processed += 1
rx_result = self._process_rx(msg)
if rx_result.frame_received:
nb_frame_received += 1
if rx_result.immediate_tx_required:
break
start_with_tx = False # it's a one-time event
self.rate_limiter.update() # Only applies to transmission. Update after rxfn because it can be blocking.
if do_tx:
first_loop = True
msg = None
while msg is not None or first_loop:
first_loop = False
tx_result = self._process_tx()
msg = tx_result.msg
if msg is not None:
msg_sent += 1
if self.logger.isEnabledFor(logging.DEBUG):
self.logger.debug("Tx: <%03X> (%02d) [ ]\t %s" % (msg.arbitration_id,
len(msg.data), binascii.hexlify(msg.data).decode('ascii')))
self.txfn(msg)
if tx_result.immediate_rx_required:
run_process = True
break
if self.logger.isEnabledFor(logging.DEBUG):
if self.last_rx_state != self.rx_state or self.last_tx_state != self.tx_state:
self.logger.debug(f"TxState={self.tx_state.name} - RxState={self.rx_state.name}")
self.last_tx_state = self.tx_state
self.last_rx_state = self.rx_state
return self.ProcessStats(
received=msg_received,
received_processed=msg_received_processed,
sent=msg_sent,
frame_received=nb_frame_received
)
def _set_rxfn(self, rxfn: "TransportLayerLogic.RxFn") -> None:
"""
Allow post init change of rxfn. This is a trick to implement the threaded Transport Layer
and keeping the ability to run the TransportLayerLogic without threads for backward compatibility
with v1.x
"""
self.rxfn = rxfn
def _check_timeouts_rx(self) -> None:
if self.timer_rx_cf.is_timed_out():
self._trigger_error(isotp.errors.ConsecutiveFrameTimeoutError("Reception of CONSECUTIVE_FRAME timed out."))
self._stop_receiving()
def _process_rx(self, msg: CanMessage) -> ProcessRxReport:
"""Process the reception of a CAN message. Moves the reception state machine accordingly and optionally"""
# Decoding of message into PDU
try:
pdu = PDU(msg, start_of_data=self.address.get_rx_prefix_size())
except Exception as e:
self._trigger_error(isotp.errors.InvalidCanDataError("Received invalid CAN frame. %s" % (str(e))))
self._stop_receiving()
return self.ProcessRxReport(immediate_tx_required=False, frame_received=False)
# Process Flow Control message
if pdu.type == PDU.Type.FLOW_CONTROL:
self.last_flow_control_frame = pdu # Given to _process_tx method. Queue of 1 message depth
# Nothing else to be done with FlowControl. Return and run _process_tx right away
return self.ProcessRxReport(immediate_tx_required=True, frame_received=False)
frame_complete = False
if pdu.type == PDU.Type.SINGLE_FRAME:
if pdu.can_dl > 8 and pdu.escape_sequence == False:
self._trigger_error(isotp.errors.MissingEscapeSequenceError(
'For SingleFrames conveyed on a CAN message with data length (CAN_DL) > 8, length should be encoded on byte #1 and byte #0 should be 0x00'))
return self.ProcessRxReport(immediate_tx_required=False, frame_received=False)
immediate_tx_msg_required = False
# Process the state machine
if self.rx_state == self.RxState.IDLE:
self.rx_frame_length = 0
self.timer_rx_cf.stop()
if pdu.type == PDU.Type.SINGLE_FRAME:
if pdu.data is not None:
frame_complete = True
self.rx_queue.put(bytearray(pdu.data))
elif pdu.type == PDU.Type.FIRST_FRAME:
started = self._start_reception_after_first_frame_if_valid(pdu)
immediate_tx_msg_required = immediate_tx_msg_required or started
elif pdu.type == PDU.Type.CONSECUTIVE_FRAME:
self._trigger_error(isotp.errors.UnexpectedConsecutiveFrameError('Received a ConsecutiveFrame while reception was idle. Ignoring'))
elif self.rx_state == self.RxState.WAIT_CF:
if pdu.type == PDU.Type.SINGLE_FRAME:
if pdu.data is not None:
frame_complete = True
self.rx_queue.put(bytearray(pdu.data))
self.rx_state = self.RxState.IDLE
self._trigger_error(isotp.errors.ReceptionInterruptedWithSingleFrameError(
'Reception of IsoTP frame interrupted with a new SingleFrame'))
elif pdu.type == PDU.Type.FIRST_FRAME:
started = self._start_reception_after_first_frame_if_valid(pdu)
immediate_tx_msg_required = immediate_tx_msg_required or started
self._trigger_error(isotp.errors.ReceptionInterruptedWithFirstFrameError(
'Reception of IsoTP frame interrupted with a new FirstFrame'))
elif pdu.type == PDU.Type.CONSECUTIVE_FRAME:
expected_seqnum = (self.last_seqnum + 1) & 0xF
if pdu.seqnum == expected_seqnum:
bytes_to_receive = (self.rx_frame_length - len(self.rx_buffer))
if pdu.rx_dl != self.actual_rxdl and pdu.rx_dl < bytes_to_receive:
self._trigger_error(isotp.errors.ChangingInvalidRXDLError(
"Received a ConsecutiveFrame with RX_DL=%s while expected RX_DL=%s. Ignoring frame" % (pdu.rx_dl, self.actual_rxdl)))
return self.ProcessRxReport(immediate_tx_required=False, frame_received=False)
self._start_rx_cf_timer() # Received a CF message. Restart counter. Timeout handled above.
self.last_seqnum = pdu.seqnum
self._append_rx_data(pdu.data[:bytes_to_receive]) # Python handle overflow
if len(self.rx_buffer) >= self.rx_frame_length:
frame_complete = True
self.rx_queue.put(copy(self.rx_buffer)) # Data complete
self._stop_receiving() # Go back to IDLE. Reset all variables and timers.
else:
self.rx_block_counter += 1
if self.params.blocksize > 0 and (self.rx_block_counter % self.params.blocksize) == 0:
self._request_tx_flowcontrol(PDU.FlowStatus.ContinueToSend) # Sets a flag to 1. _process_tx will send it for use.
# We stop the timer until the flow control message is gone. This timer is reactivated in the _process_tx().
self.timer_rx_cf.stop()
immediate_tx_msg_required = True
else:
self._stop_receiving()
received = str(None)
if pdu.seqnum is not None:
received = "0x%02X" % pdu.seqnum
self._trigger_error(isotp.errors.WrongSequenceNumberError(
'Received a ConsecutiveFrame with wrong SequenceNumber. Expecting 0x%02X, Received %s' % (expected_seqnum, received)))
if self.pending_flow_control_tx:
immediate_tx_msg_required = True
return self.ProcessRxReport(immediate_tx_required=immediate_tx_msg_required, frame_received=frame_complete)
def _process_tx(self) -> ProcessTxReport:
"""Process the transmit state machine"""
output_msg = None # Value outputted. If None, no subsequent call to _process_tx will be done.
allowed_bytes = self.rate_limiter.allowed_bytes()
# Sends flow control if _process_rx requested it
if self.pending_flow_control_tx:
self.pending_flow_control_tx = False
if self.pending_flowcontrol_status == PDU.FlowStatus.ContinueToSend:
self._start_rx_cf_timer() # We tell the sending party that it can continue to send data, so we start checking the timeout again
if not self.params.listen_mode: # Inhibit Flow Control in listen mode.
flow_control_msg = self._make_flow_control(flow_status=self.pending_flowcontrol_status)
return self.ProcessTxReport(msg=flow_control_msg, immediate_rx_required=True) # No need to wait. Needs to run _process_rx right away
# Handle flow control reception
flow_control_frame = self.last_flow_control_frame # Reads the last message received and clears it. (Dequeue message)
self.last_flow_control_frame = None
if flow_control_frame is not None:
if flow_control_frame.flow_status == PDU.FlowStatus.Overflow: # Needs to stop sending.
self._stop_sending(success=False)
self._trigger_error(isotp.errors.OverflowError('Received a FlowControl PDU indicating an Overflow. Stopping transmission.'))
return self.ProcessTxReport(msg=None, immediate_rx_required=False)
if self.tx_state == self.TxState.IDLE:
self._trigger_error(isotp.errors.UnexpectedFlowControlError('Received a FlowControl message while transmission was Idle. Ignoring'))
else:
if flow_control_frame.flow_status == PDU.FlowStatus.Wait:
if self.params.wftmax == 0:
self._trigger_error(isotp.errors.UnsupportedWaitFrameError(
'Received a FlowControl requesting to wait, but wftmax is set to 0'))
elif self.wft_counter >= self.params.wftmax:
self._trigger_error(isotp.errors.MaximumWaitFrameReachedError(
'Received %d wait frame which is the maximum set in params.wftmax' % (self.wft_counter)))
self._stop_sending(success=False)
else:
self.wft_counter += 1
if self.tx_state in [self.TxState.WAIT_FC, self.TxState.TRANSMIT_CF]:
self.tx_state = self.TxState.WAIT_FC
self._start_rx_fc_timer()
elif flow_control_frame.flow_status == PDU.FlowStatus.ContinueToSend and not self.timer_rx_fc.is_timed_out():
self.wft_counter = 0
self.timer_rx_fc.stop()
assert flow_control_frame.stmin_sec is not None
if self.params.override_receiver_stmin is not None:
self.timer_tx_stmin.set_timeout(self.params.override_receiver_stmin)
else:
self.timer_tx_stmin.set_timeout(flow_control_frame.stmin_sec)
self.remote_blocksize = flow_control_frame.blocksize
if self.tx_state == self.TxState.WAIT_FC:
self.tx_block_counter = 0
self.timer_tx_stmin.start()
elif self.tx_state == self.TxState.TRANSMIT_CF:
pass
self.tx_state = self.TxState.TRANSMIT_CF
# ======= Timeouts ======
if self.timer_rx_fc.is_timed_out():
self._trigger_error(isotp.errors.FlowControlTimeoutError('Reception of FlowControl timed out. Stopping transmission'))
self._stop_sending(success=False)
# ======= FSM ======
# Check this first as we may have another isotp frame to send and we need to handle it right away without waiting for next "process()" call
if self.tx_state != self.TxState.IDLE:
assert self.active_send_request is not None
if self.active_send_request.generator.depleted() and self.tx_standby_msg is None: # No transmission in progress
self._stop_sending(success=True)
immediate_rx_msg_required = False
if self.tx_state == self.TxState.IDLE:
read_tx_queue = True # Read until we get non-empty frame to send
while read_tx_queue:
read_tx_queue = False
if not self.tx_queue.empty():
self.active_send_request = self.tx_queue.get()
if self.active_send_request.generator.depleted():
read_tx_queue = True # Read another frame from tx_queue
self.active_send_request.complete(True)
else:
size_on_first_byte = (self.active_send_request.generator.remaining_size() + len(self.address.get_tx_payload_prefix())) <= 7
size_offset = 1 if size_on_first_byte else 2
try:
# Single frame
total_size = self.active_send_request.generator.total_length()
if total_size <= self.params.tx_data_length - size_offset - len(self.address.get_tx_payload_prefix()):
# Will raise if size is not what was requested
payload = self.active_send_request.generator.consume(total_size, enforce_exact=True)
if size_on_first_byte:
msg_data = self.address.get_tx_payload_prefix() + bytearray([0x0 | len(payload)]) + payload
else:
msg_data = self.address.get_tx_payload_prefix() + bytearray([0x0, len(payload)]) + payload
arbitration_id = self.address.get_tx_arbitration_id(self.active_send_request.target_address_type)
msg_temp = self._make_tx_msg(arbitration_id, msg_data)
if len(msg_data) > allowed_bytes:
self.tx_standby_msg = msg_temp
self.tx_state = self.TxState.TRANSMIT_SF_STANDBY
else:
output_msg = msg_temp
# Multi frame - First Frame
else:
self.tx_frame_length = total_size
encode_length_on_2_first_bytes = True if self.tx_frame_length <= 0xFFF else False
if encode_length_on_2_first_bytes:
data_length = self.params.tx_data_length - 2 - len(self.address.get_tx_payload_prefix())
payload = self.active_send_request.generator.consume(data_length, enforce_exact=True)
msg_data = self.address.get_tx_payload_prefix() + \
bytearray([0x10 | ((self.tx_frame_length >> 8) & 0xF), self.tx_frame_length & 0xFF]) + payload
else:
data_length = self.params.tx_data_length - 6 - len(self.address.get_tx_payload_prefix())
payload = self.active_send_request.generator.consume(data_length, enforce_exact=True)
msg_data = self.address.get_tx_payload_prefix() + bytearray([0x10, 0x00, (self.tx_frame_length >> 24) & 0xFF, (self.tx_frame_length >> 16) & 0xFF, (
self.tx_frame_length >> 8) & 0xFF, (self.tx_frame_length >> 0) & 0xFF]) + payload
arbitration_id = self.address.get_tx_arbitration_id()
self.tx_seqnum = 1
msg_temp = self._make_tx_msg(arbitration_id, msg_data)
if len(msg_data) <= allowed_bytes:
output_msg = msg_temp
self.tx_state = self.TxState.WAIT_FC
self._start_rx_fc_timer()
else:
self.tx_standby_msg = msg_temp
self.tx_state = self.TxState.TRANSMIT_FF_STANDBY
except isotp.errors.BadGeneratorError as e:
self._trigger_error(e)
self._stop_sending(success=False)
elif self.tx_state in [self.TxState.TRANSMIT_SF_STANDBY, self.TxState.TRANSMIT_FF_STANDBY]:
# This states serves if the rate limiter prevent from starting a new transmission.
# We need to pop the isotp frame to know if the rate limiter must kick, but since the data is already popped,
# we can't stay in IDLE state. So we come here until the rate limiter gives us permission to proceed.
if self.tx_standby_msg is not None:
if len(self.tx_standby_msg.data) <= allowed_bytes:
output_msg = self.tx_standby_msg
self.tx_standby_msg = None
if self.tx_state == self.TxState.TRANSMIT_FF_STANDBY:
self._start_rx_fc_timer()
self.tx_state = self.TxState.WAIT_FC # After a first frame, we wait for flow control
else:
self.tx_state = self.TxState.IDLE # After a single frame, there's nothing to do
elif self.tx_state == self.TxState.WAIT_FC:
pass # Nothing to do. Flow control will make the FSM switch state by calling init_tx_consecutive_frame
elif self.tx_state == self.TxState.TRANSMIT_CF:
assert self.remote_blocksize is not None
assert self.active_send_request is not None
if self.timer_tx_stmin.is_timed_out():
data_length = self.params.tx_data_length - 1 - len(self.address.get_tx_payload_prefix())
payload_length = min(data_length, self.active_send_request.generator.remaining_size())
if payload_length <= allowed_bytes:
# We may have less data than requested
payload = self.active_send_request.generator.consume(payload_length, enforce_exact=False)
if len(payload) > 0: # Corner case. If generator size is a multiple of ll_data_length, we will get an empty payload on last frame.
msg_data = self.address.get_tx_payload_prefix() + bytearray([0x20 | self.tx_seqnum]) + payload
arbitration_id = self.address.get_tx_arbitration_id()
output_msg = self._make_tx_msg(arbitration_id, msg_data)
self.tx_seqnum = (self.tx_seqnum + 1) & 0xF
self.timer_tx_stmin.start()
self.tx_block_counter += 1
if (self.active_send_request.generator.depleted()):
if self.active_send_request.generator.remaining_size() > 0:
self._trigger_error(isotp.errors.BadGeneratorError("Generator depleted before reaching specified size"))
self._stop_sending(success=False)
else:
self._stop_sending(success=True)
elif self.remote_blocksize != 0 and self.tx_block_counter >= self.remote_blocksize:
self.tx_state = self.TxState.WAIT_FC
immediate_rx_msg_required = True
self._start_rx_fc_timer()
else:
pass # We are rate limited. Standby
if output_msg is not None:
self.rate_limiter.inform_byte_sent(len(output_msg.data))
return self.ProcessTxReport(msg=output_msg, immediate_rx_required=immediate_rx_msg_required)
def set_sleep_timing(self, idle: float, wait_fc: float) -> None:
"""
Sets values in seconds that can be passed to ``time.sleep()`` when the stack is processed in a different thread.
:param idle: Time when rx state machine is idle
:type idle: float
:param wait_fc: Time when rx state machine is waiting for a flow control message
:type wait_fc: float
"""
self.timings = {
(self.RxState.IDLE, self.TxState.IDLE): idle,
(self.RxState.IDLE, self.TxState.WAIT_FC): wait_fc,
}
def set_address(self, address: Union[isotp.address.Address, isotp.address.AsymmetricAddress]) -> None:
"""
Sets the layer address. Can be set after initialization if needed. May cause a timeout if called while a transmission is active.
:param address: Address to use
:type address: :class:`Address<isotp.Address>` or :class:`AsymmetricAddress<isotp.AsymmetricAddress>`
"""
if not isinstance(address, (isotp.address.Address, isotp.address.AsymmetricAddress)):
raise ValueError('address must be a valid Address instance')
if address.is_partial_address():
raise ValueError('Cannot use a partially defined address. Either use a fully defined isotp.Address or an isotp.AsymmetricAddress')
self.address = address
txid = self.address.get_tx_arbitration_id(isotp.TargetAddressType.Physical)
rxid = self.address.get_rx_arbitration_id(isotp.TargetAddressType.Physical)
if (txid > 0x7F4 and txid < 0x7F6 or txid > 0x7FA and txid < 0x7FB):
self.logger.warning('Used txid overlaps the range of ID reserved by ISO-15765 (0x7F4-0x7F6 and 0x7FA-0x7FB)')
if (rxid > 0x7F4 and rxid < 0x7F6 or rxid > 0x7FA and rxid < 0x7FB):
self.logger.warning('Used rxid overlaps the range of ID reserved by ISO-15765 (0x7F4-0x7F6 and 0x7FA-0x7FB)')
def _pad_message_data(self, msg_data: bytes) -> bytes:
"""Pad a message if required with the proper padding byte according to the configuration"""
must_pad = False
padding_byte = 0xCC if self.params.tx_padding is None else self.params.tx_padding
if self.params.tx_data_length == 8:
if self.params.tx_data_min_length is None:
if self.params.tx_padding is not None: # ISO-15765:2016 - 10.4.2.1
must_pad = True
target_length = 8
else: # ISO-15765:2016 - 10.4.2.2
pass
else: # issue #27
must_pad = True
target_length = self.params.tx_data_min_length
elif self.params.tx_data_length > 8:
if self.params.tx_data_min_length is None: # ISO-15765:2016 - 10.4.2.3
target_length = self._get_nearest_can_fd_size(len(msg_data))
must_pad = True
else: # Issue #27
must_pad = True
target_length = max(self.params.tx_data_min_length, self._get_nearest_can_fd_size(len(msg_data)))
if must_pad and len(msg_data) < target_length:
return msg_data + bytes([padding_byte & 0xFF] * (target_length - len(msg_data)))
return msg_data
def _empty_rx_buffer(self) -> None:
self.rx_buffer = bytearray()
def _start_rx_fc_timer(self) -> None:
self.timer_rx_fc = Timer(timeout=float(self.params.rx_flowcontrol_timeout) / 1000)
self.timer_rx_fc.start()
def _start_rx_cf_timer(self) -> None:
self.timer_rx_cf = Timer(timeout=float(self.params.rx_consecutive_frame_timeout) / 1000)
self.timer_rx_cf.start()
def _append_rx_data(self, data: Union[bytes, bytearray]) -> None:
self.rx_buffer.extend(data)
def _request_tx_flowcontrol(self, status: int = PDU.FlowStatus.ContinueToSend) -> None:
self.pending_flow_control_tx = True
self.pending_flowcontrol_status = status
def _stop_sending_flow_control(self) -> None:
self.pending_flow_control_tx = False
self.last_flow_control_frame = None
def _make_tx_msg(self, arbitration_id: int, data: bytes) -> CanMessage:
data = self._pad_message_data(data)
return CanMessage(
arbitration_id=arbitration_id,
dlc=self._get_dlc(data, validate_tx=True),
data=data,
extended_id=self.address.is_tx_29bits(),
is_fd=self.params.can_fd,
bitrate_switch=self.params.bitrate_switch
)
def _get_dlc(self, data: bytes, validate_tx: bool = False) -> int:
# DLC cannot be smaller than 2 as per ISO-15765-2. Each messages has a PDU type (SF, FF, CF, FC) + at least one data byte.
fdlen = self._get_nearest_can_fd_size(len(data))
if validate_tx:
if self.params.tx_data_length == 8:
if fdlen < 2 or fdlen > 8:
raise ValueError("Impossible DLC size for payload of %d bytes with tx_data_length of %d" %
(len(data), self.params.tx_data_length))
if fdlen >= 2 and fdlen <= 8: return fdlen
elif fdlen == 12: return 9
elif fdlen == 16: return 10
elif fdlen == 20: return 11
elif fdlen == 24: return 12
elif fdlen == 32: return 13
elif fdlen == 48: return 14
elif fdlen == 64: return 15
raise ValueError("Impossible DLC size for payload of %d bytes with tx_data_length of %d" % (len(data), self.params.tx_data_length))
def _get_nearest_can_fd_size(self, size: int) -> int:
if size <= 8:
return size
if size <= 12: return 12
if size <= 16: return 16
if size <= 20: return 20
if size <= 24: return 24
if size <= 32: return 32
if size <= 48: return 48
if size <= 64: return 64
raise ValueError("Impossible data size for CAN FD : %d " % (size))
def _make_flow_control(self, flow_status: int = PDU.FlowStatus.ContinueToSend, blocksize: Optional[int] = None, stmin: Optional[int] = None) -> CanMessage:
if blocksize is None:
blocksize = self.params.blocksize
if stmin is None:
stmin = self.params.stmin
data = PDU.craft_flow_control_data(flow_status, blocksize, stmin)
return self._make_tx_msg(self.address.get_tx_arbitration_id(), self.address.get_tx_payload_prefix() + data)
def stop_sending(self) -> None:
"""
Request the TransportLayer object to stop transmitting, clear the transmit buffer and put back its transmit state machine to idle state.
"""
self._stop_sending(success=False)
def _stop_sending(self, success: bool) -> None:
if self.active_send_request is not None:
self.active_send_request.complete(success)
self.active_send_request = None
self.tx_state = self.TxState.IDLE
self.tx_frame_length = 0
self.timer_rx_fc.stop()
self.timer_tx_stmin.stop()
self.remote_blocksize = None
self.tx_block_counter = 0
self.tx_seqnum = 0
self.wft_counter = 0
self.tx_standby_msg = None
def stop_receiving(self) -> None:
"""
Request the TransportLayer object to stop receiving, clear the reception buffer and put back its receive state machine to idle state.
If a reception is ongoing, the following messages will be discared and considered like garbage
"""
self._stop_receiving()
def _stop_receiving(self) -> None:
self.actual_rxdl = None
self.rx_state = self.RxState.IDLE
self._empty_rx_buffer()
self._stop_sending_flow_control()
self.timer_rx_cf.stop()
def clear_rx_queue(self) -> None:
while not self.rx_queue.empty():
self.rx_queue.get_nowait()
def clear_tx_queue(self) -> None:
while not self.tx_queue.empty():
self.tx_queue.get_nowait()
# Init the reception of a multi-pdu frame.
def _start_reception_after_first_frame_if_valid(self, pdu: PDU) -> bool:
assert pdu.length is not None
self._empty_rx_buffer()
if pdu.rx_dl not in [8, 12, 16, 20, 24, 32, 48, 64]:
self._trigger_error(isotp.errors.InvalidCanFdFirstFrameRXDL(
"Received a FirstFrame with a RX_DL value of %d which is invalid according to ISO-15765-2" % (pdu.rx_dl)))
self._stop_receiving()
return False
self.actual_rxdl = pdu.rx_dl
started = False
if pdu.length > self.params.max_frame_size:
self._trigger_error(isotp.errors.FrameTooLongError(
"Received a First Frame with a length of %d bytes, but params.max_frame_size is set to %d bytes. Ignoring" % (pdu.length, self.params.max_frame_size)))
self._request_tx_flowcontrol(PDU.FlowStatus.Overflow)
self.rx_state = self.RxState.IDLE
else:
self.rx_state = self.RxState.WAIT_CF
self.rx_frame_length = pdu.length
self._append_rx_data(pdu.data)
self._request_tx_flowcontrol(PDU.FlowStatus.ContinueToSend)
self._start_rx_cf_timer()
started = True
self.last_seqnum = 0
self.rx_block_counter = 0
return started
def _trigger_error(self, error: isotp.errors.IsoTpError) -> None:
if self.error_handler is not None:
if hasattr(self.error_handler, '__call__') and isinstance(error, isotp.errors.IsoTpError):
self.error_handler(error)
else:
self.logger.warning('Given error handler is not a callable object.')
self.logger.warning(str(error))
# Clears everything within the layer.
def reset(self) -> None:
"""
Reset the layer: Empty all buffers, set the internal state machines to Idle
"""
self.clear_rx_queue()
self.clear_tx_queue()
self._stop_sending(success=False)
self._stop_receiving()
self.rate_limiter.reset()
# Gives a time to pass to time.sleep() based on the state of the FSM. Avoid using too much CPU
def sleep_time(self) -> float:
"""
Returns a value in seconds that can be passed to ``time.sleep()`` when the stack is processed in a different thread.
The value will change according to the internal state machine state, sleeping longer while idle and shorter when active.
"""
key = (self.rx_state, self.tx_state)
if key in self.timings:
return self.timings[key]
else:
return 0.001
def is_tx_throttled(self) -> bool:
"""Tells if the transmission is actively being slowed down by the rate limited"""
return self.tx_state in [self.TxState.TRANSMIT_SF_STANDBY, self.TxState.TRANSMIT_FF_STANDBY]
def is_rx_active(self) -> bool:
return self.rx_state != self.RxState.IDLE
def is_tx_transmitting_cf(self) -> bool:
return self.tx_state == self.TxState.TRANSMIT_CF
def next_cf_delay(self) -> Optional[float]:
if not self.is_tx_transmitting_cf():
return None
if self.timer_tx_stmin.is_timed_out():
return 0
return self.timer_tx_stmin.remaining()
# Inheritance of TransportLayerLogic instead of using composition is a design choice to ease backward compatibility at the expense of a more crowded interface.
[docs]class TransportLayer(TransportLayerLogic):
"""
An IsoTP transport layer implementation that runs in a separate thread. The main public interface are ``start``, ``stop``, ``send``, ``recv``.
For backward compatibility, this class will behave similarly as a V1 TransportLayer if start/stop are never called; meaning that ``process()`` can be called by the user
:param rxfn: Function to be called by the transport layer to read the CAN layer. Must return a :class:`isotp.CanMessage<isotp.CanMessage>` or None if no message has been received.
For optimal performance, this function should perform a blocking read that waits on IO
:type rxfn: Callable : expected signature: ``my_rxfn(timeout:float) -> Optional[isotp.CanMessage]``
:param txfn: Function to be called by the transport layer to send a message on the CAN layer. This function should receive a :class:`isotp.CanMessage<isotp.CanMessage>`
:type txfn: Callable : expected signature: ``my_txfn(msg:isotp.CanMessage) -> None``
:param address: The address information of CAN messages. Includes the addressing mode, txid/rxid, source/target address and address extension. See :class:`isotp.Address<isotp.Address>` for more details.
:type address: isotp.Address
:param error_handler: A function to be called when an error has been detected.
An :class:`isotp.IsoTpError<isotp.IsoTpError>` (inheriting Exception class) will be given as sole parameter. See the :ref:`Error section<Errors>`
When started, the error handler will be called from a different thread than the user thread, make sure to consider thread safety if the error handler is more complex than a log.
:type error_handler: Callable
:param params: Dict of parameters for the transport layer. See :ref:`the list of parameters<parameters>`
:type params: dict
:param read_timeout: Default blocking read timeout passed down to the ``rxfn``. Affects only the reading thread time granularity which can affect timing performance.
A value between 20ms-500ms should generally be good. MEaningless if the provided ``rxfn`` ignores its timeout parameter
:type read_timeout: float
"""
class Events:
main_thread_ready: threading.Event
relay_thread_ready: threading.Event
stop_requested: threading.Event
reset_tx: threading.Event
reset_rx: threading.Event
reset_tx_complete: threading.Event
reset_rx_complete: threading.Event
def __init__(self) -> None:
self.main_thread_ready = threading.Event()
self.relay_thread_ready = threading.Event()
self.stop_requested = threading.Event()
self.reset_tx = threading.Event()
self.reset_tx_complete = threading.Event()
self.reset_rx = threading.Event()
self.reset_rx_complete = threading.Event()
started: bool
main_thread: Optional[threading.Thread]
relay_thread: Optional[threading.Thread]
default_read_timeout: float
events: Events
rx_relay_queue: "queue.Queue[Optional[CanMessage]]"
user_rxfn: TransportLayerLogic.RxFn
def __init__(self,
rxfn: TransportLayerLogic.RxFn,
txfn: TransportLayerLogic.TxFn,
address: isotp.Address,
error_handler: Optional[TransportLayerLogic.ErrorHandler] = None,
params: Optional[Dict[str, Any]] = None,
read_timeout: float = 0.05) -> None:
self.rx_relay_queue = queue.Queue()
self.started = False
self.main_thread = None
self.default_read_timeout = read_timeout
self.events = self.Events()
self.user_rxfn = rxfn # Used by the relay thread
def post_send_callback(send_request: TransportLayer.SendRequest) -> None:
# This callback wakeup the main thread from its blocking read if necessary
self.rx_relay_queue.put(None)
TransportLayerLogic.__init__(self, rxfn, txfn, address, error_handler, params, post_send_callback)
def _read_relay_queue(self, timeout: Optional[float]) -> Optional[CanMessage]:
try:
return self.rx_relay_queue.get(block=True, timeout=timeout)
except queue.Empty:
return None
[docs] def start(self) -> None:
"""Starts the IsoTP layer. Starts internal threads that handle the IsoTP communication."""
self.logger.debug(f"Starting {self.__class__.__name__}")
if self.started:
raise RuntimeError("Transport Layer is already started")
self._set_rxfn(self._read_relay_queue)
self.main_thread = threading.Thread(target=self._main_thread_fn, daemon=True)
self.relay_thread = threading.Thread(target=self._relay_thread_fn, daemon=True)
self.events.main_thread_ready.clear()
self.events.relay_thread_ready.clear()
self.events.stop_requested.clear()
self.events.reset_tx.clear()
self.events.reset_rx.clear()
self.events.reset_tx_complete.clear()
self.events.reset_rx_complete.clear()
self.main_thread.start()
self.relay_thread.start()
self.events.main_thread_ready.wait(0.5)
if not self.events.main_thread_ready.is_set():
self.stop()
raise RuntimeError("Failed to start the Transport Layer")
self.events.relay_thread_ready.wait(0.5)
if not self.events.relay_thread_ready.is_set():
self.stop()
raise RuntimeError("Failed to start the Transport Layer")
self.started = True
[docs] def stop(self) -> None:
"""Stops the IsoTP layer. Stops the internal threads that handle the IsoTP communication and reset the layer state."""
self.logger.debug(f"Stopping {self.__class__.__name__}")
self.events.stop_requested.set()
self.rx_relay_queue.put(None)
if self.main_thread is not None:
self.main_thread.join(1.0)
if self.main_thread.is_alive():
self.logger.error("Failed to stop the main thread")
self.main_thread = None
if self.relay_thread is not None:
wait_time = max(self.default_read_timeout + 0.5, 1.0)
self.relay_thread.join(timeout=wait_time)
if self.relay_thread.is_alive():
self.logger.warning("Failed to stop the reading thread. Does your rxfn callback block without respecting the timeout parameter?")
self.relay_thread = None
self.events.main_thread_ready.clear()
self.events.relay_thread_ready.clear()
self.events.stop_requested.clear()
self.events.reset_tx.clear()
self.events.reset_rx.clear()
self.events.reset_tx_complete.clear()
self.events.reset_rx_complete.clear()
super().reset()
while not self.rx_relay_queue.empty():
self.rx_relay_queue.get()
self._set_rxfn(self.user_rxfn) # Switch back to the given user rxfn. Backward compatibility with v1.x
self.started = False
self.logger.debug(f"{self.__class__.__name__} Stopped")
def _relay_thread_fn(self) -> None:
"""Internal function executed by the relay thread. Reads the user rxfn and put any results in a queue."""
self.logger.debug("Relay thread has started")
assert self.user_rxfn is not None
self.events.relay_thread_ready.set()
while not self.events.stop_requested.is_set():
rx_timeout = 0.0 if self.is_tx_throttled() else self.default_read_timeout
t1 = time.perf_counter()
data = self.user_rxfn(rx_timeout)
diff = time.perf_counter() - t1
if data is not None:
self.rx_relay_queue.put(data)
else: # No data received. Sleep if user is not blocking
if not self.events.stop_requested.is_set():
if not self.blocking_rxfn or diff < rx_timeout * 0.5:
time.sleep(max(0, min(self.sleep_time(), rx_timeout - diff)))
def _main_thread_fn(self) -> None:
"""Internal function executed by the main thread. """
self.logger.debug("Main thread has started")
self.events.main_thread_ready.set()
try:
while not self.events.stop_requested.is_set():
if not self.is_rx_active() and self.is_tx_transmitting_cf():
delay = self.next_cf_delay()
assert delay is not None # Confirmed by is_tx_transmitting_cf()
if delay > 0:
self.params.wait_func(delay) # If we are transmitting CFs, no need to call rxfn, we can stream those CF with short sleep
if not self.events.stop_requested.is_set():
super().process(do_rx=False, do_tx=True)
else:
rx_timeout = 0.0 if self.is_tx_throttled() else self.default_read_timeout
super().process(rx_timeout)
if self.events.reset_tx.is_set():
self._stop_sending(success=False)
self.events.reset_tx.clear()
self.events.reset_tx_complete.set()
if self.events.reset_rx.is_set():
self._stop_receiving()
self.events.reset_rx.clear()
self.events.reset_rx_complete.set()
finally:
super().reset()
self.logger.debug("Main thread is exiting")
[docs] @is_documented_by(TransportLayerLogic.stop_sending)
def stop_sending(self) -> None:
if self.started:
if not self.events.stop_requested.is_set():
if self.main_thread is not None and self.main_thread.is_alive():
self.events.reset_tx.set()
self.events.reset_tx_complete.wait(1.0)
if not self.events.reset_tx_complete.is_set():
self.logger.error("Main thread failed to stop sending when requested.")
else:
self._stop_sending(success=False)
[docs] @is_documented_by(TransportLayerLogic.stop_receiving)
def stop_receiving(self) -> None:
if self.started:
if not self.events.stop_requested.is_set():
if self.main_thread is not None and self.main_thread.is_alive():
self.events.reset_rx.set()
self.rx_relay_queue.put(None) # Wakeup from blocking read
self.events.reset_rx_complete.wait(1.0)
if not self.events.reset_rx_complete.is_set():
self.logger.error("Main thread failed to stop receiving when requested.")
else:
self._stop_receiving()
# Protect against usage of non thread-safe methods while threads are running. We don't hide those method for backward compatibility
[docs] @is_documented_by(TransportLayerLogic.process)
def process(self, rx_timeout: float = 0.0, do_rx: bool = True, do_tx: bool = True) -> TransportLayerLogic.ProcessStats:
if self.started:
raise RuntimeError("Cannot call process() after a start(). See documentation and notes about backward compatibility.")
return super().process(rx_timeout=rx_timeout, do_rx=do_rx, do_tx=do_tx)
[docs] @is_documented_by(TransportLayerLogic.reset)
def reset(self) -> None:
if self.started:
raise RuntimeError("Cannot call reset() after a start(). See documentation and notes about backward compatibility.")
super().reset()
class BusOwner:
bus: "can.BusABC"
def python_can_tx_canbus_3plus(owner: BusOwner, msg: CanMessage) -> None:
owner.bus.send(can.Message(arbitration_id=msg.arbitration_id, data=msg.data,
is_extended_id=msg.is_extended_id, is_fd=msg.is_fd, bitrate_switch=msg.bitrate_switch))
def python_can_tx_canbus_3minus(owner: BusOwner, msg: CanMessage) -> None:
owner.bus.send(can.Message(arbitration_id=msg.arbitration_id, data=msg.data, # type:ignore
extended_id=msg.is_extended_id, is_fd=msg.is_fd, bitrate_switch=msg.bitrate_switch)) # type:ignore
def _make_python_can_tx_func(owner: BusOwner) -> Callable[[CanMessage], None]:
message_input_args = inspect.signature(can.Message.__init__).parameters
if 'is_extended_id' in message_input_args:
return functools.partial(python_can_tx_canbus_3plus, owner)
else:
return functools.partial(python_can_tx_canbus_3minus, owner)
def _python_can_to_isotp_message(msg: Optional["can.Message"]) -> Optional[CanMessage]:
if msg is None:
return None
if msg.is_error_frame or msg.is_remote_frame:
return None
return CanMessage(arbitration_id=msg.arbitration_id, data=msg.data, extended_id=msg.is_extended_id, is_fd=msg.is_fd, bitrate_switch=msg.bitrate_switch)
[docs]class CanStack(TransportLayer, BusOwner):
"""
The IsoTP transport layer pre configured to use `python-can <https://python-can.readthedocs.io>`__ as CAN layer. python-can must be installed in order to use this class.
All parameters except the ``bus`` parameter will be given to the :class:`TransportLayer<isotp.TransportLayer>` constructor
This class directly calls ``bus.recv``, consuming the message from the receive queue, potentially starving other application. Consider using the :class:`NotifierBasedCanStack<isotp.NotifierBasedCanStack>`
to avoid starvation issues
:param bus: A python-can bus object implementing ``recv`` and ``send``
:type bus: can.BusABC
:param args: Passed down to :class:`TransportLayer<isotp.TransportLayer>`. ``rxfn`` and ``txfn`` are predefined
:type args: N/A
:param kwargs: Passed down to :class:`TransportLayer<isotp.TransportLayer>`. ``rxfn`` and ``txfn`` are predefined
:type kwargs: N/A
"""
bus: "can.BusABC"
def _rx_canbus(self, timeout: float) -> Optional[CanMessage]:
msg = self.bus.recv(timeout)
return _python_can_to_isotp_message(msg)
def __init__(self, bus: "can.BusABC", *args: Any, **kwargs: Any):
if not _can_available:
raise RuntimeError(f"python-can is not installed in this environment and is required for the {self.__class__.__name__} object.")
self.set_bus(bus)
kwargs.update(dict(
rxfn=self._rx_canbus,
txfn=_make_python_can_tx_func(self),
))
super().__init__(*args, **kwargs)
def set_bus(self, bus: "can.BusABC") -> None:
if not isinstance(bus, can.BusABC):
raise ValueError('bus must be a python-can BusABC object')
self.bus = bus
[docs]class NotifierBasedCanStack(TransportLayer, BusOwner):
"""
The IsoTP transport layer pre configured to use `python-can <https://python-can.readthedocs.io>`__ as CAN layer and reading through a ``can.Notifier``. python-can must be installed in order to use this class.
All parameters except the ``bus`` and the ``notifier`` parameter will be given to the :class:`TransportLayer<isotp.TransportLayer>` constructor
This class reads by registering a listener to the given notifier and sends by calling ``bus.recv``.
:param bus: A python-can Bus object implementing ``send`` used for transmission
:type bus: can.BusABC
:param notifier: A python-can Notifier object onto which a new listener will be added
:type notifier: can.Notifier
:param args: Passed down to :class:`TransportLayer<isotp.TransportLayer>`. ``rxfn`` and ``txfn`` are predefined
:type args: N/A
:param kwargs: Passed down to :class:`TransportLayer<isotp.TransportLayer>`. ``rxfn`` and ``txfn`` are predefined
:type kwargs: N/A
"""
buffered_reader: Optional["can.BufferedReader"]
notifier: "can.Notifier"
bus: "can.BusABC"
def _rx_canbus(self, timeout: float) -> Optional[CanMessage]:
if self.buffered_reader is None:
return None
msg = self.buffered_reader.get_message(timeout=timeout)
return _python_can_to_isotp_message(msg)
def __init__(self, bus: "can.BusABC", notifier: "can.Notifier", *args: Any, **kwargs: Any) -> None:
if not _can_available:
raise RuntimeError(f"python-can is not installed in this environment and is required for the {self.__class__.__name__} object.")
if not isinstance(bus, can.BusABC):
raise ValueError('bus must be a python-can BusABC object')
if not isinstance(notifier, can.Notifier):
raise ValueError("notifier must be a valid can.Notifier object")
self.bus = bus
self.notifier = notifier
self.buffered_reader = None
kwargs.update(dict(
rxfn=self._rx_canbus,
txfn=_make_python_can_tx_func(self),
))
super().__init__(*args, **kwargs)
def start(self) -> None:
self.buffered_reader = can.BufferedReader()
self.notifier.add_listener(self.buffered_reader)
super().start()
def stop(self) -> None:
try:
if self.buffered_reader is not None:
self.notifier.remove_listener(self.buffered_reader)
except Exception:
pass
self.buffered_reader = None
super().stop()