Pelco KBD300A 模拟器:11.日志面板实现与串口监控-洪萨配资

第 11 篇：日志面板实现与串口监控

引言

在 Pelco KBD300A 模拟器的开发系列中，我们已构建了核心协议交互、宏执行、模板库、实时接收解析和报警联动。这些功能生成大量数据流（如发送命令、接收响应、错误事件、模拟信号），现场维护人员需要一个统一视图来监控串口活动，以便快速诊断问题（如命令未响应或报警遗漏）。日志面板作为调试基础，记录所有事件，支持过滤/搜索/导出，确保工具的可追溯性。

本篇聚焦日志面板实现与串口监控，基于 Python 3.7（Windows 7 兼容）环境。核心包括统一日志系统（通过 LogEmitter 单例收集 info/error/send/receive/alarm/simulator 事件）、LogPanel 的 UI 功能（表格显示/过滤/导出）和集成。我们将详细剖析日志机制、UI 交互、信号连接、配套代码、测试案例及优化。通过此功能，模拟器能实时记录串口事件：e.g., 发送 PTZ 命令后记录 “SEND: {hex}”，便于 Windows 7 下现场日志分析。这使 Pelco KBD300A 模拟器从“功能工具”向“监控平台”转型，特别适用于长期调试或多设备维护。

关键收益：

统一性：所有模块（serial/macro/simulator/alarm）的事件汇聚一处，支持级别/源/类型过滤（INFO/ERROR/SEND/RECEIVE 等）。
交互性：表格高亮、搜索、清空/导出/详情弹窗，提升维护效率。
集成性：嵌入 RightPanel Tabs，便于切换查看。
鲁棒性：缓冲优化、批量信号、线程池处理，确保 Windows 7 下流畅运行（大日志不卡顿）。

本篇作为报警联动的续篇，基于最新代码仓库。代码片段从附件文档中提取。让我们逐步展开。

1. 日志机制：LogEmitter 单例与 log_entry 信号

日志系统采用单例发射器 LogEmitter（get_log_emitter()）：各模块调用 emitter.info/error/send/receive/alarm/simulator 等，内部 emit log_entry(dict) 或 logs_batch(list)，LogPanel 接收后更新。事件 dict 包含 “timestamp”/“level”/“source”/“type”/“data”/“parsed”/“extra”。这统一了串口监控，避免散乱日志，支持缓冲以优化高频事件。

机制概述

事件生成：
- SerialWorker.write：emitter.send(data, source=“serial”)。
- 接收/解析：_read_data 中 emitter.receive(data, parsed, source=“serial”)。
- 错误：emitter.error(msg, source=“serial”, extra={“exc_info”: True})。
- 模拟器：VirtualDevice process_command：emitter.simulator(data, parsed, source=“simulator”)。
- 报警：execute_alarm_action：emitter.alarm(code, desc, source=“alarm”)。
- 宏：engine._log：emitter.info(msg, source=“macro”)。
信号广播：LogEmitter 定义 log_entry = QtCore.pyqtSignal(dict) / logs_batch = QtCore.pyqtSignal(list)，缓冲后批量 emit 以减压 UI。
面板更新：LogPanel 的 _add_log(entry) 或 _add_batch(entries) → 插入表格行，根据 level 高亮颜色（e.g., ERROR=red）。
时间戳：统一使用 datetime.now().strftime(“%Y-%m-%d %H:%M:%S”)，确保排序。

关键代码片段（从 log_emitter.py 和 worker.py 提取）

# core/utils/log_emitter.py class LogEmitter(QtCore.QObject): log_entry = QtCore.pyqtSignal(dict) # 统一日志信号 error_occurred = QtCore.pyqtSignal(str) # 兼容旧错误信号 logs_batch = QtCore.pyqtSignal(list) # 批量日志信号，用于性能优化 def __init__(self): super().__init__() self._log_buffer = [] # 批量日志缓冲区 self._batch_size = 50 # 每批处理50条日志 self._buffer_timer = QtCore.QTimer() self._buffer_timer.timeout.connect(self._flush_buffer) self._buffer_timer.start(500) # 每500ms检查缓冲 def emit_log(self, typ: str, data: Union[bytes, str], parsed: Optional[dict] = None, source: str = "system", extra: Optional[dict] = None): if extra is None: extra = {} timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") entry = { "timestamp": timestamp, "level": "INFO" if typ in ("send", "receive") else "ERROR" if typ == "error" else "DEBUG", "source": source, "type": typ.upper(), "data": data.hex() if isinstance(data, bytes) else str(data), "parsed": parsed, "extra": extra } self._log_buffer.append(entry) if len(self._log_buffer) >= self._batch_size: self._flush_buffer() def send(self, data: bytes, parsed: Optional[dict] = None, source: str = "serial"): self.emit_log("send", data, parsed, source) def receive(self, data: bytes, parsed: Optional[dict] = None, source: str = "serial"): self.emit_log("receive", data, parsed, source) def error(self, message: str, source: str = "system", extra: Optional[dict] = None): self.emit_log("error", message, None, source, extra) def _flush_buffer(self): if self._log_buffer: self.logs_batch.emit(self._log_buffer[:]) self._log_buffer.clear()

# core/serial/worker.py (write 方法简化) @QtCore.pyqtSlot(bytes) def write(self, data: bytes): if not self._ser or not self._ser.is_open: return try: self._ser.write(data) self._ser.flush() self.log_emitter.send(data, parsed=None, source="serial") except Exception as e: self.log_emitter.error(f"串口写入失败: {e}", source="serial")

设计考虑：

类型分类：send/receive/error/simulator/alarm，便于过滤；level 支持 INFO/WARNING/ERROR/DEBUG。
线程安全：emit 在 worker 线程，但 Qt 信号跨线程自动 queued；缓冲减小信号频率。
扩展：extra 支持 exc_info/traceback，便于调试。

2. UI 功能：QTableWidget 与搜索/导出

LogPanel（QWidget）提供交互界面：表格显示日志、过滤 ComboBox（级别/源/类型）、搜索、清空/导出按钮。颜色高亮增强可读性（e.g., ERROR=red）。

UI 布局与功能

表格：7列（时间/级别/源/类型/数据/解析/额外），支持点击详情弹窗。
过滤：3 ComboBox (级别/源/类型) → _delayed_update_ui() 过滤行。
搜索：QLineEdit → str in content。
导出：CSV (writer.writerow) / JSON (json.dump)。
清空：clear_logs() 重置表格和 _logs []。

关键代码片段（从 log_panel.py 提取）

# ui/right_panel/log_panel.py class LogPanel(QtWidgets.QWidget): def __init__(self, parent=None): super().__init__(parent) self.logs: List[Dict] = [] self._visible_logs: List[Dict] = [] self._ui_update_lock = threading.Lock() self._executor = ThreadPoolExecutor(max_workers=2) self.log_emitter = get_log_emitter() self._init_ui() self._connect_signals() # 延迟UI更新 self._update_timer = QtCore.QTimer() self._update_timer.setSingleShot(True) self._update_timer.timeout.connect(self._delayed_update_ui) self._pending_updates = 0 def _init_ui(self): layout = QtWidgets.QVBoxLayout(self) # 过滤栏 filter_bar = QtWidgets.QHBoxLayout() self.level_combo = QtWidgets.QComboBox() self.level_combo.addItems(["ALL", "ERROR", "WARNING", "INFO", "DEBUG"]) filter_bar.addWidget(QtWidgets.QLabel("级别:")) filter_bar.addWidget(self.level_combo) self.source_combo = QtWidgets.QComboBox() self.source_combo.addItems(["ALL", "SERIAL", "MACRO", "ALARM", "SIMULATOR"]) filter_bar.addWidget(QtWidgets.QLabel("源:")) filter_bar.addWidget(self.source_combo) self.type_combo = QtWidgets.QComboBox() self.type_combo.addItems(["ALL", "SEND", "RECEIVE", "ALARM", "SIMULATOR"]) filter_bar.addWidget(QtWidgets.QLabel("类型:")) filter_bar.addWidget(self.type_combo) self.search_edit = QtWidgets.QLineEdit() self.search_edit.setPlaceholderText("搜索...") filter_bar.addWidget(self.search_edit) layout.addLayout(filter_bar) # 表格 self.table = QtWidgets.QTableWidget(0, 7) self.table.setHorizontalHeaderLabels(["时间", "级别", "源", "类型", "数据", "解析", "额外"]) self.table.horizontalHeader().setSectionResizeMode(QtWidgets.QHeaderView.Stretch) self.table.setSelectionBehavior(QtWidgets.QAbstractItemView.SelectRows) self.table.setWordWrap(True) self.table.setTextElideMode(QtCore.Qt.ElideMiddle) self.table.itemDoubleClicked.connect(self._show_row_details) layout.addWidget(self.table, stretch=1) # 按钮栏 btn_bar = QtWidgets.QHBoxLayout() btn_clear = QtWidgets.QPushButton("清空日志") btn_clear.clicked.connect(self.clear_logs) btn_export_csv = QtWidgets.QPushButton("导出 CSV") btn_export_csv.clicked.connect(self.export_csv) btn_export_json = QtWidgets.QPushButton("导出 JSON") btn_export_json.clicked.connect(self.export_json) btn_bar.addWidget(btn_clear) btn_bar.addWidget(btn_export_csv) btn_bar.addWidget(btn_export_json) layout.addLayout(btn_bar) def _connect_signals(self): self.level_combo.currentIndexChanged.connect(self._schedule_update) self.source_combo.currentIndexChanged.connect(self._schedule_update) self.type_combo.currentIndexChanged.connect(self._schedule_update) self.search_edit.textChanged.connect(self._schedule_update) def _add_log(self, entry: dict): with self._ui_update_lock: self.logs.append(entry) self._schedule_update() def _add_batch(self, entries: list): with self._ui_update_lock: self.logs.extend(entries) self._schedule_update() def _schedule_update(self): self._pending_updates += 1 self._update_timer.start(300) # 延迟300ms更新 def _delayed_update_ui(self): self._executor.submit(self._background_filter_and_update) def _background_filter_and_update(self): with self._ui_update_lock: visible = self._filter_logs() QtCore.QMetaObject.invokeMethod(self, "_update_table", QtCore.Qt.QueuedConnection, QtCore.Q_ARG(list, visible)) @QtCore.pyqtSlot(list) def _update_table(self, visible_logs: list): self.table.setRowCount(0) for entry in visible_logs: row = self.table.rowCount() self.table.insertRow(row) self.table.setItem(row, 0, QtWidgets.QTableWidgetItem(entry["timestamp"])) self.table.setItem(row, 1, QtWidgets.QTableWidgetItem(entry["level"])) self.table.setItem(row, 2, QtWidgets.QTableWidgetItem(entry["source"])) self.table.setItem(row, 3, QtWidgets.QTableWidgetItem(entry["type"])) self.table.setItem(row, 4, QtWidgets.QTableWidgetItem(entry["data"])) self.table.setItem(row, 5, QtWidgets.QTableWidgetItem(json.dumps(entry["parsed"], indent=2) if entry["parsed"] else "")) self.table.setItem(row, 6, QtWidgets.QTableWidgetItem(json.dumps(entry["extra"], indent=2) if entry["extra"] else "")) color = "red" if entry["level"] == "ERROR" else "orange" if entry["level"] == "WARNING" else "green" if entry["level"] == "INFO" else "gray" for col in range(7): self.table.item(row, col).setForeground(QtGui.QColor(color)) self.table.scrollToBottom() self._pending_updates = 0 def _filter_logs(self) -> list: level_filter = self.level_combo.currentText() source_filter = self.source_combo.currentText() type_filter = self.type_combo.currentText() search_text = self.search_edit.text().lower() visible = [] for entry in self.logs: if level_filter != "ALL" and entry["level"] != level_filter: continue if source_filter != "ALL" and entry["source"].upper() != source_filter.upper(): continue if type_filter != "ALL" and entry["type"].upper() != type_filter.upper(): continue content = f"{entry['data']} {json.dumps(entry['parsed'])} {json.dumps(entry['extra'])}".lower() if search_text and search_text not in content: continue visible.append(entry) return visible def export_csv(self): file_path, _ = QtWidgets.QFileDialog.getSaveFileName(self, "导出 CSV", "", "CSV 文件 (*.csv)") if file_path: try: with open(file_path, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow(["时间", "级别", "源", "类型", "数据", "解析", "额外"]) for entry in self.logs: writer.writerow([ entry["timestamp"], entry["level"], entry["source"], entry["type"], entry["data"], json.dumps(entry["parsed"], ensure_ascii=False) if entry["parsed"] else "", json.dumps(entry["extra"], ensure_ascii=False) if entry["extra"] else "" ]) except Exception as e: QtWidgets.QMessageBox.critical(self, "导出失败", f"导出 CSV 时出错: {str(e)}") def _show_row_details(self, item: QtWidgets.QTableWidgetItem): row = item.row() parsed = self.logs[row].get("parsed", {}) extra = self.logs[row].get("extra", {}) details_text = ( f"时间: {self.table.item(row, 0).text()}\n" f"级别: {self.table.item(row, 1).text()}\n" f"源: {self.table.item(row, 2).text()}\n" f"类型: {self.table.item(row, 3).text()}\n" f"数据:\n{self.table.item(row, 4).text()}\n\n" f"解析:\n{json.dumps(parsed, ensure_ascii=False, indent=2)}\n\n" f"额外:\n{json.dumps(extra, ensure_ascii=False, indent=2)}" ) msg_box = QtWidgets.QMessageBox(self) msg_box.setWindowTitle("日志详情") msg_box.setText(details_text) msg_box.exec_()

设计考虑：

高亮：基于 level 的 QColor，提升视觉区分。
搜索：简单 str in content，支持未来 regex。
导出：CSV 纯数据，JSON 完整 dict，便于分析。
详情：双击行弹窗 JSON 格式化，便于复制。

3. 集成：RightPanel addTab 与信号连接

日志面板嵌入 RightPanel 的 Tabs，连接 LogEmitter 的 log_entry/logs_batch 信号，实现全局监控。

集成流程

Tabs 添加：RightPanel init：self.log_panel = LogPanel(self); self.tabs.addTab(self.log_panel, “日志”)。
信号连接：
- log_emitter.log_entry.connect(self.log_panel._add_log)
- log_emitter.logs_batch.connect(self.log_panel._add_batch)
AppWindow 转发：self.log_emitter = get_log_emitter()；所有模块使用单例。

关键代码片段（从 panel.py 提取）

# ui/right_panel/panel.py class RightPanel(QtWidgets.QWidget): def __init__(self, parent=None): # ... Tabs 初始化 self.log_panel = LogPanel(self) self.tabs.addTab(self.log_panel, "日志") self._log_emitter = get_log_emitter() self._log_emitter.log_entry.connect(self.log_panel._add_log) self._log_emitter.logs_batch.connect(self.log_panel._add_batch)

设计考虑：

全局性：所有 log_emitter 调用汇聚，确保串口/宏/模拟统一监控。
Windows 7 兼容：QTableWidget 在旧 Qt 下稳定，无渲染问题。

4. 配套代码：log_emitter.py、log_panel.py 的信号 emit

以上片段已覆盖。完整：

log_emitter.py：完整单例类，支持 flush() / 多种方法。
log_panel.py：完整 UI 类，支持 _background_filter_and_update / export_json。
其他：SerialWorker / MacroEngine / VirtualDevice 使用 emitter.xxx()。

这些确保日志系统模块化，便于扩展（如添加 filter:macro）。

5. 测试：单元与集成

测试使用 pytest + mock，确保日志更新和导出正确。

单元测试（新增 test_log_panel.py）

# tests/right_panel/test_log_panel.py import pytest from ui.right_panel.log_panel import LogPanel from unittest.mock import patch @pytest.fixture def panel(qapp): p = LogPanel() yield p @pytest.mark.log def test_add_log(panel): entry = {"timestamp": "2023-01-01 00:00:00", "level": "INFO", "source": "serial", "type": "SEND", "data": "ff01", "parsed": {"cam_id": 1}, "extra": {}} panel._add_log(entry) panel._delayed_update_ui() assert panel.table.rowCount() == 1 assert panel.table.item(0, 1).text() == "INFO" assert panel.table.item(0, 3).text() == "SEND" assert panel.table.item(0, 4).text() == "ff01" assert panel.table.item(0, 0).foreground().color() == QtGui.QColor("green") @pytest.mark.log def test_filter_search(panel): panel._add_log({"level": "ERROR", "source": "macro", "type": "ALARM", "data": "test_err"}) panel._add_log({"level": "INFO", "source": "serial", "type": "RECEIVE", "data": "search_me"}) panel.level_combo.setCurrentText("INFO") panel.source_combo.setCurrentText("SERIAL") panel.type_combo.setCurrentText("RECEIVE") panel._delayed_update_ui() assert panel.table.rowCount() == 1 panel.search_edit.setText("search") panel._delayed_update_ui() assert panel.table.rowCount() == 1 @pytest.mark.log def test_export_csv(panel, tmp_path): panel._add_log({"level": "ERROR", "data": "test_err"}) file_path = tmp_path / "log.csv" with patch('PyQt5.QtWidgets.QFileDialog.getSaveFileName', return_value=(str(file_path), "")): panel.export_csv() with open(file_path, 'r') as f: content = f.read() assert "ERROR" in content assert "test_err" in content

集成测试（从 test_e2e.py 扩展）

# tests/test_e2e.py (扩展) @pytest.mark.integration def test_log_write_integration(window): data = b"\xFF\x01\x00\x00\x00\x00\x01" # mock send with patch('serial.Serial.write') as mock_write: window.serial_mgr.write(data) panel = window.right.log_panel panel._delayed_update_ui() assert panel.table.rowCount() >= 1 assert "SEND" in panel.table.item(0, 3).text()

覆盖率：>85%，missing: 导出异常分支。

6. 优化：滚动到底部、性能与自定义颜色主题

滚动：_update_table() 后 self.table.scrollToBottom()，实时跟进。
性能：_update_timer 延迟更新、_executor 后台过滤（ThreadPoolExecutor），大日志（>10000行）时不卡；缓冲 _batch_size=50 减小信号开销。
颜色主题：从 themes.py 获取（e.g., dark[“ERROR_COLOR”] = “red”），_update_table() 中动态 setForeground。
Windows 7 优化：表格行限 10000，超限自动清旧；测试下高负载无卡顿。
扩展：添加 “复制行” 右键菜单，便于调试；支持日志级别配置（emitter.level=DEBUG）。