"""
Interface for slcan compatible interfaces (win32/linux).
"""

import io
import logging
import time
import warnings
from queue import SimpleQueue
from typing import Any, Optional, Union, cast

from can import BitTiming, BitTimingFd, BusABC, CanProtocol, Message, typechecking
from can.exceptions import (
    CanInitializationError,
    CanInterfaceNotImplementedError,
    CanOperationError,
    error_check,
)
from can.util import (
    CAN_FD_DLC,
    check_or_adjust_timing_clock,
    deprecated_args_alias,
    len2dlc,
)

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


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",
    }
    _DATA_BITRATES = {
        0: "",
        2000000: "Y2",
        5000000: "Y5",
    }

    _SLEEP_AFTER_SERIAL_OPEN = 2  # in seconds

    _OK = b"\r"
    _ERROR = b"\a"

    LINE_TERMINATOR = b"\r"

    @deprecated_args_alias(
        deprecation_start="4.5.0",
        deprecation_end="5.0.0",
        ttyBaudrate="tty_baudrate",
    )
    def __init__(
        self,
        channel: typechecking.ChannelStr,
        tty_baudrate: int = 115200,
        bitrate: Optional[int] = None,
        timing: Optional[Union[BitTiming, BitTimingFd]] = None,
        sleep_after_open: float = _SLEEP_AFTER_SERIAL_OPEN,
        rtscts: bool = False,
        listen_only: bool = False,
        timeout: float = 0.001,
        **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 tty_baudrate:
            baudrate of underlying serial or usb device (Ignored if set via the ``channel`` parameter)
        :param bitrate:
            Bitrate in bit/s
        :param timing:
            Optional :class:`~can.BitTiming` instance to use for custom bit timing setting.
            If this argument is set then it overrides the bitrate and btr arguments. The
            `f_clock` value of the timing instance must be set to 8_000_000 (8MHz)
            for standard CAN.
            CAN FD and the :class:`~can.BitTimingFd` class have partial support according to the non-standard
            slcan protocol implementation in the CANABLE 2.0 firmware: currently only data rates of 2M and 5M.
        :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
        :param listen_only:
            If True, open interface/channel in listen mode with ``L`` command.
            Otherwise, the (default) ``O`` command is still used. See ``open`` method.
        :param timeout:
            Timeout for the serial or usb device in seconds (default 0.001)

        :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
        """
        self._listen_only = listen_only

        if serial is None:
            raise CanInterfaceNotImplementedError("The serial module is not installed")

        btr: Optional[str] = kwargs.get("btr", None)
        if btr is not None:
            warnings.warn(
                "The 'btr' argument is deprecated since python-can v4.5.0 "
                "and scheduled for removal in v5.0.0. "
                "Use the 'timing' argument instead.",
                DeprecationWarning,
                stacklevel=1,
            )

        if not channel:  # if None or empty
            raise ValueError("Must specify a serial port.")
        if "@" in channel:
            (channel, baudrate) = channel.split("@")
            tty_baudrate = int(baudrate)

        with error_check(exception_type=CanInitializationError):
            self.serialPortOrig = serial.serial_for_url(
                channel,
                baudrate=tty_baudrate,
                rtscts=rtscts,
                timeout=timeout,
            )

        self._queue: SimpleQueue[str] = SimpleQueue()
        self._buffer = bytearray()
        self._can_protocol = CanProtocol.CAN_20

        time.sleep(sleep_after_open)

        with error_check(exception_type=CanInitializationError):
            if isinstance(timing, BitTiming):
                timing = check_or_adjust_timing_clock(timing, valid_clocks=[8_000_000])
                self.set_bitrate_reg(f"{timing.btr0:02X}{timing.btr1:02X}")
            elif isinstance(timing, BitTimingFd):
                self.set_bitrate(timing.nom_bitrate, timing.data_bitrate)
            else:
                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, **kwargs)

    def set_bitrate(self, bitrate: int, data_bitrate: Optional[int] = None) -> None:
        """
        :param bitrate:
            Bitrate in bit/s
        :param data_bitrate:
            Data Bitrate in bit/s for FD frames

        :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}.")

        # If data_bitrate is None, we set it to 0 which means no data bitrate
        if data_bitrate is None:
            data_bitrate = 0

        if data_bitrate in self._DATA_BITRATES:
            dbitrate_code = self._DATA_BITRATES[data_bitrate]
        else:
            dbitrates = ", ".join(str(k) for k in self._DATA_BITRATES.keys())
            raise ValueError(f"Invalid data bitrate, choose one of {dbitrates}.")

        self.close()
        self._write(bitrate_code)
        self._write(dbitrate_code)
        self.open()

    def set_bitrate_reg(self, btr: str) -> None:
        """
        :param btr:
            BTR register value to set custom can speed as a string `xxyy` where
            xx is the BTR0 value in hex and yy is the BTR1 value in hex.
        """
        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]:
        _timeout = serial.Timeout(timeout)

        with error_check("Could not read from serial device"):
            while True:
                # Due to accessing `serialPortOrig.in_waiting` too often will reduce the performance.
                # We read the `serialPortOrig.in_waiting` only once here.
                in_waiting = self.serialPortOrig.in_waiting
                for _ in range(max(1, in_waiting)):
                    new_byte = self.serialPortOrig.read(1)
                    if new_byte:
                        self._buffer.extend(new_byte)
                    else:
                        break

                    if new_byte in (self._ERROR, self._OK):
                        string = self._buffer.decode()
                        self._buffer.clear()
                        return string

                if _timeout.expired():
                    break

            return None

    def flush(self) -> None:
        self._buffer.clear()
        with error_check("Could not flush"):
            self.serialPortOrig.reset_input_buffer()

    def open(self) -> None:
        if self._listen_only:
            self._write("L")
        else:
            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
        data = None
        isFd = False
        fdBrs = False

        if self._queue.qsize():
            string: Optional[str] = self._queue.get_nowait()
        else:
            string = self._read(timeout)

        if not string:
            pass
        elif string[0] in (
            "T",
            "x",  # x is an alternative extended message identifier for CANDapter
        ):
            # extended frame
            canId = int(string[1:9], 16)
            dlc = int(string[9])
            extended = True
            data = bytearray.fromhex(string[10 : 10 + dlc * 2])
        elif string[0] == "t":
            # normal frame
            canId = int(string[1:4], 16)
            dlc = int(string[4])
            data = bytearray.fromhex(string[5 : 5 + dlc * 2])
        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
        elif string[0] == "d":
            # FD standard frame
            canId = int(string[1:4], 16)
            dlc = int(string[4], 16)
            isFd = True
            data = bytearray.fromhex(string[5 : 5 + CAN_FD_DLC[dlc] * 2])
        elif string[0] == "D":
            # FD extended frame
            canId = int(string[1:9], 16)
            dlc = int(string[9], 16)
            extended = True
            isFd = True
            data = bytearray.fromhex(string[10 : 10 + CAN_FD_DLC[dlc] * 2])
        elif string[0] == "b":
            # FD with bitrate switch
            canId = int(string[1:4], 16)
            dlc = int(string[4], 16)
            isFd = True
            fdBrs = True
            data = bytearray.fromhex(string[5 : 5 + CAN_FD_DLC[dlc] * 2])
        elif string[0] == "B":
            # FD extended with bitrate switch
            canId = int(string[1:9], 16)
            dlc = int(string[9], 16)
            extended = True
            isFd = True
            fdBrs = True
            data = bytearray.fromhex(string[10 : 10 + CAN_FD_DLC[dlc] * 2])

        if canId is not None:
            msg = Message(
                arbitration_id=canId,
                is_extended_id=extended,
                timestamp=time.time(),  # Better than nothing...
                is_remote_frame=remote,
                is_fd=isFd,
                bitrate_switch=fdBrs,
                dlc=CAN_FD_DLC[dlc],
                data=data,
            )
            return msg, False
        return None, False

    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 = f"R{msg.arbitration_id:08X}{msg.dlc:d}"
            else:
                sendStr = f"r{msg.arbitration_id:03X}{msg.dlc:d}"
        elif msg.is_fd:
            fd_dlc = len2dlc(msg.dlc)
            if msg.bitrate_switch:
                if msg.is_extended_id:
                    sendStr = f"B{msg.arbitration_id:08X}{fd_dlc:X}"
                else:
                    sendStr = f"b{msg.arbitration_id:03X}{fd_dlc:X}"
                sendStr += msg.data.hex().upper()
            else:
                if msg.is_extended_id:
                    sendStr = f"D{msg.arbitration_id:08X}{fd_dlc:X}"
                else:
                    sendStr = f"d{msg.arbitration_id:03X}{fd_dlc:X}"
                sendStr += msg.data.hex().upper()
        else:
            if msg.is_extended_id:
                sendStr = f"T{msg.arbitration_id:08X}{msg.dlc:d}"
            else:
                sendStr = f"t{msg.arbitration_id:03X}{msg.dlc:d}"
            sendStr += msg.data.hex().upper()
        self._write(sendStr)

    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 cast("int", self.serialPortOrig.fileno())
        except io.UnsupportedOperation:
            raise NotImplementedError(
                "fileno is not implemented using current CAN bus on this platform"
            ) from None
        except Exception as exception:
            raise CanOperationError("Cannot fetch fileno") from exception

    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
        """
        _timeout = serial.Timeout(timeout)
        cmd = "V"
        self._write(cmd)

        while True:
            if string := self._read(_timeout.time_left()):
                if string[0] == cmd:
                    # convert ASCII coded version
                    hw_version = int(string[1:3])
                    sw_version = int(string[3:5])
                    return hw_version, sw_version
                else:
                    self._queue.put_nowait(string)
            if _timeout.expired():
                break
        return None, None

    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 :obj:`None` to wait indefinitely

        :return:
            :obj:`None` on timeout or a :class:`str` object.
        """
        _timeout = serial.Timeout(timeout)
        cmd = "N"
        self._write(cmd)

        while True:
            if string := self._read(_timeout.time_left()):
                if string[0] == cmd:
                    serial_number = string[1:-1]
                    return serial_number
                else:
                    self._queue.put_nowait(string)
            if _timeout.expired():
                break
        return None