diff --git a/can/interfaces/__init__.py b/can/interfaces/__init__.py index 2f00d0309..e4db379ec 100644 --- a/can/interfaces/__init__.py +++ b/can/interfaces/__init__.py @@ -24,6 +24,7 @@ "canalystii": ("can.interfaces.canalystii", "CANalystIIBus"), "systec": ("can.interfaces.systec", "UcanBus"), "seeedstudio": ("can.interfaces.seeedstudio", "SeeedBus"), + "usbtin": ("can.interfaces.usbtin", "USBtinBus"), } BACKENDS.update( diff --git a/can/interfaces/usbtin.py b/can/interfaces/usbtin.py new file mode 100644 index 000000000..a3db004b3 --- /dev/null +++ b/can/interfaces/usbtin.py @@ -0,0 +1,77 @@ +""" +Interface for the USBtin module + +Implementation references: +* Uses the pyUSBTin API (https://github.com/fishpepper/pyUSBtin) +* Interfaces with the USBtin module (https://www.fischl.de/usbtin/) +""" + +from typing import List, Optional, Tuple, Any + +import logging +import pyusbtin # type: ignore +import can + +logger = logging.getLogger(__name__) + + +class USBtinBus(can.BusABC): + """ + usbtin interface + """ + + def __init__(self, channel: str, baudrate: int): + super().__init__(channel, baudrate=baudrate) + + self.usbtin = pyusbtin.usbtin.USBtin() + self.baudrate = baudrate + self.usbtin.connect(channel) + self.rx_fifo: List[Any] = [] + self.channel_info = "usbtin" + + self.usbtin.open_can_channel(self.baudrate, pyusbtin.usbtin.USBtin.ACTIVE) + self.usbtin.add_message_listener(self._bridge_cb) + + def _bridge_cb(self, msg: Any) -> None: + self.rx_fifo.append(msg) + + def _recv_internal( + self, timeout: Optional[float] + ) -> Tuple[Optional[can.Message], bool]: + if len(self.rx_fifo) == 0: + return None, False + + canmsg = self.rx_fifo.pop(0) + data = [ + canmsg[0], + canmsg[1], + canmsg[2], + canmsg[3], + canmsg[4], + canmsg[5], + canmsg[6], + canmsg[7], + ] + msg = can.Message( + arbitration_id=canmsg.mid, + dlc=canmsg.dlc, + data=data[: canmsg.dlc], + is_remote_frame=canmsg.rtr, + ) + + return msg, False + + def send(self, msg: can.Message, timeout: Optional[float] = None) -> None: + data = list(msg.data) + pymsg = pyusbtin.canmessage.CANMessage( + mid=msg.arbitration_id, dlc=msg.dlc, data=data + ) + self.usbtin.send(pymsg) + + def shutdown(self) -> None: + self.usbtin.close_can_channel() + self.usbtin.disconnect() + + @staticmethod + def _detect_available_configs() -> List[can.typechecking.AutoDetectedConfig]: + raise NotImplementedError() diff --git a/doc/interfaces.rst b/doc/interfaces.rst index bd7a0d1df..1e534a002 100644 --- a/doc/interfaces.rst +++ b/doc/interfaces.rst @@ -27,6 +27,7 @@ The available interfaces are: interfaces/canalystii interfaces/systec interfaces/seeedstudio + interfaces/usbtin Additional interfaces can be added via a plugin interface. An external package can register a new interface by using the ``can.interface`` entry point in its setup.py. diff --git a/doc/interfaces/usbtin.rst b/doc/interfaces/usbtin.rst new file mode 100644 index 000000000..b1b522bca --- /dev/null +++ b/doc/interfaces/usbtin.rst @@ -0,0 +1,12 @@ +USBtin +===== + +Interface for `Thomas Fischl's`_ USBtin device. + + +Bus +--- + +.. autoclass:: can.interfaces.usbtin.USBtinBus + +.. _`Thomas Fischl's`: https://www.fischl.de/usbtin/ diff --git a/setup.py b/setup.py index 8327b1432..6e7e5fb83 100644 --- a/setup.py +++ b/setup.py @@ -109,6 +109,7 @@ "filelock", "mypy_extensions >= 0.4.0, < 0.5.0", 'pywin32;platform_system=="Windows"', + "pyusbtin @ git+https://github.com/fishpepper/pyUSBtin@master#egg=pyusbtin", ], setup_requires=pytest_runner, extras_require=extras_require, diff --git a/test/test_usbtin.py b/test/test_usbtin.py new file mode 100644 index 000000000..d09d88365 --- /dev/null +++ b/test/test_usbtin.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python +# coding: utf-8 + +""" +Test for usbtin interface +""" + +from unittest import TestCase, mock + +import can +import pyusbtin +from can.interfaces.usbtin import USBtinBus + + +class TestUSBtinBus(TestCase): + @mock.patch("can.interfaces.usbtin.pyusbtin.usbtin") + def setUp(self, pyusbtin_usbtin): + # setup the bus + test_port = "test_port" + test_baudrate = 100 + self.bus = bus = can.Bus( + bustype="usbtin", channel=test_port, baudrate=test_baudrate + ) + + # test the bus + self.assertEqual(bus.baudrate, test_baudrate) + bus.usbtin.connect.assert_called_once_with(test_port) + self.assertEqual(bus.channel_info, "usbtin") + bus.usbtin.open_can_channel.assert_called_once_with( + test_baudrate, pyusbtin.usbtin.USBtin.ACTIVE + ) + bus.usbtin.add_message_listener.assert_called_once_with(bus._bridge_cb) + + # setup some test messages + test_id = 1 + test_rtr = False + test_dlc = 2 + test_data = [3, 4] + self.test_can_message = can.Message( + arbitration_id=test_id, + is_remote_frame=test_rtr, + dlc=test_dlc, + data=bytearray(test_data), + ) + self.test_pyusbtin_can_message = pyusbtin.canmessage.CANMessage( + mid=test_id, dlc=test_dlc, data=test_data + ) + + def test_send(self): + self.bus.send(self.test_can_message) + + # TODO: add addTypeEqualityFunc() to make things cleaner + self.assertEqual( + self.test_pyusbtin_can_message.mid, self.bus.usbtin.send.call_args[0][0].mid + ) + self.assertEqual( + self.test_pyusbtin_can_message.dlc, self.bus.usbtin.send.call_args[0][0].dlc + ) + self.assertEqual( + self.test_pyusbtin_can_message.rtr, self.bus.usbtin.send.call_args[0][0].rtr + ) + self.assertEqual( + self.test_pyusbtin_can_message._data, + self.bus.usbtin.send.call_args[0][0]._data, + ) + + def test_shutdown(self): + self.bus.shutdown() + self.bus.usbtin.close_can_channel.assert_called_once() + self.bus.usbtin.disconnect.assert_called_once() + + def test__recv_internal_empty_fifo(self): + self.assertEqual(self.bus._recv_internal(0), (None, False)) + + def test__recv_internal_fifo_with_entries(self): + self.bus.rx_fifo = [self.test_pyusbtin_can_message] + resp = self.bus._recv_internal(0) + + self.assertEqual(resp[0].arbitration_id, self.test_can_message.arbitration_id) + self.assertEqual(resp[0].dlc, self.test_can_message.dlc) + self.assertEqual(resp[0].is_remote_frame, self.test_can_message.is_remote_frame) + self.assertEqual(resp[0].data, self.test_can_message.data) + self.assertEqual(resp[1], False)