509 lines
17 KiB
Python
509 lines
17 KiB
Python
"""
|
|
EGM 输电线路绕击跳闸率计算程序 - Pywebview 界面
|
|
使用 Vue 3 + Quasar + TypeScript + Tailwind CSS 作为前端
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import math
|
|
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, para,
|
|
func_ng, min_i, rs_fun, rc_fun, rg_fun,
|
|
bd_area, thunder_density, arc_possibility,
|
|
rg_line_function_factory, solve_circle_intersection,
|
|
solve_circle_line_intersection, Draw
|
|
)
|
|
import numpy as np
|
|
|
|
|
|
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 EGMWebApp:
|
|
"""EGM 计算程序的 Web 界面后端"""
|
|
|
|
def __init__(self):
|
|
self.window = None
|
|
self.web_handler = None
|
|
self.logs: List[Dict[str, str]] = []
|
|
|
|
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)
|
|
|
|
def get_logs(self) -> List[Dict[str, str]]:
|
|
"""获取日志列表"""
|
|
logs = self.logs.copy()
|
|
self.logs = []
|
|
return logs
|
|
|
|
def calculate(self, params: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
执行 EGM 计算
|
|
|
|
Args:
|
|
params: 包含 parameter, advance, optional 的字典
|
|
|
|
Returns:
|
|
计算结果字典
|
|
"""
|
|
self.logs = [] # 清空日志
|
|
self.add_log("info", "开始 EGM 计算...")
|
|
|
|
try:
|
|
# 解析参数
|
|
parameter_data = params.get('parameter', {})
|
|
advance_data = params.get('advance', {})
|
|
optional_data = params.get('optional', {})
|
|
|
|
# 更新全局参数对象
|
|
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 = list(parameter_data.get('gc_x', [17.9, 17]))
|
|
para.ground_angels = [
|
|
angel / 180 * math.pi
|
|
for angel in parameter_data.get('ground_angels', [0])
|
|
]
|
|
para.h_arm = list(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.voltage_n = int(optional_data.get('voltage_n', 3))
|
|
para.max_i = float(optional_data.get('max_i', 200))
|
|
|
|
self.add_log("info", f"参数: 额定电压={para.rated_voltage}kV, 雷暴日={para.td}d, 海拔={para.altitude}m")
|
|
|
|
logger.info("开始执行 _do_calculate...")
|
|
|
|
# 执行实际计算
|
|
self.add_log("info", "EGM 计算中...")
|
|
result = self._do_calculate()
|
|
|
|
# 调试输出
|
|
logger.info(f"_do_calculate 返回 keys: {list(result.keys())}")
|
|
logger.info(f"日志列表: {self.logs}")
|
|
|
|
# 将日志添加到结果中（在返回之前添加最后一条日志）
|
|
self.add_log("info", "EGM 计算完成")
|
|
|
|
# 创建一个新的返回值，确保 logs 字段被包含
|
|
final_result = {
|
|
"success": result.get("success", True),
|
|
"message": result.get("message", "计算完成"),
|
|
"data": result.get("data", {}),
|
|
"logs": self.logs,
|
|
"DEBUG_VERSION": "v2" # 标记版本
|
|
}
|
|
|
|
logger.info(f"最终返回结果 keys: {list(final_result.keys())}")
|
|
logger.info(f"日志数量: {len(self.logs)}")
|
|
logger.info(f"DEBUG: self.logs = {self.logs}")
|
|
|
|
return final_result
|
|
|
|
except Exception as e:
|
|
self.add_log("error", f"计算失败: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return {
|
|
"success": False,
|
|
"message": f"计算失败: {str(e)}",
|
|
"error": str(e),
|
|
"logs": self.logs
|
|
}
|
|
|
|
def _do_calculate(self) -> Dict[str, Any]:
|
|
"""执行实际的EGM计算"""
|
|
try:
|
|
h_whole = para.h_arm[0]
|
|
except Exception as e:
|
|
self.add_log("error", f"获取参数失败: {str(e)}")
|
|
raise
|
|
string_g_len = para.string_g_len
|
|
string_c_len = para.string_c_len
|
|
h_g_sag = para.h_g_sag
|
|
h_c_sag = para.h_c_sag
|
|
gc_x = para.gc_x.copy()
|
|
h_arm = para.h_arm
|
|
|
|
gc_y = [
|
|
h_whole - string_g_len - h_g_sag * 2 / 3,
|
|
]
|
|
if len(h_arm) > 1:
|
|
for hoo in h_arm[1:]:
|
|
gc_y.append(hoo - string_c_len - h_c_sag * 2 / 3)
|
|
|
|
if len(gc_y) > 2:
|
|
phase_n = 3
|
|
else:
|
|
phase_n = 1
|
|
|
|
td = para.td
|
|
ng = func_ng(td)
|
|
avr_n_sf = 0
|
|
ground_angels = para.ground_angels
|
|
voltage_n = para.voltage_n
|
|
n_sf_phases = np.zeros((phase_n, voltage_n))
|
|
|
|
results = []
|
|
|
|
for ground_angel in ground_angels:
|
|
self.add_log("info", f"地面倾角 {ground_angel / math.pi * 180:.3f}°")
|
|
rg_type = None
|
|
rg_x = None
|
|
rg_y = None
|
|
|
|
for phase_conductor_foo in range(phase_n):
|
|
rs_x = gc_x[phase_conductor_foo]
|
|
rs_y = gc_y[phase_conductor_foo]
|
|
rc_x = gc_x[phase_conductor_foo + 1]
|
|
rc_y = gc_y[phase_conductor_foo + 1]
|
|
|
|
if phase_n == 1:
|
|
rg_type = "g"
|
|
if phase_n > 1:
|
|
if phase_conductor_foo < 2:
|
|
rg_type = "c"
|
|
rg_x = gc_x[phase_conductor_foo + 2]
|
|
rg_y = gc_y[phase_conductor_foo + 2]
|
|
else:
|
|
rg_type = "g"
|
|
|
|
rated_voltage = para.rated_voltage
|
|
|
|
for u_bar in range(voltage_n):
|
|
u_ph = rated_voltage / 1.732
|
|
|
|
insulator_c_len = para.insulator_c_len
|
|
i_min = min_i(insulator_c_len, u_ph)
|
|
_min_i = i_min
|
|
_max_i = para.max_i
|
|
|
|
i_max = _min_i
|
|
|
|
for i_bar in np.linspace(_min_i, _max_i, int((_max_i - _min_i) / 1)):
|
|
rs = rs_fun(i_bar)
|
|
rc = rc_fun(i_bar, u_ph)
|
|
rg = rg_fun(i_bar, rc_y, u_ph, typ=rg_type)
|
|
|
|
rg_line_func = None
|
|
if rg_type == "g":
|
|
rg_line_func = rg_line_function_factory(rg, ground_angel)
|
|
|
|
rs_rc_circle_intersection = solve_circle_intersection(
|
|
rs, rc, rs_x, rs_y, rc_x, rc_y
|
|
)
|
|
i_max = i_bar
|
|
|
|
if not rs_rc_circle_intersection:
|
|
continue
|
|
|
|
circle_rc_or_rg_line_intersection = None
|
|
if rg_type == "g":
|
|
circle_rc_or_rg_line_intersection = solve_circle_line_intersection(
|
|
rc, rc_x, rc_y, rg_line_func
|
|
)
|
|
elif rg_type == "c":
|
|
circle_rc_or_rg_line_intersection = solve_circle_intersection(
|
|
rg, rc, rg_x, rg_y, rc_x, rc_y
|
|
)
|
|
|
|
if not circle_rc_or_rg_line_intersection:
|
|
if rg_type == "g":
|
|
if rg_line_func(rc_x) > rc_y:
|
|
i_min = i_bar
|
|
continue
|
|
else:
|
|
continue
|
|
|
|
min_distance_intersection = (
|
|
np.sum(
|
|
(
|
|
np.array(rs_rc_circle_intersection)
|
|
- np.array(circle_rc_or_rg_line_intersection)
|
|
)
|
|
** 2
|
|
)
|
|
** 0.5
|
|
)
|
|
|
|
if min_distance_intersection < 0.1:
|
|
break
|
|
|
|
self.add_log("info", f"最大电流为 {i_max:.2f}, 最小电流为 {i_min:.2f}")
|
|
|
|
curt_fineness = 0.1
|
|
if i_min > i_max or abs(i_min - i_max) < curt_fineness:
|
|
self.add_log("info", "最大电流小于等于最小电流，没有暴露弧。")
|
|
continue
|
|
|
|
curt_segment_n = int((i_max - i_min) / curt_fineness)
|
|
i_curt_samples, d_curt = np.linspace(
|
|
i_min, i_max, curt_segment_n + 1, retstep=True
|
|
)
|
|
|
|
bd_area_vec = np.vectorize(bd_area)
|
|
ip_a = para.Ip_a
|
|
ip_b = para.Ip_b
|
|
|
|
bd_area_vec_result = bd_area_vec(
|
|
i_curt_samples,
|
|
u_ph,
|
|
rc_x,
|
|
rc_y,
|
|
rs_x,
|
|
rs_y,
|
|
rg_x,
|
|
rg_y,
|
|
ground_angel,
|
|
rg_type,
|
|
)
|
|
thunder_density_result = thunder_density(
|
|
i_curt_samples, td, ip_a, ip_b
|
|
)
|
|
cal_bd_np = bd_area_vec_result * thunder_density_result
|
|
calculus = np.sum(cal_bd_np[:-1] + cal_bd_np[1:]) / 2 * d_curt
|
|
|
|
n_sf = (
|
|
2
|
|
* ng
|
|
/ 10
|
|
* calculus
|
|
* arc_possibility(rated_voltage, insulator_c_len)
|
|
)
|
|
avr_n_sf += n_sf / voltage_n
|
|
n_sf_phases[phase_conductor_foo][u_bar] = n_sf
|
|
|
|
self.add_log("info", f"相{phase_conductor_foo + 1}, 跳闸率: {n_sf:.16f} 次/(100km·a)")
|
|
|
|
result = {
|
|
"ground_angle": f"{ground_angel / math.pi * 180:.3f}°",
|
|
"tripping_rate": avr_n_sf,
|
|
"phases": np.mean(n_sf_phases, axis=1).tolist()
|
|
}
|
|
results.append(result)
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "计算完成",
|
|
"data": {
|
|
"tripping_rate": f"{avr_n_sf:.16f} 次/(100km·a)",
|
|
"results": results,
|
|
"parameters": {
|
|
"rated_voltage": para.rated_voltage,
|
|
"td": para.td,
|
|
"altitude": para.altitude,
|
|
"ground_angels": [a / math.pi * 180 for a in para.ground_angels],
|
|
"max_i": para.max_i
|
|
}
|
|
}
|
|
}
|
|
|
|
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 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 开发服务器
|
|
# 在生产环境中使用构建后的文件
|
|
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=1200,
|
|
height=900,
|
|
resizable=True,
|
|
min_size=(800, 600)
|
|
)
|
|
|
|
# 将窗口对象传递给 API
|
|
api.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() |