"""
Interface for slcan compatible interfaces (win32/linux).
"""
from typing import Any, Optional, Tuple
import io
import time
import logging
from can import BusABC, Message
from ..exceptions import (
CanInterfaceNotImplementedError,
CanInitializationError,
CanOperationError,
error_check,
)
from can import typechecking
logger = logging.getLogger(__name__)
try:
import serial
except ImportError:
logger.warning(
"You won't be able to use the slcan can backend without "
"the serial module installed!"
)
serial = None
[docs]class slcanBus(BusABC):
"""
slcan interface
"""
# the supported bitrates and their commands
_BITRATES = {
10000: "S0",
20000: "S1",
50000: "S2",
100000: "S3",
125000: "S4",
250000: "S5",
500000: "S6",
750000: "S7",
1000000: "S8",
83300: "S9",
}
_SLEEP_AFTER_SERIAL_OPEN = 2 # in seconds
_OK = b"\r"
_ERROR = b"\a"
LINE_TERMINATOR = b"\r"
def __init__(
self,
channel: typechecking.ChannelStr,
ttyBaudrate: int = 115200,
bitrate: Optional[int] = None,
btr: Optional[str] = None,
sleep_after_open: float = _SLEEP_AFTER_SERIAL_OPEN,
rtscts: bool = False,
**kwargs: Any,
) -> None:
"""
:param str channel:
port of underlying serial or usb device (e.g. ``/dev/ttyUSB0``, ``COM8``, ...)
Must not be empty. Can also end with ``@115200`` (or similarly) to specify the baudrate.
:param int ttyBaudrate:
baudrate of underlying serial or usb device (Ignored if set via the ``channel`` parameter)
:param bitrate:
Bitrate in bit/s
:param btr:
BTR register value to set custom can speed
:param poll_interval:
Poll interval in seconds when reading messages
:param sleep_after_open:
Time to wait in seconds after opening serial connection
:param rtscts:
turn hardware handshake (RTS/CTS) on and off
:raise ValueError: if both ``bitrate`` and ``btr`` are set or the channel is invalid
:raise CanInterfaceNotImplementedError: if the serial module is missing
:raise CanInitializationError: if the underlying serial connection could not be established
"""
if serial is None:
raise CanInterfaceNotImplementedError("The serial module is not installed")
if not channel: # if None or empty
raise ValueError("Must specify a serial port.")
if "@" in channel:
(channel, baudrate) = channel.split("@")
ttyBaudrate = int(baudrate)
with error_check(exception_type=CanInitializationError):
self.serialPortOrig = serial.serial_for_url(
channel, baudrate=ttyBaudrate, rtscts=rtscts
)
self._buffer = bytearray()
time.sleep(sleep_after_open)
with error_check(exception_type=CanInitializationError):
if bitrate is not None and btr is not None:
raise ValueError("Bitrate and btr mutually exclusive.")
if bitrate is not None:
self.set_bitrate(bitrate)
if btr is not None:
self.set_bitrate_reg(btr)
self.open()
super().__init__(
channel, ttyBaudrate=115200, bitrate=None, rtscts=False, **kwargs
)
[docs] def set_bitrate(self, bitrate: int) -> None:
"""
:param bitrate:
Bitrate in bit/s
:raise ValueError: if ``bitrate`` is not among the possible values
"""
if bitrate in self._BITRATES:
bitrate_code = self._BITRATES[bitrate]
else:
bitrates = ", ".join(str(k) for k in self._BITRATES.keys())
raise ValueError(f"Invalid bitrate, choose one of {bitrates}.")
self.close()
self._write(bitrate_code)
self.open()
[docs] def set_bitrate_reg(self, btr: str) -> None:
"""
:param btr:
BTR register value to set custom can speed
"""
self.close()
self._write("s" + btr)
self.open()
def _write(self, string: str) -> None:
with error_check("Could not write to serial device"):
self.serialPortOrig.write(string.encode() + self.LINE_TERMINATOR)
self.serialPortOrig.flush()
def _read(self, timeout: Optional[float]) -> Optional[str]:
with error_check("Could not read from serial device"):
# first read what is already in receive buffer
while self.serialPortOrig.in_waiting:
self._buffer += self.serialPortOrig.read()
# if we still don't have a complete message, do a blocking read
start = time.time()
time_left = timeout
while not (
ord(self._OK) in self._buffer or ord(self._ERROR) in self._buffer
):
self.serialPortOrig.timeout = time_left
byte = self.serialPortOrig.read()
if byte:
self._buffer += byte
# if timeout is None, try indefinitely
if timeout is None:
continue
# try next one only if there still is time, and with
# reduced timeout
else:
time_left = timeout - (time.time() - start)
if time_left > 0:
continue
else:
return None
# return first message
for i in range(len(self._buffer)):
if self._buffer[i] == ord(self._OK) or self._buffer[i] == ord(self._ERROR):
string = self._buffer[: i + 1].decode()
del self._buffer[: i + 1]
break
return string
def flush(self) -> None:
del self._buffer[:]
with error_check("Could not flush"):
while self.serialPortOrig.in_waiting:
self.serialPortOrig.read()
def open(self) -> None:
self._write("O")
def close(self) -> None:
self._write("C")
def _recv_internal(
self, timeout: Optional[float]
) -> Tuple[Optional[Message], bool]:
canId = None
remote = False
extended = False
frame = []
string = self._read(timeout)
if not string:
pass
elif string[0] == "T":
# extended frame
canId = int(string[1:9], 16)
dlc = int(string[9])
extended = True
for i in range(0, dlc):
frame.append(int(string[10 + i * 2 : 12 + i * 2], 16))
elif string[0] == "t":
# normal frame
canId = int(string[1:4], 16)
dlc = int(string[4])
for i in range(0, dlc):
frame.append(int(string[5 + i * 2 : 7 + i * 2], 16))
elif string[0] == "r":
# remote frame
canId = int(string[1:4], 16)
dlc = int(string[4])
remote = True
elif string[0] == "R":
# remote extended frame
canId = int(string[1:9], 16)
dlc = int(string[9])
extended = True
remote = True
if canId is not None:
msg = Message(
arbitration_id=canId,
is_extended_id=extended,
timestamp=time.time(), # Better than nothing...
is_remote_frame=remote,
dlc=dlc,
data=frame,
)
return msg, False
return None, False
[docs] def send(self, msg: Message, timeout: Optional[float] = None) -> None:
if timeout != self.serialPortOrig.write_timeout:
self.serialPortOrig.write_timeout = timeout
if msg.is_remote_frame:
if msg.is_extended_id:
sendStr = "R%08X%d" % (msg.arbitration_id, msg.dlc)
else:
sendStr = "r%03X%d" % (msg.arbitration_id, msg.dlc)
else:
if msg.is_extended_id:
sendStr = "T%08X%d" % (msg.arbitration_id, msg.dlc)
else:
sendStr = "t%03X%d" % (msg.arbitration_id, msg.dlc)
sendStr += "".join(["%02X" % b for b in msg.data])
self._write(sendStr)
[docs] def shutdown(self) -> None:
super().shutdown()
self.close()
with error_check("Could not close serial socket"):
self.serialPortOrig.close()
def fileno(self) -> int:
try:
return self.serialPortOrig.fileno()
except io.UnsupportedOperation:
raise NotImplementedError(
"fileno is not implemented using current CAN bus on this platform"
)
except Exception as exception:
raise CanOperationError("Cannot fetch fileno") from exception
[docs] def get_version(
self, timeout: Optional[float]
) -> Tuple[Optional[int], Optional[int]]:
"""Get HW and SW version of the slcan interface.
:param timeout:
seconds to wait for version or None to wait indefinitely
:returns: tuple (hw_version, sw_version)
WHERE
int hw_version is the hardware version or None on timeout
int sw_version is the software version or None on timeout
"""
cmd = "V"
self._write(cmd)
start = time.time()
time_left = timeout
while True:
string = self._read(time_left)
if not string:
pass
elif string[0] == cmd and len(string) == 6:
# convert ASCII coded version
hw_version = int(string[1:3])
sw_version = int(string[3:5])
return hw_version, sw_version
# if timeout is None, try indefinitely
if timeout is None:
continue
# try next one only if there still is time, and with
# reduced timeout
else:
time_left = timeout - (time.time() - start)
if time_left > 0:
continue
else:
return None, None
[docs] def get_serial_number(self, timeout: Optional[float]) -> Optional[str]:
"""Get serial number of the slcan interface.
:param timeout:
seconds to wait for serial number or ``None`` to wait indefinitely
:return:
``None`` on timeout or a :class:`~builtin.str` object.
"""
cmd = "N"
self._write(cmd)
start = time.time()
time_left = timeout
while True:
string = self._read(time_left)
if not string:
pass
elif string[0] == cmd and len(string) == 6:
serial_number = string[1:-1]
return serial_number
# if timeout is None, try indefinitely
if timeout is None:
continue
# try next one only if there still is time, and with
# reduced timeout
else:
time_left = timeout - (time.time() - start)
if time_left > 0:
continue
else:
return None