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

"""配风计划计算过程的口径判定与自洽核验。

计划里同一个物理量可能有不同口径（"断面"是有效断面积还是净断面积），而计划往往
不标注。这里负责两件事，都是**确定性**的：

1. 变体选择：口径标记命中时按声明走；命中不到时用计划**自己写出的结果**反推它用的是
   哪个变体（`_infer_variant`），并留下痕迹供人工核实。
2. 自洽核验：拿计划写出的代入值，在规则库允许的**任一**口径下复现它写出的结果
   （`_reproduce_with_candidates`）。一个都复现不出来才算真的算错。

关键在于候选集**只来自规则库**，不是任意表达式——所以"逐个试算"并没有放宽判定，
只是不再用单一假设去否定一份正确的计划。
"""
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,
    TARGET_RETURN_SYSTEM,
)
from algorithms.airflow_review.engine.expression import try_evaluate
from algorithms.airflow_review.rules.loader import RuleSet

STATUS_OK = "ok"
STATUS_INSUFFICIENT = "insufficient"
STATUS_INCONSISTENT = "inconsistent"


def _candidate_expressions(
    formula: Mapping[str, Any],
    facts: Mapping[str, Any],
    rules: RuleSet,
) -> list[tuple[str, Any]]:
    """按优先顺序给出该公式的全部候选表达式：显式命中的变体优先，其余变体次之，基础表达式最后。

    候选集**只来自规则库**，不是任意表达式——所以"逐个试算"并没有放宽判定。
    """
    selected = rules.select_formula(dict(formula), facts)
    if selected.get("expression") is None:
        return []
    ordered: list[tuple[str, Any]] = []
    if selected.get("variant"):
        ordered.append((str(selected["expression"]), selected["variant"]))
    for variant in formula.get("variants") or []:
        expression = variant.get("expression")
        if expression and all(expression != existing for existing, _ in ordered):
            ordered.append((str(expression), variant.get("when")))
    base = formula.get("expression")
    if base and all(base != existing for existing, _ in ordered):
        ordered.append((str(base), None))
    return ordered

def _infer_variant(
    formula: Mapping[str, Any],
    params: Mapping[str, Any],
    stated_result: float | None,
) -> dict[str, Any] | None:
    """用计划写出的结果反推变体；返回命中的表达式与候选清单，命中不到返回 None。

    容差按计划结果的 0.5% 与 1 m³/min 取大者，避免因计划取整而判不中。
    """
    variants = formula.get("variants") or []
    if not variants or not isinstance(stated_result, (int, float)):
        return None
    limit = max(abs(float(stated_result)) * 0.005, 1.0)
    candidates: list[dict[str, Any]] = []
    for variant in variants:
        expression = variant.get("expression")
        if not expression:
            continue
        value, _ = try_evaluate(str(expression), params)
        entry = {"expression": expression, "when": variant.get("when"),
                 "value_m3_min": None if value is None else round(value, 4)}
        candidates.append(entry)
        if value is not None and abs(value - float(stated_result)) <= limit:
            return {"expression": expression, "variant": variant.get("when"),
                    "candidates": candidates}
    return None

def _reproduce_with_candidates(
    formula: Mapping[str, Any],
    merged: Mapping[str, Any],
    values: Mapping[str, Any],
    stated: Mapping[str, Any],
    resolved: Mapping[str, Any] | None,
    rules: RuleSet,
) -> dict[str, Any] | None:
    """自洽核验：计划写出的代入值，能否在规则库允许的某个口径下复现它写出的结果。

    计划往往不标注参数口径（如"断面"到底是有效断面积还是净断面积），这时逐个候选变体
    试算：**某一个**能复现即认为计划自洽；一个都复现不出来才是真的算错。这既避免把正确
    计划误判成计算错误，也没有放宽判定——候选集来自规则库。计划没写结果时返回 None，
    交由调用方按基础表达式走常规路径。
    """
    stated_result = stated.get("stated_result_m3_min")
    if not isinstance(stated_result, (int, float)):
        return None
    candidates = _candidate_expressions(formula, merged, rules)
    if not candidates:
        return None
    params = {**merged, **values}
    if resolved is not None:
        params.setdefault("component_required", list(((resolved or {}).get("planned") or {}).values()))
        params.setdefault("reserve_coefficient",
                          values.get("reserve_coefficient", rules.constant("reserve_coefficient")))
    limit = max(abs(float(stated_result)) * 0.005, 1.0)
    evaluated: list[dict[str, Any]] = []
    for expression, variant in candidates:
        value, reason = try_evaluate(expression, params)
        evaluated.append({"expression": expression, "variant": variant,
                          "value_m3_min": None if value is None else round(value, 4),
                          "reason": reason})
        if value is not None and abs(value - float(stated_result)) <= limit:
            return {
                "status": STATUS_OK, "expression": expression, "source": "rule_library",
                "variant": variant,
                "variant_selection": ("declared" if variant else "base"),
                "recomputed_m3_min": round(value, 4),
                "substituted_values": dict(values),
                "stated_result_m3_min": stated_result,
                "difference_m3_min": round(float(stated_result) - value, 4),
                "candidates": evaluated,
            }
    return {
        "status": STATUS_INCONSISTENT,
        "reason": ("计划写出的代入值在规则库允许的任何口径下都得不到它写出的结果："
                   + "；".join(f"{entry['expression']} → {entry['value_m3_min']}" for entry in evaluated)),
        "expression": candidates[0][0],
        "candidates": evaluated,
        "substituted_values": dict(values),
        "stated_result_m3_min": stated_result,
    }

def _reproduce_stated(
    item: Mapping[str, Any],
    facts: Mapping[str, Any],
    rules: RuleSet,
    resolved: Mapping[str, Any] | None = None,
) -> dict[str, Any] | None:
    """自洽核验：计划写的代入值，按它自己引用的公式应该得到什么结果。"""
    stated = item.get("stated_calculation")
    if not isinstance(stated, Mapping):
        return None
    values = as_mapping(stated.get("substituted_values"))
    # 汇总类对象的"代入值"就是它列出的各子项计划值
    if "component_required" in str(stated.get("expression") or "") or "sum(" in str(stated.get("expression") or ""):
        planned = (resolved or {}).get("planned") or {}
        if planned:
            values.setdefault("component_required", list(planned.values()))
    if not values and not stated.get("expression"):
        return None
    expression = None
    source = "plan_stated"
    formula_id = stated.get("formula_id") or item.get("planned_formula")
    if formula_id:
        formula = _find_formula(rules, str(formula_id), str(facts.get("location_type") or ""))
        if formula is not None:
            merged = {**facts, **values}
            outcome = _reproduce_with_candidates(formula, merged, values, stated, resolved, rules)
            if outcome is not None:
                return outcome
            selected = rules.select_formula(dict(formula), merged)
            if selected.get("expression"):
                expression = str(selected["expression"])
                source = "rule_library"
    if expression is None:
        expression = stated.get("expression")
    if not expression:
        return {"status": STATUS_INSUFFICIENT, "reason": "计划未给出可复算的计算式",
                "stated_result_m3_min": stated.get("stated_result_m3_min")}
    merged_params = {**facts, **values}
    if resolved is not None:
        merged_params.setdefault("component_required", list(((resolved or {}).get("planned") or {}).values()))
        merged_params.setdefault("reserve_coefficient", facts.get("reserve_coefficient",
                                                                 rules.constant("reserve_coefficient")))
    value, reason = try_evaluate(str(expression), merged_params)
    if value is None:
        return {"status": STATUS_INSUFFICIENT, "reason": reason, "expression": expression,
                "stated_result_m3_min": stated.get("stated_result_m3_min")}
    stated_result = stated.get("stated_result_m3_min")
    payload = {"status": STATUS_OK, "expression": expression, "source": source,
               "recomputed_m3_min": round(value, 4),
               "substituted_values": values,
               "stated_result_m3_min": stated_result}
    if isinstance(stated_result, (int, float)):
        payload["difference_m3_min"] = round(float(stated_result) - value, 4)
    return payload

def _find_formula(rules: RuleSet, formula_id: str, location_type: str) -> dict[str, Any] | None:
    for bucket in (location_type, LOC_WORKPLACE_COAL, LOC_WORKPLACE_HEADING, LOC_CHAMBER,
                   LOC_OTHER, LOC_STANDBY, TARGET_RETURN_SYSTEM, "mine_total"):
        for formula in rules.formulas(str(bucket)) or rules.formulas_for_target(str(bucket)):
            if formula.get("id") == formula_id:
                return formula
    return None