""" 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 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 EGMWebApp: """EGM 计算程序的 Web 界面后端""" def __init__(self): self.window = None def calculate(self, params: Dict[str, Any]) -> Dict[str, Any]: """ 执行 EGM 计算 Args: params: 包含 parameter, advance, optional 的字典 Returns: 计算结果字典 """ try: logger.info("开始 EGM 计算...") # 解析参数 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)) para.rated_voltage = float(parameter_data.get('rated_voltage', 750)) 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)) logger.info(f"参数: 额定电压={para.rated_voltage}kV, 雷暴日={para.td}d, 海拔={para.altitude}m") # 执行实际计算 result = self._do_calculate() logger.info("EGM 计算完成") return result except Exception as e: logger.error(f"计算失败: {str(e)}") import traceback traceback.print_exc() return { "success": False, "message": f"计算失败: {str(e)}", "error": str(e) } def _do_calculate(self) -> Dict[str, Any]: """执行实际的EGM计算""" h_whole = para.h_arm[0] 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: logger.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 logger.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: logger.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 logger.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 export_config(self, params: Dict[str, Any]) -> str: """ 导出配置为 JSON 字符串 Args: params: 参数字典 Returns: JSON 字符串 """ return json.dumps(params, indent=2, ensure_ascii=False) 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) ) # 启动 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()