hil2.can_helper

 1from typing import Optional
 2
 3import os
 4import logging
 5
 6import cantools.database as cantools_db
 7
 8
 9# Helper functions --------------------------------------------------------------------#
10def load_can_dbcs(dbc_fpath: str) -> dict[str, cantools_db.Database]:
11    """
12    Scans a folder for DBC files and loads them as CAN databases.
13
14    :param dbc_fpath: The path to the CAN DBC folder
15    :return: A dictionary of CAN databases, keyed by DBC file name
16    """
17    dbs: dict[str, cantools_db.Database] = {}
18
19    if not dbc_fpath or not os.path.isdir(dbc_fpath):
20        logging.warning(f"Invalid DBC folder path: {dbc_fpath}")
21        return dbs
22    else:
23        for file in os.listdir(dbc_fpath):
24            if file.endswith(".dbc"):
25                logging.debug(f"Loading DBC file: {os.path.join(dbc_fpath, file)}")
26                db = cantools_db.load_file(os.path.join(dbc_fpath, file))
27                if db.messages:
28                    dbs[file] = db
29                else:
30                    logging.error(f"No messages found in DBC file: {file}")
31
32    return dbs
33
34
35# CAN Message struct ------------------------------------------------------------------#
36class CanMessage:
37    """Represents a parsed/decoded CAN message"""
38
39    def __init__(self, signal: str | int, data: dict):
40        """
41        :param signal: The signal name or message ID
42        :param data: The data contained in the CAN message
43        """
44
45        self.signal: str | int = signal
46        self.data: dict = data
47
48
49# CAN Message Manager class -----------------------------------------------------------#
50class CanMessageManager:
51    """Manages a collection of CAN messages"""
52
53    def __init__(self):
54        self._messages: list[CanMessage] = []
55
56    def add_multiple(self, messages: list[CanMessage]) -> None:
57        """
58        :param messages: The list of CAN messages to add
59        """
60        self._messages.extend(messages)
61
62    def get_last(self, signal: Optional[str | int]) -> Optional[CanMessage]:
63        """
64        :param signal: The signal name or message ID to get. If None, the last message
65                       will be returned (if any) regardless of the signal/id
66        :return: The last CAN message with the specified signal, or None if not found
67        """
68        if signal is None:
69            return self._messages[-1] if self._messages else None
70        for msg in reversed(self._messages):
71            if msg.signal == signal:
72                return msg
73        return None
74
75    def get_all(self, signal: Optional[str | int] = None) -> list[CanMessage]:
76        """
77        :param signal: The signal name or message ID to get. If None, all messages will
78                       be returned (if any) regardless of the signal/id
79        :return: A list of all CAN messages with the specified signal (or all)
80        """
81        return list(
82            filter(lambda msg: signal is None or msg.signal == signal, self._messages)
83        )
84
85    def clear(self, signal: Optional[str | int] = None) -> None:
86        """
87        :param signal: The signal name or message ID to clear. If None, all messages
88                       will be cleared (if any) regardless of the signal/id
89        """
90        if signal is None:
91            self._messages.clear()
92        else:
93            self._messages = list(
94                filter(lambda msg: msg.signal != signal, self._messages)
95            )