Remove can bus Recorder

src/can_explorer/app.py

@@ -7,8 +7,7 @@
 import can
 import dearpygui.dearpygui as dpg
 
-from can_explorer import can_bus
-from can_explorer.configs import Default
+from can_explorer.configs import CANBus, Default
 from can_explorer.controllers import Controller
 from can_explorer.models import PlotModel
 from can_explorer.tags import Tag
@@ -50,8 +49,8 @@ def setup(self):
 
         main_window = self.view.ui.build()
 
-        self.view.settings.set_interface_options(can_bus.INTERFACES)
-        self.view.settings.set_baudrate_options(can_bus.BAUDRATES)
+        self.view.settings.set_interface_options(CANBus.INTERFACES)
+        self.view.settings.set_baudrate_options(CANBus.BAUDRATES)
         self.view.settings.set_apply_button_callback(
             self.controller.settings_apply_button_callback
         )

src/can_explorer/configs.py

@@ -1,5 +1,7 @@
 from typing import Final
 
+from can.interfaces import VALID_INTERFACES
+
 from can_explorer.resources import DIR_PATH as RESOURCES_DIR
 from can_explorer.resources import HOST_OS
 
@@ -21,3 +23,9 @@ class Default:
     TITLE: Final = "CAN Explorer"
     FONT: Final = RESOURCES_DIR / "Inter-Medium.ttf"
     FOOTER_OFFSET: Final = 50 if HOST_OS == "linux" else 85
+
+
+class CANBus:
+    INTERFACES: Final = sorted(list(VALID_INTERFACES))
+    _BAUDRATES = [33_333, 125_000, 250_000, 500_000, 1_000_000]
+    BAUDRATES: Final = [format(i, "_d") for i in _BAUDRATES]

src/can_explorer/controllers.py

@@ -1,16 +1,12 @@
 from __future__ import annotations
 
 import enum
-from threading import current_thread
-from typing import cast
 
 import can
 from can.bus import BusABC
 
-from can_explorer.can_bus import Recorder
 from can_explorer.configs import Default
 from can_explorer.models import PlotModel
-from can_explorer.resources import StoppableThread
 from can_explorer.views import MainView
 
 
@@ -25,38 +21,30 @@ def __init__(
         model: PlotModel,
         view: MainView,
         bus: BusABC | None = None,
-        recorder: Recorder | None = None,
         refresh_rate: float | None = Default.REFRESH_RATE,
     ) -> None:
         self.model = model
         self.view = view
-        self.recorder = recorder or Recorder()
+        self.notififer: can.Notifier | None = None
 
         self._bus = bus
         self._rate = refresh_rate
         self._state = State.STOPPED
-        self._worker: StoppableThread | None = None
 
     @property
     def state(self) -> State:
         return self._state
 
     @property
-    def bus(self) -> BusABC | None:
+    def bus(self) -> BusABC:
         if self._bus is None:
             raise RuntimeError("Must apply settings before starting")
         return self._bus
 
-    @property
-    def worker(self) -> StoppableThread:
-        if self._worker is None:
-            raise RuntimeError("Worker not set.")
-        return self._worker
-
     def is_active(self) -> bool:
         return bool(self.state)
 
-    def set_bus(self, bus: can.BusABC) -> None:
+    def set_bus(self, bus: BusABC) -> None:
         """
         Set CAN bus to use during controller loop.
         """
@@ -73,14 +61,8 @@ def start(self) -> None:
         if self.state == State.RUNNING:
             raise RuntimeError("App is already running")
 
-        self.recorder.set_bus(self.bus)
-        self.recorder.start()
-
-        self.create_worker_thread()
-        self.worker.start()
-
+        self.notifier = can.Notifier(self.bus, [self._on_message_received])
         self.view.set_main_button_label(True)
-
         self._state = State.RUNNING
 
     def stop(self) -> None:
@@ -90,38 +72,25 @@ def stop(self) -> None:
         if self.state == State.STOPPED:
             return
 
-        self.recorder.stop()
-        self.worker.stop()
-
+        self.notifier.stop()
         self.view.set_main_button_label(False)
-
         self._state = State.STOPPED
 
-    def _worker_loop(self) -> None:
-        thread = cast(StoppableThread, current_thread())
-        while not thread.cancel.wait(self._rate):
-            current_data = self.recorder.get_data()
-
-            for can_id, payloads in current_data.items():
-                self.model.update(can_id, payloads)
-                plot_data = self.model.get_plot_data(can_id)
-                self.view.plot.update(can_id, plot_data)
-
-    def create_worker_thread(self) -> None:
-        self._worker = StoppableThread(target=self._worker_loop, daemon=True)
+    def _on_message_received(self, message: can.Message):
+        self.model.add_message(message)
+        plot_data = self.model.get_data(message.arbitration_id)
+        self.view.plot.update(message.arbitration_id, plot_data)
 
     def start_stop_button_callback(self, *args, **kwargs) -> None:
         self.stop() if self.is_active() else self.start()
 
     def clear_button_callback(self, *args, **kwargs) -> None:
-        self.model.clear()
         self.view.plot.clear()
 
     def plot_buffer_slider_callback(self, *args, **kwargs) -> None:
         self.model.set_limit(self.view.get_plot_buffer())
-        for can_id, payloads in self.recorder.get_data().items():
-            self.model.update(can_id, payloads)
-            plot_data = self.model.get_plot_data(can_id)
+        for can_id in self.view.plot.get_rows():
+            plot_data = self.model.get_data(can_id)
             self.view.plot.update(can_id, plot_data)
 
     def plot_height_slider_callback(self, *args, **kwargs) -> None:

src/can_explorer/models.py

@@ -1,48 +1,38 @@
-from collections.abc import Collection
+from collections import defaultdict, deque
+from typing import DefaultDict
+
+import can
 
-from can_explorer.can_bus import PayloadBuffer
 from can_explorer.configs import Default
-from can_explorer.plotting import PlotData
+from can_explorer.plotting import PlotData, convert_payloads
 
 
-def convert_payloads(payloads: Collection) -> PlotData:
-    return PlotData(
-        x=tuple(range(len(payloads))),
-        y=tuple(payloads),
-    )
+class PayloadBuffer(deque):
+    def __init__(self):
+        self.MIN = Default.BUFFER_MIN
+        self.MAX = Default.BUFFER_MAX
+        super().__init__([0] * self.MAX, maxlen=self.MAX)
 
+    def __getitem__(self, index) -> tuple:  # type: ignore [override]
+        # Add ability to utilize slicing
+        # Note: must convert deque to avoid runtime error
+        if isinstance(index, slice):
+            return tuple(self)[index.start : index.stop : index.step]
+        return tuple(deque.__getitem__(self, index))
 
-class PlotModel:
-    _len = Default.BUFFER_SIZE
 
+class PlotModel:
     def __init__(self) -> None:
-        self._plot: dict[int, PlotData] = {}
+        self._data: DefaultDict[int, PayloadBuffer] = defaultdict(PayloadBuffer)
+        self._len = Default.BUFFER_SIZE
 
-    def update(self, can_id: int, payloads: PayloadBuffer) -> None:
-        """
-        Update a plot.
-
-        Args:
-            can_id (int)
-            payloads (PayloadBuffer)
-        """
-        plot_data = convert_payloads(payloads[-self._len :])
-        self._plot[can_id] = plot_data
+    def add_message(self, message: can.Message) -> None:
+        can_id = message.arbitration_id
+        val = int.from_bytes(message.data, byteorder="big")
+        self._data[can_id].append(val)
 
-    def get_plot_data(self, can_id: int) -> PlotData:
-        return self._plot[can_id]
-
-    def get_plots(self) -> dict:
-        """
-        Get all plots.
-        """
-        return self._plot.copy()
-
-    def clear(self) -> None:
-        """
-        Remove all plots.
-        """
-        self._plot.clear()
+    def get_data(self, can_id: int) -> PlotData:
+        return convert_payloads(self._data[can_id][-self._len :])
 
     def set_limit(self, limit: int) -> None:
         """

src/can_explorer/plotting.py

@@ -10,6 +10,13 @@
 from can_explorer.tags import generate_tag
 
 
+def convert_payloads(payloads: Collection) -> PlotData:
+    return PlotData(
+        x=tuple(range(len(payloads))),
+        y=tuple(payloads),
+    )
+
+
 @dataclass
 class PlotData:
     x: Collection