662 lines
21 KiB
Python
662 lines
21 KiB
Python
"""
|
||
EGM 输电线路绕击跳闸率计算程序 - Pywebview 界面
|
||
使用 Vue 3 + Quasar + TypeScript + Tailwind CSS 作为前端
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import math
|
||
import threading
|
||
import queue
|
||
import tomllib
|
||
from pathlib import Path
|
||
from typing import Dict, Any, List
|
||
from datetime import datetime
|
||
import webview
|
||
from loguru import logger
|
||
|
||
# 添加项目根目录到路径
|
||
project_root = Path(__file__).parent
|
||
sys.path.insert(0, str(project_root))
|
||
|
||
from core import Parameter
|
||
from main import parameter_display, run_egm
|
||
|
||
|
||
class WebAnimation:
|
||
"""
|
||
Web 动画类，将 Python 端的 Animation 调用映射到前端 JavaScript
|
||
对应 Animation.vue 的功能
|
||
|
||
注意：动画的启用/禁用由前端用户通过"启用动画"开关控制，
|
||
后端只负责发送绘制指令，前端根据 enabled 状态决定是否执行
|
||
"""
|
||
|
||
def __init__(self, window=None):
|
||
self._window = window
|
||
|
||
def set_window(self, window):
|
||
"""设置窗口对象"""
|
||
self._window = window
|
||
|
||
def enable(self, enabled: bool):
|
||
"""
|
||
启用/禁用动画（由前端用户控制）
|
||
此方法保留以兼容接口，但实际启用状态由前端控制
|
||
"""
|
||
if self._window:
|
||
js_code = f'if(window.animationApi){{window.animationApi.enable({str(enabled).lower()})}}'
|
||
self._window.evaluate_js(js_code)
|
||
|
||
def init_fig(self):
|
||
"""初始化画布"""
|
||
if not self._window:
|
||
return
|
||
js_code = 'if(window.animationApi){window.animationApi.initFig()}'
|
||
self._window.evaluate_js(js_code)
|
||
|
||
def add_rs(self, rs: float, rs_x: float, rs_y: float):
|
||
"""
|
||
添加地线保护弧（RS 圆）
|
||
对应 animation.py 的 add_rs 方法
|
||
"""
|
||
if not self._window:
|
||
return
|
||
js_code = f'if(window.animationApi){{window.animationApi.addRs({rs}, {rs_x}, {rs_y})}}'
|
||
self._window.evaluate_js(js_code)
|
||
|
||
def add_rc(self, rc: float, rc_x: float, rc_y: float):
|
||
"""
|
||
添加导线暴露弧（RC 圆）
|
||
对应 animation.py 的 add_rc 方法
|
||
"""
|
||
if not self._window:
|
||
return
|
||
js_code = f'if(window.animationApi){{window.animationApi.addRc({rc}, {rc_x}, {rc_y})}}'
|
||
self._window.evaluate_js(js_code)
|
||
|
||
def add_rg_line(self, line_func):
|
||
"""
|
||
添加地面线（RG 线）
|
||
对应 animation.py 的 add_rg_line 方法
|
||
|
||
Args:
|
||
line_func: 一个函数，接收 x 返回 y
|
||
"""
|
||
if not self._window:
|
||
return
|
||
# 生成线上的点，传递给前端
|
||
# 由于无法直接传递函数，我们预先计算一些点
|
||
import numpy as np
|
||
x_points = np.linspace(0, 300, 50)
|
||
y_points = [line_func(x) for x in x_points]
|
||
points = list(zip(x_points.tolist(), y_points))
|
||
|
||
js_code = f'''
|
||
if(window.animationApi){{
|
||
window.animationApi.addRgLine({json.dumps(points)})
|
||
}}
|
||
'''
|
||
self._window.evaluate_js(js_code)
|
||
|
||
def add_expose_area(
|
||
self,
|
||
rc_x: float,
|
||
rc_y: float,
|
||
intersection_x1: float,
|
||
intersection_y1: float,
|
||
intersection_x2: float,
|
||
intersection_y2: float
|
||
):
|
||
"""
|
||
添加暴露弧区域（两条红线）
|
||
对应 animation.py 的 add_expose_area 方法
|
||
"""
|
||
if not self._window:
|
||
return
|
||
js_code = f'''if(window.animationApi){{
|
||
window.animationApi.addExposeArea(
|
||
{rc_x}, {rc_y},
|
||
{intersection_x1}, {intersection_y1},
|
||
{intersection_x2}, {intersection_y2}
|
||
)
|
||
}}'''
|
||
self._window.evaluate_js(js_code)
|
||
|
||
def clear(self):
|
||
"""清除画布"""
|
||
if not self._window:
|
||
return
|
||
js_code = 'if(window.animationApi){window.animationApi.clear()}'
|
||
self._window.evaluate_js(js_code)
|
||
|
||
def pause(self):
|
||
"""
|
||
暂停并刷新
|
||
对应 animation.py 的 pause 方法
|
||
"""
|
||
if not self._window:
|
||
return
|
||
js_code = 'if(window.animationApi){window.animationApi.pause()}'
|
||
self._window.evaluate_js(js_code)
|
||
# 添加延迟以便动画可见
|
||
import time
|
||
time.sleep(0.1) # 增加延迟，让用户看清动画
|
||
|
||
|
||
class WebHandler:
|
||
"""Web日志处理器"""
|
||
def __init__(self, callback=None):
|
||
self.callback = callback
|
||
self.logs: List[Dict[str, str]] = []
|
||
|
||
def write(self, message):
|
||
if message.strip():
|
||
log_entry = {
|
||
"level": "info",
|
||
"time": datetime.now().strftime("%H:%M:%S"),
|
||
"message": message.strip()
|
||
}
|
||
self.logs.append(log_entry)
|
||
if self.callback:
|
||
self.callback(log_entry)
|
||
|
||
|
||
class LoguruWebHandler:
|
||
"""Loguru 日志处理器，将 loguru 日志转发到 Web 界面"""
|
||
|
||
def __init__(self, log_queue: queue.Queue):
|
||
self.log_queue = log_queue
|
||
|
||
def write(self, message):
|
||
"""loguru handler 的写入方法"""
|
||
record = message.record
|
||
level = record['level'].name.lower()
|
||
|
||
# 映射 loguru 级别到前端级别
|
||
level_map = {
|
||
'trace': 'debug',
|
||
'debug': 'debug',
|
||
'info': 'info',
|
||
'success': 'info',
|
||
'warning': 'warning',
|
||
'error': 'error',
|
||
'critical': 'error'
|
||
}
|
||
frontend_level = level_map.get(level, 'info')
|
||
|
||
# 提取消息文本
|
||
msg = record['message']
|
||
if msg.strip():
|
||
log_entry = {
|
||
"level": frontend_level,
|
||
"time": datetime.now().strftime("%H:%M:%S"),
|
||
"message": msg
|
||
}
|
||
# 将日志放入队列，由主线程处理
|
||
self.log_queue.put(log_entry)
|
||
|
||
|
||
class EGMWebApp:
|
||
"""EGM 计算程序的 Web 界面后端"""
|
||
|
||
def __init__(self):
|
||
self.window = None
|
||
self.web_handler = None
|
||
self.logs: List[Dict[str, str]] = []
|
||
self._loguru_handler_id = None
|
||
self._log_queue: queue.Queue = queue.Queue()
|
||
self._running = False
|
||
self.animation = WebAnimation() # Web 动画实例
|
||
|
||
def _process_log_queue(self):
|
||
"""处理日志队列，在主线程中定时调用"""
|
||
if not self._running:
|
||
return
|
||
|
||
try:
|
||
# 处理队列中的所有日志
|
||
while not self._log_queue.empty():
|
||
try:
|
||
log_entry = self._log_queue.get_nowait()
|
||
self._push_log_to_frontend(log_entry)
|
||
except queue.Empty:
|
||
break
|
||
except Exception as e:
|
||
print(f"处理日志队列失败: {e}")
|
||
|
||
# 继续定时检查
|
||
if self._running:
|
||
threading.Timer(0.05, self._process_log_queue).start()
|
||
|
||
def _push_log_to_frontend(self, log_entry: Dict[str, str]):
|
||
"""推送单条日志到前端"""
|
||
if self.window:
|
||
try:
|
||
js_code = f'if(window.addLogFromBackend){{window.addLogFromBackend({json.dumps(log_entry)})}}'
|
||
self.window.evaluate_js(js_code)
|
||
except Exception as e:
|
||
print(f"推送日志到前端失败: {e}")
|
||
|
||
def add_log(self, level: str, message: str):
|
||
"""添加日志并实时推送到前端"""
|
||
log_entry = {
|
||
"level": level,
|
||
"time": datetime.now().strftime("%H:%M:%S"),
|
||
"message": message
|
||
}
|
||
self.logs.append(log_entry)
|
||
|
||
# 将日志放入队列，由主线程处理
|
||
self._log_queue.put(log_entry)
|
||
|
||
def get_logs(self) -> List[Dict[str, str]]:
|
||
"""获取日志列表"""
|
||
logs = self.logs.copy()
|
||
self.logs = []
|
||
return logs
|
||
|
||
def _setup_loguru_handler(self):
|
||
"""设置 loguru 处理器，捕获所有 logger 调用"""
|
||
self._loguru_handler_id = logger.add(
|
||
LoguruWebHandler(self._log_queue),
|
||
format="{message}",
|
||
level="DEBUG"
|
||
)
|
||
|
||
def _remove_loguru_handler(self):
|
||
"""移除 loguru 处理器"""
|
||
if self._loguru_handler_id is not None:
|
||
logger.remove(self._loguru_handler_id)
|
||
self._loguru_handler_id = None
|
||
|
||
def calculate(self, params: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
执行 EGM 计算（启动后台线程，立即返回）
|
||
|
||
Args:
|
||
params: 包含 parameter, advance, optional 的字典
|
||
|
||
Returns:
|
||
计算状态字典
|
||
"""
|
||
self.logs = [] # 清空日志
|
||
self._log_queue = queue.Queue() # 重置队列
|
||
|
||
# 启动日志队列处理器
|
||
self._running = True
|
||
self._process_log_queue()
|
||
|
||
# 启动后台线程执行计算
|
||
thread = threading.Thread(target=self._calculate_thread, args=(params,))
|
||
thread.daemon = True
|
||
thread.start()
|
||
|
||
return {"status": "started", "message": "计算已启动"}
|
||
|
||
def _calculate_thread(self, params: Dict[str, Any]):
|
||
"""后台线程中执行计算"""
|
||
# 设置 loguru 处理器，捕获所有 logger.info/debug 等调用
|
||
self._setup_loguru_handler()
|
||
|
||
self.add_log("info", "开始 EGM 计算...")
|
||
|
||
try:
|
||
# 解析参数
|
||
parameter_data = params.get('parameter', {})
|
||
advance_data = params.get('advance', {})
|
||
optional_data = params.get('optional', {})
|
||
|
||
# 创建局部参数对象
|
||
para = Parameter()
|
||
para.h_g_sag = float(parameter_data.get('h_g_sag', 11.67))
|
||
para.h_c_sag = float(parameter_data.get('h_c_sag', 14.43))
|
||
para.td = int(parameter_data.get('td', 20))
|
||
para.insulator_c_len = float(parameter_data.get('insulator_c_len', 7.02))
|
||
para.string_c_len = float(parameter_data.get('string_c_len', 9.2))
|
||
para.string_g_len = float(parameter_data.get('string_g_len', 0.5))
|
||
# 确保数组元素转换为数字类型
|
||
para.gc_x = [float(x) for x in parameter_data.get('gc_x', [17.9, 17])]
|
||
para.ground_angels = [
|
||
float(angel) / 180 * math.pi
|
||
for angel in parameter_data.get('ground_angels', [0])
|
||
]
|
||
para.h_arm = [float(h) for h in parameter_data.get('h_arm', [150, 130])]
|
||
para.altitude = int(parameter_data.get('altitude', 1000))
|
||
# 解析电压等级字符串，如 "500kV" -> 500
|
||
rated_voltage_str = str(parameter_data.get('rated_voltage', '500kV'))
|
||
para.rated_voltage = float(rated_voltage_str.replace('kV', '').replace('±', ''))
|
||
|
||
para.ng = float(advance_data.get('ng', -1))
|
||
para.Ip_a = float(advance_data.get('Ip_a', -1))
|
||
para.Ip_b = float(advance_data.get('Ip_b', -1))
|
||
para.u_50 = float(advance_data.get('u_50', -1))
|
||
|
||
para.voltage_n = int(optional_data.get('voltage_n', 3))
|
||
para.max_i = float(optional_data.get('max_i', 200))
|
||
para.z_0 = float(parameter_data.get('z_0', 300))
|
||
para.z_c = float(parameter_data.get('z_c', 251))
|
||
|
||
# 设置 ac_or_dc 参数
|
||
ac_or_dc_value = str(parameter_data.get('ac_or_dc', 'AC'))
|
||
para.ac_or_dc = 'DC' if 'DC' in ac_or_dc_value.upper() else 'AC'
|
||
|
||
# 调用 main.py 的参数显示函数，日志会被 loguru handler 捕获
|
||
# parameter_display(para)
|
||
|
||
logger.info("开始执行 EGM 计算...")
|
||
|
||
# 根据前端动画启用状态决定是否传递 animation 对象
|
||
animation_enabled = params.get('animation_enabled', False)
|
||
animation_obj = self.animation if animation_enabled else None
|
||
|
||
# 调用 main.py 的核心计算函数
|
||
result = run_egm(para, animation_obj)
|
||
|
||
self.add_log("info", "EGM 计算完成")
|
||
|
||
# 推送结果到前端
|
||
self._send_result_to_frontend(result)
|
||
|
||
# 等待队列中的日志处理完毕
|
||
import time
|
||
time.sleep(0.1)
|
||
while not self._log_queue.empty():
|
||
time.sleep(0.05)
|
||
|
||
# 停止日志队列处理器
|
||
self._running = False
|
||
|
||
# 移除 loguru 处理器
|
||
self._remove_loguru_handler()
|
||
|
||
except Exception as e:
|
||
self.add_log("error", f"计算失败: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
# 停止日志队列处理器
|
||
self._running = False
|
||
|
||
# 移除 loguru 处理器
|
||
self._remove_loguru_handler()
|
||
|
||
# 推送错误到前端
|
||
self._send_result_to_frontend({
|
||
"success": False,
|
||
"message": f"计算失败: {str(e)}",
|
||
"error": str(e)
|
||
})
|
||
|
||
def _send_result_to_frontend(self, result: Dict[str, Any]):
|
||
"""将计算结果推送到前端"""
|
||
if self.window:
|
||
try:
|
||
js_code = f'if(window.receiveResult){{window.receiveResult({json.dumps(result)})}}'
|
||
self.window.evaluate_js(js_code)
|
||
except Exception as e:
|
||
logger.error(f"推送结果到前端失败: {e}")
|
||
|
||
def dict_to_toml(self, obj: Dict[str, Any], indent: str = '') -> str:
|
||
"""
|
||
将字典转换为 TOML 格式字符串
|
||
|
||
Args:
|
||
obj: 参数字典
|
||
indent: 缩进字符串
|
||
|
||
Returns:
|
||
TOML 格式字符串
|
||
"""
|
||
result = ''
|
||
|
||
for key, value in obj.items():
|
||
if value is None:
|
||
continue
|
||
|
||
if isinstance(value, list):
|
||
result += f'{indent}{key} = [{", ".join(str(v) for v in value)}]\n'
|
||
elif isinstance(value, dict):
|
||
result += f'\n{indent}[{key}]\n'
|
||
result += self.dict_to_toml(value, indent)
|
||
elif isinstance(value, str):
|
||
result += f'{indent}{key} = "{value}"\n'
|
||
elif isinstance(value, bool):
|
||
result += f'{indent}{key} = {str(value).lower()}\n'
|
||
else:
|
||
result += f'{indent}{key} = {value}\n'
|
||
|
||
return result
|
||
|
||
def export_config(self, params: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
导出配置为 TOML 文件，弹出保存对话框
|
||
|
||
Args:
|
||
params: 参数字典
|
||
|
||
Returns:
|
||
包含保存状态和路径的字典
|
||
"""
|
||
try:
|
||
# 生成默认文件名
|
||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||
default_filename = f'egm_config_{timestamp}.toml'
|
||
|
||
# 打开保存文件对话框
|
||
result = self.window.create_file_dialog(
|
||
webview.SAVE_DIALOG,
|
||
directory='',
|
||
save_filename=default_filename,
|
||
file_types=('TOML Files (*.toml)', 'All files (*.*)')
|
||
)
|
||
|
||
if result and len(result) > 0:
|
||
file_path = result[0]
|
||
|
||
# 转换为 TOML 格式
|
||
toml_content = self.dict_to_toml(params)
|
||
|
||
# 写入文件
|
||
with open(file_path, 'w', encoding='utf-8') as f:
|
||
f.write(toml_content)
|
||
|
||
return {
|
||
"success": True,
|
||
"message": f"配置已保存到: {file_path}",
|
||
"file_path": file_path
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": "用户取消了保存操作"
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"导出配置失败: {str(e)}")
|
||
return {
|
||
"success": False,
|
||
"message": f"保存失败: {str(e)}"
|
||
}
|
||
|
||
def export_log(self, log_text: str) -> Dict[str, Any]:
|
||
"""
|
||
导出日志为 TXT 文件，弹出保存对话框
|
||
|
||
Args:
|
||
log_text: 日志文本内容
|
||
|
||
Returns:
|
||
包含保存状态和路径的字典
|
||
"""
|
||
try:
|
||
# 生成默认文件名
|
||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||
default_filename = f'egm_log_{timestamp}.txt'
|
||
|
||
# 打开保存文件对话框
|
||
result = self.window.create_file_dialog(
|
||
webview.SAVE_DIALOG,
|
||
directory='',
|
||
save_filename=default_filename,
|
||
file_types=('Text Files (*.txt)', 'All files (*.*)')
|
||
)
|
||
|
||
if result and len(result) > 0:
|
||
file_path = result[0]
|
||
|
||
# 写入文件
|
||
with open(file_path, 'w', encoding='utf-8') as f:
|
||
f.write(log_text)
|
||
|
||
return {
|
||
"success": True,
|
||
"message": f"日志已保存到: {file_path}",
|
||
"file_path": file_path
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": "用户取消了保存操作"
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"导出日志失败: {str(e)}")
|
||
return {
|
||
"success": False,
|
||
"message": f"保存失败: {str(e)}"
|
||
}
|
||
|
||
def import_config(self) -> Dict[str, Any]:
|
||
"""
|
||
导入配置从 TOML 文件，弹出打开对话框
|
||
|
||
Returns:
|
||
包含解析后的参数和文件路径的字典
|
||
"""
|
||
try:
|
||
# 打开文件选择对话框
|
||
result = self.window.create_file_dialog(
|
||
webview.OPEN_DIALOG,
|
||
directory='',
|
||
file_types=('TOML Files (*.toml)', 'All files (*.*)')
|
||
)
|
||
|
||
if result and len(result) > 0:
|
||
file_path = result[0]
|
||
|
||
# 读取并解析 TOML 文件
|
||
with open(file_path, 'rb') as f:
|
||
toml_data = tomllib.load(f)
|
||
|
||
return {
|
||
"success": True,
|
||
"message": f"成功导入配置",
|
||
"file_path": file_path,
|
||
"params": toml_data
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": "用户取消了选择"
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"导入配置失败: {str(e)}")
|
||
return {
|
||
"success": False,
|
||
"message": f"导入失败: {str(e)}"
|
||
}
|
||
|
||
def get_default_config(self) -> Dict[str, Any]:
|
||
"""
|
||
获取默认配置
|
||
|
||
Returns:
|
||
默认配置字典
|
||
"""
|
||
return {
|
||
"parameter": {
|
||
"rated_voltage": 750,
|
||
"h_c_sag": 14.43,
|
||
"h_g_sag": 11.67,
|
||
"insulator_c_len": 7.02,
|
||
"string_c_len": 9.2,
|
||
"string_g_len": 0.5,
|
||
"h_arm": [150, 130],
|
||
"gc_x": [17.9, 17],
|
||
"ground_angels": [0],
|
||
"altitude": 1000,
|
||
"td": 20
|
||
},
|
||
"advance": {
|
||
"ng": -1,
|
||
"Ip_a": -1,
|
||
"Ip_b": -1
|
||
},
|
||
"optional": {
|
||
"voltage_n": 3,
|
||
"max_i": 200
|
||
}
|
||
}
|
||
|
||
|
||
def start_webview():
|
||
"""启动 pywebview 界面"""
|
||
|
||
# 确定前端 URL
|
||
# 在开发环境中使用 Vite 开发服务器
|
||
# 在生产环境中使用构建后的文件
|
||
|
||
# 检查是否在打包环境中运行
|
||
if getattr(sys, 'frozen', False):
|
||
# 打包环境：强制使用生产模式，禁用调试
|
||
dev_mode = False
|
||
else:
|
||
# 开发环境：通过环境变量控制
|
||
dev_mode = os.getenv('EGM_DEV_MODE', 'true').lower() == 'true'
|
||
|
||
if dev_mode:
|
||
# 开发模式：使用 Vite 开发服务器
|
||
url = 'http://localhost:5173'
|
||
logger.info(f"开发模式：使用 Vite 开发服务器 {url}")
|
||
logger.info("请先在 webui 目录中运行: npm install && npm run dev")
|
||
else:
|
||
# 生产模式：使用构建后的文件
|
||
dist_path = project_root / 'webui' / 'dist'
|
||
if not dist_path.exists():
|
||
logger.error(f"构建目录不存在: {dist_path}")
|
||
logger.error("请先运行: cd webui && npm run build")
|
||
sys.exit(1)
|
||
url = f'file://{dist_path / "index.html"}'
|
||
logger.info(f"生产模式：使用构建文件 {url}")
|
||
|
||
# 创建 API 实例
|
||
api = EGMWebApp()
|
||
|
||
# 创建窗口
|
||
window = webview.create_window(
|
||
title='EGM 输电线路绕击跳闸率计算',
|
||
url=url,
|
||
js_api=api,
|
||
width=1500,
|
||
height=900,
|
||
resizable=True,
|
||
min_size=(800, 600)
|
||
)
|
||
|
||
# 将窗口对象传递给 API
|
||
api.window = window
|
||
api.animation.set_window(window)
|
||
|
||
# 启动
|
||
logger.info("启动 EGM Web 界面...")
|
||
webview.start(debug=dev_mode)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# 配置日志
|
||
logger.remove()
|
||
logger.add(sys.stderr, level="INFO")
|
||
logger.add("egm_webui.log", rotation="10 MB", retention="7 days")
|
||
|
||
# 启动界面
|
||
start_webview() |