vent_agent_2.0/algorithms/airflow_review/engine/recalc.py

"""需风量复算：按「用风地点 → 回风系统 → 矿井总需风量」自下而上逐层计算。

两种核验同时进行，二者的失败含义不同：

1. 独立复算（standard）：用规则库公式 + 计划给定参数独立算一遍，与计划值比对。
   差异说明公式适用性、参数取值或口径有问题。
2. 自洽核验（stated）：用计划自己写出的代入数值，按规则库公式重算，
   与计划写出的结果比对。差异说明计划本身算错了。

两种结果都写进计算链，报告里逐项展示。
"""
from __future__ import annotations

from typing import Any, Mapping

from algorithms.airflow_review.contracts import (
    LOC_CHAMBER,
    as_mapping,
    LOC_OTHER,
    LOC_STANDBY,
    LOC_WORKPLACE_COAL,
    LOC_WORKPLACE_HEADING,
    PLAN_TARGET_TYPES,
    TARGET_MINE_INTAKE,
    TARGET_MINE_RETURN,
    TARGET_RETURN_SYSTEM,
    unique,
)
from algorithms.airflow_review.engine.expression import try_evaluate
from algorithms.airflow_review.engine.variants import (
    STATUS_INCONSISTENT,
    _infer_variant,
    _reproduce_stated,
)
from algorithms.airflow_review.rules.loader import RuleSet, derive_params

STATUS_OK = "ok"
STATUS_INSUFFICIENT = "insufficient"
STATUS_NOT_APPLICABLE = "not_applicable"

# 复算顺序：自下而上（§2.6 第 4 条）
_ORDER = {LOC_WORKPLACE_COAL: 0, LOC_WORKPLACE_HEADING: 0, LOC_STANDBY: 0,
          LOC_CHAMBER: 0, LOC_OTHER: 0, TARGET_RETURN_SYSTEM: 1, "mine_total": 2}


def recalc_plan_item(
    item: Mapping[str, Any],
    rules: RuleSet,
    resolved: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """复算单个计划对象。resolved 为子项汇总所需的两路需风量（见 _resolve_components）。"""
    target_type = str(item.get("target_type") or "")
    location_type = item.get("location_type") or _location_type_for(target_type, item)
    facts = as_mapping(item.get("planned_parameters"))
    facts.setdefault("target_id", item.get("target_id"))
    facts.setdefault("location_type", location_type)
    if item.get("chamber_type"):
        facts.setdefault("chamber_type", item["chamber_type"])
    if item.get("roadway_type"):
        facts.setdefault("roadway_type", item["roadway_type"])
    if item.get("mining_method"):
        facts.setdefault("mining_method", item["mining_method"])

    result: dict[str, Any] = {
        "target_id": item.get("target_id"),
        "target_type": target_type,
        "location_type": location_type,
        "target_name": item.get("target_name") or item.get("target_id"),
        "planned_required_airflow_m3_min": item.get("planned_required_airflow_m3_min"),
        "aggregation_basis": "planned_components" if target_type in (TARGET_RETURN_SYSTEM, "mine_total") else "",
        "methods": [],
        "checks": [],
        "warnings": [],
        "evidence_ids": unique(list(item.get("evidence_ids") or []) + [item.get("evidence_id")]),
    }

    if location_type == LOC_STANDBY:
        result["methods"].append(_standby_method(item, rules, resolved))
    elif target_type in (TARGET_RETURN_SYSTEM, "mine_total"):
        result["methods"].append(_aggregate_method(item, rules, target_type, resolved))
    else:
        stated_result = (item.get("stated_calculation") or {}).get("stated_result_m3_min")
        stated_result = stated_result if isinstance(stated_result, (int, float)) else None
        for formula in rules.formulas(location_type):
            result["methods"].append(_evaluate_formula(formula, facts, rules, stated_result))
        inferred = [method["method_id"] for method in result["methods"]
                    if method.get("variant_selection") == "inferred_from_stated_calculation"]
        if inferred:
            result["warnings"].append(
                "参数口径未标注或无法识别，以下计算方法的公式变体是按计划写出的结果反推的，"
                "建议核实计划采用的参数口径：" + "、".join(inferred))

    candidates = [method["value_m3_min"] for method in result["methods"]
                  if method["status"] == STATUS_OK and method.get("value_m3_min") is not None]
    if candidates:
        result["calculated_required_airflow_m3_min"] = round(max(candidates), 4)
        result["calculation_status"] = STATUS_OK
        result["governing_method_id"] = next(
            method["method_id"] for method in result["methods"]
            if method.get("value_m3_min") == max(candidates))
    else:
        result["calculated_required_airflow_m3_min"] = None
        result["calculation_status"] = STATUS_INSUFFICIENT
        result["governing_method_id"] = None
        result["warnings"].append("所有计算方法均缺少参数，无法复算需风量")

    result["checks"] = _run_checks(item, facts, rules, location_type, result)
    result["min_airflow_m3_min"], result["max_airflow_m3_min"] = _bounds(result)
    result["stated_reproduction"] = _reproduce_stated(item, facts, rules, resolved)
    return result


def recalc_all(items: list[Mapping[str, Any]], rules: RuleSet) -> dict[str, dict[str, Any]]:
    """按层级顺序复算全部对象。

    父级汇总以子项的**计划值**为基准：父级的正确性 = 它的声称值是否等于其子项
    声称值的汇总（这正是"汇总错误/漏算/重复计算"的检验）。子项各自的复算值另外
    汇总一路，用于观察误差是否向上传播。
    """
    ordered = sorted(items, key=lambda item: (
        _ORDER.get(str(item.get("target_type")), 9),
        _ORDER.get(str(item.get("location_type") or ""), 0),
        str(item.get("target_id")),
    ))
    results: dict[str, dict[str, Any]] = {}
    for item in ordered:
        resolved = _resolve_components(item, items, results)
        results[str(item.get("target_id"))] = recalc_plan_item(item, rules, resolved)
    return results


def _resolve_components(
    item: Mapping[str, Any],
    items: list[Mapping[str, Any]],
    results: dict[str, dict[str, Any]],
) -> dict[str, Any]:
    """收集组成对象的两路需风量：计划值（汇总核验基准）与复算值（传播观察）。"""
    if str(item.get("target_type")) not in (TARGET_RETURN_SYSTEM, "mine_total"):
        return {}
    by_id = {str(entry.get("target_id")): entry for entry in items}
    child_ids = list(item.get("component_workplace_ids") or item.get("component_return_system_ids")
                     or item.get("component_target_ids") or [])
    planned: dict[str, float] = {}
    recalculated: dict[str, float] = {}
    missing_planned: list[str] = []
    for child_id in child_ids:
        child_id = str(child_id)
        child = by_id.get(child_id) or {}
        value = child.get("planned_required_airflow_m3_min")
        if isinstance(value, (int, float)):
            planned[child_id] = float(value)
        else:
            missing_planned.append(child_id)
        child_result = results.get(child_id)
        if child_result and child_result.get("calculated_required_airflow_m3_min") is not None:
            recalculated[child_id] = child_result["calculated_required_airflow_m3_min"]
    return {
        "planned": planned,
        "recalculated": recalculated,
        "recalculated_complete": len(recalculated) == len([cid for cid in child_ids if cid]),
        "missing_planned": missing_planned,
        "child_count": len(child_ids),
    }


def _location_type_for(target_type: str, item: Mapping[str, Any]) -> str:
    if target_type in (TARGET_MINE_INTAKE, TARGET_MINE_RETURN, "mine_total"):
        return "矿井"
    if target_type == TARGET_RETURN_SYSTEM:
        return TARGET_RETURN_SYSTEM
    return str(item.get("location_type") or "")


def _evaluate_formula(
    formula: Mapping[str, Any],
    facts: Mapping[str, Any],
    rules: RuleSet,
    stated_result: float | None = None,
) -> dict[str, Any]:
    """按公式（含变体）计算，并记录变体是怎么选出来的。

    口径标记（如 section_area_basis）写错或没写时，如果直接退回基础表达式，会把一份
    正确的计划算成"与计划不符"从而整盘阻断。因此这里做一次**确定性反推**：拿计划自己
    写出的结果去对每个候选表达式，命中哪个就采用哪个，并把"这个变体是反推来的"记在
    计算链里、同时给出提示，让人能复核。计划没写结果时才退回基础表达式。
    """
    selected = rules.select_formula(dict(formula), facts)
    method: dict[str, Any] = {
        "method_id": formula.get("id"),
        "name": formula.get("name"),
        "citation": formula.get("citation"),
        "aggregation": formula.get("aggregation"),
        "expression": selected.get("expression"),
        "variant": selected.get("variant"),
        "variant_selection": "declared" if selected.get("variant") else "default",
        "note": selected.get("note", ""),
    }
    if selected.get("expression") is None:
        method.update({"status": STATUS_NOT_APPLICABLE, "value_m3_min": None,
                       "reason": selected.get("reason", "公式不适用")})
        return method
    params, missing, derived = derive_params(dict(formula), facts, rules)
    method["params"] = {name: params.get(name) for name in (formula.get("params") or {})}
    method["derived"] = derived
    if missing:
        method.update({"status": STATUS_INSUFFICIENT, "value_m3_min": None,
                       "missing_params": missing,
                       "reason": "缺少参数：" + "、".join(missing)})
        return method

    if not selected.get("variant"):
        inferred = _infer_variant(formula, params, stated_result)
        if inferred is not None:
            method.update({"expression": inferred["expression"], "variant": inferred["variant"],
                           "variant_selection": "inferred_from_stated_calculation",
                           "variant_candidates": inferred["candidates"],
                           "note": ("参数口径未标注或无法识别，已按计划写出的结果反推采用该变体；"
                                    "建议核实计划采用的参数口径")})
    value, reason = try_evaluate(str(method["expression"]), params)
    if value is None:
        method.update({"status": STATUS_INSUFFICIENT, "value_m3_min": None, "reason": reason})
        return method
    method.update({"status": STATUS_OK, "value_m3_min": round(value, 4)})
    return method


def _standby_method(
    item: Mapping[str, Any],
    rules: RuleSet,
    resolved: Mapping[str, Any] | None,
) -> dict[str, Any]:
    """备用工作面：规则给出的是下限，因此用「参照采面需风量 × 50%」作为核验线。"""
    formula = (rules.formulas(LOC_STANDBY) or [{}])[0]
    facts = as_mapping(item.get("planned_parameters"))
    ref_id = item.get("referenced_workplace_id")
    referenced = None
    if ref_id and resolved:
        referenced = (resolved.get("planned") or {}).get(str(ref_id))
    if referenced is None:
        referenced = facts.get("referenced_required_airflow")
    method = {
        "method_id": formula.get("id"),
        "name": formula.get("name"),
        "citation": formula.get("citation"),
        "aggregation": "min_bound",
        "expression": formula.get("expression"),
        "params": {"referenced_required_airflow": referenced},
        "reference_workplace_id": ref_id,
    }
    if referenced is None:
        method.update({"status": STATUS_INSUFFICIENT, "value_m3_min": None,
                       "reason": "缺少参照采煤工作面的需风量"})
        return method
    value, reason = try_evaluate(str(formula.get("expression")), {"referenced_required_airflow": float(referenced)})
    if value is None:
        method.update({"status": STATUS_INSUFFICIENT, "value_m3_min": None, "reason": reason})
        return method
    method.update({"status": STATUS_OK, "value_m3_min": round(value, 4)})
    return method


def _aggregate_method(
    item: Mapping[str, Any],
    rules: RuleSet,
    target_type: str,
    resolved: Mapping[str, Any] | None,
) -> dict[str, Any]:
    """回风系统与矿井总需风量的汇总公式。

    汇总核验以子项计划值为基准（检验漏算、重复计算与算错）；子项复算值另走一路，
    完整时给出 propagation_value_m3_min，用于观察复算误差是否向上传播。
    """
    formula = (rules.formulas_for_target(str(target_type)) or [{}])[0]
    facts = as_mapping(item.get("planned_parameters"))
    planned = dict((resolved or {}).get("planned") or {})
    recalculated = dict((resolved or {}).get("recalculated") or {})
    child_count = (resolved or {}).get("child_count") or 0
    method: dict[str, Any] = {
        "method_id": formula.get("id"),
        "name": formula.get("name"),
        "citation": formula.get("citation"),
        "aggregation": formula.get("aggregation"),
        "expression": formula.get("expression"),
        "basis": "planned_components",
        "component_count": child_count,
        "component_values": planned,
        "missing_component_ids": list((resolved or {}).get("missing_planned") or []),
        "params": {"reserve_coefficient": facts.get("reserve_coefficient",
                                                     rules.constant("reserve_coefficient"))},
    }
    if not planned:
        method.update({"status": STATUS_INSUFFICIENT, "value_m3_min": None,
                       "reason": "缺少组成对象的计划需风量，无法校验汇总关系"})
        return method
    factor = method["params"]["reserve_coefficient"]
    value, reason = try_evaluate(str(formula.get("expression")),
                                 {"component_required": list(planned.values()),
                                  "reserve_coefficient": factor})
    if value is None:
        method.update({"status": STATUS_INSUFFICIENT, "value_m3_min": None, "reason": reason})
        return method
    method.update({"status": STATUS_OK, "value_m3_min": round(value, 4)})
    if len(recalculated) == child_count and child_count:
        propagated, propagated_reason = try_evaluate(
            str(formula.get("expression")),
            {"component_required": list(recalculated.values()), "reserve_coefficient": factor},
        )
        if propagated is None:
            method["propagation_note"] = f"复算值汇总失败：{propagated_reason}"
        else:
            method["propagation_value_m3_min"] = round(propagated, 4)
            method["recalculated_component_values"] = recalculated
    else:
        method["propagation_note"] = (
            f"仅 {len(recalculated)}/{child_count} 个组成对象完成复算，未给出复算传播值"
        )
    return method


def _run_checks(
    item: Mapping[str, Any],
    facts: Mapping[str, Any],
    rules: RuleSet,
    location_type: str,
    result: Mapping[str, Any],
) -> list[dict[str, Any]]:
    """验算：把规则里的不等式取右边的计算值，与左侧主体值比较。"""
    subject = {
        "required_airflow_m3_min": result.get("calculated_required_airflow_m3_min")
        if result.get("calculated_required_airflow_m3_min") is not None
        else item.get("planned_required_airflow_m3_min"),
        "planned_required_airflow_m3_min": item.get("planned_required_airflow_m3_min"),
    }
    subject.update({key: value for key, value in facts.items() if value is not None})
    checks: list[dict[str, Any]] = []
    for spec in rules.verification(location_type):
        check: dict[str, Any] = {
            "check_id": spec.get("id"),
            "name": spec.get("name"),
            "operator": spec.get("operator"),
            "citation": spec.get("citation"),
            "expression": spec.get("expression"),
        }
        if spec.get("applicable_when") and not all(
            facts.get(key) == value for key, value in spec["applicable_when"].items()
        ):
            check.update({"status": STATUS_NOT_APPLICABLE, "reason": "验算条件不适用"})
            checks.append(check)
            continue
        left_value = subject.get(str(spec.get("left")))
        params, missing, derived = derive_params(dict(spec), facts, rules)
        check["derived"] = derived
        if missing:
            check.update({"status": STATUS_INSUFFICIENT, "reason": "缺少参数：" + "、".join(missing)})
            checks.append(check)
            continue
        right_value, reason = try_evaluate(str(spec.get("expression")), params)
        if right_value is None:
            check.update({"status": STATUS_INSUFFICIENT, "reason": reason})
            checks.append(check)
            continue
        check["left_value"] = left_value
        check["limit_value"] = round(right_value, 4)
        check["params"] = {name: params.get(name) for name in (spec.get("params") or {})}
        if not isinstance(left_value, (int, float)):
            check.update({"status": STATUS_INSUFFICIENT, "reason": "左侧主体值缺失"})
            checks.append(check)
            continue
        check["status"] = "pass" if _compare(spec.get("operator"), float(left_value), right_value) else "fail"
        checks.append(check)
    return checks


def _compare(operator: Any, left: float, right: float) -> bool:
    tolerance = abs(right) * 1e-9 + 1e-9
    if operator == ">=":
        return left >= right - tolerance
    if operator == "<=":
        return left <= right + tolerance
    if operator == ">":
        return left > right
    if operator == "<":
        return left < right
    return False


def _bounds(result: Mapping[str, Any]) -> tuple[float | None, float | None]:
    """从验算项里抽出需风量的上下限，供比对与报告引用。"""
    low: float | None = None
    high: float | None = None
    for check in result.get("checks") or []:
        if check.get("status") != STATUS_OK and check.get("limit_value") is None:
            continue
        limit = check.get("limit_value")
        if limit is None:
            continue
        if check.get("operator") in (">=", ">"):
            low = max(low, limit) if low is not None else limit
        elif check.get("operator") in ("<=", "<"):
            high = min(high, limit) if high is not None else limit
    return low, high


def within_tolerance(
    planned: float | None,
    calculated: float | None,
    tolerance: Mapping[str, float],
) -> tuple[bool, float | None, float | None]:
    """绝对与相对容差需同时满足；用于吸收计划取整（如 1220.4 → 1221）。"""
    if planned is None or calculated is None:
        return False, None, None
    difference = round(float(calculated) - float(planned), 4)
    absolute_limit = float(tolerance.get("absolute_m3_min", 1.0))
    relative_limit = abs(float(planned)) * float(tolerance.get("relative_percent", 0.5)) / 100.0
    limit = max(absolute_limit, relative_limit)
    return abs(difference) <= limit, difference, round(limit, 4)


__all__ = ["STATUS_OK", "STATUS_INSUFFICIENT", "STATUS_NOT_APPLICABLE", "STATUS_INCONSISTENT",
           "recalc_plan_item", "recalc_all", "within_tolerance", "PLAN_TARGET_TYPES"]