vent_agent_2.0/extensions/main_fan/pipeline.py

"""七阶段业务流程：确定性执行，保存阶段结果，支持从指定阶段继续。"""
from pathlib import Path
import json

import config
from algorithms.main_fan.equivalence import (RulePolicy, normalize_and_validate_fan_data, compare_basic_parameters,
    analyze_fan_performance_curve, analyze_fan_switch_event, match_fan_rules_and_knowledge,
    assess_fan_equivalent_capability, trace_fan_evidence_and_review, envelope, evidence_id)
from algorithms.main_fan.models import ScreeningInput, Screening
from extensions.main_fan.pair_reporting import render_fan_assessment_report

STAGES = ("data_preparation", "basic_parameters", "performance_curves", "switch_events", "rules",
          "overall_assessment", "report")


def new_stages():
    return [{"name":name,"status":"pending","result":None} for name in STAGES]


def load_rules():
    configured=config.get_str("MAIN_FAN_RULES_FILE")
    path=Path(configured) if configured else Path(__file__).resolve().parents[2]/'examples/main_fan/equivalence_rules.json'
    if configured and not path.is_absolute():
        path=config.ROOT/path
    return RulePolicy.model_validate(json.loads(path.read_text(encoding="utf-8")))


def run_pipeline(context, *, stages=None, start_stage=1, save=None):
    if start_stage not in range(1,8):
        raise ValueError("重试阶段必须在1～7之间")
    stages=stages or new_stages()
    if any(s["result"] is None for s in stages[:start_stage-1]):
        raise ValueError("上游阶段尚未完成，不能跳过执行")
    for stage in stages[start_stage-1:]:
        stage.update(status="pending",result=None)
    data=ScreeningInput.model_validate(context["data"])
    analysis=Screening.model_validate(context["analysis"])
    policy=RulePolicy.model_validate(context["policy"])
    dataset=normalize_and_validate_fan_data(data,policy,context.get("fan_names"))

    def execute(number, function):
        stage=stages[number-1]
        if number<start_stage:
            return stage["result"]
        stage["status"]="running"
        if save:
            save(stages)
        try:
            value=function()
            stage.update(status="insufficient" if value.get("status")=="insufficient" else "completed",result=value)
        except Exception as exc:
            stage.update(status="failed",error=str(exc)[:240])
            if save:
                save(stages)
            raise
        if save:
            save(stages)
        return value

    prepared=execute(1,lambda:envelope({"status":"completed" if analysis.has_in_scope_data else "insufficient", "fan_pair":[{"fan_id":f['fan_id'],"fan_name":f['fan_name'],
        "report_id":f['report'].report_id if f['report'] else None} for f in dataset.fans],
        "document_count":len(context.get('document_audits',[])),"operating_series_count":len(data.operating_series),
        "warnings":dataset.warnings},policy))
    basic=execute(2,lambda:compare_basic_parameters(dataset))
    curves=execute(3,lambda:analyze_fan_performance_curve(dataset))
    switches=execute(4,lambda:analyze_fan_switch_event(dataset))
    rules=execute(5,lambda:match_fan_rules_and_knowledge(dataset,basic,curves,switches))
    overall=execute(6,lambda:{"status":"completed", "decision":assess_fan_equivalent_capability(dataset,basic,curves,switches,rules),
        "trace":trace_fan_evidence_and_review(dataset,basic,curves,switches)})
    assessment={"fan_pair":prepared['fan_pair'],"basic_comparison":basic,"performance_comparison":curves,
        "switch_comparison":switches,"rule_match":rules,**overall}
    report=execute(7,lambda:envelope({"status":"completed", "body":render_fan_assessment_report(dataset,assessment,analysis)},policy))
    assessment["stages"]=[{"name":s['name'],"status":s['status']} for s in stages]
    return assessment, report['body'], stages


def component(workflow, tool):
    stages=workflow['stages']
    mapping={"parse_fan_evidence_documents":0,"query_fan_operating_data":0,"normalize_and_validate_fan_data":0,
        "analyze_fan_performance_curve":2,"analyze_fan_switch_event":3,"match_fan_rules_and_knowledge":4,
        "assess_fan_equivalent_capability":5,"trace_fan_evidence_and_review":5,"render_fan_assessment_report":6}
    value=stages[mapping[tool]]['result']
    if value is None:
        raise ValueError("该阶段尚无完成结果")
    if tool=='parse_fan_evidence_documents':
        value={**value,"reports":workflow['context']['data']['reports']}
    elif tool=='query_fan_operating_data':
        value={**value,**{k:workflow['context']['data'][k] for k in ('operating_series','switch_records','fan_state_records')}}
    elif tool=='normalize_and_validate_fan_data':
        value={**value,"basic_comparison":stages[1]['result']}
    elif tool=='trace_fan_evidence_and_review':
        value=value['trace']
    elif tool=='assess_fan_equivalent_capability':
        value=value['decision']
    elif tool=='render_fan_assessment_report':
        value={**value,"analysis_id":(workflow['result'] or {}).get('analysis_id','')}
    return {"workflow_id":workflow['id'],**value}