WIP: isotp processing according to specs

Dockerfile

@@ -4,7 +4,8 @@ ENV TZ="Europe/Berlin"
 RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
 
 RUN apt -y update
-RUN apt -y install git vim python3 python3-pip pipenv 
+RUN apt -y install git vim python3 python3-pip 
+# pipenv 
 
 RUN apt -y update
 RUN apt -y install libglib2.0-0 libglu1-mesa-dev libxkbcommon-x11-0 build-essential libgl1-mesa-dev libdbus-1-dev libxcb-*

can_explorer/gui/about_dialog.py

@@ -33,8 +33,8 @@ def _connect_signals(self):
 
     def _copy_info_to_clipboard(self):
         info = AboutDialog._get_program_info()
-        build_info = '\n\n'.join([f'{key}: {value}'for key, value in info.items()])
-        logger.info(f'Copy to clipboard {build_info=}')
+        build_info = '\n\n'.join([f'{key}: {value}' for key, value in info.items()])
+        logger.info(f'Copied to clipboard: {build_info=}')
         clipboard = self._app.clipboard()
         clipboard.setText(build_info)
 
@@ -45,6 +45,5 @@ def _get_program_info():
             'version': f'{PROJECT_NAME} ({__version__})',
             'build': f'{PROJECT_PLATFORM}-{__version__}, built on {PROJECT_BUILD_DATE}',
             'runtime': f'Python Runtime version: {sys.version}\nQt Version {QT_VERSION}, PyQt Version: {PYQT_VERSION}',
-            'copyright': f'Copyright @2024-2024 {PROJECT_NAME}'
+            'copyright': f'Copyright @2024-2024 {PROJECT_NAME}',
         }
-

can_explorer/gui/new_connection_dialog.py

@@ -1,9 +1,8 @@
 import logging
 
 from PyQt6.QtCore import pyqtSignal
-from PyQt6.QtWidgets import QDialog, QComboBox, QLineEdit, QCheckBox
+from PyQt6.QtWidgets import QDialog, QComboBox
 from PyQt6.uic import loadUi
-from matplotlib.pyplot import connect
 
 from can_explorer.util.canutils import CanConfiguration
 from can_explorer.util.gui import get_res_path
@@ -41,7 +40,6 @@ def _connect_signals(self):
         pass
 
     def accept(self):
-        QComboBox().currentText()
         can_configuration = CanConfiguration(
             connection_name=self.connection_name_box.text(),
             bitrate=int(self.bitrate_box.currentText(), base=10),

can_explorer/transport/base_protocol.py

@@ -1,4 +1,100 @@
 from abc import ABC
+import asyncio
+import logging
 
-class BaseCanProtocol(ABC):
-    pass
+logger = logging.getLogger(__name__)
+
+
+class BaseCanProtocol(ABC, asyncio.Protocol):
+    pass
+
+
+class CanProtocol(BaseCanProtocol):
+    """Interface for stream protocol.
+
+    The user should implement this interface.  They can inherit from
+    this class but don't need to.  The implementations here do
+    nothing (they don't raise exceptions).
+
+    When the user wants to requests a transport, they pass a protocol
+    factory to a utility function (e.g., EventLoop.create_connection()).
+
+    When the connection is made successfully, connection_made() is
+    called with a suitable transport object.  Then data_received()
+    will be called 0 or more times with data (bytes) received from the
+    transport; finally, connection_lost() will be called exactly once
+    with either an exception object or None as an argument.
+
+    State machine of calls:
+
+      start -> CM [-> DR*] [-> ER?] -> CL -> end
+
+    * CM: connection_made()
+    * DR: data_received()
+    * ER: eof_received()
+    * CL: connection_lost()
+    """
+
+    def data_received(self, data):
+        """Called when some data is received.
+
+        The argument is a bytes object.
+        """
+        logger.info('Data received')
+
+    def eof_received(self):
+        """Called when the other end calls write_eof() or equivalent.
+
+        If this returns a false value (including None), the transport
+        will close itself.  If it returns a true value, closing the
+        transport is up to the protocol.
+        """
+        logger.info('EOF received')
+
+    def connection_made(self, transport):
+        """Called when a connection is made.
+
+        The argument is the transport representing the pipe connection.
+        To receive data, wait for data_received() calls.
+        When the connection is closed, connection_lost() is called.
+        """
+        logger.info('Connection made')
+
+    def connection_lost(self, exc):
+        """Called when the connection is lost or closed.
+
+        The argument is an exception object or None (the latter
+        meaning a regular EOF is received or the connection was
+        aborted or closed).
+        """
+        logger.info('Connection lost')
+
+    def pause_writing(self):
+        """Called when the transport's buffer goes over the high-water mark.
+
+        Pause and resume calls are paired -- pause_writing() is called
+        once when the buffer goes strictly over the high-water mark
+        (even if subsequent writes increases the buffer size even
+        more), and eventually resume_writing() is called once when the
+        buffer size reaches the low-water mark.
+
+        Note that if the buffer size equals the high-water mark,
+        pause_writing() is not called -- it must go strictly over.
+        Conversely, resume_writing() is called when the buffer size is
+        equal or lower than the low-water mark.  These end conditions
+        are important to ensure that things go as expected when either
+        mark is zero.
+
+        NOTE: This is the only Protocol callback that is not called
+        through EventLoop.call_soon() -- if it were, it would have no
+        effect when it's most needed (when the app keeps writing
+        without yielding until pause_writing() is called).
+        """
+        logger.info('Paused writing')
+
+    def resume_writing(self):
+        """Called when the transport's buffer drains below the low-water mark.
+
+        See pause_writing() for details.
+        """
+        logger.info('Resumed writing')

can_explorer/transport/can_connection.py

@@ -1,30 +1,30 @@
 import urllib
-import asyncio
-import logging
-from typing import Union, Tuple, Optional
 import urllib.parse
 from functools import partial
-from can_explorer.transport.canopen import *
-from can_explorer.transport.fdcan import *
 from can_explorer.transport.isocan import *
-from can_explorer.transport.isotp import *
-from can_explorer.transport.j1939 import *
+from can_explorer.transport.isotp.old_transport import *
 from can_explorer.transport.base_protocol import BaseCanProtocol
 
 logger = logging.getLogger(__name__)
 
+
 def connection_for_can(loop: asyncio.AbstractEventLoop, protocol_factory: BaseCanProtocol, can_bus: can.BusABC):
     # protocol = protocol_factory()
     transport = IsoCanTransport(bus=can_bus)
     protocol = transport.get_protocol()
     return protocol, transport
-
-def create_can_connection(loop: asyncio.AbstractEventLoop, protocol_factory: BaseCanProtocol, url: Optional[str], *args, **kwargs):
+
+
+def create_can_connection(
+    loop: asyncio.AbstractEventLoop, protocol_factory: BaseCanProtocol, url: Optional[str], *args, **kwargs
+):
     logger.info(f'Creating can connection with: {args}, {kwargs}')
     parsed_url = urllib.parse.urlparse(url=url)
     bus_instance = partial(can.Bus, *args, **kwargs)
     if parsed_url.scheme == 'socket':
-        transport, protocol =  loop.run_until_complete(loop.create_connection(protocol_factory, parsed_url.hostname, parsed_url.port))
+        transport, protocol = loop.run_until_complete(
+            loop.create_connection(protocol_factory, parsed_url.hostname, parsed_url.port)
+        )
     else:
         transport, protocol = connection_for_can(loop, protocol_factory, bus_instance())
-    return transport, protocol
+    return transport, protocol

can_explorer/transport/can_message.py

@@ -0,0 +1,39 @@
+import logging
+import asyncio
+from dataclasses import dataclass
+from can_explorer.transport.isotp.addressing import AddressInfo
+import can
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(slots=True)
+class CanMessage:
+    arbitration_id: int
+    dlc: int
+    data: bytearray
+    is_extended_id: bool
+    is_fd: bool
+    DLC_MAP = list(range(0, 9))
+
+    @classmethod
+    def from_can(cls, msg: can.Message) -> 'CanMessage':
+        address_info = AddressInfo()
+        return cls(
+            arbitration_id=msg.arbitration_id,
+            dlc=msg.dlc,
+            data=msg.data,
+            is_extended_id=msg.is_fd,
+            is_fd=msg.is_fd,
+        )
+
+    @classmethod
+    def export(cls) -> can.Message:
+        return can.Message()
+
+    def decode_dlc(self, dlc: int) -> int:
+        dlc_map = self.DLC_MAP
+        if self.is_fd:
+            dlc_map = dlc_map + [12, 16, 20, 24, 32, 48, 64]
+        assert dlc < len(dlc_map), f'Given {dlc=} out of range: {dlc_map=}'
+        return dlc_map[dlc]

can_explorer/transport/isotp/addressing.py

@@ -0,0 +1,93 @@
+import logging
+import asyncio
+import enum
+
+from can_explorer.util.validator import is_byte
+
+logger = logging.getLogger(__name__)
+
+
+@enum.unique
+class TargetAddressingType(enum.IntEnum):
+    PHYSICAL = enum.auto()  # 1 to 1 communication
+    FUNCTIONAL = enum.auto()  # 1 to n communication
+
+
+@enum.unique
+class AddressingType(enum.IntEnum):
+    NORMAL = enum.auto()
+    MIXED_EXTENDED = enum.auto()
+
+
+@enum.unique
+class MessageType(enum.IntEnum):
+    """
+    The parameter Mtype shall be used to identify the type
+    and range of address information parameters included
+    in a service call. The intention is that users of
+    this part of the ISO 15665 can extend the range
+    of values by specifying other types and combinations
+    of address information parameters to be used with the
+    network layer specified as part of the ISO 15765.
+    - If Mtype = diagnotics, the Address information
+    N_AI shall consist of the parameters N_SA, N_TA and
+    N_TAtype
+    - If Mtype = remote diagnotics, then the address
+    information N_AI shall consist of the parameters
+    N_SA, N_TA, N_TAtype and N_AE
+    """
+
+    DIAGNOSTIC = enum.auto()
+    REMOTE_DIAGNOSTIC = enum.auto()
+
+
+class AddressInfo:
+    def __init__(
+        self,
+        source_address: int | None,
+        target_address: int | None,
+        address_extension: int | None,
+        target_address_type: TargetAddressingType,
+        btr: bool = False,
+        max_payload_length: int = 8,
+        is_fd: bool = True,
+        is_extended: bool = False,
+        addressing_type: AddressingType = AddressingType.NORMAL,
+    ):
+        """
+        :param source_address:
+        :param target_address:
+        :param address_extension:
+        :param target_address_type:
+        :param btr: BRS bit which is part of a CAN FD frame
+        and used to determine if the data phase is to be
+        transmitted at a different bit rate than the arbi-
+        tration phase. The bitrate of the data phase is
+        defined to be equal or higher than the arbitration
+        bitrate. Bitrate switching does not influence the
+        transport protocol itself. See ISO-15765-2-2016
+        Page 6
+        :param max_payload_length: The maximum allowed
+        payload length (CAN_DL, 8...64 bytes) ISO-15765-2-2016
+        Page 6
+        """
+        self.source_address = source_address
+        self.target_address = target_address
+        self.address_extension = address_extension
+        self.target_address_type = target_address_type
+        self.btr = btr
+        self.maximum_payload_length = max_payload_length
+        self.is_fd = is_fd
+        self.is_extended = is_extended
+        self.addressing_type = addressing_type
+        is_byte.validate(source_address)
+        is_byte.validate(target_address)
+        is_byte.validate(address_extension)
+
+    @property
+    def is_normal_addressing(self) -> bool:
+        return self.addressing_type == AddressingType.NORMAL
+
+    @property
+    def arbitration_id(self) -> int:
+        return self.target_address << 4 | self.source_address

can_explorer/transport/isotp/errors.py

@@ -0,0 +1,34 @@
+import logging
+import enum
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+@enum.unique
+class NResult(enum.IntEnum):
+    N_OK = 0x00
+    N_TIMEOUT_A = 0x01
+    N_TIMEOUT_Bs = 0x02
+    N_TIMEOUT_Cr = 0x03
+    N_WRONG_SN = 0x04
+    N_INVALID_FS = 0x05
+    N_UNEXP_PDU = 0x06
+    N_WFT_OVRN = 0x07
+    N_BUFFER_OVFLW = 0x08
+    N_ERROR = 0x09
+
+
+class IsoTpError(Exception):
+    def __init__(self, result: NResult, msg: Optional[str]):
+        self._result = result
+        self._msg = msg
+
+
+@enum.unique
+class IsoTpWarning(enum.IntEnum):
+    pass
+
+
+class IsoTpException(Exception):
+    pass