"""Reusable CAN monitor controls for host applications.""" from datetime import datetime from pathlib import Path from PySide6.QtCore import QTimer from PySide6.QtWidgets import (QWidget, QHBoxLayout, QVBoxLayout, QPushButton, QCheckBox, QComboBox, QLabel, QLineEdit, QTableWidget, QTableWidgetItem, QMessageBox, QApplication, QAbstractItemView, QFileDialog) from ..can_transport import build_frame from ..can_monitor import CanSession, PROTOCOLS, parse_filter, format_data, decode_protocol 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.save_button = QPushButton("Сохранить…") self.save_button.setToolTip("Сохранить весь буфер сессии в CSV (данные HEX), включая кадры вне таблицы") self.save_button.clicked.connect(self.save_session) layout.addWidget(self.save_button) 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()) self.protocol.currentIndexChanged.connect(lambda *_: self.refilter()) self.raw.toggled.connect(lambda *_: self.refilter()) 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.setToolTip("Фильтр только нижней таблицы; сверху отображаются все ID") 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) headers = ("Последний приём",) + CanSession.HEADER[1:6] + ("Счётчик", "Данные", "Расшифровка") self.history = QTableWidget(0, len(headers)) self.history.setHorizontalHeaderLabels(headers) self.history.setToolTip("Уникальные кадры и число их повторений с момента сброса") self.history.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers) self.history.horizontalHeader().setStretchLastSection(True) self.history.setColumnWidth(0, 190) self.history.setColumnWidth(7, 220) self.history.setColumnWidth(8, 500) 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 save_session(self): if not self.session.counts: QMessageBox.information(self, "Сохранение CAN", "Буфер сессии пуст.") return try: directory = Path(self.directory()) directory.mkdir(parents=True, exist_ok=True) suggested = directory / datetime.now().strftime("session_%Y%m%d_%H%M%S.csv") path, _ = QFileDialog.getSaveFileName( self, "Сохранить буфер CAN", str(suggested), "CSV (*.csv)") if not path: return saved = self.session.save(path) except (OSError, ValueError) as exc: QMessageBox.critical(self, "Сохранение CAN", str(exc)) return self.save_button.setToolTip("Сохранено: " + str(saved)) QMessageBox.information(self, "Сохранение CAN", "Сохранено: " + str(saved)) 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) unique_count = sum(1 for values, _ in self.session.unique_frames.values() if ids is None or int(values[3], 16) in ids) self.count.setText("Кадров: %d; уникальных: %d" % (count, unique_count)) self.history.horizontalHeaderItem(7).setText( "Данные (DEC)" if self.data_format.isChecked() else "Данные (HEX)") self.history.setColumnHidden(8, self.raw.isChecked()) self.history.setVisible(ids is not None) if ids is None: self.history.setRowCount(0) return frames = [(values, count) for values, count in self.session.unique_frames.values() if int(values[3], 16) in ids] self.history.setRowCount(len(frames)) for row, (values, count) in enumerate(frames): data = bytes.fromhex(values[6]) summary = "" if not self.raw.isChecked(): frame = build_frame(int(values[3], 16), data, ide=values[2] == "EXT", rtr=bool(values[4]), to_can=values[1] == "TX") summary = decode_protocol(frame, self.protocol.currentData()).summary display = values[:6] + (count, self.format_data(data), summary) for column, value in enumerate(display): item = self.history.item(row, column) if item is None: self.history.setItem(row, column, QTableWidgetItem(str(value))) elif item.text() != str(value): item.setText(str(value)) if column == 8: self.history.item(row, column).setToolTip(summary)