"""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)))