From c7294830f9bb15fa780ae821c2182ef96f46b73c Mon Sep 17 00:00:00 2001 From: Andrey Kruchinkin Date: Sat, 3 Oct 2026 22:53:51 +0300 Subject: [PATCH] Add shared CAN monitor sessions, protocol selection and HEX/DEC display --- python/set_devices/can_monitor.py | 108 ++++++++++++++ python/set_devices/can_transport.py | 5 + python/set_devices/qt_ports/can_monitor.py | 158 +++++++++++++++++++++ 3 files changed, 271 insertions(+) create mode 100644 python/set_devices/can_monitor.py create mode 100644 python/set_devices/qt_ports/can_monitor.py diff --git a/python/set_devices/can_monitor.py b/python/set_devices/can_monitor.py new file mode 100644 index 0000000..43c839f --- /dev/null +++ b/python/set_devices/can_monitor.py @@ -0,0 +1,108 @@ +"""Shared CAN protocol selection and disk-backed session buffer (no Qt).""" +import csv +import tempfile +from collections import Counter, deque +from datetime import datetime +from pathlib import Path +from protocan import protocan as pc +from .can_bridge import decode_frame + +PROTOCOLS = (("auto", "Авто (ProtoCAN / Balsam)"), + ("devboard", "DevBoard_V1 / SETCAN")) + + +def decode_protocol(frame, protocol="auto"): + if protocol == "devboard": + return pc.decode(frame.can_id, frame.data, ide=frame.ide, rtr=frame.rtr) + return decode_frame(frame) + + +def protocol_values(identifier, extended): + if not extended: + return ("",) * 7 + value = pc.ProtoCanId.parse(identifier) + return (str(value.priority), str(value.pm), str(value.device_type), + str(value.device), value.device_name, + "0x%X — %s" % (value.msg_type, value.msg_type_name), + "0x%04X" % value.body) + + +def parse_filter(text): + if not text.strip(): + return None + try: + ids = {int(part.strip(), 16) for part in text.replace(";", ",").split(",")} + if any(value < 0 or value > 0x1FFFFFFF for value in ids): + raise ValueError() + return ids + except ValueError: + raise ValueError("Введите CAN ID в hex, например 701 или 0x701, 201") from None + + +class CanSession: + """All frames spool to disk; only the latest 5000 stay in display memory. + + Starting recording includes the current session buffer. Reset clears the + buffer/counters but never removes frames already written to a recording. + """ + HEADER = ("Время", "Канал", "Тип", "ID", "RTR", "DLC", "Данные") + + def __init__(self): + self.recent = deque(maxlen=5000) + self.counts = Counter() + self.output = None + self.path = None + self.buffer = tempfile.TemporaryFile(mode="w+", encoding="utf-8", newline="") + + def append(self, identifier, data, extended, rtr=False, source="CAN", timestamp=None): + row = ((datetime.fromtimestamp(timestamp) if timestamp is not None else datetime.now()).isoformat(timespec="milliseconds"), + source, "EXT" if extended else "STD", ("%08X" if extended else "%03X") % identifier, + int(rtr), len(data), bytes(data).hex(" ").upper()) + csv.writer(self.buffer, delimiter=";").writerow(row) + self.recent.append((identifier, row)) + self.counts[identifier] += 1 + if self.output is not None: + csv.writer(self.output, delimiter=";").writerow(row) + + def start(self, directory): + directory = Path(directory) / datetime.now().strftime("session_%Y%m%d_%H%M%S_%f") + directory.mkdir(parents=True, exist_ok=False) + path = directory / "frames.csv" + output = path.open("w", encoding="utf-8-sig", newline="") + try: + csv.writer(output, delimiter=";").writerow(self.HEADER) + self.buffer.seek(0) + while chunk := self.buffer.read(65536): + output.write(chunk) + output.flush() + except Exception: + output.close() + raise + finally: + self.buffer.seek(0, 2) + self.output, self.path = output, path + return path + + def flush(self): + if self.output is not None: + self.output.flush() + + def stop(self): + if self.output is not None: + self.output.close() + self.output = None + + def reset(self): + self.buffer.seek(0) + self.buffer.truncate() + self.recent.clear() + self.counts.clear() + + def close(self): + self.stop() + self.buffer.close() + + +def format_data(data: bytes, decimal: bool = False) -> str: + """Format individual payload bytes without changing their order or width.""" + return " ".join(str(byte) if decimal else "%02X" % byte for byte in data) diff --git a/python/set_devices/can_transport.py b/python/set_devices/can_transport.py index 9b2834b..de082e0 100644 --- a/python/set_devices/can_transport.py +++ b/python/set_devices/can_transport.py @@ -259,3 +259,8 @@ def parse_hex(text: str) -> bytes: def hex_str(data: bytes) -> str: return " ".join("%02X" % b for b in data) + + +def raw_frame_text(data: bytes) -> str: + """Show the same raw bytes in hexadecimal and decimal on separate lines.""" + return hex_str(data) + "\n" + " ".join(str(byte) for byte in data) diff --git a/python/set_devices/qt_ports/can_monitor.py b/python/set_devices/qt_ports/can_monitor.py new file mode 100644 index 0000000..4d98452 --- /dev/null +++ b/python/set_devices/qt_ports/can_monitor.py @@ -0,0 +1,158 @@ +"""Reusable CAN monitor controls for host applications.""" +from PySide6.QtCore import QTimer +from PySide6.QtWidgets import (QWidget, QHBoxLayout, QVBoxLayout, QPushButton, + QCheckBox, QComboBox, QLabel, QLineEdit, QTableWidget, QTableWidgetItem, + QMessageBox, QApplication, QAbstractItemView) +from ..can_monitor import CanSession, PROTOCOLS, parse_filter, format_data + + +class MonitorControls(QWidget): + def __init__(self, owner, directory, default="auto"): + super().__init__(owner) + self.owner = owner + self.directory = directory + self.session = CanSession() + self._dirty = False + layout = QHBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + self.protocol = QComboBox() + for key, label in PROTOCOLS: + self.protocol.addItem(label, key) + self.protocol.setCurrentIndex(self.protocol.findData(default)) + layout.addWidget(QLabel("Протокол")) + layout.addWidget(self.protocol) + self.raw = QCheckBox("Без парса") + layout.addWidget(self.raw) + self.reset_button = QPushButton("Сброс") + self.reset_button.clicked.connect(owner.clear) + self.reset_button.setToolTip("Очистить таблицы, счётчики и буфер; уже записанный CSV сохраняется") + layout.addWidget(self.reset_button) + self.pause = QPushButton("Пауза") + self.pause.clicked.connect(owner._toggle_pause) + self.pause.setToolTip("Пауза экрана; буфер сессии и запись продолжаются") + layout.addWidget(self.pause) + self.record = QPushButton("● Запись") + self.record.clicked.connect(self.toggle_record) + self.record.setToolTip("Записать буфер сессии и последующие кадры в CSV") + layout.addWidget(self.record) + self.data_format = QPushButton("Data: HEX") + self.data_format.setCheckable(True) + self.data_format.setToolTip("Переключить отображение байтов Data: HEX / DEC") + self.data_format.toggled.connect(self._change_data_format) + layout.addWidget(self.data_format) + self.status = QLabel("Буфер: 0") + layout.addWidget(self.status) + layout.addStretch(1) + self.timer = QTimer(self) + self.timer.setInterval(250) + self.timer.timeout.connect(self.refresh) + self.timer.start() + QApplication.instance().aboutToQuit.connect(self.session.close) + self.destroyed.connect(lambda: self.session.close()) + + def format_data(self, data): + return format_data(data, self.data_format.isChecked()) + + def _change_data_format(self, decimal): + self.data_format.setText("Data: DEC" if decimal else "Data: HEX") + self.owner._refresh_data_format() + self.refilter() + + def filter_panel(self): + panel = QWidget() + layout = QVBoxLayout(panel) + layout.setContentsMargins(0, 0, 0, 0) + row = QHBoxLayout() + row.addWidget(QLabel("Фильтр ID (hex)")) + self.filter = QLineEdit() + self.filter.setPlaceholderText("701 или 0x701, 201") + self.filter.setClearButtonEnabled(True) + self.filter.textChanged.connect(self.refilter) + row.addWidget(self.filter) + self.count = QLabel("Кадров: 0") + row.addWidget(self.count) + layout.addLayout(row) + self.history = QTableWidget(0, len(CanSession.HEADER)) + self.history.setHorizontalHeaderLabels(CanSession.HEADER) + self.history.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers) + self.history.horizontalHeader().setStretchLastSection(True) + self.history.setColumnWidth(0, 190) + self.history.verticalHeader().setVisible(False) + self.history.setMinimumHeight(180) + self.history.setVisible(False) + layout.addWidget(self.history) + return panel + + def ingest(self, identifier, data, extended, rtr=False, source="CAN", timestamp=None): + try: + self.session.append(identifier, data, extended, rtr, source, timestamp) + except OSError as exc: + self.status.setText("Ошибка записи: " + str(exc)) + self.session.stop() + self.record.setText("● Запись") + self._dirty = True + + def reset(self): + self.session.reset() + self._dirty = True + self.refilter() + self.status.setText("Буфер: 0") + + def toggle_record(self): + try: + if self.session.output is None: + path = self.session.start(self.directory()) + self.record.setText("■ Стоп записи") + self.record.setToolTip(str(path)) + else: + self.session.stop() + self.record.setText("● Запись") + except OSError as exc: + QMessageBox.critical(self, "Запись CAN", str(exc)) + + def refresh(self): + try: + self.session.flush() + except OSError as exc: + self.status.setText("Ошибка записи: " + str(exc)) + return + if not self.owner._paused and self._dirty: + self._dirty = False + self.status.setText("Буфер: %d" % sum(self.session.counts.values())) + self.refilter() + + def refilter(self): + if not hasattr(self, "filter"): + return + try: + ids = parse_filter(self.filter.text()) + except ValueError as exc: + self.count.setText(str(exc)) + ids = set() + else: + count = sum(value for key, value in self.session.counts.items() + if ids is None or key in ids) + self.count.setText("Кадров: %d (показаны последние ≤5000)" % count) + self.history.horizontalHeaderItem(6).setText( + "Данные (DEC)" if self.data_format.isChecked() else "Данные (HEX)") + self.history.setVisible(ids is not None) + self.history.setRowCount(0) + table = getattr(self.owner, "online_table", None) + if table is None: + table = getattr(self.owner, "live_table", None) + if table is not None: + for row in range(table.rowCount()): + item = table.item(row, 0) + if item is not None: + table.setRowHidden(row, ids is not None and int(item.text(), 16) not in ids) + if ids is None: + return + for identifier, values in self.session.recent: + if identifier not in ids: + continue + row = self.history.rowCount() + self.history.insertRow(row) + for column, value in enumerate(values): + if column == 6: + value = self.format_data(bytes.fromhex(value)) + self.history.setItem(row, column, QTableWidgetItem(str(value)))