#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ pPowerManger SQLite 数据查看工具 (tkinter GUI, 仅使用 Python 标准库) 功能: 1. 数据浏览 - 打开 power_data.db, 浏览各表, 支持行数限制/时间筛选/条件过滤/人性化解码 2. SQL 查询 - 执行任意只读 SQL, 结果表格展示, 导出 CSV 3. 帧解码 - 解析 msg_log 中的原始 hex 帧, 按 msg_id 还原结构体字段(中文标注) 4. 趋势图 - 选中数值列, 按时间绘制折线图(支持时间桶聚合) 依赖: tkinter(系统包 python3-tk), 其余均为标准库(sqlite3/csv/struct/datetime) 运行: python3 tools/pmpower_db.py """ import csv import os import struct import sqlite3 from datetime import datetime, timedelta try: import tkinter as tk from tkinter import ttk, filedialog, messagebox except Exception: # tkinter 未安装时仍可导入本模块用于逻辑测试 tk = None ttk = None APP_TITLE = "pPowerManger 数据查看工具" # 由 setup_fonts() 在启动时设置的实际可用中文字体族 _UI_SANS = "TkDefaultFont" _UI_MONO = "TkFixedFont" # ===================================================================== # 枚举映射 # ===================================================================== BREAKER_MAP = {0x00: "无效", 0x55: "合闸", 0xAA: "分闸", 0x5A: "故障"} SIGNAL_MAP = {0x00: "无效", 0x55: "正常", 0xAA: "故障"} INSUL_MAP = {0x00: "无效", 0x55: "正常", 0xAA: "绝缘低"} ONOFF_MAP = {0x00: "无效", 0x55: "启动", 0xAA: "停止", 0x5A: "故障"} ENUMS = { "breaker": BREAKER_MAP, "signal": SIGNAL_MAP, "insulation": INSUL_MAP, "onoff": ONOFF_MAP, "work_condition": { 0: "工况切换中", 1: "待机", 2: "岸基备航", 3: "水中备航", 4: "遥控工况", 5: "巡航模式", 6: "高速模式", 7: "上浮下潜", 8: "浮调模式", 9: "水下侦查", 10: "水面侦查", 11: "DJ模式", 12: "调试模式", }, "fcu_cmd": { 0x01: "自检", 0x02: "启动", 0x03: "停机", 0x04: "复位", 0x05: "急停", 0x06: "补给", 0x07: "预热", 0x08: "维护", }, "fcu_state": { 0x01: "初始化", 0x02: "待机", 0x03: "自检异常", 0x04: "就绪", 0x05: "启动", 0x06: "运行", 0x07: "停机", 0x08: "故障", 0x09: "补给", 0x0A: "预热", 0x0B: "预热完成", 0x0C: "维护", 0x0D: "维护完成", }, "battery_cmd": {0x00: "保持", 0x10: "启动", 0x20: "停止"}, "fc_mode": {0x00: "无效", 0x01: "调试", 0x02: "自动", 0x03: "补给"}, "fc_fault_level": {0x00: "无故障", 0x10: "一级", 0x20: "二级", 0x30: "三级", 0x40: "四级"}, "charge_status": {0x00: "无效", 0x10: "充电", 0x20: "放电"}, "emergency_status": {0x00: "无报警", 0x3A: "4级报警", 0x3B: "3级报警", 0x3C: "2级报警", 0x3D: "1级报警"}, "flame": {0x00: "无效", 0x01: "报警", 0x02: "故障"}, "direction": {0: "接收", 1: "发送"}, "checksum_ok": {0: "错误", 1: "正确"}, } # ===================================================================== # 列元数据: 表 -> {列名: (中文名, 缩放倍数, 单位, 枚举key 或 None)} # ===================================================================== COL_META = {} COL_META["ccuColCmd"] = { "workCondition": ("工况", 1, "", "work_condition"), "fcuCmd": ("燃料电池指令", 1, "", "fcu_cmd"), "fcuPowerEfficiency": ("燃料电池功率效率", 1, "%", None), "insChargeEnable": ("仪表电池充电使能", 1, "", None), "insBatCmd": ("仪表电池指令", 1, "", "battery_cmd"), "powerBatCmd": ("动力电池指令", 1, "", "battery_cmd"), "powerBatEfficiency": ("动力电池效率", 1, "%", None), "heartbeat": ("心跳计数", 1, "", None), "masterState": ("主机状态", 1, "", "hex"), "paceholder1": ("占位字段1", 1, "", "hex"), "paceholder2": ("占位字段2", 1, "", "hex"), } COL_META["ccuSetParmCmd"] = { "insSocHold1": ("仪表SOC保持1", 1, "%", None), "insSocHold2": ("仪表SOC保持2", 1, "%", None), "insSocHold3": ("仪表SOC保持3", 1, "%", None), "fcuToInsLimit1": ("燃电到仪表限值1", 1, "", None), "fcuToInsLimit2": ("燃电到仪表限值2", 1, "", None), "powerSocHold1": ("动力SOC保持1", 1, "%", None), "powerSocHold2": ("动力SOC保持2", 1, "%", None), "powerSocHold3": ("动力SOC保持3", 1, "%", None), "fcuToPowerLimit1": ("燃电到动力限值1", 1, "", None), "fcuToPowerLimit2": ("燃电到动力限值2", 1, "", None), } COL_META["ccuSetParmFb"] = { "flag": ("标志", 1, "", "hex"), "failureCode": ("故障码", 1, "", "hex"), } # --- ccuState 复合管控状态反馈 --- COL_META["ccuState"] = { "fc_mode": ("燃料电池运行模式", 1, "", "fc_mode"), "fc_status": ("燃料电池状态", 1, "", "fcu_state"), "fc_fault_level": ("燃料电池故障等级", 1, "", "fc_fault_level"), "output_power_limit": ("输出功率限制", 1, "W", None), "total_generation_time": ("累计发电时间", 0.1, "h", None), "total_generation_power": ("累计发电功率", 1, "W", None), "fault_level_1": ("一级故障位", 1, "", "hex"), "fault_level_2": ("二级故障位", 1, "", "hex"), "fault_level_3": ("三级故障位", 1, "", "hex"), "fault_level_4": ("四级故障位", 1, "", "hex"), "hydrogen_capacity": ("储氢罐剩余容量", 1, "%", None), "liquid_oxygen_capacity": ("液氧罐剩余容量", 1, "%", None), "palladium_temp": ("钯膜最高温度", 1, "℃", None), "main_pipe_pressure": ("主水路压力", 0.01, "kPa", None), "aux_pipe_pressure": ("辅水路压力", 0.01, "kPa", None), "cabin_pressure1": ("舱室1压力", 0.01, "kPa", None), "cabin_pressure2": ("舱室2压力", 0.01, "kPa", None), "cabin_temp1": ("舱室1温度", 0.01, "℃", None), "cabin_temp2": ("舱室2温度", 0.01, "℃", None), "cabin_humidity1": ("舱室1湿度", 0.01, "%RH", None), "cabin_humidity2": ("舱室2湿度", 0.01, "%RH", None), "flame_detector1": ("火焰探测器1", 1, "", "flame"), "flame_detector2": ("火焰探测器2", 1, "", "flame"), "h2_concentration1": ("舱室H2浓度1", 0.01, "%LEL", None), "h2_concentration2": ("舱室H2浓度2", 0.01, "%LEL", None), "h2_concentration3": ("舱室H2浓度3", 0.01, "%LEL", None), "o2_concentration1": ("舱室O2浓度1", 0.01, "%LEL", None), "o2_concentration2": ("舱室O2浓度2", 0.01, "%LEL", None), "ch3oh_concentration1": ("舱室甲醇浓度1", 0.01, "%LEL", None), "ch3oh_concentration2": ("舱室甲醇浓度2", 0.01, "%LEL", None), "reserved1_1": ("保留1_1", 1, "", None), "reserved1_2": ("保留1_2", 1, "", None), "emergency_battery1_voltage": ("应急电池1总电压", 1, "V", None), "emergency_battery1_current": ("应急电池1总电流", 1, "A", None), "emergency_battery1_max_temp": ("应急电池1最高温度", 1, "℃", None), "emergency_battery1_fault_word": ("应急电池1故障字", 1, "", "hex"), "emergency_battery2_voltage": ("应急电池2总电压", 1, "V", None), "emergency_battery2_current": ("应急电池2总电流", 1, "A", None), "emergency_battery2_max_temp": ("应急电池2最高温度", 1, "℃", None), "emergency_battery2_fault_word": ("应急电池2故障字", 1, "", "hex"), "cabin_ox_concentration": ("仪表舱氧气浓度", 1, "%", None), "cabin_temperature": ("仪表舱温度", 1, "℃", None), "cabin_humidity": ("仪表舱湿度", 1, "%RH", None), "cabin_pressure": ("仪表舱大气压", 1, "kPa", None), "power_cabin_ox_concentration": ("动力舱氧气浓度", 1, "%", None), "power_cabin_temperature": ("动力舱温度", 1, "℃", None), "power_cabin_humidity": ("动力舱湿度", 1, "%RH", None), "power_cabin_pressure": ("动力舱大气压", 1, "kPa", None), "reserved2": ("保留2", 1, "", None), "dyn_alarm_flag1": ("动力电池报警1", 1, "", "hex"), "dyn_alarm_flag2": ("动力电池报警2", 1, "", "hex"), "dyn_alarm_flag3": ("动力电池报警3", 1, "", "hex"), "dyn_alarm_flag4": ("动力电池报警4", 1, "", "hex"), "dyn_alarm_flag5": ("动力电池报警5", 1, "", "hex"), "dyn_alarm_flag6": ("动力电池报警6", 1, "", "hex"), "ins_alarm_flag1": ("仪表电池报警1", 1, "", "hex"), "ins_alarm_flag2": ("仪表电池报警2", 1, "", "hex"), "ins_alarm_flag3": ("仪表电池报警3", 1, "", "hex"), "ins_alarm_flag4": ("仪表电池报警4", 1, "", "hex"), "ins_alarm_flag5": ("仪表电池报警5", 1, "", "hex"), "ins_alarm_flag6": ("仪表电池报警6", 1, "", "hex"), "ins_relay_status1": ("仪表继电器正极/充电", 1, "", "hex"), "ins_relay_status2": ("仪表继电器预充/负极", 1, "", "hex"), "dyn_relay_status1": ("动力继电器正极/充电", 1, "", "hex"), "dyn_relay_status2": ("动力继电器预充/负极", 1, "", "hex"), "ins_max_discharge_power": ("仪表电池最大放电功率", 0.05, "kW", None), "dyn_max_discharge_power": ("动力电池最大放电功率", 0.05, "kW", None), "ins_soc": ("仪表锂电池SOC", 1, "%", None), "dyn_soc": ("动力锂电池SOC", 1, "%", None), "ins_total_energy": ("仪表电池总能量", 1, "kWh", None), "dyn_total_energy": ("动力电池总能量", 1, "kWh", None), "ins_power_input": ("仪表电池输入功率", 1, "kW", None), "dyn_power_input": ("动力电池输入功率", 1, "kW", None), "ins_charge_status": ("仪表电池充放电", 1, "", "charge_status"), "dyn_charge_status": ("动力电池充放电", 1, "", "charge_status"), "ins_voltage_link": ("仪表电池LINK电压", 0.01, "V", None), "ins_voltage_pack": ("仪表电池PACK电压", 0.01, "V", None), "ins_current": ("仪表电池电流", 0.01, "A", None), "ins_resistance_pos": ("仪表电池正极绝缘电阻", 1, "kΩ", None), "ins_resistance_neg": ("仪表电池负极绝缘电阻", 1, "kΩ", None), "dyn_voltage_link": ("动力电池LINK电压", 0.01, "V", None), "dyn_voltage_pack": ("动力电池PACK电压", 0.01, "V", None), "dyn_current": ("动力电池电流", 0.01, "A", None), "dyn_resistance_pos": ("动力电池正极绝缘电阻", 1, "kΩ", None), "dyn_resistance_neg": ("动力电池负极绝缘电阻", 1, "kΩ", None), "ins_emergency_status": ("仪表电池紧急状态", 1, "", "emergency_status"), "dyn_emergency_status": ("动力电池紧急状态", 1, "", "emergency_status"), "reserved3_1": ("保留3_1", 1, "", None), "reserved3_2": ("保留3_2", 1, "", None), "ins_soc_threshold1": ("仪表电池SOC门限1", 1, "%", None), "ins_soc_threshold2": ("仪表电池SOC门限2", 1, "%", None), "ins_soc_threshold3": ("仪表电池SOC门限3", 1, "%", None), "dyn_soc_threshold1": ("动力电池SOC门限1", 1, "%", None), "dyn_soc_threshold2": ("动力电池SOC门限2", 1, "%", None), "dyn_soc_threshold3": ("动力电池SOC门限3", 1, "%", None), "ins_power_limit1": ("仪表电池功率限制1", 1, "%", None), "ins_power_limit2": ("仪表电池功率限制2", 1, "%", None), "dyn_power_limit1": ("动力电池功率限制1", 1, "%", None), "dyn_power_limit2": ("动力电池功率限制2", 1, "%", None), "device_online_flag1": ("设备在线标志1", 1, "", "hex"), "device_online_flag2": ("设备在线标志2", 1, "", "hex"), "reserved4": ("保留4", 1, "", None), "heartbeat": ("心跳计数", 1, "", None), } # --- 高压母线配电指令 --- COL_META["disHighVolBusCmd"] = { "fuelCellCircuitBreaker": ("燃料电池断路器", 1, "", "breaker"), "powerLithiumBatteryCircuitBreaker": ("动力锂电池断路器", 1, "", "breaker"), "propulsionMotorCircuitBreaker": ("推进电机断路器", 1, "", "breaker"), "lithiumBatteryGroupInstrumentCircuitBreaker": ("锂电池组仪表断路器", 1, "", "breaker"), "dcDc5ModuleCircuitBreaker": ("DC/DC5模块断路器", 1, "", "breaker"), "bowHighVoltageDistributionBoxCircuitBreaker": ("艏部高压配电箱断路器", 1, "", "breaker"), "sternFTDevice45CircuitBreaker": ("艉部FT装置4/5断路器", 1, "", "breaker"), "sternRudderSwitch1CircuitBreaker": ("艉舵开关1断路器", 1, "", "breaker"), "sternRudderSwitch2CircuitBreaker": ("艉舵开关2断路器", 1, "", "breaker"), "fbReservedCircuitBreaker": ("FB预留断路器", 1, "", "breaker"), "tyzReservedCircuitBreaker": ("TYZ预留断路器", 1, "", "breaker"), "reservedCircuitBreaker": ("预留断路器", 1, "", "breaker"), "dcdc5Module": ("DC/DC5模块启动", 1, "", "onoff"), "coolingSystem": ("冷却系统启动", 1, "", "onoff"), } # --- 高压母线A配电指令 --- COL_META["disHighAVolBusCmd"] = { "bowFTDevice123CircuitBreaker": ("艏部FT装置1/2/3断路器", 1, "", "breaker"), "actuatorCircuitBreaker": ("启闭机构断路器", 1, "", "breaker"), "mastSteeringGearControlBoxCircuitBreaker": ("桅杆舵机控制箱断路器", 1, "", "breaker"), "xczCircuitBreaker": ("XCZ断路器", 1, "", "breaker"), "bowRudderControlBoxCircuitBreaker": ("艏舵控制箱断路器", 1, "", "breaker"), "openWaterCoverStartCylinderCircuitBreaker": ("敞水盖启动电缸断路器", 1, "", "breaker"), } # --- 低压主母排配电指令 --- COL_META["disLowMainBusCmd"] = { "lithiumBatteryGroupInstrumentCircuitBreaker": ("锂电池组(仪表)断路器", 1, "", "breaker"), "bowLowVoltageDistributionBoxCircuitBreaker": ("艏部低压配电箱断路器", 1, "", "breaker"), "unit4InstrumentDC48VCircuitBreaker": ("UNIT4仪表DC48V断路器", 1, "", "breaker"), "dcC1DCDistributionPanelCircuitBreaker": ("DC-C1直流配电板断路器", 1, "", "breaker"), "reservedCircuitBreaker1": ("预留断路器1", 1, "", "breaker"), "reservedCircuitBreaker2": ("预留断路器2", 1, "", "breaker"), "emergencyLithiumBatteryGroup2CircuitBreaker": ("应急锂电池组2断路器", 1, "", "breaker"), } # --- 低压母线配电指令 --- COL_META["disLowBusCmd"] = { "unit1CircuitBreaker": ("UNIT1断路器", 1, "", "breaker"), "unit2CircuitBreaker": ("UNIT2断路器", 1, "", "breaker"), "unit3CircuitBreaker": ("UNIT3断路器", 1, "", "breaker"), "unit5CircuitBreaker": ("UNIT5断路器", 1, "", "breaker"), "bowPZDeviceCircuitBreaker": ("艏部PZ装置断路器", 1, "", "breaker"), "reservedCircuitBreaker1": ("预留断路器1", 1, "", "breaker"), "reservedCircuitBreaker2": ("预留断路器2", 1, "", "breaker"), "emergencyLithiumBatteryGroup1CircuitBreaker": ("应急锂电池组1断路器", 1, "", "breaker"), } # --- 高压母线配电反馈 --- COL_META["disHighVolBusState"] = { "fuelCellCircuitBreaker": ("燃料电池断路器", 1, "", "breaker"), "powerLithiumBatteryCircuitBreaker": ("动力锂电池断路器", 1, "", "breaker"), "propulsionMotorCircuitBreaker": ("推进电机断路器", 1, "", "breaker"), "lithiumBatteryGroupInstrumentCircuitBreaker": ("锂电池组仪表断路器", 1, "", "breaker"), "dcDc5ModuleCircuitBreaker": ("DC/DC5模块断路器", 1, "", "breaker"), "bowHighVoltageDistributionBoxCircuitBreaker": ("艏部高压配电箱断路器", 1, "", "breaker"), "sternFTDevice45CircuitBreaker": ("艉部FT装置4/5断路器", 1, "", "breaker"), "sternRudderSwitch1CircuitBreaker": ("艉舵开关1断路器", 1, "", "breaker"), "sternRudderSwitch2CircuitBreaker": ("艉舵开关2断路器", 1, "", "breaker"), "fbReservedCircuitBreaker": ("FB预留断路器", 1, "", "breaker"), "tyzReservedCircuitBreaker": ("TYZ预留断路器", 1, "", "breaker"), "reservedCircuitBreaker": ("预留断路器", 1, "", "breaker"), "dcDcFaultWord1": ("DC/DC5故障字低", 1, "", "hex"), "dcDcFaultWord2": ("DC/DC5故障字高", 1, "", "hex"), "powerBusInsulationStatus": ("动力母排绝缘状态", 1, "", "insulation"), "aBusInsulationStatus": ("A汇流排绝缘状态", 1, "", "insulation"), "meterPowerLossSignal": ("仪表电源状态", 1, "", "signal"), "emergencyPowerLossSignal": ("应急电源状态", 1, "", "signal"), "powerBusVoltage": ("动力汇流排电压", 1, "", None), "dcDc5ModuleCurrent": ("DC/DC5模块电流", 1, "", None), "powerBusCurrent": ("动力汇流排电流", 1, "", None), "busbarAVoltage": ("A汇流排电压", 1, "", None), "propulsionMotorControlBoxCurrent": ("推进电机控制箱电流", 1, "", None), "busbarACurrent": ("A汇流排电流", 1, "", None), "lithiumBatteryGroupMeterCurrent": ("锂电池组(仪表)电流", 1, "", None), "bowHighVoltageDistributionBoxCurrent": ("艏部高压配电箱电流", 1, "", None), "sternFTDevice45Current": ("艉部FT装置4/5电流", 1, "", None), "tyzReservedSwitchCurrent": ("TYZ预留开关电流", 1, "", None), "sternRudderSwitch1Current": ("艉舵开关1电流", 1, "", None), "reservedCurrent1": ("预留电流1", 1, "", None), "sternRudderSwitch2Current": ("艉舵开关2电流", 1, "", None), "reservedCurrent2": ("预留电流2", 1, "", None), "fbReservedSwitchCurrent": ("FB预留电流", 1, "", None), "reservedCurrent3": ("预留电流3", 1, "", None), "coolWaterPressure": ("冷却水压力", 1, "", None), } # --- 高压母线A配电反馈 --- COL_META["disHighAVolBusState"] = { "bowFTDevice123CircuitBreaker": ("艏部FT装置1/2/3断路器", 1, "", "breaker"), "actuatorCircuitBreaker": ("启闭机构断路器", 1, "", "breaker"), "mastSteeringGearControlBoxCircuitBreaker": ("桅杆舵机控制箱断路器", 1, "", "breaker"), "xczCircuitBreaker": ("XCZ断路器", 1, "", "breaker"), "bowRudderControlBoxCircuitBreaker": ("艏舵控制箱断路器", 1, "", "breaker"), "openWaterCoverStartCylinderCircuitBreaker": ("敞水盖启动电缸断路器", 1, "", "breaker"), "instrumentPowerFailureSignal": ("仪表电源失电信号", 1, "", "signal"), "emergencyPowerFailureSignal": ("应急电源失电信号", 1, "", "signal"), "waterIngressionAlarm": ("浸水报警", 1, "", "signal"), "busbarBVoltage": ("汇流排B电压", 1, "", None), "busbarBCurrent": ("汇流排B电流", 1, "", None), "bowFTDevice123Current": ("艏部FT装置1/2/3电流", 1, "", None), "actuatorCurrent": ("启闭机构电流", 1, "", None), "mastSteeringGearControlBoxCurrent": ("桅杆舵机控制箱电流", 1, "", None), "xczCurrent": ("XCZ电流", 1, "", None), "bowRudderControlBoxCurrent": ("艏舵控制箱电流", 1, "", None), "openWaterCoverStartCylinderCurrent": ("敞水盖启动电缸电流", 1, "", None), } # --- 低压主母排配电反馈 --- COL_META["disLowMainBusState"] = { "lithiumBatteryGroupInstrumentCircuitBreaker": ("锂电池组(仪表)断路器", 1, "", "breaker"), "bowLowVoltageDistributionBoxCircuitBreaker": ("艏部低压配电箱断路器", 1, "", "breaker"), "unit4InstrumentDC48VCircuitBreaker": ("UNIT4仪表DC48V断路器", 1, "", "breaker"), "dcC1DCDistributionPanelCircuitBreaker": ("DC-C1直流配电板断路器", 1, "", "breaker"), "reservedCircuitBreaker1": ("预留断路器1", 1, "", "breaker"), "reservedCircuitBreaker2": ("预留断路器2", 1, "", "breaker"), "emergencyLithiumBatteryGroup2CircuitBreaker": ("应急锂电池组2断路器", 1, "", "breaker"), "instrumentPowerFailureSignal": ("仪表电源失电信号", 1, "", "signal"), "emergencyPowerFailureSignal": ("应急电源失电信号", 1, "", "signal"), "instrumentBusbarInsulationLow": ("仪表汇流排绝缘低", 1, "", "insulation"), "instrumentBusbarVoltage": ("仪表母排电压", 1, "", None), "bowLowVoltageDistributionBoxCurrent": ("艏部低压配电箱电流", 1, "", None), "dcC1InstrumentDC48VCurrent": ("DC-C1仪表DC48V电流", 1, "", None), "reservedCurrent1": ("预留电流1", 1, "", None), "emergencyLithiumBatteryGroup2Current": ("应急锂电池组2电流", 1, "", None), "lithiumBatteryGroupInstrumentCurrent": ("锂电池组(仪表)电流", 1, "", None), "unit4InstrumentDC48VCurrent": ("UNIT4仪表DC48V电流", 1, "", None), "reservedCurrent2": ("预留电流2", 1, "", None), "emergency2BusbarVoltage": ("应急2汇流排电压", 1, "", None), "compositeEnergyManagementSystemEmergencyDC48VCurrent": ("复合能源管控应急DC48V电流", 1, "", None), "fuelCellSecuritySystemEmergencyDC48VCurrent": ("燃料电池安全系统应急DC48V电流", 1, "", None), "plcControlPowerCurrent": ("PLC控制电源电流", 1, "", None), } # --- 低压母线配电反馈 --- COL_META["disLowBusState"] = { "unit1CircuitBreaker": ("UNIT1断路器", 1, "", "breaker"), "unit2CircuitBreaker": ("UNIT2断路器", 1, "", "breaker"), "unit3CircuitBreaker": ("UNIT3断路器", 1, "", "breaker"), "unit5CircuitBreaker": ("UNIT5断路器", 1, "", "breaker"), "bowPZDeviceCircuitBreaker": ("艏部PZ装置断路器", 1, "", "breaker"), "reservedCircuitBreaker1": ("预留断路器1", 1, "", "breaker"), "reservedCircuitBreaker2": ("预留断路器2", 1, "", "breaker"), "emergencyLithiumBatteryGroup1CircuitBreaker": ("应急锂电池组1断路器", 1, "", "breaker"), "powerFailureSignal": ("仪表电源失电信号", 1, "", "signal"), "emergencyPowerFailureSignal": ("应急电源失电信号", 1, "", "signal"), "waterIngressionAlarm": ("浸水报警", 1, "", "signal"), "instrumentBusbarVoltage": ("仪表母排电压", 1, "", None), "instrumentBusbarCurrent": ("仪表母排电流", 1, "", None), "unit1Current": ("UNIT1电流", 1, "", None), "unit2Current": ("UNIT2电流", 1, "", None), "unit3Current": ("UNIT3电流", 1, "", None), "unit5Current": ("UNIT5电流", 1, "", None), "bowPZDeviceCurrent": ("艏部PZ装置电流", 1, "", None), "emergencyLithiumBatteryGroup1Current": ("应急锂电池组1电流", 1, "", None), "emergency1BusbarVoltage": ("应急1汇流排电压", 1, "", None), } # --- 原始收发帧日志 --- COL_META["msg_log"] = { "ts": ("时间戳(unix秒)", 1, "", None), "ts_local": ("本地时间", 1, "", None), "direction": ("方向", 1, "", "direction"), "msg_id": ("消息ID", 1, "", "hex"), "msg_name": ("消息名", 1, "", None), "length": ("帧长度", 1, "", None), "payload_hex": ("载荷(hex)", 1, "", None), "checksum_ok": ("校验和", 1, "", "checksum_ok"), "note": ("备注", 1, "", None), } # 反馈表 feedback_data.db COL_META["fb_log"] = { "ts": ("时间戳(unix秒)", 1, "", None), "msg": ("消息类型", 1, "", None), "payload": ("载荷", 1, "", None), } # ===================================================================== # 帧解码定义: msg_id -> (消息名, [(字段名, 格式, 中文名, 枚举key)]) # 格式: B=u8, H=u16, I=u32 # ===================================================================== FRAME_DEFS = {} FRAME_DEFS[0x01] = ("disHighVolBusCmd 高压母线配电指令", [ ("fuelCellCircuitBreaker", "B", "燃料电池断路器", "breaker"), ("powerLithiumBatteryCircuitBreaker", "B", "动力锂电池断路器", "breaker"), ("propulsionMotorCircuitBreaker", "B", "推进电机断路器", "breaker"), ("lithiumBatteryGroupInstrumentCircuitBreaker", "B", "锂电池组仪表断路器", "breaker"), ("dcDc5ModuleCircuitBreaker", "B", "DC/DC5模块断路器", "breaker"), ("bowHighVoltageDistributionBoxCircuitBreaker", "B", "艏部高压配电箱断路器", "breaker"), ("sternFTDevice45CircuitBreaker", "B", "艉部FT装置4/5断路器", "breaker"), ("fbReservedCircuitBreaker", "B", "FB预留断路器", "breaker"), ("sternRudderSwitch1CircuitBreaker", "B", "艉舵开关1断路器", "breaker"), ("sternRudderSwitch2CircuitBreaker", "B", "艉舵开关2断路器", "breaker"), ("tyzReservedCircuitBreaker", "B", "TYZ预留断路器", "breaker"), ("reservedCircuitBreaker", "B", "预留断路器", "breaker"), ("dcdc5Module", "B", "DC/DC5模块启动", "onoff"), ("coolingSystem", "B", "冷却系统启动", "onoff"), ]) FRAME_DEFS[0x02] = ("disHighAVolBusCmd 高压母线A配电指令", [ ("bowFTDevice123CircuitBreaker", "B", "艏部FT装置1/2/3断路器", "breaker"), ("actuatorCircuitBreaker", "B", "启闭机构断路器", "breaker"), ("mastSteeringGearControlBoxCircuitBreaker", "B", "桅杆舵机控制箱断路器", "breaker"), ("xczCircuitBreaker", "B", "XCZ断路器", "breaker"), ("bowRudderControlBoxCircuitBreaker", "B", "艏舵控制箱断路器", "breaker"), ("openWaterCoverStartCylinderCircuitBreaker", "B", "敞水盖启动电缸断路器", "breaker"), ]) FRAME_DEFS[0x03] = ("disLowMainBusCmd 低压主母排配电指令", [ ("lithiumBatteryGroupInstrumentCircuitBreaker", "B", "锂电池组(仪表)断路器", "breaker"), ("bowLowVoltageDistributionBoxCircuitBreaker", "B", "艏部低压配电箱断路器", "breaker"), ("unit4InstrumentDC48VCircuitBreaker", "B", "UNIT4仪表DC48V断路器", "breaker"), ("dcC1DCDistributionPanelCircuitBreaker", "B", "DC-C1直流配电板断路器", "breaker"), ("reservedCircuitBreaker1", "B", "预留断路器1", "breaker"), ("reservedCircuitBreaker2", "B", "预留断路器2", "breaker"), ("emergencyLithiumBatteryGroup2CircuitBreaker", "B", "应急锂电池组2断路器", "breaker"), ]) FRAME_DEFS[0x04] = ("disLowBusCmd 低压母线配电指令", [ ("unit1CircuitBreaker", "B", "UNIT1断路器", "breaker"), ("unit2CircuitBreaker", "B", "UNIT2断路器", "breaker"), ("unit3CircuitBreaker", "B", "UNIT3断路器", "breaker"), ("unit5CircuitBreaker", "B", "UNIT5断路器", "breaker"), ("bowPZDeviceCircuitBreaker", "B", "艏部PZ装置断路器", "breaker"), ("reservedCircuitBreaker1", "B", "预留断路器1", "breaker"), ("reservedCircuitBreaker2", "B", "预留断路器2", "breaker"), ("emergencyLithiumBatteryGroup1CircuitBreaker", "B", "应急锂电池组1断路器", "breaker"), ]) FRAME_DEFS[0x40] = ("ccuColCmd 复合管控操作指令", [ ("year", "H", "年", None), ("month", "B", "月", None), ("day", "B", "日", None), ("hour", "B", "时", None), ("minute", "B", "分", None), ("second", "B", "秒", None), ("millisecond", "B", "毫秒(×10)", None), ("fcmode", "B", "工况", "work_condition"), ("fcuCmd", "B", "燃料电池指令", "fcu_cmd"), ("fcuPowerEfficiency", "B", "燃料电池功率效率", None), ("insChargeEnable", "B", "仪表电池充电使能", None), ("insBatCmd", "B", "仪表电池指令", "battery_cmd"), ("powerBatCmd", "B", "动力电池指令", "battery_cmd"), ("powerBatEfficiency", "H", "动力电池效率", None), ("heartbeat", "B", "心跳计数", None), ("masterState", "B", "主机状态", "hex"), ("paceholder1", "I", "占位字段1", "hex"), ("paceholder2", "I", "占位字段2", "hex"), ]) FRAME_DEFS[0x41] = ("ccuSetParmCmd 复合管控参数设定指令", [ ("insSocHold1", "B", "仪表SOC保持1", None), ("insSocHold2", "B", "仪表SOC保持2", None), ("insSocHold3", "B", "仪表SOC保持3", None), ("fcuToInsLimit1", "B", "燃电到仪表限值1", None), ("fcuToInsLimit2", "B", "燃电到仪表限值2", None), ("powerSocHold1", "B", "动力SOC保持1", None), ("powerSocHold2", "B", "动力SOC保持2", None), ("powerSocHold3", "B", "动力SOC保持3", None), ("fcuToPowerLimit1", "B", "燃电到动力限值1", None), ("fcuToPowerLimit2", "B", "燃电到动力限值2", None), ]) # 配电状态反馈 (0x05~0x08), 供未来状态帧解码使用 FRAME_DEFS[0x05] = ("disHighVolBusState 高压母线配电反馈", [ ("fuelCellCircuitBreaker", "B", "燃料电池断路器", "breaker"), ("powerLithiumBatteryCircuitBreaker", "B", "动力锂电池断路器", "breaker"), ("propulsionMotorCircuitBreaker", "B", "推进电机断路器", "breaker"), ("lithiumBatteryGroupInstrumentCircuitBreaker", "B", "锂电池组仪表断路器", "breaker"), ("dcDc5ModuleCircuitBreaker", "B", "DC/DC5模块断路器", "breaker"), ("bowHighVoltageDistributionBoxCircuitBreaker", "B", "艏部高压配电箱断路器", "breaker"), ("sternFTDevice45CircuitBreaker", "B", "艉部FT装置4/5断路器", "breaker"), ("fbReservedCircuitBreaker", "B", "FB预留断路器", "breaker"), ("sternRudderSwitch1CircuitBreaker", "B", "艉舵开关1断路器", "breaker"), ("sternRudderSwitch2CircuitBreaker", "B", "艉舵开关2断路器", "breaker"), ("tyzReservedCircuitBreaker", "B", "TYZ预留断路器", "breaker"), ("reservedCircuitBreaker", "B", "预留断路器", "breaker"), ("dcDcFaultWord1", "B", "DC/DC5故障字低", "hex"), ("dcDcFaultWord2", "B", "DC/DC5故障字高", "hex"), ("powerBusInsulationStatus", "B", "动力母排绝缘状态", "insulation"), ("aBusInsulationStatus", "B", "A汇流排绝缘状态", "insulation"), ("meterPowerLossSignal", "B", "仪表电源状态", "signal"), ("emergencyPowerLossSignal", "B", "应急电源状态", "signal"), ("dcDcTemperature", "B", "DC/DC温度", None), ("powerBusVoltage", "H", "动力汇流排电压", None), ("dcDc5ModuleCurrent", "H", "DC/DC5模块电流", None), ("powerBusCurrent", "H", "动力汇流排电流", None), ("busbarAVoltage", "H", "A汇流排电压", None), ("propulsionMotorControlBoxCurrent", "H", "推进电机控制箱电流", None), ("busbarACurrent", "H", "A汇流排电流", None), ("lithiumBatteryGroupMeterCurrent", "H", "锂电池组(仪表)电流", None), ("bowHighVoltageDistributionBoxCurrent", "H", "艏部高压配电箱电流", None), ("sternFTDevice45Current", "H", "艉部FT装置4/5电流", None), ("tyzReservedSwitchCurrent", "H", "TYZ预留开关电流", None), ("sternRudderSwitch1Current", "H", "艉舵开关1电流", None), ("reservedCurrent1", "H", "预留电流1", None), ("sternRudderSwitch2Current", "H", "艉舵开关2电流", None), ("reservedCurrent2", "H", "预留电流2", None), ("fbReservedSwitchCurrent", "H", "FB预留电流", None), ("reservedCurrent3", "H", "预留电流3", None), ("coolWaterPressure", "H", "冷却水压力", None), ]) FRAME_DEFS[0x06] = ("disHighAVolBusState 高压母线A配电反馈", [ ("bowFTDevice123CircuitBreaker", "B", "艏部FT装置1/2/3断路器", "breaker"), ("actuatorCircuitBreaker", "B", "启闭机构断路器", "breaker"), ("mastSteeringGearControlBoxCircuitBreaker", "B", "桅杆舵机控制箱断路器", "breaker"), ("xczCircuitBreaker", "B", "XCZ断路器", "breaker"), ("bowRudderControlBoxCircuitBreaker", "B", "艏舵控制箱断路器", "breaker"), ("openWaterCoverStartCylinderCircuitBreaker", "B", "敞水盖启动电缸断路器", "breaker"), ("instrumentPowerFailureSignal", "B", "仪表电源失电信号", "signal"), ("emergencyPowerFailureSignal", "B", "应急电源失电信号", "signal"), ("waterIngressionAlarm", "B", "浸水报警", "signal"), ("busbarBVoltage", "H", "汇流排B电压", None), ("busbarBCurrent", "H", "汇流排B电流", None), ("bowFTDevice123Current", "H", "艏部FT装置1/2/3电流", None), ("actuatorCurrent", "H", "启闭机构电流", None), ("mastSteeringGearControlBoxCurrent", "H", "桅杆舵机控制箱电流", None), ("xczCurrent", "H", "XCZ电流", None), ("bowRudderControlBoxCurrent", "H", "艏舵控制箱电流", None), ("openWaterCoverStartCylinderCurrent", "H", "敞水盖启动电缸电流", None), ]) FRAME_DEFS[0x07] = ("disLowMainBusState 低压主母排配电反馈", [ ("lithiumBatteryGroupInstrumentCircuitBreaker", "B", "锂电池组(仪表)断路器", "breaker"), ("bowLowVoltageDistributionBoxCircuitBreaker", "B", "艏部低压配电箱断路器", "breaker"), ("unit4InstrumentDC48VCircuitBreaker", "B", "UNIT4仪表DC48V断路器", "breaker"), ("dcC1DCDistributionPanelCircuitBreaker", "B", "DC-C1直流配电板断路器", "breaker"), ("reservedCircuitBreaker1", "B", "预留断路器1", "breaker"), ("reservedCircuitBreaker2", "B", "预留断路器2", "breaker"), ("emergencyLithiumBatteryGroup2CircuitBreaker", "B", "应急锂电池组2断路器", "breaker"), ("instrumentPowerFailureSignal", "B", "仪表电源失电信号", "signal"), ("emergencyPowerFailureSignal", "B", "应急电源失电信号", "signal"), ("instrumentBusbarInsulationLow", "B", "仪表汇流排绝缘低", "insulation"), ("instrumentBusbarVoltage", "H", "仪表母排电压", None), ("bowLowVoltageDistributionBoxCurrent", "H", "艏部低压配电箱电流", None), ("dcC1InstrumentDC48VCurrent", "H", "DC-C1仪表DC48V电流", None), ("reservedCurrent1", "H", "预留电流1", None), ("emergencyLithiumBatteryGroup2Current", "H", "应急锂电池组2电流", None), ("lithiumBatteryGroupInstrumentCurrent", "H", "锂电池组(仪表)电流", None), ("unit4InstrumentDC48VCurrent", "H", "UNIT4仪表DC48V电流", None), ("reservedCurrent2", "H", "预留电流2", None), ("emergency2BusbarVoltage", "H", "应急2汇流排电压", None), ("compositeEnergyManagementSystemEmergencyDC48VCurrent", "H", "复合能源管控应急DC48V电流", None), ("fuelCellSecuritySystemEmergencyDC48VCurrent", "H", "燃料电池安全系统应急DC48V电流", None), ("plcControlPowerCurrent", "H", "PLC控制电源电流", None), ]) FRAME_DEFS[0x08] = ("disLowBusState 低压母线配电反馈", [ ("unit1CircuitBreaker", "B", "UNIT1断路器", "breaker"), ("unit2CircuitBreaker", "B", "UNIT2断路器", "breaker"), ("unit3CircuitBreaker", "B", "UNIT3断路器", "breaker"), ("unit5CircuitBreaker", "B", "UNIT5断路器", "breaker"), ("bowPZDeviceCircuitBreaker", "B", "艏部PZ装置断路器", "breaker"), ("reservedCircuitBreaker1", "B", "预留断路器1", "breaker"), ("reservedCircuitBreaker2", "B", "预留断路器2", "breaker"), ("emergencyLithiumBatteryGroup1CircuitBreaker", "B", "应急锂电池组1断路器", "breaker"), ("powerFailureSignal", "B", "仪表电源失电信号", "signal"), ("emergencyPowerFailureSignal", "B", "应急电源失电信号", "signal"), ("waterIngressionAlarm", "B", "浸水报警", "signal"), ("instrumentBusbarVoltage", "H", "仪表母排电压", None), ("instrumentBusbarCurrent", "H", "仪表母排电流", None), ("unit1Current", "H", "UNIT1电流", None), ("unit2Current", "H", "UNIT2电流", None), ("unit3Current", "H", "UNIT3电流", None), ("unit5Current", "H", "UNIT5电流", None), ("bowPZDeviceCurrent", "H", "艏部PZ装置电流", None), ("emergencyLithiumBatteryGroup1Current", "H", "应急锂电池组1电流", None), ("emergency1BusbarVoltage", "H", "应急1汇流排电压", None), ]) # ===================================================================== # 数据库访问 # ===================================================================== def connect_db(path): """只读连接数据库, 兼容 WAL, 程序运行期间可安全读取。""" con = sqlite3.connect(path) con.execute("PRAGMA query_only=ON") con.row_factory = sqlite3.Row return con def list_tables(con): """返回 [(表名, 行数, 时间列名, 最小时间, 最大时间)]""" out = [] rows = con.execute( "SELECT name FROM sqlite_master WHERE type='table' " "AND name NOT LIKE 'sqlite_%' ORDER BY name" ).fetchall() for (name,) in rows: try: n = con.execute("SELECT COUNT(*) FROM %s" % name).fetchone()[0] except Exception: n = -1 tcol, tmin, tmax = detect_time_range(con, name) out.append((name, n, tcol, tmin, tmax)) return out def table_columns(con, table): return [r[1] for r in con.execute("PRAGMA table_info(%s)" % table).fetchall()] def detect_time_range(con, table): cols = table_columns(con, table) tcol = None for cand in ("timestamp", "ts_local", "ts"): if cand in cols: tcol = cand break if tcol is None: return None, None, None try: r = con.execute("SELECT MIN(%s), MAX(%s) FROM %s" % (tcol, tcol, table)).fetchone() return tcol, r[0], r[1] except Exception: return tcol, None, None def format_time_value(tcol, val): if val is None: return "" if tcol == "ts" and isinstance(val, (int, float)): return datetime.fromtimestamp(val).strftime("%Y-%m-%d %H:%M:%S") return str(val) def parse_time_input(s): s = s.strip() if not s: return None for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"): try: return datetime.strptime(s, fmt) except ValueError: continue raise ValueError("时间格式应为 YYYY-MM-DD [HH:MM:SS]: " + s) # ===================================================================== # 数值格式化 / 人性化解码 # ===================================================================== _reverse_meta = None def _build_reverse(): global _reverse_meta if _reverse_meta is None: _reverse_meta = {} for tbl, cols in COL_META.items(): for col in cols: _reverse_meta.setdefault(col, (tbl, cols[col])) return _reverse_meta def fmt_value(table, col, raw, human): """把原始值转为展示字符串。human=True 时应用缩放/单位/枚举解码。""" if raw is None: return "" if not human: return str(raw) meta = COL_META.get(table, {}).get(col) if meta is None: rev = _build_reverse().get(col) if rev is not None: meta = rev[1] if meta is None: return str(raw) _desc, scale, unit, enum_key = meta if enum_key == "hex" and isinstance(raw, int): return "0x%X (%d)" % (raw, raw) if enum_key and isinstance(raw, int): m = ENUMS.get(enum_key, {}) if raw in m: if enum_key in ("breaker", "signal", "insulation", "onoff"): return "%s (0x%02X)" % (m[raw], raw) return "%s (%d)" % (m[raw], raw) if isinstance(raw, (int, float)) and scale != 1: v = raw * scale if isinstance(v, float) and v == int(v): v = int(v) return "%s%s" % (v, unit) if unit: return "%s%s" % (raw, unit) return str(raw) def fmt_frame_value(raw, fmt, enum_key): size = {"B": 1, "H": 2, "I": 4}[fmt] hexs = "0x%0*X" % (size * 2, raw) s = "%d (%s)" % (raw, hexs) if enum_key == "hex": return hexs if enum_key and raw in ENUMS.get(enum_key, {}): s += " = " + ENUMS[enum_key][raw] return s def decode_frame(hexstr): """解析单个 hex 帧, 返回字符串行列表。""" hs = hexstr.strip().replace(" ", "").replace("\n", "") if not hs: return ["(空输入)"] try: b = bytes.fromhex(hs) except ValueError: return ["非法 hex 字符串: %s" % hexstr.strip()] if len(b) < 6: return ["长度不足(至少6字节): %s (%d字节)" % (hs, len(b))] start1, start2 = b[0], b[1] msg_id = struct.unpack_from(" len(data): lines.append(" [字段越界] %-32s (剩余 %d 字节)" % (nm, len(data) - off)) break raw = struct.unpack_from("<" + fmt, data, off)[0] off += sz lines.append(" %-34s %-22s = %s" % (nm, desc, fmt_frame_value(raw, fmt, ekey))) lines.append(" 校验和: %s 数据区(hex): %s" % (checksum.hex().upper(), data.hex().upper())) return lines # ===================================================================== # 默认数据库路径 # ===================================================================== def default_db_path(): base = os.path.dirname(os.path.abspath(__file__)) for p in ( os.path.join(base, "..", "bin", "power_data.db"), os.path.join(base, "..", "src", "pPowerMangerHost", "feedback_data.db"), os.path.join(base, "..", "bin", "power_data.db"), ): p = os.path.normpath(p) if os.path.exists(p): return p return os.path.normpath(os.path.join(base, "..", "bin", "power_data.db")) # ===================================================================== # GUI # ===================================================================== class PowerDbApp: def __init__(self, root, db_path): self.root = root self.conn = None self.db_path = db_path self.current_table = None self.current_cols = [] self.current_raw_rows = [] self.time_col = None self.time_col_type = "text" root.title(APP_TITLE) root.geometry("1280x800") self._build_toolbar() self._build_notebook() self._build_statusbar() if db_path and os.path.exists(db_path): self.set_status("请点击「打开数据库...」选择数据库文件（已默认填入: %s）" % db_path) else: self.set_status("请点击「打开数据库...」选择数据库文件") # ---------- 工具栏 ---------- def _build_toolbar(self): bar = ttk.Frame(self.root, padding=(6, 4)) bar.pack(fill="x") self.db_var = tk.StringVar(value=self.db_path) ttk.Button(bar, text="打开数据库...", command=self.choose_db).pack(side="left") ent = ttk.Entry(bar, textvariable=self.db_var) ent.pack(side="left", fill="x", expand=True, padx=6) self.human_var = tk.BooleanVar(value=True) ttk.Checkbutton(bar, text="人性化显示", variable=self.human_var, command=self._on_human_toggle).pack(side="left", padx=6) def _on_human_toggle(self): if self.current_table: self.load_table(self.current_table) # ---------- 主标签页 ---------- def _build_notebook(self): nb = ttk.Notebook(self.root) nb.pack(fill="both", expand=True) self.nb = nb self.tab_browse = ttk.Frame(nb) self.tab_query = ttk.Frame(nb) self.tab_decode = ttk.Frame(nb) self.tab_plot = ttk.Frame(nb) nb.add(self.tab_browse, text="数据浏览") nb.add(self.tab_query, text="SQL 查询") nb.add(self.tab_decode, text="帧解码") nb.add(self.tab_plot, text="趋势图") self._build_browse_tab() self._build_query_tab() self._build_decode_tab() self._build_plot_tab() # ---------- 状态栏 ---------- def _build_statusbar(self): self.status_var = tk.StringVar(value="就绪") ttk.Label(self.root, textvariable=self.status_var, relief="sunken", anchor="w", padding=(6, 2)).pack(fill="x", side="bottom") def set_status(self, msg): self.status_var.set(msg) # ---------- 数据库打开 ---------- def choose_db(self): p = filedialog.askopenfilename( title="选择 SQLite 数据库", filetypes=[("SQLite 数据库", "*.db *.sqlite *.db3"), ("所有文件", "*.*")], ) if p: self.open_db(p) def open_db(self, path): try: if self.conn: self.conn.close() self.conn = connect_db(path) self.db_path = path self.db_var.set(path) self._refresh_tables() self._refresh_plot_tables() self.set_status("已打开: %s" % path) except Exception as e: messagebox.showerror("打开失败", str(e)) def _refresh_tables(self): for i in self.tables_tree.get_children(): self.tables_tree.delete(i) if not self.conn: return for name, n, tcol, tmin, tmax in list_tables(self.conn): tr = "%s ~ %s" % (format_time_value(tcol, tmin), format_time_value(tcol, tmax)) if tmin is not None else "" self.tables_tree.insert("", "end", iid=name, values=(name, n, tr)) # ---------- 数据浏览 ---------- def _build_browse_tab(self): left = ttk.Frame(self.tab_browse, width=340) left.pack(side="left", fill="y") left.pack_propagate(False) ttk.Label(left, text="数据表").pack(anchor="w", padx=4, pady=2) self.tables_tree = ttk.Treeview(left, columns=("name", "rows", "range"), show="headings", height=30) self.tables_tree.heading("name", text="表名") self.tables_tree.heading("rows", text="行数") self.tables_tree.heading("range", text="时间范围") self.tables_tree.column("name", width=150) self.tables_tree.column("rows", width=70, anchor="e") self.tables_tree.column("range", width=180) tsb = ttk.Scrollbar(left, orient="vertical", command=self.tables_tree.yview) self.tables_tree.configure(yscrollcommand=tsb.set) self.tables_tree.pack(side="left", fill="both", expand=True, padx=(4, 0), pady=2) tsb.pack(side="left", fill="y") self.tables_tree.bind("<>", self._on_table_select) right = ttk.Frame(self.tab_browse) right.pack(side="left", fill="both", expand=True) filter_bar = ttk.Frame(right) filter_bar.pack(fill="x", padx=4, pady=4) ttk.Label(filter_bar, text="条件(WHERE)").pack(side="left") self.where_var = tk.StringVar() ttk.Entry(filter_bar, textvariable=self.where_var, width=30).pack(side="left", padx=4) ttk.Label(filter_bar, text="起始").pack(side="left", padx=(8, 0)) self.since_var = tk.StringVar() ttk.Entry(filter_bar, textvariable=self.since_var, width=18).pack(side="left", padx=2) ttk.Label(filter_bar, text="结束").pack(side="left") self.until_var = tk.StringVar() ttk.Entry(filter_bar, textvariable=self.until_var, width=18).pack(side="left", padx=2) ttk.Label(filter_bar, text="行数").pack(side="left", padx=(8, 0)) self.limit_var = tk.StringVar(value="500") ttk.Combobox(filter_bar, textvariable=self.limit_var, width=8, values=("100", "500", "1000", "5000", "10000", "全部")).pack(side="left", padx=2) ttk.Button(filter_bar, text="查询", command=self._refresh_browse).pack(side="left", padx=6) ttk.Button(filter_bar, text="导出CSV", command=self._export_browse).pack(side="left", padx=2) tree_area = ttk.Frame(right) tree_area.pack(fill="both", expand=True) self.data_tree = ttk.Treeview(tree_area, show="headings") ysb = ttk.Scrollbar(tree_area, orient="vertical", command=self.data_tree.yview) xsb = ttk.Scrollbar(tree_area, orient="horizontal", command=self.data_tree.xview) self.data_tree.configure(yscrollcommand=ysb.set, xscrollcommand=xsb.set) self.data_tree.grid(row=0, column=0, sticky="nsew") ysb.grid(row=0, column=1, sticky="ns") xsb.grid(row=1, column=0, sticky="ew") tree_area.rowconfigure(0, weight=1) tree_area.columnconfigure(0, weight=1) def _on_table_select(self, _event): sel = self.tables_tree.selection() if not sel: return self.load_table(sel[0]) def load_table(self, table): self.current_table = table self._refresh_browse() def _build_where(self): parts = [] where = self.where_var.get().strip() if where: parts.append("(" + where + ")") since = self.since_var.get().strip() until = self.until_var.get().strip() if (since or until) and self.time_col: col = self.time_col if self.time_col_type == "real": if since: parts.append("%s >= %f" % (col, parse_time_input(since).timestamp())) if until: parts.append("%s <= %f" % (col, parse_time_input(until).timestamp())) else: if since: parts.append("%s >= '%s'" % (col, parse_time_input(since).strftime("%Y-%m-%d %H:%M:%S"))) if until: parts.append("%s <= '%s'" % (col, parse_time_input(until).strftime("%Y-%m-%d %H:%M:%S"))) return (" WHERE " + " AND ".join(parts)) if parts else "" def _refresh_browse(self): if not self.conn or not self.current_table: return table = self.current_table self.data_tree.delete(*self.data_tree.get_children()) cols = table_columns(self.conn, table) self.current_cols = cols # 时间列检测 self.time_col = None self.time_col_type = "text" for cand in ("timestamp", "ts_local", "ts"): if cand in cols: self.time_col = cand self.time_col_type = "real" if cand == "ts" else "text" break self.data_tree["columns"] = cols for c in cols: meta = COL_META.get(table, {}).get(c) title = meta[0] if meta else c self.data_tree.heading(c, text=title) self.data_tree.column(c, width=120, stretch=True, anchor="w") sql = "SELECT * FROM %s" % table try: sql += self._build_where() sql += " ORDER BY 1 DESC" limit = self.limit_var.get().strip() if limit != "全部": sql += " LIMIT %d" % int(limit) t0 = datetime.now() rows = self.conn.execute(sql).fetchall() dt = (datetime.now() - t0).total_seconds() * 1000 except Exception as e: messagebox.showerror("查询失败", str(e)) return self.current_raw_rows = rows human = self.human_var.get() for r in rows: vals = [fmt_value(table, c, r[i], human) for i, c in enumerate(cols)] self.data_tree.insert("", "end", values=vals) total = self.conn.execute("SELECT COUNT(*) FROM %s" % table).fetchone()[0] self.set_status("表 %s: 显示 %d / 共 %d 行, 耗时 %.1f ms" % (table, len(rows), total, dt)) def _export_browse(self): if not self.current_raw_rows: messagebox.showinfo("提示", "没有可导出的数据, 请先查询。") return self._export_to_csv(self.current_cols, self.current_raw_rows) # ---------- SQL 查询 ---------- def _build_query_tab(self): top = ttk.Frame(self.tab_query) top.pack(fill="x", padx=4, pady=4) ttk.Label(top, text="SQL 语句 (只读)").pack(anchor="w") self.sql_text = tk.Text(top, height=5) self.sql_text.pack(fill="x", pady=2) btns = ttk.Frame(top) btns.pack(fill="x") ttk.Button(btns, text="执行", command=self._run_query).pack(side="left") ttk.Button(btns, text="导出CSV", command=self._export_query).pack(side="left", padx=6) self.q_human_var = tk.BooleanVar(value=False) ttk.Checkbutton(btns, text="人性化显示", variable=self.q_human_var).pack(side="left") self.q_table_hint = ttk.Label(btns, text="") self.q_table_hint.pack(side="left", padx=10) q_area = ttk.Frame(self.tab_query) q_area.pack(fill="both", expand=True) self.q_tree = ttk.Treeview(q_area, show="headings") ysb = ttk.Scrollbar(q_area, orient="vertical", command=self.q_tree.yview) xsb = ttk.Scrollbar(q_area, orient="horizontal", command=self.q_tree.xview) self.q_tree.configure(yscrollcommand=ysb.set, xscrollcommand=xsb.set) self.q_tree.grid(row=0, column=0, sticky="nsew") ysb.grid(row=0, column=1, sticky="ns") xsb.grid(row=1, column=0, sticky="ew") q_area.rowconfigure(0, weight=1) q_area.columnconfigure(0, weight=1) self.q_rows = [] self.q_cols = [] self.q_table = None def _guess_table(self, sql): import re m = re.search(r"\bfrom\s+([a-zA-Z_][\w]*)", sql, re.I) if m: return m.group(1) return None def _run_query(self): if not self.conn: messagebox.showinfo("提示", "请先打开数据库。") return self.q_tree.delete(*self.q_tree.get_children()) sql = self.sql_text.get("1.0", "end").strip() if not sql: return try: t0 = datetime.now() cur = self.conn.execute(sql) cols = [d[0] for d in cur.description] if cur.description else [] rows = cur.fetchall() dt = (datetime.now() - t0).total_seconds() * 1000 except Exception as e: messagebox.showerror("执行失败", str(e)) return self.q_cols = cols self.q_rows = rows self.q_table = self._guess_table(sql) self.q_table_hint.config(text="(匹配表: %s)" % self.q_table if self.q_table else "") self.q_tree["columns"] = cols for c in cols: self.q_tree.heading(c, text=c) self.q_tree.column(c, width=130, stretch=True, anchor="w") table = self.q_table or "" human = self.q_human_var.get() for r in rows: vals = [fmt_value(table, c, r[i], human) for i, c in enumerate(cols)] self.q_tree.insert("", "end", values=vals) self.set_status("查询返回 %d 行, 耗时 %.1f ms" % (len(rows), dt)) def _export_query(self): if not self.q_rows: messagebox.showinfo("提示", "没有可导出的数据, 请先执行查询。") return self._export_to_csv(self.q_cols, self.q_rows) # ---------- CSV 导出 ---------- def _export_to_csv(self, cols, rows): p = filedialog.asksaveasfilename( title="导出 CSV", defaultextension=".csv", filetypes=[("CSV 文件", "*.csv")], ) if not p: return try: with open(p, "w", newline="", encoding="utf-8-sig") as f: w = csv.writer(f) w.writerow(cols) for r in rows: w.writerow(list(r)) self.set_status("已导出 %d 行到 %s" % (len(rows), p)) except Exception as e: messagebox.showerror("导出失败", str(e)) # ---------- 帧解码 ---------- def _build_decode_tab(self): top = ttk.Frame(self.tab_decode) top.pack(fill="x", padx=4, pady=4) ttk.Label(top, text="粘贴 hex 帧 (可多帧, 空格/换行分隔), 也可从 msg_log 表复制 payload_hex:").pack(anchor="w") self.decode_input = tk.Text(top, height=6) self.decode_input.pack(fill="x", pady=2) btns = ttk.Frame(top) btns.pack(fill="x") ttk.Button(btns, text="解码", command=self._run_decode).pack(side="left") ttk.Button(btns, text="从 msg_log 加载示例", command=self._load_decode_samples).pack(side="left", padx=6) ttk.Button(btns, text="清空", command=lambda: self.decode_input.delete("1.0", "end")).pack(side="left") d_area = ttk.Frame(self.tab_decode) d_area.pack(fill="both", expand=True) self.decode_out = tk.Text(d_area, wrap="none", font=(_UI_MONO, 10)) ysb = ttk.Scrollbar(d_area, orient="vertical", command=self.decode_out.yview) xsb = ttk.Scrollbar(d_area, orient="horizontal", command=self.decode_out.xview) self.decode_out.configure(yscrollcommand=ysb.set, xscrollcommand=xsb.set) self.decode_out.grid(row=0, column=0, sticky="nsew") ysb.grid(row=0, column=1, sticky="ns") xsb.grid(row=1, column=0, sticky="ew") d_area.rowconfigure(0, weight=1) d_area.columnconfigure(0, weight=1) def _run_decode(self): text = self.decode_input.get("1.0", "end") tokens = text.split() if not tokens: self.decode_out.insert("end", "请输入 hex 帧。\n") return self.decode_out.delete("1.0", "end") for t in tokens: for line in decode_frame(t): self.decode_out.insert("end", line + "\n") self.decode_out.insert("end", "-" * 70 + "\n") def _load_decode_samples(self): if not self.conn: messagebox.showinfo("提示", "请先打开数据库。") return rows = self.conn.execute( "SELECT msg_id, payload_hex FROM msg_log GROUP BY msg_id LIMIT 6" ).fetchall() if not rows: messagebox.showinfo("提示", "msg_log 表为空。") return samples = [r[1] for r in rows] self.decode_input.delete("1.0", "end") self.decode_input.insert("end", "\n".join(samples)) self.set_status("已加载 %d 个不同消息类型的示例帧" % len(samples)) # ---------- 趋势图 ---------- def _build_plot_tab(self): top = ttk.Frame(self.tab_plot) top.pack(fill="x", padx=4, pady=4) ttk.Label(top, text="表").pack(side="left") self.plot_table_var = tk.StringVar() self.plot_table_combo = ttk.Combobox(top, textvariable=self.plot_table_var, width=22, state="readonly") self.plot_table_combo.pack(side="left", padx=4) self.plot_table_combo.bind("<>", self._plot_on_table) ttk.Label(top, text="数值列").pack(side="left") self.plot_col_var = tk.StringVar() self.plot_col_combo = ttk.Combobox(top, textvariable=self.plot_col_var, width=24, state="readonly") self.plot_col_combo.pack(side="left", padx=4) ttk.Label(top, text="时间桶").pack(side="left") self.plot_bucket_var = tk.StringVar(value="原始") ttk.Combobox(top, textvariable=self.plot_bucket_var, width=10, state="readonly", values=("原始", "1分钟", "10分钟", "1小时", "1天")).pack(side="left", padx=4) ttk.Button(top, text="绘图", command=self._run_plot).pack(side="left", padx=6) self.plot_canvas = tk.Canvas(self.tab_plot, bg="white") self.plot_canvas.pack(fill="both", expand=True, padx=4, pady=4) def _plot_on_table(self, _event=None): table = self.plot_table_var.get() if not self.conn or not table: return cols = table_columns(self.conn, table) # 只挑数值列 num_cols = [] for c in cols: r = self.conn.execute("SELECT %s FROM %s WHERE %s IS NOT NULL LIMIT 1" % (c, table, c)).fetchone() if r and isinstance(r[0], (int, float)): num_cols.append(c) self.plot_col_combo["values"] = num_cols if num_cols: self.plot_col_var.set(num_cols[0]) def _refresh_plot_tables(self): if not self.conn: return names = [t[0] for t in list_tables(self.conn)] self.plot_table_combo["values"] = names if names: self.plot_table_var.set(names[0]) self._plot_on_table() def _run_plot(self): if not self.conn: return table = self.plot_table_var.get() col = self.plot_col_var.get() if not table or not col: return tcol = None cols = table_columns(self.conn, table) for cand in ("timestamp", "ts_local", "ts"): if cand in cols: tcol = cand break if tcol is None: messagebox.showinfo("提示", "该表没有可识别的时间列。") return try: rows = self.conn.execute( "SELECT %s, %s FROM %s WHERE %s IS NOT NULL" % (tcol, col, table, col) ).fetchall() except Exception as e: messagebox.showerror("查询失败", str(e)) return pts = [] for tv, v in rows: if not isinstance(v, (int, float)): continue try: if tcol == "ts": t = datetime.fromtimestamp(tv) else: t = datetime.strptime(str(tv), "%Y-%m-%d %H:%M:%S") except Exception: continue pts.append((t, v)) if not pts: messagebox.showinfo("提示", "没有可绘制的数据点。") return pts.sort(key=lambda x: x[0]) bucket = self.plot_bucket_var.get() if bucket != "原始": pts = self._bucket_points(pts, bucket) self._draw_plot(pts, col) def _bucket_points(self, pts, bucket): if bucket == "1分钟": sec = 60 elif bucket == "10分钟": sec = 600 elif bucket == "1小时": sec = 3600 elif bucket == "1天": sec = 86400 else: sec = 1 buckets = {} for t, v in pts: key = int(t.timestamp() // sec) * sec buckets.setdefault(key, []).append(v) out = [] for k in sorted(buckets): vs = buckets[k] out.append((datetime.fromtimestamp(k), sum(vs) / len(vs))) return out def _draw_plot(self, pts, col): c = self.plot_canvas c.delete("all") W = c.winfo_width() or 800 H = c.winfo_height() or 500 pad_l, pad_r, pad_t, pad_b = 60, 20, 20, 40 plot_w = W - pad_l - pad_r plot_h = H - pad_t - pad_b if plot_w <= 0 or plot_h <= 0: return xs = [p[0] for p in pts] ys = [p[1] for p in pts] ymin, ymax = min(ys), max(ys) if ymin == ymax: ymax = ymin + 1 xmin, xmax = xs[0], xs[-1] span = (xmax - xmin).total_seconds() or 1 def sx(t): return pad_l + (t - xmin).total_seconds() / span * plot_w def sy(v): return pad_t + (ymax - v) / (ymax - ymin) * plot_h # 网格与轴 c.create_line(pad_l, pad_t + plot_h, W - pad_r, pad_t + plot_h, fill="black") c.create_line(pad_l, pad_t, pad_l, pad_t + plot_h, fill="black") for i in range(5): yv = ymin + (ymax - ymin) * i / 4 yy = sy(yv) c.create_line(pad_l, yy, W - pad_r, yy, fill="#dddddd") c.create_text(pad_l - 5, yy, text="%.4g" % yv, anchor="e", font=("Arial", 8)) # 时间标签(首/中/尾) for frac in (0.0, 0.5, 1.0): t = xmin + timedelta(seconds=span * frac) xx = pad_l + plot_w * frac c.create_text(xx, pad_t + plot_h + 15, text=t.strftime("%m-%d %H:%M"), font=("Arial", 8)) # 折线 coords = [] for t, v in pts: coords.append(sx(t)) coords.append(sy(v)) if len(coords) >= 4: c.create_line(*coords, fill="#1f77b4", width=1.5) c.create_text(pad_l + plot_w / 2, pad_t - 8, text="%s (%s, %d 点)" % (col, self.plot_bucket_var.get(), len(pts)), font=(_UI_SANS, 10)) def _pick_font(candidates, default): """从候选字体族中选取第一个实际可用的, 否则返回默认。""" if tk is None: return default import tkinter.font as tkfont available = set(tkfont.families()) for fam in candidates: if fam in available: return fam return default def setup_fonts(root): """设置全局默认字体为支持中文的字体, 避免中文显示为方块。""" import tkinter.font as tkfont cjk = _pick_font(["Noto Sans CJK SC", "WenQuanYi Micro Hei", "WenQuanYi Zen Hei", "Microsoft YaHei", "SimHei", "Source Han Sans SC"], "TkDefaultFont") cjk_mono = _pick_font(["Noto Sans Mono CJK SC", "WenQuanYi Micro Hei Mono", "Microsoft YaHei", "SimHei"], "TkFixedFont") default_font = tkfont.nametofont("TkDefaultFont") default_font.configure(family=cjk, size=10) fixed_font = tkfont.nametofont("TkFixedFont") fixed_font.configure(family=cjk_mono, size=10) style = ttk.Style(root) try: style.theme_use("clam") except Exception: pass style.configure(".", font=(cjk, 10)) style.configure("Treeview", font=(cjk, 10), rowheight=24) style.configure("Treeview.Heading", font=(cjk, 10, "bold")) style.configure("TNotebook.Tab", font=(cjk, 10)) style.configure("TLabel", font=(cjk, 10)) style.configure("TButton", font=(cjk, 10)) style.configure("TCheckbutton", font=(cjk, 10)) style.configure("TCombobox", font=(cjk, 10)) style.configure("TEntry", font=(cjk, 10)) root.option_add("*Font", (cjk, 10)) root.option_add("*Text.Font", (cjk_mono, 10)) global _UI_SANS, _UI_MONO _UI_SANS = cjk _UI_MONO = cjk_mono return cjk, cjk_mono def main(): if tk is None: print("未检测到 tkinter。请安装系统包后重试:") print(" sudo apt install python3-tk") return 1 root = tk.Tk() setup_fonts(root) app = PowerDbApp(root, default_db_path()) root.mainloop() return 0 if __name__ == "__main__": raise SystemExit(main())