zjk
54c32a180a
- 新增 pCCU 复合管控器程序（src/pCCU）：FC/PM 双链路(UDP)、协议编解码器 (FcProtocol/PmProtocol/Frame/Message/MessageRegistry)、快照与落库(DbStore)、 内置 WebServer；挂入 src/CMakeLists.txt 与板卡 mission（h100.moos） - 惯导/深度数据：新增 uDevice_insData 消息解析，接入 UpperCommManager， pPowerMangerHost 注册监视，UpmsgTest 覆盖 - pPowerMangerHost 设备控制页增强（web/index.html） - 部署/运维脚本改为 4 个 systemd 服务（moosdb/pPowerManger/pPowerMangerHost/pCCU）： deploy/clean-data/fetch-data 同步适配 - 协议文档：归档 0824 控制主机与复合管控器通信协议 xlsx 及关联 docx，移除旧 md - 测试：新增 pccu 单元/集成测试与 disSysTest Python 配电仿真 - 清理：移除未使用的 CCU 状态调试打印 coutMsg；.gitignore 补充编译产物/临时文件忽略
86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
"""程序入口：python -m app 或 python run.py。
|
|
|
|
支持 --selftest 无 GUI 自检（协议 / 配置 / 数据库）。
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
|
|
|
|
def selftest():
|
|
ok = True
|
|
print("== 协议自检 ==")
|
|
proto_ok, report = _protocol_selftest()
|
|
print(report)
|
|
ok = ok and proto_ok
|
|
|
|
print("== 配置自检 ==")
|
|
try:
|
|
from .config import Config
|
|
c = Config()
|
|
for key in ("local", "c1", "c2", "host"):
|
|
assert key in c.data, f"缺少配置节 {key}"
|
|
print(" [OK] 配置加载/默认值")
|
|
except Exception as e:
|
|
print(f" [FAIL] 配置: {e}")
|
|
ok = False
|
|
|
|
print("== 数据库自检 ==")
|
|
try:
|
|
from .db import DB, TABLE_NAMES
|
|
from . import protocol
|
|
assert len(TABLE_NAMES) == 8, "消息表数量应为 8"
|
|
fd, path = tempfile.mkstemp(suffix=".db")
|
|
os.close(fd)
|
|
db = DB(path)
|
|
db.connect()
|
|
for cls in protocol.ALL_MESSAGES.values():
|
|
assert db.insert_message(cls()), f"{cls.__name__} 插入失败"
|
|
rows = {name: 0 for name in TABLE_NAMES.values()}
|
|
for name in TABLE_NAMES.values():
|
|
rows[name] = db.conn.execute(f'SELECT COUNT(*) FROM "{name}"').fetchone()[0]
|
|
assert all(v == 1 for v in rows.values()), f"写入行数异常: {rows}"
|
|
db.disconnect()
|
|
os.unlink(path)
|
|
print(" [OK] 建表 / 写入 / disconnect")
|
|
except Exception as e:
|
|
print(f" [FAIL] 数据库: {e}")
|
|
ok = False
|
|
|
|
print("SELFTEST:", "PASS" if ok else "FAIL")
|
|
return 0 if ok else 1
|
|
|
|
|
|
def _protocol_selftest():
|
|
from . import protocol
|
|
return protocol.self_test()
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description="配电系统调试助手")
|
|
parser.add_argument("--selftest", action="store_true",
|
|
help="运行协议/配置/数据库自检后退出")
|
|
parser.add_argument("--config", default=None, help="配置文件路径")
|
|
parser.add_argument("--version", action="store_true", help="显示版本")
|
|
args = parser.parse_args(argv)
|
|
|
|
if args.version:
|
|
from . import __version__
|
|
print(f"配电系统调试助手 v{__version__}")
|
|
return 0
|
|
if args.selftest:
|
|
return selftest()
|
|
|
|
import tkinter as tk
|
|
from .app import PowerManagerApp
|
|
root = tk.Tk()
|
|
app = PowerManagerApp(root, config_path=args.config)
|
|
root.mainloop()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|