AutoControlCourse/case_demo_functions.py

import os
import sys
# ===== 新增：OpenMP冲突兼容设置，避免PyTorch初始化报错 =====
os.environ.setdefault("KMP_DUPLICATE_LIB_OK", "TRUE")
import numpy as np
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt

MODEL_SRC_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "Model", "src")
if MODEL_SRC_PATH not in sys.path:
    # ===== 新增：将混动模型源码路径加入导入搜索路径 =====
    sys.path.insert(0, MODEL_SRC_PATH)


def _profile_points(profile_name):
    if profile_name == "高机动阶跃":
        return [
            (0.0, 1600.0, 70.0),
            (8.0, 3200.0, 220.0),
            (20.0, 2500.0, 130.0),
            (35.0, 3400.0, 250.0),
            (50.0, 1800.0, 80.0),
        ]
    if profile_name == "经济巡航":
        return [
            (0.0, 1500.0, 60.0),
            (15.0, 2100.0, 95.0),
            (35.0, 2300.0, 105.0),
            (55.0, 2000.0, 90.0),
        ]
    return [
        (0.0, 1500.0, 50.0),
        (10.0, 3000.0, 200.0),
        (30.0, 2800.0, 150.0),
        (50.0, 1800.0, 60.0),
    ]


def _target_from_profile(t, points, rpm_scale, load_scale):
    rpm, torque = points[0][1], points[0][2]
    for p_t, p_rpm, p_torque in points:
        if t >= p_t:
            rpm, torque = p_rpm, p_torque
        else:
            break
    return rpm * rpm_scale, torque * load_scale


def run_case_demo(sim_time_s, dt, initial_soc_pct, initial_engine_power_kw, profile_name, rpm_scale, load_scale):
    try:
        try:
            # ===== 新增：懒加载混动系统模型，便于捕获缺失依赖 =====
            from series_hybrid_sim import SeriesHybridSystem
        except ModuleNotFoundError as e:
            if getattr(e, "name", "") == "torch":
                return None, "算例仿真失败：缺少依赖 torch，请先在当前环境安装 PyTorch。", []
            return None, f"算例仿真失败：缺少依赖 {e.name}。", []

        sim_time_s = float(np.clip(sim_time_s, 10.0, 240.0))
        dt = float(np.clip(dt, 0.01, 0.2))
        initial_soc_pct = float(np.clip(initial_soc_pct, 10.0, 95.0))
        initial_engine_power_kw = float(np.clip(initial_engine_power_kw, 20.0, 260.0))
        rpm_scale = float(np.clip(rpm_scale, 0.5, 1.6))
        load_scale = float(np.clip(load_scale, 0.5, 1.6))

        points = _profile_points(profile_name)
        system = SeriesHybridSystem()
        system.battery.SOC = initial_soc_pct / 100.0
        system.bus_voltage = system.battery._get_ocv(system.battery.SOC)
        system.genset.set_steady_state_by_power(H_env=0.0, Ma_env=0.0, Power_target=initial_engine_power_kw)

        time_array = np.arange(0.0, sim_time_s, dt)
        log = {k: [] for k in [
            "soc", "bus_voltage", "prop_speed_rpm", "target_prop_rpm",
            "target_engine_pwr", "p_engine_out_kw", "p_drive_req_kw",
            "p_batt_actual_kw", "wf_kg_h"
        ]}

        for t in time_array:
            target_rpm, load_torque = _target_from_profile(t, points, rpm_scale, load_scale)
            res = system.step(dt, target_rpm, load_torque)
            res["target_prop_rpm"] = target_rpm
            for k in log:
                log[k].append(res[k])

        speed_error = np.array(log["target_prop_rpm"]) - np.array(log["prop_speed_rpm"])
        soc_arr = np.array(log["soc"])
        fuel_arr = np.array(log["wf_kg_h"])
        engine_pwr_arr = np.array(log["p_engine_out_kw"])
        batt_pwr_arr = np.array(log["p_batt_actual_kw"])

        fig, axes = plt.subplots(3, 1, figsize=(12, 10), sharex=True)
        axes[0].plot(time_array, log["target_prop_rpm"], "k--", lw=1.5, label="目标转速")
        axes[0].plot(time_array, log["prop_speed_rpm"], "b-", lw=1.5, label="实际转速")
        axes[0].set_ylabel("RPM")
        axes[0].set_title("推进轴转速响应")
        axes[0].grid(True, linestyle=":")
        axes[0].legend()

        axes[1].plot(time_array, log["p_drive_req_kw"], "k--", lw=1.2, label="电机需求")
        axes[1].plot(time_array, log["p_engine_out_kw"], "r-", lw=1.2, label="发动机输出")
        axes[1].plot(time_array, log["p_batt_actual_kw"], "g-", lw=1.2, label="电池功率")
        axes[1].axhline(0, color="gray", lw=1)
        axes[1].set_ylabel("kW")
        axes[1].set_title("功率分配")
        axes[1].grid(True, linestyle=":")
        axes[1].legend()

        axes[2].plot(time_array, log["bus_voltage"], "m-", lw=1.2, label="母线电压")
        axes[2].set_ylabel("V")
        axes[2].set_xlabel("时间 (s)")
        axes[2].set_title("电气状态")
        axes[2].grid(True, linestyle=":")
        ax_soc = axes[2].twinx()
        ax_soc.plot(time_array, log["soc"], "c--", lw=1.6, label="SOC")
        ax_soc.set_ylabel("SOC (%)")

        fig.tight_layout()

        summary = (
            f"### 算例结果解读\n"
            f"- 仿真时长：{sim_time_s:.1f} s，步长：{dt:.3f} s\n"
            f"- 最大转速误差：{np.max(np.abs(speed_error)):.1f} RPM\n"
            f"- SOC 变化：{soc_arr[0]:.2f}% → {soc_arr[-1]:.2f}%（最小 {np.min(soc_arr):.2f}%）\n"
            f"- 平均发动机输出：{np.mean(engine_pwr_arr):.2f} kW\n"
            f"- 平均电池功率：{np.mean(batt_pwr_arr):.2f} kW\n"
            f"- 平均燃油流量：{np.mean(fuel_arr):.2f} kg/h"
        )

        pick_idx = np.linspace(0, len(time_array) - 1, 8, dtype=int)
        table_data = []
        for idx in pick_idx:
            table_data.append([
                round(float(time_array[idx]), 2),
                round(float(log["target_prop_rpm"][idx]), 1),
                round(float(log["prop_speed_rpm"][idx]), 1),
                round(float(log["p_engine_out_kw"][idx]), 2),
                round(float(log["p_batt_actual_kw"][idx]), 2),
                round(float(log["soc"][idx]), 2),
            ])

        return fig, summary, table_data
    except Exception as e:
        return None, f"算例仿真失败：{e}", []