hil2.serial_helper

  1from typing import Optional
  2
  3import logging
  4import threading
  5import time
  6
  7import serial
  8import serial.tools.list_ports
  9
 10from . import commands
 11from . import hil_errors
 12
 13SERIAL_BAUDRATE = 115200
 14SERIAL_TIMEOUT = 0.1
 15SERIAL_RETRIES = 5
 16RESPONSE_WAIT = 0.2
 17
 18GET_TIMEOUT = 0.1
 19SLEEP_INTERVAL = 0.01
 20
 21
 22# Discover ----------------------------------------------------------------------------#
 23def discover_devices(hil_ids: list[int]) -> dict[int, tuple[serial.Serial, list[int]]]:
 24    """
 25    Attempts to find HIL devices by sending an identification command to each serial
 26    port.
 27
 28    :param hil_ids: A list of expected HIL device IDs
 29    :return: A dictionary mapping discovered HIL device IDs to their serial connections
 30             and any extra readings (that need to be parsed)
 31    """
 32
 33    devices = {}
 34
 35    com_ports = [
 36        port.device
 37        for port in serial.tools.list_ports.comports()
 38        if "USB Serial" in port.description
 39    ]
 40
 41    # For every serial port, try to see if it is a HIL device
 42    for cp in com_ports:
 43        logging.debug(f"Trying to discover HIL device on port {cp}")
 44        serial_con = serial.Serial(
 45            cp,
 46            SERIAL_BAUDRATE,
 47            timeout=SERIAL_TIMEOUT,
 48            bytesize=serial.EIGHTBITS,
 49            parity=serial.PARITY_NONE,
 50            stopbits=serial.STOPBITS_ONE,
 51            xonxoff=0,
 52            rtscts=0,
 53        )
 54        serial_con.dtr = False
 55        time.sleep(1)
 56        serial_con.reset_input_buffer()
 57        serial_con.dtr = True
 58
 59        # Need to give a little time
 60        for _ in range(SERIAL_RETRIES):
 61            match commands.read_id(serial_con, RESPONSE_WAIT):
 62                case None:
 63                    pass
 64                case (read_hil_id, extra_readings) if read_hil_id in hil_ids:
 65                    devices[read_hil_id] = (serial_con, extra_readings)
 66                    logging.info(
 67                        f"Discovered HIL device with ID {read_hil_id} on port {cp}"
 68                    )
 69                    break
 70                case _:
 71                    logging.debug(f"Found non-matching HIL ID on port {cp}")
 72            time.sleep(1)
 73        else:
 74            # If it is not a HIL device, close it
 75            serial_con.close()
 76
 77    # Check we found all devices
 78    for hil_id in hil_ids:
 79        if hil_id not in devices:
 80            error_msg = f"Failed to discover HIL device with ID {hil_id} on any port"
 81            raise hil_errors.SerialError(error_msg)
 82
 83    return devices
 84
 85
 86# Threaded serial ---------------------------------------------------------------------#
 87class ThreadedSerial:
 88    """
 89    A class that handles serial communication in a separate thread.
 90    This is needed because CAN messages are received asynchronously as opposed to
 91    command and response.
 92    """
 93
 94    def __init__(
 95        self,
 96        serial_con: serial.Serial,
 97        readings: list[int],
 98        stop_event: threading.Event,
 99    ):
100        """
101        :param serial_con: The serial connection to the HIL device
102        :param stop_event: The event used to signal the thread to stop
103        """
104        self.serial_con: serial.Serial = serial_con
105        self.stop_event: threading.Event = stop_event
106
107        # Raw readings from the serial port. Some bytes may have been already read when
108        # discovering the device.
109        self.readings: list[int] = readings
110
111        # Parsed readings. The key is the command (ex: READ_GPIO) and the value is the
112        # list of bytes
113        self.parsed_readings: dict[int, list[int]] = {}
114        # Parsed CAN messages. The key is the bus number, the value is a list of the
115        # list of bytes for each message
116        self.parsed_can_messages: dict[int, list[list[int]]] = {}
117
118        # Lock for synchronizing access to shared resources
119        self.lock = threading.Lock()
120
121        if len(self.readings) > 0:
122            # If there are any initial readings, try to process them
123            with self.lock:
124                self._process_readings()
125
126    def write(self, data: bytes) -> None:
127        """
128        Write data to the serial port. Safe to be called from another thread.
129
130        :param data: The data to write to the serial port
131        """
132        self.serial_con.write(data)
133
134    def _read(self):
135        """
136        Attempt to read a single byte from the serial port.
137        """
138        read_data = self.serial_con.read(1)
139        if len(read_data) < 1:
140            return
141        value = int.from_bytes(read_data, "big")
142        self.readings.append(value)
143
144    def _process_readings(self):
145        """
146        Attempt to process read bytes.
147        """
148        processed = True
149        while processed:
150            # If something was processed, try to process again
151            processed, self.readings = commands.parse_readings(
152                self.readings, self.parsed_readings, self.parsed_can_messages
153            )
154
155    def _get_readings(self, command: int) -> Optional[list[int]]:
156        """
157        Get the readings for a specific command. Safe to be called from a different thread.
158
159        :param command: The command to get readings for (used as key)
160        :return: The readings for the command, or None if not found
161        """
162        with self.lock:
163            val = self.parsed_readings.pop(command, None)
164            return val
165
166    def get_readings_with_timeout(
167        self,
168        command: int,
169        timeout: float = GET_TIMEOUT,
170        sleep_interval: float = SLEEP_INTERVAL,
171    ) -> Optional[list[int]]:
172        """
173        Get the readings for a command, with a delay.
174        Retries the reading at regular intervals until the timeout is reached.
175        Safe to be called from a different thread.
176
177        :param command: The command to get readings for (used as key)
178        :param timeout: The maximum time to wait for readings (seconds)
179        :param sleep_interval: The time to wait between retries (seconds)
180        :return: The readings for the command, or None if not found
181        """
182
183        deadline = time.time() + timeout
184        while time.time() < deadline:
185            if (reading := self._get_readings(command)) is not None:
186                return reading
187            time.sleep(sleep_interval)
188        return None
189
190    def get_parsed_can_messages(self, bus: int) -> list[list[int]]:
191        """
192        Get the parsed CAN messages for a specific bus.
193        Safe to be called from a different thread.
194
195        :param bus: The bus number to get messages for
196        :return: A list of parsed (but not decoded) CAN messages for the bus
197        """
198        with self.lock:
199            return self.parsed_can_messages.pop(bus, [])
200
201    def stop(self):
202        """
203        Stop the serial helper thread.
204        Safe to be called from a different thread.
205        """
206        self.stop_event.set()
207
208    def _close(self):
209        """
210        Close the serial connection.
211        """
212        self.serial_con.close()
213
214    def run(self):
215        """
216        Run the serial helper thread.
217        Constantly tries to read from the serial port and process the readings.
218        Should be run in a separate thread.
219        """
220        while not self.stop_event.is_set():
221            self._read()
222            if len(self.readings) > 0:
223                with self.lock:
224                    self._process_readings()
225
226        self._close()