16362801995

Committed 18 Jul 2025 05:17AM UTC coverage: 70.862% (+0.1%) from 70.763%

Build # 16362801995

Build Type

Pull #1920

github

Committed by

web-flow

Commit Message

Merge f9e8a3c29 into 958fc64ed

Pull Request Pull Request #1920: add FD support to slcan according to CANable 2.0 impementation

Run Details

6 of 45 new or added lines in 1 file covered. (13.33%)

838 existing lines in 35 files now uncovered.

7770 of 10965 relevant lines covered (70.86%)

13.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.05

/can/interfaces/ixxat/canlib.py

from collections.abc import Sequence
from typing import Callable, Optional, Union

import can.interfaces.ixxat.canlib_vcinpl as vcinpl
import can.interfaces.ixxat.canlib_vcinpl2 as vcinpl2
from can import (
    BusABC,
    BusState,
    CyclicSendTaskABC,
    Message,
)
from can.typechecking import AutoDetectedConfig


class IXXATBus(BusABC):
    """The CAN Bus implemented for the IXXAT interface.

    Based on the C implementation of IXXAT, two different dlls are provided by IXXAT, one to work with CAN,
    the other with CAN-FD.

    This class only delegates to related implementation (in calib_vcinpl or canlib_vcinpl2)
    class depending on fd user option.
    """

    def __init__(
        self,
        channel: int,
        can_filters=None,
        receive_own_messages: bool = False,
        unique_hardware_id: Optional[int] = None,
        extended: bool = True,
        fd: bool = False,
        rx_fifo_size: Optional[int] = None,
        tx_fifo_size: Optional[int] = None,
        bitrate: int = 500000,
        data_bitrate: int = 2000000,
        sjw_abr: Optional[int] = None,
        tseg1_abr: Optional[int] = None,
        tseg2_abr: Optional[int] = None,
        sjw_dbr: Optional[int] = None,
        tseg1_dbr: Optional[int] = None,
        tseg2_dbr: Optional[int] = None,
        ssp_dbr: Optional[int] = None,
        **kwargs,
    ):
        """
        :param channel:
            The Channel id to create this bus with.

        :param can_filters:
            See :meth:`can.BusABC.set_filters`.

        :param receive_own_messages:
            Enable self-reception of sent messages.

        :param unique_hardware_id:
            UniqueHardwareId to connect (optional, will use the first found if not supplied)

        :param extended:
            Default True, enables the capability to use extended IDs.

        :param fd:
            Default False, enables CAN-FD usage.

        :param rx_fifo_size:
            Receive fifo size (default 1024 for fd, else 16)

        :param tx_fifo_size:
            Transmit fifo size (default 128 for fd, else 16)

        :param bitrate:
            Channel bitrate in bit/s

        :param data_bitrate:
            Channel bitrate in bit/s (only in CAN-Fd if baudrate switch enabled).

        :param sjw_abr:
            Bus timing value sample jump width (arbitration). Only takes effect with fd enabled.

        :param tseg1_abr:
            Bus timing value tseg1 (arbitration). Only takes effect with fd enabled.

        :param tseg2_abr:
            Bus timing value tseg2 (arbitration). Only takes effect with fd enabled.

        :param sjw_dbr:
            Bus timing value sample jump width (data). Only takes effect with fd and baudrate switch enabled.

        :param tseg1_dbr:
            Bus timing value tseg1 (data). Only takes effect with fd and bitrate switch enabled.

        :param tseg2_dbr:
            Bus timing value tseg2 (data). Only takes effect with fd and bitrate switch enabled.

        :param ssp_dbr:
            Secondary sample point (data). Only takes effect with fd and bitrate switch enabled.

        """
        if fd:
            if rx_fifo_size is None:
                rx_fifo_size = 1024
            if tx_fifo_size is None:
                tx_fifo_size = 128
            self.bus = vcinpl2.IXXATBus(
                channel=channel,
                can_filters=can_filters,
                receive_own_messages=receive_own_messages,
                unique_hardware_id=unique_hardware_id,
                extended=extended,
                rx_fifo_size=rx_fifo_size,
                tx_fifo_size=tx_fifo_size,
                bitrate=bitrate,
                data_bitrate=data_bitrate,
                sjw_abr=sjw_abr,
                tseg1_abr=tseg1_abr,
                tseg2_abr=tseg2_abr,
                sjw_dbr=sjw_dbr,
                tseg1_dbr=tseg1_dbr,
                tseg2_dbr=tseg2_dbr,
                ssp_dbr=ssp_dbr,
                **kwargs,
            )
        else:
            if rx_fifo_size is None:
                rx_fifo_size = 16
            if tx_fifo_size is None:
                tx_fifo_size = 16
            self.bus = vcinpl.IXXATBus(
                channel=channel,
                can_filters=can_filters,
                receive_own_messages=receive_own_messages,
                unique_hardware_id=unique_hardware_id,
                extended=extended,
                rx_fifo_size=rx_fifo_size,
                tx_fifo_size=tx_fifo_size,
                bitrate=bitrate,
                **kwargs,
            )

        super().__init__(channel=channel, **kwargs)
        self._can_protocol = self.bus.protocol

    def flush_tx_buffer(self):
        """Flushes the transmit buffer on the IXXAT"""
        return self.bus.flush_tx_buffer()

    def _recv_internal(self, timeout):
        """Read a message from IXXAT device."""
        return self.bus._recv_internal(timeout)

    def send(self, msg: Message, timeout: Optional[float] = None) -> None:
        return self.bus.send(msg, timeout)

    def _send_periodic_internal(
        self,
        msgs: Union[Sequence[Message], Message],
        period: float,
        duration: Optional[float] = None,
        autostart: bool = True,
        modifier_callback: Optional[Callable[[Message], None]] = None,
    ) -> CyclicSendTaskABC:
        return self.bus._send_periodic_internal(
            msgs, period, duration, autostart, modifier_callback
        )

    def shutdown(self) -> None:
        super().shutdown()
        self.bus.shutdown()

    @property
    def state(self) -> BusState:
        """
        Return the current state of the hardware
        """
        return self.bus.state

    @staticmethod
    def _detect_available_configs() -> list[AutoDetectedConfig]:
        return vcinpl._detect_available_configs()

1	from collections.abc import Sequence	21✔
2	from typing import Callable, Optional, Union	21✔
3
4	import can.interfaces.ixxat.canlib_vcinpl as vcinpl	21✔
5	import can.interfaces.ixxat.canlib_vcinpl2 as vcinpl2	7✔
6	from can import (	7✔
7	BusABC,
8	BusState,
9	CyclicSendTaskABC,
10	Message,
11	)
12	from can.typechecking import AutoDetectedConfig	7✔
13
14
15	class IXXATBus(BusABC):	7✔
16	"""The CAN Bus implemented for the IXXAT interface.	7✔
17
18	Based on the C implementation of IXXAT, two different dlls are provided by IXXAT, one to work with CAN,
19	the other with CAN-FD.
20
21	This class only delegates to related implementation (in calib_vcinpl or canlib_vcinpl2)
22	class depending on fd user option.
23	"""
24
25	def __init__(	7✔
26	self,
27	channel: int,
28	can_filters=None,
29	receive_own_messages: bool = False,
30	unique_hardware_id: Optional[int] = None,
31	extended: bool = True,
32	fd: bool = False,
33	rx_fifo_size: Optional[int] = None,
34	tx_fifo_size: Optional[int] = None,
35	bitrate: int = 500000,
36	data_bitrate: int = 2000000,
37	sjw_abr: Optional[int] = None,
38	tseg1_abr: Optional[int] = None,
39	tseg2_abr: Optional[int] = None,
40	sjw_dbr: Optional[int] = None,
41	tseg1_dbr: Optional[int] = None,
42	tseg2_dbr: Optional[int] = None,
43	ssp_dbr: Optional[int] = None,
44	**kwargs,
45	):
46	"""
47	:param channel:
48	The Channel id to create this bus with.
49
50	:param can_filters:
51	See :meth:`can.BusABC.set_filters`.
52
53	:param receive_own_messages:
54	Enable self-reception of sent messages.
55
56	:param unique_hardware_id:
57	UniqueHardwareId to connect (optional, will use the first found if not supplied)
58
59	:param extended:
60	Default True, enables the capability to use extended IDs.
61
62	:param fd:
63	Default False, enables CAN-FD usage.
64
65	:param rx_fifo_size:
66	Receive fifo size (default 1024 for fd, else 16)
67
68	:param tx_fifo_size:
69	Transmit fifo size (default 128 for fd, else 16)
70
71	:param bitrate:
72	Channel bitrate in bit/s
73
74	:param data_bitrate:
75	Channel bitrate in bit/s (only in CAN-Fd if baudrate switch enabled).
76
77	:param sjw_abr:
78	Bus timing value sample jump width (arbitration). Only takes effect with fd enabled.
79
80	:param tseg1_abr:
81	Bus timing value tseg1 (arbitration). Only takes effect with fd enabled.
82
83	:param tseg2_abr:
84	Bus timing value tseg2 (arbitration). Only takes effect with fd enabled.
85
86	:param sjw_dbr:
87	Bus timing value sample jump width (data). Only takes effect with fd and baudrate switch enabled.
88
89	:param tseg1_dbr:
90	Bus timing value tseg1 (data). Only takes effect with fd and bitrate switch enabled.
91
92	:param tseg2_dbr:
93	Bus timing value tseg2 (data). Only takes effect with fd and bitrate switch enabled.
94
95	:param ssp_dbr:
96	Secondary sample point (data). Only takes effect with fd and bitrate switch enabled.
97
98	"""
99	if fd:	7✔
100	if rx_fifo_size is None:	7✔
101	rx_fifo_size = 1024	7✔
102	if tx_fifo_size is None:	7✔
103	tx_fifo_size = 128	7✔
104	self.bus = vcinpl2.IXXATBus(	7✔
105	channel=channel,
106	can_filters=can_filters,
107	receive_own_messages=receive_own_messages,
108	unique_hardware_id=unique_hardware_id,
109	extended=extended,
110	rx_fifo_size=rx_fifo_size,
111	tx_fifo_size=tx_fifo_size,
112	bitrate=bitrate,
113	data_bitrate=data_bitrate,
114	sjw_abr=sjw_abr,
115	tseg1_abr=tseg1_abr,
116	tseg2_abr=tseg2_abr,
117	sjw_dbr=sjw_dbr,
118	tseg1_dbr=tseg1_dbr,
119	tseg2_dbr=tseg2_dbr,
120	ssp_dbr=ssp_dbr,
121	**kwargs,
122	)
123	else:
124	if rx_fifo_size is None:	7✔
125	rx_fifo_size = 16	7✔
126	if tx_fifo_size is None:	7✔
127	tx_fifo_size = 16	7✔
128	self.bus = vcinpl.IXXATBus(	7✔
129	channel=channel,
130	can_filters=can_filters,
131	receive_own_messages=receive_own_messages,
132	unique_hardware_id=unique_hardware_id,
133	extended=extended,
134	rx_fifo_size=rx_fifo_size,
135	tx_fifo_size=tx_fifo_size,
136	bitrate=bitrate,
137	**kwargs,
138	)
139
140	super().__init__(channel=channel, **kwargs)	×
UNCOV 141	self._can_protocol = self.bus.protocol	×
142
143	def flush_tx_buffer(self):	7✔
144	"""Flushes the transmit buffer on the IXXAT"""
UNCOV 145	return self.bus.flush_tx_buffer()	×
146
147	def _recv_internal(self, timeout):	7✔
148	"""Read a message from IXXAT device."""
UNCOV 149	return self.bus._recv_internal(timeout)	×
150
151	def send(self, msg: Message, timeout: Optional[float] = None) -> None:	7✔
UNCOV 152	return self.bus.send(msg, timeout)	×
153
154	def _send_periodic_internal(	7✔
155	self,
156	msgs: Union[Sequence[Message], Message],
157	period: float,
158	duration: Optional[float] = None,
159	autostart: bool = True,
160	modifier_callback: Optional[Callable[[Message], None]] = None,
161	) -> CyclicSendTaskABC:
UNCOV 162	return self.bus._send_periodic_internal(	×
163	msgs, period, duration, autostart, modifier_callback
164	)
165
166	def shutdown(self) -> None:	7✔
167	super().shutdown()	7✔
168	self.bus.shutdown()	7✔
169
170	@property	7✔
171	def state(self) -> BusState:	7✔
172	"""
173	Return the current state of the hardware
174	"""
UNCOV 175	return self.bus.state	×
176
177	@staticmethod	7✔
178	def _detect_available_configs() -> list[AutoDetectedConfig]:	7✔
179	return vcinpl._detect_available_configs()	7✔

hardbyte / python-can / 16362801995

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous