"""电源管理/配电系统调试主程序。 主窗口组织： - 菜单：通信选项 / 配电操作 / 配置 / DCDC状态 / 复合管控面板(预留) / 软件说明 - 四个母线状态页（Notebook），数据驱动刷新 - 系统日志区 数据通路： 真实设备/UDP 反馈 -> on_feedback -> 状态缓存 + DB + (可选)转发上位机 配电指令(合/分) -> send_breaker_cmd -> UDP 下发 + DB + (可选)模拟回显 模拟模式 -> Simulator.update -> 注入同一 on_feedback 通路 """ import threading import time import tkinter as tk from tkinter import ttk from . import protocol from .config import Config from .db import DB from .logger import AppLogger from .panels import (BUS_SPECS, CMD_SPECS, build_bus_page, update_bus_page, build_control_panel, build_dcdc_display, update_dcdc_display) from .simulator import Simulator from .udp_comm import UdpComm from .protocol import BREAKER_ON MSG_ID_BUS = {0x0005: "high_vol", 0x0006: "high_a_vol", 0x0007: "low_main", 0x0008: "low_bus"} CMD_ID_DEST = {0x0001: "c1", 0x0002: "c1", 0x0003: "c2", 0x0004: "c2"} class PowerManagerApp: def __init__(self, root, config_path=None): self.root = root self.config = Config(config_path) self.root.title("配电系统调试助手") self.root.tk.call("tk", "scaling", 1.5) self.root.minsize(860, 640) # 菜单开关状态 self.Forward_Feedback = tk.BooleanVar(value=False) self.Transparent_Command = tk.BooleanVar(value=False) self.Simulation_Mode = tk.BooleanVar(value=False) self.Start_Connection = tk.BooleanVar(value=True) self.Record_Data = tk.BooleanVar(value=True) self.Simulation_Mode.trace_add("write", self._on_sim_toggle) self.db = DB(self.config.db_path) try: self.db.connect() except Exception as e: print(f"数据库打开失败: {e}") self.sim = Simulator() self._sim_thread = None self._sim_running = False self._data_lock = threading.Lock() # 各母线最新反馈（原始 Message 对象），None 表示尚无数据 self.raw_msgs = {key: None for key in BUS_SPECS} self.bus_refs = {} self.dcdc_refs = None # DCDC 弹窗 refs self.dcdc_popup = None self._build_ui() self.logger = AppLogger(widget=self.log_text) self.udp = UdpComm(logger=self.logger) self.udp.on_feedback = self._on_feedback self.udp.on_command = self._on_command self.udp.on_raw = self._on_raw try: self.udp.bind(self.config.local_ip, self.config.local_port) except OSError as e: self.logger.error(f"绑定本机 {self.config.local_ip}:{self.config.local_port} 失败: {e}") if self.Start_Connection.get() and self.udp.bound: try: self.udp.start() self.logger.info("通信连接已启动") except RuntimeError as e: self.logger.error(str(e)) self.logger.info(f"配电系统调试助手启动（配置文件: {self.config.path}）") self.root.protocol("WM_DELETE_WINDOW", self.on_closing) self.root.after(self.config.refresh_ms, self._refresh_ui) # ------------------------------------------------------------------ # 界面构建 # ------------------------------------------------------------------ def _build_ui(self): menubar = tk.Menu(self.root) comm_menu = tk.Menu(menubar, tearoff=0) comm_menu.add_checkbutton(label="转发反馈", variable=self.Forward_Feedback) comm_menu.add_checkbutton(label="透传指令", variable=self.Transparent_Command) comm_menu.add_checkbutton(label="模拟模式", variable=self.Simulation_Mode) menubar.add_cascade(label="通信选项", menu=comm_menu) op_menu = tk.Menu(menubar, tearoff=0) op_menu.add_command(label="动力母线操作", command=lambda: self.show_control_window("hvb")) op_menu.add_command(label="艏部高压操作", command=lambda: self.show_control_window("hvba")) op_menu.add_command(label="仪表母线操作", command=lambda: self.show_control_window("lvmb")) op_menu.add_command(label="艏部低压操作", command=lambda: self.show_control_window("lvb")) menubar.add_cascade(label="配电操作", menu=op_menu) cfg_menu = tk.Menu(menubar, tearoff=0) cfg_menu.add_checkbutton(label="启动连接", variable=self.Start_Connection, command=self._on_connection_toggle) cfg_menu.add_checkbutton(label="记录数据", variable=self.Record_Data) cfg_menu.add_command(label="通信配置", command=self.show_config_window) cfg_menu.add_command(label="数据库配置", command=self.show_db_config_window) menubar.add_cascade(label="配置", menu=cfg_menu) menubar.add_command(label="DCDC状态", command=self.show_dcdc_window) menubar.add_command(label="复合管控面板", command=lambda: self.show_placeholder_window("复合管控面板", "CCU 复合管控功能预留，待后续版本实现。")) menubar.add_command(label="软件说明", command=self.show_help_window) self.root.config(menu=menubar) main_frame = ttk.Frame(self.root, padding=8) main_frame.pack(fill=tk.BOTH, expand=True) main_frame.grid_rowconfigure(0, weight=1) main_frame.grid_columnconfigure(0, weight=1) self.notebook = ttk.Notebook(main_frame) self.notebook.grid(row=0, column=0, sticky="nsew") for key, spec in BUS_SPECS.items(): page = build_bus_page(self.notebook, spec) self.bus_refs[key] = page self.notebook.add(page["frame"], text=spec.title) log_frame = ttk.LabelFrame(main_frame, text="系统日志", padding=6) log_frame.grid(row=1, column=0, sticky="nsew", pady=(6, 0)) log_frame.grid_rowconfigure(0, weight=1) log_frame.grid_columnconfigure(0, weight=1) self.log_text = tk.Text(log_frame, height=8, state=tk.DISABLED) self.log_text.grid(row=0, column=0, sticky="nsew") sb = ttk.Scrollbar(log_frame, orient=tk.VERTICAL, command=self.log_text.yview) sb.grid(row=0, column=1, sticky="ns") self.log_text.config(yscrollcommand=sb.set) self.status_bar = ttk.Label(main_frame, text="未连接", anchor="w", relief="sunken") self.status_bar.grid(row=2, column=0, sticky="ew", pady=(4, 0)) # ------------------------------------------------------------------ # UDP 回调（运行在接收线程） # ------------------------------------------------------------------ def _on_feedback(self, msg, addr): bus_key = MSG_ID_BUS.get(msg.MSG_ID) if bus_key is None: return with self._data_lock: self.raw_msgs[bus_key] = msg if self.Record_Data.get(): self.db.insert_message(msg) def _on_command(self, msg, addr): if self.Simulation_Mode.get(): self.sim.apply_command(msg) def _on_raw(self, data, addr): msg_id = protocol.msg_id_of(data) if msg_id is None: return if self.Forward_Feedback.get() and msg_id in protocol.FB_MESSAGES: self.udp.sendto(data, self.config.host_ip, self.config.host_port) elif self.Transparent_Command.get() and msg_id in protocol.CMD_MESSAGES: self._forward_raw(data, msg_id) def _forward_raw(self, data, msg_id): dest = CMD_ID_DEST.get(msg_id) if dest is None: return ip, port = self._dest(dest) self.udp.sendto(data, ip, port) def _dest(self, key): if key == "c1": return self.config.c1_ip, self.config.c1_port return self.config.c2_ip, self.config.c2_port # ------------------------------------------------------------------ # 指令下发 # ------------------------------------------------------------------ def send_breaker_cmd(self, spec, field, state): cmd = spec.msg_cls() setattr(cmd, field, state) if self.Simulation_Mode.get(): self.sim.apply_command(cmd) ip, port = self._dest(spec.dest) ok = self.udp.send_message(cmd, ip, port) if self.Record_Data.get(): self.db.insert_message(cmd) label = next((l for l, f in spec.actions if f == field), field) txt = "合" if state == BREAKER_ON else "分" self.logger.info(f"发送{spec.title}「{label}」{txt} → {ip}:{port}" + ("" if ok else "（发送失败）")) # ------------------------------------------------------------------ # 开关回调 # ------------------------------------------------------------------ def _on_connection_toggle(self): if self.Start_Connection.get(): if not self.udp.bound: self.logger.error("本机地址未绑定，无法启动") self.Start_Connection.set(False) return try: self.udp.start() except RuntimeError as e: self.logger.error(str(e)) self.Start_Connection.set(False) return self.logger.info("通信连接已启动") else: self.udp.stop() self.logger.info("通信连接已停止") def _on_sim_toggle(self, *args): if self.Simulation_Mode.get(): self._start_simulation() else: self._stop_simulation() def _start_simulation(self): if self._sim_running: return self.sim.reset() self._sim_running = True self._sim_thread = threading.Thread(target=self._sim_loop, daemon=True) self._sim_thread.start() self.logger.info("模拟模式已启动") def _stop_simulation(self): was_running = self._sim_running self._sim_running = False if self._sim_thread: self._sim_thread.join(timeout=1.5) self._sim_thread = None if was_running: self.logger.info("模拟模式已停止") def _sim_loop(self): while self._sim_running: msgs = self.sim.update(self.config.sim_interval) for msg in msgs: self._on_feedback(msg, ("sim", 0)) if self.Forward_Feedback.get(): self.udp.sendto(msg.pack(), self.config.host_ip, self.config.host_port) time.sleep(self.config.sim_interval) # ------------------------------------------------------------------ # 界面刷新（主线程定时器） # ------------------------------------------------------------------ def _refresh_ui(self): self.logger.drain() for key, refs in self.bus_refs.items(): with self._data_lock: msg = self.raw_msgs[key] update_bus_page(BUS_SPECS[key], refs, msg) # DCDC 弹窗刷新 if self.dcdc_refs is not None: with self._data_lock: m = self.raw_msgs.get("high_vol") if m is not None: update_dcdc_display(self.dcdc_refs, m.dcDcFaultWord1, m.dcDcFaultWord2) # 状态栏 state = "已连接" if self.udp.running else "未连接" self.status_bar.config( text=f"{state} 收包: {self.udp.rx_count} 发包: {self.udp.tx_count}" f" 模拟: {'开' if self.Simulation_Mode.get() else '关'}" f" 记录: {'开' if self.Record_Data.get() else '关'}") self.root.after(self.config.refresh_ms, self._refresh_ui) # ------------------------------------------------------------------ # 弹窗 # ------------------------------------------------------------------ def show_control_window(self, spec_key): spec = CMD_SPECS[spec_key] popup = tk.Toplevel(self.root) popup.title(spec.title) frame = ttk.Frame(popup, padding=8) frame.pack(fill="both", expand=True) build_control_panel(frame, spec, self.send_breaker_cmd) ttk.Button(popup, text="关闭", command=popup.destroy).pack(pady=8) def show_dcdc_window(self): if self.dcdc_popup is not None and self.dcdc_popup.winfo_exists(): self.dcdc_popup.lift() return popup = tk.Toplevel(self.root) popup.title("DC/DC 状态") frame = ttk.Frame(popup, padding=8) frame.pack(fill="both", expand=True, padx=5, pady=5) self.dcdc_refs = build_dcdc_display(frame, "DC/DC 故障字") self.dcdc_refs["widget"].pack(fill="both", expand=True) self.dcdc_popup = popup def on_close(): self.dcdc_refs = None self.dcdc_popup = None popup.destroy() ttk.Button(popup, text="关闭", command=on_close).pack(pady=8) def show_placeholder_window(self, title, text): popup = tk.Toplevel(self.root) popup.title(title) popup.geometry("420x200") frame = ttk.Frame(popup, padding=20) frame.pack(fill="both", expand=True) ttk.Label(frame, text=text, wraplength=360, justify="left").pack() ttk.Button(popup, text="关闭", command=popup.destroy).pack(pady=10) def show_help_window(self): popup = tk.Toplevel(self.root) popup.title("程序说明") popup.geometry("640x460") frame = ttk.Frame(popup, padding=10) frame.pack(fill="both", expand=True) text = tk.Text(frame, wrap="word", font=("Microsoft YaHei", 10)) sb = ttk.Scrollbar(frame, orient="vertical", command=text.yview) text.configure(yscrollcommand=sb.set) text.pack(side="left", fill="both", expand=True) sb.pack(side="right", fill="y") help_text = f"""配电系统调试助手 功能概述： 1. 高压母线 / 艏部高压 / 仪表母线 / 艏部低压 四路状态实时监控 2. 断路器配电指令远程合/分操作 3. 转发反馈：把收到的反馈转发给上位机(信息主机) 4. 透传指令：把上位机下发的指令透传给对应配电板 5. 模拟模式：无需真实设备即可本地联调 6. 数据记录：反馈与指令自动存入 SQLite（可开关） 通信端点（可在 配置-通信配置 修改）： 本机监听: {self.config.local_ip}:{self.config.local_port} 高压配电板: {self.config.c1_ip}:{self.config.c1_port} 仪表配电板: {self.config.c2_ip}:{self.config.c2_port} 上位机: {self.config.host_ip}:{self.config.host_port} 数据库: {self.config.db_path} 指示灯说明： 绿色=合闸(0x55) 黑色=分闸(0xAA) 红色=故障(0x5A) 灰色=无数据/未知 操作提示： 1. 「启动连接」控制 UDP 接收线程（默认开启） 2. 勾选「模拟模式」后，配电操作会同步回写模拟反馈 3. 重要操作请以现场实际设备状态为准""" text.insert("1.0", help_text) text.config(state="disabled") ttk.Button(popup, text="关闭", command=popup.destroy).pack(pady=8) # ------------------------------------------------------------------ # 配置弹窗 # ------------------------------------------------------------------ def show_config_window(self): win = tk.Toplevel(self.root) win.title("通信配置") win.geometry("420x360") rows = [ ("本机 IP", "local_ip"), ("本机端口", "local_port"), ("高压配电板 IP", "c1_ip"), ("高压配电板端口", "c1_port"), ("仪表配电板 IP", "c2_ip"), ("仪表配电板端口", "c2_port"), ("上位机 IP", "host_ip"), ("上位机端口", "host_port"), ] entries = {} for i, (label, key) in enumerate(rows): ttk.Label(win, text=label).grid(row=i, column=0, padx=10, pady=4, sticky="w") e = ttk.Entry(win, width=20) e.grid(row=i, column=1, padx=10, pady=4) val = getattr(self.config, key) e.insert(0, str(val)) entries[key] = e def save(): try: self.config.set_endpoints( entries["local_ip"].get(), int(entries["local_port"].get()), entries["c1_ip"].get(), int(entries["c1_port"].get()), entries["c2_ip"].get(), int(entries["c2_port"].get()), entries["host_ip"].get(), int(entries["host_port"].get())) self.config.save() # 重新绑定本机 if self.udp.running: self.udp.stop() self.udp.bind(self.config.local_ip, self.config.local_port) if self.Start_Connection.get(): self.udp.start() self.logger.info("通信配置已保存并生效") win.destroy() except Exception as e: self.logger.error(f"配置保存失败: {e}") ttk.Button(win, text="保存", command=save).grid(row=8, column=0, columnspan=2, pady=10) ttk.Button(win, text="取消", command=win.destroy).grid(row=9, column=0, columnspan=2) def show_db_config_window(self): win = tk.Toplevel(self.root) win.title("数据库配置") win.geometry("420x140") ttk.Label(win, text="数据库路径:").grid(row=0, column=0, padx=10, pady=8, sticky="w") e = ttk.Entry(win, width=32) e.grid(row=0, column=1, padx=10, pady=8) e.insert(0, self.config.db_path) def save(): path = e.get().strip() try: self.db.disconnect() self.db = DB(path) self.db.connect() self.config.db_path = path self.config.save() self.logger.info(f"数据库已切换: {path}") win.destroy() except Exception as ex: self.logger.error(f"数据库切换失败: {ex}") ttk.Button(win, text="保存", command=save).grid(row=1, column=0, padx=10, pady=6) ttk.Button(win, text="取消", command=win.destroy).grid(row=1, column=1, pady=6) # ------------------------------------------------------------------ def on_closing(self): self._stop_simulation() self.udp.stop() self.db.disconnect() self.root.destroy()