205 lines
9.8 KiB
Python
205 lines
9.8 KiB
Python
"""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.setMinimumContentsLength(28)
|
||
self.protocol.setSizeAdjustPolicy(QComboBox.SizeAdjustPolicy.AdjustToContents)
|
||
self.protocol.setToolTip(
|
||
"Выберите формат CAN вручную. УМП / Balsam — legacy CAN; "
|
||
"карта 167/v1 — имена слов 0…127 и 0x5000…0x507F в ProtoCAN; "
|
||
"новые ID — назначенные диапазоны 5000…5127 и 6000…6127.")
|
||
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)
|