vent_deep_agent/tools/calc_tools.py

# -*- coding: utf-8 -*-
"""
煤矿配风计划 - 需风量计算工具模块

基于《煤矿安全规程》2025 版和 AQ 1056-2008《煤矿通风能力核定标准》，
实现各用风地点类型的需风量计算工具。

每个工具函数返回结构化 JSON，包含：
- formula: 计算公式（文字描述）
- expression: 计算表达式（含数值代入）
- steps: 计算步骤列表
- result: 计算结果
- unit: 单位

工具列表:
- 采煤工作面: calc_face_by_gas / calc_face_by_workers / calc_face_by_wind_speed / calc_face_air_volume_max
- 掘进工作面: calc_tunnel_by_gas / calc_tunnel_by_explosives / calc_tunnel_by_workers / calc_tunnel_by_wind_speed / calc_tunnel_by_vehicle / calc_tunnel_air_volume_max
- 硐室: calc_chamber_by_equipment / calc_chamber_by_wind_speed
- 其他巷道: calc_other_by_wind_speed
- 辅助: calc_effective_area / calc_total_air_volume
- 反算: calc_gas_emission_from_wind（从测风报表风量×浓度反算涌出量）
"""

import json


# ============================================================
# 辅助工具
# ============================================================

def calc_effective_area(max_control_distance: float,
                         min_control_distance: float,
                         avg_mining_height: float) -> str:
    """计算工作面有效断面积。

    公式: S = (L_max + L_min) / 2 × H
    其中 L_max 为最大控顶距，L_min 为最小控顶距，H 为平均采高。

    Args:
        max_control_distance: 最大控顶距，单位 m
        min_control_distance: 最小控顶距，单位 m
        avg_mining_height: 平均采高，单位 m

    Returns:
        JSON格式的计算过程与结果
    """
    avg_distance = (max_control_distance + min_control_distance) / 2
    area = avg_distance * avg_mining_height

    return json.dumps({
        "formula": "S = (L_max + L_min) / 2 × H",
        "expression": f"S = ({max_control_distance} + {min_control_distance}) / 2 × {avg_mining_height}",
        "steps": [
            f"平均控顶距 = ({max_control_distance} + {min_control_distance}) / 2 = {avg_distance:.2f} m",
            f"有效断面积 = {avg_distance:.2f} × {avg_mining_height} = {area:.2f} m²",
        ],
        "result": round(area, 2),
        "unit": "m²",
    }, ensure_ascii=False)


# ============================================================
# 采煤工作面需风量计算
# ============================================================

def calc_face_by_gas(q_gas: float, k_gas: float = 1.5) -> str:
    """按瓦斯涌出量计算采煤工作面需风量。

    公式: Q = 100 × q_gas × K_gas
    依据: 按瓦斯涌出量计算，K_gas 为瓦斯涌出不均衡系数（1.2~1.6）

    Args:
        q_gas: 平均绝对瓦斯涌出量，单位 m³/min
        k_gas: 瓦斯涌出不均衡系数，默认 1.5

    Returns:
        JSON格式的计算过程与结果
    """
    result = 100 * q_gas * k_gas

    return json.dumps({
        "formula": "Q_cf = 100 × q_gas × K_gas",
        "expression": f"Q_cf = 100 × {q_gas} × {k_gas}",
        "steps": [
            f"Q_cf = 100 × {q_gas} × {k_gas}",
            f"Q_cf = {100 * q_gas:.2f} × {k_gas}",
            f"Q_cf = {result:.2f} m³/min",
        ],
        "result": round(result, 2),
        "unit": "m³/min",
    }, ensure_ascii=False)


def calc_face_by_workers(n: int) -> str:
    """按同时工作最多人数计算采煤工作面需风量。

    公式: Q = 4 × N
    依据: 每人供风量不小于 4 m³/min

    Args:
        n: 同时工作最多人数

    Returns:
        JSON格式的计算过程与结果
    """
    result = 4 * n

    return json.dumps({
        "formula": "Q_cf = 4 × N",
        "expression": f"Q_cf = 4 × {n}",
        "steps": [
            f"Q_cf = 4 × {n}",
            f"Q_cf = {result:.2f} m³/min",
        ],
        "result": round(result, 2),
        "unit": "m³/min",
    }, ensure_ascii=False)


def calc_face_by_wind_speed(v: float, s: float) -> str:
    """按风速验算采煤工作面需风量。

    公式: Q = 60 × S × v
    验算条件: 0.25 ≤ v ≤ 4.0 m/s（采煤工作面最低/最高风速）

    Args:
        v: 设计风速，单位 m/s
        s: 有效断面积，单位 m²（可使用 calc_effective_area 计算）

    Returns:
        JSON格式的计算过程与结果
    """
    result = 60 * s * v

    return json.dumps({
        "formula": "Q_cf = 60 × S × v",
        "expression": f"Q_cf = 60 × {s} × {v}",
        "steps": [
            f"Q_cf = 60 × {s} × {v}",
            f"Q_cf = {60 * s:.2f} × {v}",
            f"Q_cf = {result:.2f} m³/min",
        ],
        "result": round(result, 2),
        "unit": "m³/min",
        "wind_speed_check": {
            "v_min": 0.25,
            "v_max": 4.0,
            "passed": 0.25 <= v <= 4.0,
        },
    }, ensure_ascii=False)


def calc_face_air_volume_max(q_gas: float = 0,
                               q_co2: float = 0,
                               k_gas: float = 1.5,
                               n_workers: int = 0,
                               v_wind: float = 0,
                               s_area: float = 0) -> str:
    """采煤工作面需风量综合计算 —— 取各方法最大值。

    按瓦斯、二氧化碳、人数、风速分别计算，取最大值作为最终需风量。
    同时验算风速是否在 0.25~4.0 m/s 范围内。

    Args:
        q_gas: 平均绝对瓦斯涌出量，m³/min
        q_co2: 平均绝对二氧化碳涌出量，m³/min
        k_gas: 瓦斯涌出不均衡系数，默认 1.5
        n_workers: 同时工作最多人数
        v_wind: 设计风速，m/s
        s_area: 有效断面积，m²

    Returns:
        JSON格式的综合计算过程与结果
    """
    results = {}

    if q_gas > 0:
        results["按瓦斯涌出量"] = 100 * q_gas * k_gas

    if q_co2 > 0:
        # 二氧化碳涌出量：按 67 × q_co2 × K
        results["按二氧化碳涌出量"] = 67 * q_co2 * k_gas

    if n_workers > 0:
        results["按人数"] = 4 * n_workers

    if v_wind > 0 and s_area > 0:
        q_wind = 60 * s_area * v_wind
        results["按风速"] = q_wind
        v_check = 0.25 <= v_wind <= 4.0
    else:
        q_wind = 0
        v_check = None

    if not results:
        return json.dumps({
            "error": "请至少提供一种计算参数（q_gas、q_co2、n_workers、v_wind + s_area）",
        }, ensure_ascii=False)

    max_method = max(results, key=results.get)
    max_value = results[max_method]

    steps = []
    for method, val in results.items():
        steps.append(f"{method}: {val:.2f} m³/min")
    steps.append(f"取最大值: {max_method} = {max_value:.2f} m³/min")

    return json.dumps({
        "formula": "Q_cf = max(Q_gas, Q_co2, Q_workers, Q_wind)",
        "methods": {k: round(v, 2) for k, v in results.items()},
        "steps": steps,
        "result": round(max_value, 2),
        "max_method": max_method,
        "unit": "m³/min",
        "wind_speed_check": {
            "v_min": 0.25,
            "v_max": 4.0,
            "passed": v_check,
        } if v_check is not None else None,
    }, ensure_ascii=False)


# ============================================================
# 掘进工作面需风量计算
# ============================================================

def calc_tunnel_by_gas(q_gas: float, k_gas: float = 1.8) -> str:
    """按瓦斯涌出量计算掘进工作面需风量。

    公式: Q = 100 × q_gas × K_gas
    掘进工作面瓦斯不均衡系数通常取 1.8~2.0

    Args:
        q_gas: 平均绝对瓦斯涌出量，单位 m³/min
        k_gas: 瓦斯涌出不均衡系数，默认 1.8

    Returns:
        JSON格式的计算过程与结果
    """
    result = 100 * q_gas * k_gas

    return json.dumps({
        "formula": "Q_hf = 100 × q_gas × K_gas",
        "expression": f"Q_hf = 100 × {q_gas} × {k_gas}",
        "steps": [
            f"Q_hf = 100 × {q_gas} × {k_gas}",
            f"Q_hf = {100 * q_gas:.2f} × {k_gas}",
            f"Q_hf = {result:.2f} m³/min",
        ],
        "result": round(result, 2),
        "unit": "m³/min",
    }, ensure_ascii=False)


def calc_tunnel_by_explosives(a: float) -> str:
    """按炸药量计算掘进工作面需风量。

    公式: Q = 25 × A
    其中 A 为一次爆破最大炸药用量，kg

    Args:
        a: 一次爆破最大炸药用量，单位 kg

    Returns:
        JSON格式的计算过程与结果
    """
    result = 25 * a

    return json.dumps({
        "formula": "Q_hf = 25 × A",
        "expression": f"Q_hf = 25 × {a}",
        "steps": [
            f"Q_hf = 25 × {a}",
            f"Q_hf = {result:.2f} m³/min",
        ],
        "result": round(result, 2),
        "unit": "m³/min",
    }, ensure_ascii=False)


def calc_tunnel_by_workers(n: int) -> str:
    """按同时工作最多人数计算掘进工作面需风量。

    公式: Q = 4 × N

    Args:
        n: 同时工作最多人数

    Returns:
        JSON格式的计算过程与结果
    """
    result = 4 * n

    return json.dumps({
        "formula": "Q_hf = 4 × N",
        "expression": f"Q_hf = 4 × {n}",
        "steps": [
            f"Q_hf = 4 × {n}",
            f"Q_hf = {result:.2f} m³/min",
        ],
        "result": round(result, 2),
        "unit": "m³/min",
    }, ensure_ascii=False)


def calc_tunnel_by_wind_speed(v: float, s: float) -> str:
    """按风速验算掘进工作面需风量。

    公式: Q = 60 × S × v
    验算条件: 0.25 ≤ v ≤ 4.0 m/s（掘进工作面最低/最高风速）
    注意：岩巷最低风速为 0.15 m/s

    Args:
        v: 设计风速，单位 m/s
        s: 有效断面积，单位 m²

    Returns:
        JSON格式的计算过程与结果
    """
    result = 60 * s * v

    return json.dumps({
        "formula": "Q_hf = 60 × S × v",
        "expression": f"Q_hf = 60 × {s} × {v}",
        "steps": [
            f"Q_hf = 60 × {s} × {v}",
            f"Q_hf = {60 * s:.2f} × {v}",
            f"Q_hf = {result:.2f} m³/min",
        ],
        "result": round(result, 2),
        "unit": "m³/min",
        "wind_speed_check": {
            "v_min": 0.25,
            "v_max": 4.0,
            "note": "岩巷最低风速为 0.15 m/s",
            "passed": 0.25 <= v <= 4.0,
        },
    }, ensure_ascii=False)


def calc_tunnel_by_vehicle(total_power: float,
                            k_vehicle: float = 0.8,
                            vehicle_count: int = 1) -> str:
    """按防爆柴油机车（胶轮车）功率计算掘进工作面需风量。

    公式: Q = 4 × ΣP × K
    依据: 《煤矿安全规程》—— 使用防爆柴油机车（无轨胶轮车）的掘进工作面，
    需风量按同时运行车辆总功率计算，每 kW 供风量不小于 4 m³/min。
    K 为车辆同时运行系数。

    Args:
        total_power: 防爆柴油机车总功率（同时运行），单位 kW
        k_vehicle: 车辆同时运行系数，默认 0.8（0.5~1.0）
        vehicle_count: 同时运行车辆数，默认 1

    Returns:
        JSON格式的计算过程与结果
    """
    result = 4 * total_power * k_vehicle

    return json.dumps({
        "formula": "Q_hf = 4 × ΣP × K",
        "expression": f"Q_hf = 4 × {total_power} × {k_vehicle}",
        "steps": [
            f"车辆总功率 ΣP = {total_power} kW",
            f"同时运行系数 K = {k_vehicle}（{vehicle_count} 辆车）",
            f"Q_hf = 4 × {total_power} × {k_vehicle}",
            f"Q_hf = {4 * total_power:.2f} × {k_vehicle}",
            f"Q_hf = {result:.2f} m³/min",
        ],
        "result": round(result, 2),
        "unit": "m³/min",
        "params": {
            "total_power_kw": total_power,
            "k_vehicle": k_vehicle,
            "vehicle_count": vehicle_count,
        },
        "note": "每 kW 柴油机功率供风量 ≥ 4 m³/min（《煤矿安全规程》）",
    }, ensure_ascii=False)


def calc_tunnel_air_volume_max(q_gas: float = 0,
                                q_co2: float = 0,
                                k_gas: float = 1.8,
                                a_explosives: float = 0,
                                n_workers: int = 0,
                                v_wind: float = 0,
                                s_area: float = 0,
                                p_vehicle: float = 0,
                                k_vehicle: float = 0.8) -> str:
    """掘进工作面需风量综合计算 —— 取各方法最大值。

    按瓦斯、二氧化碳、炸药量、人数、风速、胶轮车分别计算，取最大值。
    同时验算风速是否在 0.25~4.0 m/s 范围内。

    Args:
        q_gas: 平均绝对瓦斯涌出量，m³/min
        q_co2: 平均绝对二氧化碳涌出量，m³/min
        k_gas: 瓦斯涌出不均衡系数，默认 1.8
        a_explosives: 一次爆破最大炸药用量，kg
        n_workers: 同时工作最多人数
        v_wind: 设计风速，m/s
        s_area: 有效断面积，m²
        p_vehicle: 防爆柴油机车（胶轮车）总功率，kW
        k_vehicle: 车辆同时运行系数，默认 0.8

    Returns:
        JSON格式的综合计算过程与结果
    """
    results = {}

    if q_gas > 0:
        results["按瓦斯涌出量"] = 100 * q_gas * k_gas

    if q_co2 > 0:
        results["按二氧化碳涌出量"] = 67 * q_co2 * k_gas

    if a_explosives > 0:
        results["按炸药量"] = 25 * a_explosives

    if n_workers > 0:
        results["按人数"] = 4 * n_workers

    if v_wind > 0 and s_area > 0:
        results["按风速"] = 60 * s_area * v_wind
        v_check = 0.25 <= v_wind <= 4.0
    else:
        v_check = None

    if p_vehicle > 0:
        results["按胶轮车"] = 4 * p_vehicle * k_vehicle

    if not results:
        return json.dumps({
            "error": "请至少提供一种计算参数",
        }, ensure_ascii=False)

    max_method = max(results, key=results.get)
    max_value = results[max_method]

    steps = []
    for method, val in results.items():
        steps.append(f"{method}: {val:.2f} m³/min")
    steps.append(f"取最大值: {max_method} = {max_value:.2f} m³/min")

    return json.dumps({
        "formula": "Q_hf = max(Q_gas, Q_co2, Q_explosives, Q_workers, Q_wind, Q_vehicle)",
        "methods": {k: round(v, 2) for k, v in results.items()},
        "steps": steps,
        "result": round(max_value, 2),
        "max_method": max_method,
        "unit": "m³/min",
        "wind_speed_check": {
            "v_min": 0.25,
            "v_max": 4.0,
            "passed": v_check,
        } if v_check is not None else None,
    }, ensure_ascii=False)


# ============================================================
# 硐室需风量计算
# ============================================================

def calc_chamber_by_equipment(total_power: float,
                               efficiency: float = 0.85,
                               temp_rise: float = 10.0) -> str:
    """按机电设备发热量计算硐室需风量。

    公式: Q = 3600 × θ × ΣN / (ρ × C_p × Δt)
    简化: Q ≈ 0.1 × ΣN  (常规硐室估算)
    或精确: Q = ΣN × (1 - η) / (C_p × ρ × Δt) × 60

    此处使用机电设备功率简化计算。

    Args:
        total_power: 硐室内机电设备总功率，kW
        efficiency: 设备平均效率，默认 0.85
        temp_rise: 允许温升，℃，默认 10

    Returns:
        JSON格式的计算过程与结果
    """
    # 发热量: θ = (1 - η) × ΣN
    heat = (1 - efficiency) * total_power  # kW 发热量
    # Q = 3600 × θ / (ρ × C_p × Δt)  → 空气密度 1.2 kg/m³, 比热 1.005 kJ/(kg·℃)
    result = 3600 * heat / (1.2 * 1.005 * temp_rise)

    return json.dumps({
        "formula": "Q = 3600 × (1 - η) × ΣN / (ρ × C_p × Δt)",
        "expression": f"Q = 3600 × (1 - {efficiency}) × {total_power} / (1.2 × 1.005 × {temp_rise})",
        "steps": [
            f"发热量 θ = (1 - {efficiency}) × {total_power} = {heat:.2f} kW",
            f"Q = 3600 × {heat:.2f} / (1.2 × 1.005 × {temp_rise})",
            f"Q = {3600 * heat:.2f} / {1.2 * 1.005 * temp_rise:.2f}",
            f"Q = {result:.2f} m³/min",
        ],
        "result": round(result, 2),
        "unit": "m³/min",
    }, ensure_ascii=False)


def calc_chamber_by_wind_speed(v: float, s: float) -> str:
    """按风速验算硐室需风量。

    依据: 机电硐室最低风速 0.15 m/s，最高 6.0 m/s（无瓦斯涌出）

    Args:
        v: 设计风速，m/s
        s: 有效断面积，m²

    Returns:
        JSON格式的计算过程与结果
    """
    result = 60 * s * v

    return json.dumps({
        "formula": "Q = 60 × S × v",
        "expression": f"Q = 60 × {s} × {v}",
        "steps": [
            f"Q = 60 × {s} × {v}",
            f"Q = {60 * s:.2f} × {v}",
            f"Q = {result:.2f} m³/min",
        ],
        "result": round(result, 2),
        "unit": "m³/min",
        "wind_speed_check": {
            "v_min": 0.15,
            "v_max": 6.0,
            "passed": 0.15 <= v <= 6.0,
        },
    }, ensure_ascii=False)


# ============================================================
# 其他巷道需风量计算
# ============================================================

def calc_other_by_wind_speed(v: float, s: float, tunnel_type: str = "other") -> str:
    """按风速计算其他巷道需风量。

    公式: Q = 60 × S × v
    风速限值因巷道类型不同：
    - 主要进回风巷: v ≤ 8 m/s
    - 其他通风人行巷道: v ≥ 0.15 m/s
    - 运输机巷、采区进回风巷: 0.25 ≤ v ≤ 6 m/s

    Args:
        v: 设计风速，m/s
        s: 有效断面积，m²
        tunnel_type: 巷道类型 (main_in: 主要进风, main_out: 主要回风, mining: 采区, other: 其他)

    Returns:
        JSON格式的计算过程与结果
    """
    result = 60 * s * v

    # 各类型风速限值
    limits = {
        "main_in": {"v_min": None, "v_max": 8.0, "desc": "主要进风巷"},
        "main_out": {"v_min": None, "v_max": 8.0, "desc": "主要回风巷"},
        "mining": {"v_min": 0.25, "v_max": 6.0, "desc": "采区进回风巷"},
        "other": {"v_min": 0.15, "v_max": None, "desc": "其他通风人行巷道"},
    }
    limit = limits.get(tunnel_type, limits["other"])

    return json.dumps({
        "formula": "Q = 60 × S × v",
        "expression": f"Q = 60 × {s} × {v}",
        "steps": [
            f"Q = 60 × {s} × {v}",
            f"Q = {60 * s:.2f} × {v}",
            f"Q = {result:.2f} m³/min",
        ],
        "result": round(result, 2),
        "unit": "m³/min",
        "tunnel_type": limit["desc"],
        "wind_speed_limits": {
            "v_min": limit["v_min"],
            "v_max": limit["v_max"],
        },
    }, ensure_ascii=False)


# ============================================================
# 瓦斯/CO2 涌出量反算（从测风报表数据推算）
# ============================================================

def calc_gas_emission_from_wind(wind_volume: float,
                                 concentration: float,
                                 gas_type: str = "CH4") -> str:
    """根据回风顺槽实测风量和浓度反算绝对瓦斯/CO2涌出量。

    公式: q = Q × C / 100
    其中 Q 为回风顺槽实测风量（m³/min），C 为瓦斯或CO2浓度（%）。

    用于数据一致性审查：将测风报表中反算的涌出量与配风计划中的数值对比。

    Args:
        wind_volume: 回风顺槽实测风量，单位 m³/min
        concentration: 瓦斯或CO2浓度，单位 %（如 0.3 表示 0.3%）
        gas_type: 气体类型，"CH4"（瓦斯）或 "CO2"（二氧化碳）

    Returns:
        JSON格式的计算过程与结果
    """
    result = wind_volume * concentration / 100.0

    gas_label = "绝对瓦斯涌出量" if gas_type.upper() == "CH4" else "绝对二氧化碳涌出量"
    gas_symbol = "q_gas" if gas_type.upper() == "CH4" else "q_co2"

    return json.dumps({
        "formula": f"{gas_symbol} = Q × C / 100",
        "expression": f"{gas_symbol} = {wind_volume} × {concentration} / 100",
        "steps": [
            f"回风顺槽实测风量 Q = {wind_volume} m³/min",
            f"{'瓦斯' if gas_type.upper() == 'CH4' else 'CO₂'}浓度 C = {concentration}%",
            f"{gas_label} = {wind_volume} × {concentration} / 100",
            f"{gas_label} = {wind_volume * concentration:.2f} / 100",
            f"{gas_label} = {result:.4f} m³/min",
        ],
        "result": round(result, 4),
        "unit": "m³/min",
        "gas_type": gas_type.upper(),
        "gas_label": gas_label,
        "source": "测风报表反算",
    }, ensure_ascii=False)


# ============================================================
# 汇总工具
# ============================================================

def calc_total_air_volume(air_volumes: list[dict]) -> str:
    """汇总各用风地点需风量，计算矿井总需风量。

    将各用风地点的需风量求和，得到矿井总需风量。

    Args:
        air_volumes: 各用风地点需风量列表，格式 [{name: str, q: float}, ...]

    Returns:
        JSON格式的汇总结果
    """
    if not air_volumes:
        return json.dumps({"error": "请提供各用风地点需风量数据"}, ensure_ascii=False)

    total = sum(item.get("q", 0) for item in air_volumes)
    details = [f"{item.get('name', '未知')}: {item.get('q', 0):.2f} m³/min" for item in air_volumes]

    return json.dumps({
        "formula": "Q_total = ΣQ_i",
        "steps": details + [f"总需风量 = {total:.2f} m³/min"],
        "result": round(total, 2),
        "location_count": len(air_volumes),
        "unit": "m³/min",
    }, ensure_ascii=False)