hil2.test_device

  1from typing import Any, Optional
  2
  3import json
  4import os
  5import threading
  6
  7import cantools.database.can.database as cantools_db
  8
  9from . import action
 10from . import can_helper
 11from . import commands
 12from . import dut_cons
 13from . import hil_errors
 14from . import serial_helper
 15
 16
 17# Peripheral configuration ------------------------------------------------------------#
 18class AdcConfig:
 19    """Configuration for an ADC (Analog-to-Digital Converter)."""
 20
 21    def __init__(self, adc_config: dict):
 22        """
 23        :param adc_config: The ADC configuration dictionary
 24        """
 25        match adc_config:
 26            # PCB
 27            case {
 28                "bit_resolution": br,
 29                "adc_reference_v": ar,
 30                "5v_reference_v": v5r,
 31                "24v_reference_v": v24r,
 32            }:
 33                self.bit_resolution: int = br
 34                self.adc_reference_v: float = ar
 35                self.five_v_reference_v: Optional[float] = v5r
 36                self.twenty_four_v_reference_v: Optional[float] = v24r
 37            # Breadboard
 38            case {
 39                "bit_resolution": br,
 40                "adc_reference_v": ar,
 41            }:
 42                self.bit_resolution: int = br
 43                self.adc_reference_v: float = ar
 44                self.five_v_reference_v: Optional[float] = None
 45                self.twenty_four_v_reference_v: Optional[float] = None
 46            case _:
 47                raise hil_errors.ConfigurationError("Invalid ADC configuration")
 48
 49    def raw_to_v(self, raw_value: int) -> float:
 50        """
 51        Convert a raw ADC value to a voltage.
 52
 53        :param raw_value: The raw ADC value to convert
 54        :return: The converted voltage value
 55        """
 56        if raw_value < 0 or raw_value > (2**self.bit_resolution - 1):
 57            raise hil_errors.RangeError(f"ADC raw value {raw_value} out of range")
 58        return (raw_value / (2**self.bit_resolution - 1)) * self.adc_reference_v
 59
 60    def raw_to_5v(self, raw_value: int) -> float:
 61        """
 62        Convert a raw ADC value to a voltage using the 5V reference.
 63        If on a PCB, AI5 means it when through a voltage divider.
 64
 65        :param raw_value: The raw ADC value to convert
 66        :return: The converted voltage value
 67        """
 68        match self.five_v_reference_v:
 69            case None:
 70                error_msg = "5V reference voltage not configured"
 71                raise hil_errors.ConfigurationError(error_msg)
 72            case v5r:
 73                return (self.raw_to_v(raw_value) / v5r) * 5.0
 74
 75    def raw_to_24v(self, raw_value: int) -> float:
 76        """
 77        Convert a raw ADC value to a voltage using the 24V reference.
 78        If on a PCB, AI24 means it when through a voltage divider.
 79
 80        :param raw_value: The raw ADC value to convert
 81        :return: The converted voltage value
 82        """
 83        match self.twenty_four_v_reference_v:
 84            case None:
 85                error_msg = "24V reference voltage not configured"
 86                raise hil_errors.ConfigurationError(error_msg)
 87            case v24r:
 88                return (self.raw_to_v(raw_value) / v24r) * 24.0
 89
 90
 91class DacConfig:
 92    """Configuration for a DAC (Digital-to-Analog Converter)."""
 93
 94    def __init__(self, dac_config: dict):
 95        """
 96        :param dac_config: The DAC configuration dictionary
 97        """
 98        match dac_config:
 99            case {"bit_resolution": br, "reference_v": rv}:
100                self.bit_resolution = br
101                self.reference_v = rv
102            case _:
103                raise hil_errors.ConfigurationError("Invalid DAC configuration")
104
105    def v_to_raw(self, value: float) -> int:
106        """
107        Convert a voltage value to a raw DAC value.
108
109        :param value: The voltage value to convert
110        :return: The converted raw DAC value
111        """
112        if value < 0 or value > self.reference_v:
113            raise hil_errors.RangeError(f"DAC value {value} out of range")
114        return int((value / self.reference_v) * (2**self.bit_resolution - 1))
115
116
117class PotConfig:
118    """Configuration for a POT (Potentiometer)."""
119
120    def __init__(self, pot_config: dict):
121        """
122        :param pot_config: The POT configuration dictionary
123        """
124        match pot_config:
125            case {"bit_resolution": br, "reference_ohms": r, "wiper_ohms": w}:
126                self.bit_resolution = br
127                self.reference_ohms = r
128                self.wiper_ohms = w
129            case _:
130                raise hil_errors.ConfigurationError("Invalid POT configuration")
131
132    def ohms_to_raw(self, value: float) -> int:
133        """
134        Convert an ohm value to a raw POT value.
135
136        :param value: The ohm value to convert
137        :return: The converted raw POT value
138        """
139        if value < self.wiper_ohms or value > self.reference_ohms + self.wiper_ohms:
140            raise hil_errors.RangeError(f"POT value {value} out of range")
141        steps = self.bit_resolution**2 - 1
142        return int((steps * (value - self.wiper_ohms)) / self.reference_ohms)
143
144
145# Interface configuration -------------------------------------------------------------#
146class Port:
147    """Configuration for a port."""
148
149    def __init__(self, port: dict):
150        """
151        :param port: The port configuration dictionary
152        """
153        match port:
154            case {"name": name, "port": port, "mode": mode}:
155                self.name: str = name
156                self.port: int = port
157                self.mode: str = mode
158            case _:
159                raise hil_errors.ConfigurationError("Invalid Port configuration")
160
161
162class Mux:
163    """Configuration for a MUX (Multiplexer)."""
164
165    def __init__(self, mux: dict):
166        """
167        :param mux: The MUX configuration dictionary
168        """
169        match mux:
170            case {
171                "name": name,
172                "mode": mode,
173                "select_ports": select_ports,
174                "port": port,
175            }:
176                self.name: str = name
177                self.mode: str = mode
178                self.select_ports: list[int] = select_ports
179                self.port: int = port
180            case _:
181                raise hil_errors.ConfigurationError("Invalid Mux configuration")
182
183    def select_from_name(self, name: str) -> Optional["MuxSelect"]:
184        """
185        Attempt to see if self is the base mux that is being referenced.
186        For example: DMUX_6 means DMUX is the base mux and 6 is the select line.
187
188        :param name: The name of the MUX select line (ex: DMUX_6)
189        :return: The MuxSelect instance if found, None otherwise
190        """
191        name_parts = name.rsplit("_", 1)
192        if len(name_parts) < 2:
193            return None
194        if name_parts[0] != self.name:
195            return None
196        try:
197            return MuxSelect(self, int(name_parts[1]))
198        except ValueError:
199            return None
200
201
202class MuxSelect:
203    """Represents a selected MUX (Multiplexer) line."""
204
205    def __init__(self, mux: Mux, select: int):
206        """
207        :param mux: The MUX instance
208        :param select: The selected line number (0 indexed)
209        """
210        self.mux: Mux = mux
211        self.select: int = select
212
213
214class CanBus:
215    """Configuration for a CAN (Controller Area Network) bus."""
216
217    def __init__(self, can_bus: dict):
218        """
219        :param can_bus: The CAN bus configuration dictionary
220        """
221        match can_bus:
222            case {"name": name, "bus": bus, "dbc_file": dbc_file}:
223                self.name: str = name
224                self.bus: int = bus
225                self.dbc_file: str = dbc_file
226            case _:
227                raise hil_errors.ConfigurationError("Invalid CAN Bus configuration")
228
229    def find_dbc(
230        self, can_dbcs: dict[str, cantools_db.Database]
231    ) -> cantools_db.Database:
232        """
233        Attempt to find the CAN DBC database for this CAN bus. If not found, raise an error.
234
235        :param can_dbcs: The dictionary of loaded CAN databases, keyed by DBC file name
236        :return: The CAN database if found
237        """
238        match can_dbcs.get(self.dbc_file, None):
239            case None:
240                error_msg = (
241                    f"DBC file {self.dbc_file} not found for CAN bus {self.name}"
242                )
243                raise hil_errors.ConfigurationError(error_msg)
244            case db:
245                return db
246
247
248# Test device -------------------------------------------------------------------------#
249class TestDevice:
250    # Init ----------------------------------------------------------------------------#
251    def __init__(
252        self,
253        hil_id: int,
254        name: str,
255        ports: dict[str, Port],
256        muxs: dict[str, Mux],
257        can_busses: dict[str, CanBus],
258        adc_config: AdcConfig,
259        dac_config: Optional[DacConfig],
260        pot_config: Optional[PotConfig],
261    ):
262        """
263        :param hil_id: The HIL ID of the device
264        :param name: The name of the device
265        :param ports: The port configurations
266        :param muxs: The MUX configurations
267        :param can_busses: The CAN bus configurations
268        :param adc_config: The ADC configuration
269        :param dac_config: The DAC configuration
270        :param pot_config: The potentiometer configuration
271        """
272        self.hil_id: int = hil_id
273        self._name: str = name
274        self._ports: dict[str, Port] = ports
275        self._muxs: dict[str, Mux] = muxs
276        self._can_busses: dict[str, CanBus] = can_busses
277        self._adc_config: AdcConfig = adc_config
278        self._dac_config: Optional[DacConfig] = dac_config
279        self._pot_config: Optional[PotConfig] = pot_config
280
281        # Please use set_serial() to set the serial!
282        self._ser: Optional[serial_helper.ThreadedSerial] = None
283
284        self.device_can_busses: dict[int, can_helper.CanMessageManager] = dict(
285            map(
286                lambda c: (c.bus, can_helper.CanMessageManager()),
287                self._can_busses.values(),
288            )
289        )
290
291    @classmethod
292    def from_json(cls, hil_id: int, name: str, device_config_path: str):
293        """
294        Create a TestDevice instance from a JSON configuration file.
295
296        :param hil_id: The HIL ID of the device
297        :param name: The name of the device
298        :param device_config_path: The path to the device configuration JSON file
299        """
300        with open(device_config_path, "r") as device_config_path:
301            device_config = json.load(device_config_path)
302
303        ports = dict(
304            map(lambda p: (p.get("name"), Port(p)), device_config.get("ports", []))
305        )
306        muxs = dict(
307            map(lambda m: (m.get("name"), Mux(m)), device_config.get("muxs", []))
308        )
309        can_busses = dict(
310            map(lambda c: (c.get("name"), CanBus(c)), device_config.get("can", []))
311        )
312
313        match device_config:
314            case {"adc_config": adc_config_data}:
315                adc_config = AdcConfig(adc_config_data)
316            case _:
317                error_msg = f"ADC configuration missing for device {name}"
318                raise hil_errors.ConfigurationError(error_msg)
319
320        match device_config:
321            case {"dac_config": dac_config_data}:
322                dac_config = DacConfig(dac_config_data)
323            case _:
324                dac_config = None
325
326        match device_config:
327            case {"pot_config": pot_config_data}:
328                pot_config = PotConfig(pot_config_data)
329            case _:
330                pot_config = None
331
332        return cls(
333            hil_id,
334            name,
335            ports,
336            muxs,
337            can_busses,
338            adc_config,
339            dac_config,
340            pot_config,
341        )
342
343    def set_serial(self, ser: serial_helper.ThreadedSerial) -> None:
344        """
345        Set the serial connection for the TestDevice.
346        The caller is responsible for starting the serial connection's thread.
347
348        :param ser: The serial connection to set
349        """
350        self._ser = ser
351
352    def close(self) -> None:
353        """
354        Close the serial connection for the TestDevice.
355        """
356        match self._ser:
357            case None:
358                error_msg = f"Cannot close TestDevice {self._name}: serial not set"
359                raise hil_errors.EngineError(error_msg)
360            case ser:
361                ser.stop()
362
363    # Command handling ----------------------------------------------------------------#
364    def _select_mux(self, mux_select: MuxSelect) -> None:
365        """
366        Select a MUX (Multiplexer) line.
367
368        :param mux_select: The MUX selection information
369        """
370        for i, p in enumerate(mux_select.mux.select_ports):
371            select_bit = True if (mux_select.select & (1 << i)) else False
372            self._set_do(p, select_bit)
373
374    def _set_do(self, pin: int, value: bool) -> None:
375        """
376        Set a digital output (DO) pin.
377
378        :param pin: The pin number to set
379        :param value: The value to set the pin to (low = False, high = True)
380        """
381        match self._ser:
382            case None:
383                error_msg = f"Cannot set DO on TestDevice {self._name}: serial not set"
384                raise hil_errors.EngineError(error_msg)
385            case ser:
386                commands.write_gpio(ser, pin, value)
387
388    def _hiZ_do(self, pin: int) -> None:
389        """
390        Set a digital output (DO) pin to high impedance (HiZ).
391
392        :param pin: The pin number to set
393        """
394        match self._ser:
395            case None:
396                error_msg = (
397                    f"Cannot set HiZ DO on TestDevice {self._name}: serial not set"
398                )
399                raise hil_errors.EngineError(error_msg)
400            case ser:
401                commands.hiZ_gpio(ser, pin)
402
403    def _get_di(self, pin: int) -> bool:
404        """
405        Get the digital input (DI) state of a pin.
406
407        :param pin: The pin number to read
408        :return: The state of the pin (True for high, False for low)
409        """
410        match self._ser:
411            case None:
412                error_msg = f"Cannot get DI on TestDevice {self._name}: serial not set"
413                raise hil_errors.EngineError(error_msg)
414            case ser:
415                return commands.read_gpio(ser, pin)
416
417    def _set_ao(self, pin: int, value: float) -> None:
418        """
419        Set an analog output (AO) pin after converting the volts value to raw.
420
421        :param pin: The pin/offset number to set
422        :param value: The voltage value to set the pin to
423        """
424        match (self._ser, self._dac_config):
425            case (ser, dac_config) if ser is not None and dac_config is not None:
426                raw_value = dac_config.v_to_raw(value)
427                commands.write_dac(ser, pin, raw_value)
428            case _:
429                error_msg = f"Cannot set AO on TestDevice {self._name}: serial or DAC config not set"
430                raise hil_errors.EngineError(error_msg)
431
432    def _hiZ_ao(self, pin: int) -> None:
433        """
434        Set an analog output (AO) pin to high impedance (HiZ).
435
436        :param pin: The pin/offset number to set
437        """
438        match self._ser:
439            case None:
440                error_msg = (
441                    f"Cannot set HiZ AO on TestDevice {self._name}: serial not set"
442                )
443                raise hil_errors.EngineError(error_msg)
444            case ser:
445                commands.hiZ_dac(ser, pin)
446
447    def _get_ai(self, pin: int, mode: str) -> float:
448        """
449        Get an analog input (AI) reading from a pin and convert the reading to volts.
450
451        :param pin: The pin number to read
452        :param mode: The mode to use for the reading (AI5, AI24, or AI)
453        :return: The voltage value read from the pin
454        """
455        match self._ser:
456            case None:
457                error_msg = f"Cannot get AI on TestDevice {self._name}: serial not set"
458                raise hil_errors.EngineError(error_msg)
459            case ser:
460                raw_value = commands.read_adc(ser, pin)
461
462        if mode == "AI5":
463            return self._adc_config.raw_to_5v(raw_value)
464        elif mode == "AI24":
465            return self._adc_config.raw_to_24v(raw_value)
466        elif mode == "AI":
467            return self._adc_config.raw_to_v(raw_value)
468        else:
469            raise ValueError(f"Unsupported AI mode: {mode}")
470
471    def _set_pot(self, pin: int, value: float) -> None:
472        """
473        Set a potentiometer (POT) pin after converting the ohms value to raw.
474
475        :param pin: The pin/offset to set
476        :param value: The resistance value to set the pin to (in ohms)
477        """
478        match (self._ser, self._pot_config):
479            case (ser, pot_config) if ser is not None and pot_config is not None:
480                raw_value = pot_config.ohms_to_raw(value)
481                commands.write_pot(ser, pin, raw_value)
482            case _:
483                error_msg = f"Cannot set POT on TestDevice {self._name}: serial not set"
484                raise hil_errors.EngineError(error_msg)
485
486    def _update_can_messages(self, bus: int, can_dbc: cantools_db.Database) -> None:
487        """
488        Update the CAN message store by decoding the saved parsed can messages from the Serial.
489
490        :param bus: The CAN bus to update
491        :param can_dbc: The CAN database to use for decoding
492        """
493        match self._ser:
494            case None:
495                error_msg = (
496                    f"Cannot update CAN messages on TestDevice {self._name}: "
497                    "serial not set"
498                )
499                raise hil_errors.EngineError(error_msg)
500            case ser:
501                self.device_can_busses[bus].add_multiple(
502                    commands.parse_can_messages(ser, bus, can_dbc)
503                )
504
505    def _send_can(
506        self, bus: int, signal: str | int, data: dict, can_dbc: cantools_db.Database
507    ) -> None:
508        """
509        Send a CAN message on the specified bus.
510
511        :param bus: The CAN bus to send the message on
512        :param signal: The CAN signal to send
513        :param data: The data to include in the CAN message
514        :param can_dbc: The CAN database to use for encoding
515        """
516        raw_data = list(can_dbc.encode_message(signal, data))
517        if isinstance(signal, int):
518            msg_id = can_dbc.get_message_by_frame_id(signal).frame_id
519        else:
520            msg_id = can_dbc.get_message_by_name(signal).frame_id
521
522        match self._ser:
523            case None:
524                error_msg = (
525                    f"Cannot send CAN message on TestDevice {self._name}: "
526                    "serial not set"
527                )
528                raise hil_errors.EngineError(error_msg)
529            case ser:
530                commands.send_can(ser, bus, msg_id, raw_data)
531
532    # Action --------------------------------------------------------------------------#
533    def do_action(self, action_type: action.ActionType, port: str) -> Any:
534        """
535        Perform a HIL action on a specific port.
536
537        :param action_type: The type of action to perform (+ includes all needed info)
538        :param port: The HIL port to perform the action on
539        :return: depends on the action type
540        """
541
542        maybe_port = self._ports.get(port, None)
543        maybe_mux_select = next(
544            (
545                val
546                for m in self._muxs.values()
547                if (val := m.select_from_name(port)) is not None
548            ),
549            None,
550        )
551        maybe_can_bus = self._can_busses.get(port, None)
552
553        match (action_type, maybe_port, maybe_mux_select, maybe_can_bus):
554            # Set DO + direct port
555            case (action.SetDo(value), mp, _, _) if mp is not None and mp.mode == "DO":
556                self._set_do(mp.port, value)
557            # Set DO + mux select
558            case (action.SetDo(value), _, mms, _) if (
559                mms is not None and mms.mux.mode == "DO"
560            ):
561                self._select_mux(mms)
562                self._set_do(mms.mux.port, value)
563            # HiZ DO + direct port
564            case (action.HiZDo(), mp, _, _) if mp is not None and mp.mode == "DO":
565                self._hiZ_do(mp.port)
566            # HiZ DO + mux select
567            case (action.HiZDo(), _, mms, _) if (
568                mms is not None and mms.mux.mode == "DO"
569            ):
570                self._select_mux(mms)
571                self._hiZ_do(mms.mux.port)
572            # Get DI + direct port
573            case (action.GetDi(), mp, _, _) if mp is not None and mp.mode == "DI":
574                return self._get_di(mp.port)
575            # Get DI + mux select
576            case (action.GetDi(), _, mms, _) if (
577                mms is not None and mms.mux.mode == "DI"
578            ):
579                self._select_mux(mms)
580                return self._get_di(mms.mux.port)
581            # Set AO + direct port
582            case (action.SetAo(value), mp, _, _) if mp is not None and mp.mode == "AO":
583                self._set_ao(mp.port, value)
584            # HiZ AO + direct port
585            case (action.HiZAo(), mp, _, _) if mp is not None and mp.mode == "AO":
586                self._hiZ_ao(mp.port)
587            # Get AI + direct port
588            case (action.GetAi(), mp, _, _) if mp is not None and mp.mode.startswith(
589                "AI"
590            ):
591                return self._get_ai(mp.port, mp.mode)
592            # Get AI + mux select
593            case (
594                action.GetAi(),
595                _,
596                mms,
597                _,
598            ) if mms is not None and mms.mux.mode.startswith("AI"):
599                self._select_mux(mms)
600                return self._get_ai(mms.mux.port, mms.mux.mode)
601            # Set Pot + direct port
602            case (action.SetPot(value), mp, _, _) if (
603                mp is not None and mp.mode == "POT"
604            ):
605                self._set_pot(mp.port, value)
606            # Send CAN msg + can bus name
607            case (action.SendCan(signal, data, can_dbcs), _, _, mcb) if mcb is not None:
608                can_dbc = mcb.find_dbc(can_dbcs)
609                self._update_can_messages(mcb.bus, can_dbc)
610                self._send_can(mcb.bus, signal, data, can_dbc)
611            # Get last CAN msg + can bus name
612            case (action.GetLastCan(signal, can_dbcs), _, _, mcb) if mcb is not None:
613                can_dbc = mcb.find_dbc(can_dbcs)
614                self._update_can_messages(mcb.bus, can_dbc)
615                return self.device_can_busses[mcb.bus].get_last(signal)
616            # Get all CAN msgs + can bus name
617            case (action.GetAllCan(signal, can_dbcs), _, _, mcb) if mcb is not None:
618                can_dbc = mcb.find_dbc(can_dbcs)
619                self._update_can_messages(mcb.bus, can_dbc)
620                return self.device_can_busses[mcb.bus].get_all(signal)
621            # Clear CAN msgs + can bus name
622            case (action.ClearCan(signal, can_dbcs), _, _, mcb) if mcb is not None:
623                can_dbc = mcb.find_dbc(can_dbcs)
624                self._update_can_messages(mcb.bus, can_dbc)
625                self.device_can_busses[mcb.bus].clear(signal)
626            # Unsupported action
627            case _:
628                error_msg = (
629                    f"Action {type(action)} not supported for "
630                    f"port {port} on device {self._name}"
631                )
632                raise hil_errors.EngineError(error_msg)
633
634
635# Test device manager -----------------------------------------------------------------#
636class TestDeviceManager:
637    """
638    Manages test devices for HIL (Hardware-in-the-Loop) simulation.
639    """
640
641    def __init__(self, test_devices: dict[str, TestDevice]):
642        """
643        :param test_devices: A dictionary of test devices managed by this manager.
644                             key = device name, value = TestDevice instance
645        """
646        self._test_devices: dict[str, TestDevice] = test_devices
647
648    @classmethod
649    def from_json(
650        cls, test_config_path: str, device_config_fpath: str
651    ) -> "TestDeviceManager":
652        """
653        Create a TestDeviceManager instance from JSON configuration files.
654        Is responsible for starting all of the ThreadedSerial instances.
655
656        :param test_config_path: The path to the test configuration JSON file.
657        :param device_config_fpath: The file path to the directory containing device
658                                    configuration files.
659        :return: A TestDeviceManager instance.
660        """
661
662        with open(test_config_path, "r") as test_config_file:
663            test_config = json.load(test_config_file)
664
665        hil_ids = []
666        stop_events = {}
667        test_devices = {}
668        match test_config:
669            case {"hil_devices": hil_devices}:
670                for device in hil_devices:
671                    match device:
672                        case {
673                            "id": hil_id,
674                            "name": name,
675                            "config": config_file_name,
676                        } if (not hil_id in hil_ids):
677                            hil_ids.append(hil_id)
678                            stop_events[hil_id] = threading.Event()
679                            test_devices[name] = TestDevice.from_json(
680                                hil_id,
681                                name,
682                                os.path.join(device_config_fpath, config_file_name),
683                            )
684                        case {"id": hil_id}:
685                            error_msg = f"Duplicate HIL device ID found: {hil_id}"
686                            raise hil_errors.ConfigurationError(error_msg)
687                        case _:
688                            error_msg = f"Invalid HIL device configuration: {device}"
689                            raise hil_errors.ConfigurationError(error_msg)
690            case _:
691                error_msg = "Invalid test configuration: missing 'hil_devices' key"
692                raise hil_errors.ConfigurationError(error_msg)
693
694        hil_devices = serial_helper.discover_devices(hil_ids)
695
696        sers = {
697            hil_id: serial_helper.ThreadedSerial(
698                hil_devices[hil_id][0],
699                hil_devices[hil_id][1],
700                stop_events[hil_id],
701            )
702            for hil_id in hil_ids
703        }
704        for test_device in test_devices.values():
705            ser = sers[test_device.hil_id]
706            t = threading.Thread(target=ser.run)
707            t.start()
708            test_device.set_serial(ser)
709
710        return cls(test_devices)
711
712    def maybe_hil_con_from_net(
713        self, board: str, net: str
714    ) -> Optional[dut_cons.HilDutCon]:
715        """
716        Check to see if a board is a HIL device.
717
718        :param board: The name of the board to check.
719        :param net: The network to use for the connection.
720        :return: A HilDutCon instance if the board is a HIL device, None otherwise.
721        """
722        if board in self._test_devices:
723            return dut_cons.HilDutCon(board, net)
724        else:
725            return None
726
727    def do_action(
728        self, action_type: action.ActionType, hil_dut_con: dut_cons.HilDutCon
729    ) -> Any:
730        """
731        Perform an action on a HIL device.
732
733        :param action_type: The type of action to perform.
734        :param hil_dut_con: The HIL DUT connection information.
735        :return: The result of the action (if any).
736        """
737        if hil_dut_con.device in self._test_devices:
738            return self._test_devices[hil_dut_con.device].do_action(
739                action_type, hil_dut_con.port
740            )
741        else:
742            error_msg = f"Device {hil_dut_con.device} not found"
743            raise hil_errors.ConnectionError(error_msg)
744
745    def close(self) -> None:
746        """
747        Close all HIL devices.
748        """
749        for device in self._test_devices.values():
750            device.close()