PIPESIM管网管径优化自动化Python模板

# Python3.9解释器路径，无需自动切换填None
PYTHON39_EXE = r"C:\Python39\python.exe"
# PIPESIM PTK完整目录
PTK_PATH = r"C:\Users\xxx\Documents\PIPESIM Python Toolkit"
# .pips仿真模型文件完整路径
MODEL_PATH = r"D:\OilField\Model\GatheringNetwork.pips"
# 需要优化的主干管线名称（与模型内名称完全一致）
TARGET_PIPE = "FL"
# 管网末端处理厂节点，过滤该节点数据
TERMINAL_NODE = "处理厂"
# 批量扫描管径（单位：英寸）
TEST_DIAMETERS = [4.0, 6.0, 8.0, 10.0]

import sys
import os
import platform
import pandas as pd
import matplotlib.pyplot as plt
# 适用于PIPESIM2022和Python3.9，需要手动安装PIPESIM的Python ToolKit
# =========================================================================
# 配置区（使用前请根据实际情况修改以下所有参数）
# =========================================================================
# --- Python 环境配置 ---
# 如果当前 Python 版本不符合要求，脚本会自动切换到下方指定的解释器路径
# 仅 Windows 需要修改此项；如不需要自动切换，可将此项设为 None
PYTHON39_EXE = None  # 示例: r"C:\Users\你的用户名\AppData\Local\Programs\Python\Python39\python.exe"
# --- PIPESIM Python Toolkit 路径 ---
# 填写 PIPESIM Python Toolkit 文件夹的完整路径
PTK_PATH = r"请填写PIPESIM Python Toolkit的完整路径"
# 示例: r"C:\Users\你的用户名\Documents\PIPESIM Python Toolkit"
# --- 模型文件路径 ---
# 填写 .pips 模型文件的完整路径
MODEL_PATH = r"请填写.pips模型文件的完整路径"
# 示例: r"D:\Projects\MyModel\model.pips"
# --- 管网对象名称 ---
# 要进行管径扫描的主干线对象名称（与 PIPESIM 模型中的对象名一致）
TARGET_PIPE = "FL"
# 管网终点节点名称（需要从结果中排除的节点及主干线，避免大数干扰图表）
TERMINAL_NODE = "处理厂"
# --- 扫描参数 ---
# 要依次测试的管道内径列表，单位：英寸
TEST_DIAMETERS = [4.0, 6.0, 8.0, 10.0]
# --- 字体配置 ---
# 优先使用前面的字体；如果运行环境没有中文字体，在列表末尾已添加通用回退字体
_FONT_CANDIDATES = {
    "Windows": ["Microsoft YaHei", "SimHei", "DejaVu Sans"],
    "Darwin":  ["PingFang SC", "Heiti TC", "Arial Unicode MS", "DejaVu Sans"],
    "Linux":   ["WenQuanYi Micro Hei", "Noto Sans CJK SC", "DejaVu Sans"],
}
plt.rcParams["font.sans-serif"] = _FONT_CANDIDATES.get(platform.system(), ["DejaVu Sans"])
plt.rcParams["axes.unicode_minus"] = False
# =========================================================================
# 环境检查与路径挂载
# =========================================================================
REQUIRED_PYTHON = (3, 9)
if sys.version_info[:2] != REQUIRED_PYTHON:
    if PYTHON39_EXE and os.path.exists(PYTHON39_EXE):
        print(f"[INFO] 当前 Python {sys.version.split()[0]}，切换至指定解释器: {PYTHON39_EXE}")
        os.execv(PYTHON39_EXE, [PYTHON39_EXE, __file__, *sys.argv[1:]])
    print(f"[ERROR] 需要 Python {REQUIRED_PYTHON[0]}.{REQUIRED_PYTHON[1]}，当前为 {sys.version.split()[0]}。")
    if PYTHON39_EXE:
        print(f"[ERROR] 未找到指定的解释器路径: {PYTHON39_EXE}")
    sys.exit(1)
if not os.path.exists(PTK_PATH):
    print(f"[ERROR] PIPESIM Python Toolkit 路径不存在: {PTK_PATH}")
    sys.exit(1)
if PTK_PATH not in sys.path:
    sys.path.insert(0, PTK_PATH)
print(f"[OK] 已加载 PIPESIM Python Toolkit: {PTK_PATH}")
from sixgill.pipesim import Model
from sixgill.definitions import Parameters
# =========================================================================
# 辅助函数
# =========================================================================
# 输出目录与主干线模型文件同级
WORKSPACE_DIR = os.path.dirname(MODEL_PATH)
results = []
def collect_network_results(result, diameter):
    """
    从仿真结果中提取各支线的压降与温降，并完成单位换算。
    自动排除主干线（TARGET_PIPE）和终点节点（TERMINAL_NODE），
    保留上游纯支线数据，确保图表纵向分辨率。
    单位换算规则：
        压降: psia  → MPa  (× 0.00689476)
        温降: °F差  → °C差 (× 5/9)
    """
    pressure_data    = result.system.get("SystemPressureLoss")
    temperature_data = result.system.get("SystemTemperatureDifference")
    if pressure_data is None or temperature_data is None:
        raise KeyError("仿真结果缺少 SystemPressureLoss 或 SystemTemperatureDifference。")
    exclude = {TARGET_PIPE, TERMINAL_NODE, "Unit"}
    branches = sorted((set(pressure_data) | set(temperature_data)) - exclude)
    if not branches:
        raise KeyError("过滤后没有可汇总的支线分支。")
    rows = []
    for branch in branches:
        raw_p = pressure_data.get(branch)
        raw_t = temperature_data.get(branch)
        rows.append({
            "Diameter":          diameter,
            "Branch":            branch,
            "PressureDrop":      raw_p * 0.00689476  if raw_p is not None else None,
            "PressureDropUnit":  "MPa",
            "TemperatureDrop":   raw_t * (5.0 / 9.0) if raw_t is not None else None,
            "TemperatureDropUnit": "℃",
            "Status":            "Success",
        })
    return rows
def set_pipe_diameter(model, sections, diameter):
    """
    将管径写入模型。优先逐段赋值；无段信息时依次回退到
    '{TARGET_PIPE}.Section.1' 和 TARGET_PIPE 本身。
    若三种方式均失败则抛出异常，避免静默赋值错误。
    """
    if sections:
        for section in sections:
            model.set_value(context=section, parameter="InnerDiameter", value=diameter)
        return
    for ctx in [f"{TARGET_PIPE}.Section.1", TARGET_PIPE]:
        try:
            model.set_value(context=ctx, parameter="InnerDiameter", value=diameter)
            return
        except Exception:
            continue
    raise RuntimeError(f"无法对管道 '{TARGET_PIPE}' 完成内径赋值，请检查模型对象名称。")
def get_simulation_task(model):
    """
    自动探测 model.tasks 中的网络仿真任务句柄，
    兼容不同版本 sixgill 库的属性命名差异。
    """
    for attr in ["networksimulation", "network_simulation", "network", "ptask"]:
        if hasattr(model.tasks, attr):
            return getattr(model.tasks, attr)
    raise AttributeError("在 model.tasks 中找不到网络仿真任务，请确认 sixgill 版本。")
# =========================================================================
# 主流程：参数化管径扫描
# =========================================================================
print(f"\n正在打开管网模型: {MODEL_PATH}")
model = Model.open(MODEL_PATH)
try:
    print(f"[INFO] 正在检索管道 '{TARGET_PIPE}' 的管段结构...")
    try:
        sections = model.get_sections(context=TARGET_PIPE)
    except Exception:
        sections = []
    sim_task = get_simulation_task(model)
    for d in TEST_DIAMETERS:
        print(f"\n[运行中] 管径 = {d} 英寸")
        try:
            set_pipe_diameter(model, sections, d)
            run_output = sim_task.run()
            branch_rows = collect_network_results(run_output, d)
            results.extend(branch_rows)
            print(f"   [OK] 收敛成功，汇总 {len(branch_rows)} 条支线数据。")
        except RuntimeError as e:
            print(f"   [ERROR] 管径赋值失败: {e}")
            results.append({"Diameter": d, "Status": f"Failed (赋值): {e}"})
        except Exception as e:
            print(f"   [ERROR] 仿真收敛失败: {e}")
            results.append({"Diameter": d, "Status": f"Failed (仿真): {e}"})
finally:
    print("\n正在关闭 PIPESIM 后台服务...")
    model.close()
# =========================================================================
# 数据整理与 Excel 导出
# =========================================================================
df       = pd.DataFrame(results)
df_clean = df[df["Status"] == "Success"].copy()
output_excel = os.path.join(WORKSPACE_DIR, f"{TARGET_PIPE}_Network_Pipeline_Validation.xlsx")
with pd.ExcelWriter(output_excel) as writer:
    df.to_excel(writer, sheet_name="RawResults", index=False)
    if not df_clean.empty:
        df_clean.pivot_table(
            index="Diameter", columns="Branch", values="PressureDrop", aggfunc="first"
        ).to_excel(writer, sheet_name="PressureDropPivot")
        df_clean.pivot_table(
            index="Diameter", columns="Branch", values="TemperatureDrop", aggfunc="first"
        ).to_excel(writer, sheet_name="TemperatureDropPivot")
print(f"\n[OK] Excel 报表已保存: {output_excel}")
# =========================================================================
# 绘图：各支线压降 / 温降随主干线管径变化曲线
# =========================================================================
if df_clean.empty:
    print("[WARN] 无成功收敛数据，跳过绘图。")
    sys.exit()
fig, (ax_p, ax_t) = plt.subplots(2, 1, figsize=(13, 10), dpi=150, sharex=True)
for branch, group in df_clean.groupby("Branch"):
    group = group.sort_values("Diameter")
    ax_p.plot(group["Diameter"], group["PressureDrop"],    marker="o", linewidth=1.6, label=branch)
    ax_t.plot(group["Diameter"], group["TemperatureDrop"], marker="s", linewidth=1.6, label=branch)
ax_p.set_ylabel("总压降 (MPa)", fontsize=11, fontweight="bold")
ax_p.set_title("集输管网各管线压降随外输干线管径变化图", fontsize=12, fontweight="bold")
ax_p.grid(True, linestyle="--", alpha=0.45)
ax_t.set_xlabel(f"{TARGET_PIPE} 管道设计内径 (inches)", fontsize=11, fontweight="bold")
ax_t.set_ylabel("总温降 (℃)", fontsize=11, fontweight="bold")
ax_t.set_title("集输管网各管线温降随外输干线管径变化图", fontsize=12, fontweight="bold")
ax_t.grid(True, linestyle="--", alpha=0.45)
handles, labels = ax_p.get_legend_handles_labels()
fig.legend(handles, labels, loc="center left", bbox_to_anchor=(1.0, 0.5), frameon=True, title="图例")
fig.tight_layout(rect=(0, 0, 0.82, 1))
output_img = os.path.join(WORKSPACE_DIR, f"{TARGET_PIPE}_network_sensitivity_curve.png")
plt.savefig(output_img, bbox_inches="tight")
print(f"[OK] 曲线图已生成: {output_img}")
plt.close(fig)