vent_agent_2.0/algorithms/main_fan/equivalence.py

"""主通风机双机报告参数、性能和切换状态比较；所有指标由受控函数计算。"""
from bisect import bisect_right
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime
import hashlib
import json
import math
import re
from statistics import mean, pstdev
import unicodedata

from algorithms.main_fan.analysis import LOCAL_ZONE, UNITS, fan_numbers, local_time, same_mine
from algorithms.main_fan.models import FactModel, ScreeningInput, SwitchRecord
from pydantic import Field, model_validator

TOOL_VERSION = "main-fan-pair-1.2"
SKILL_VERSION = "main-fan-equivalent-capability-assessment-1.2"


class CurvePolicy(FactModel):
    minimum_points: int = Field(default=4, ge=2)
    minimum_common_flow_coverage: float = Field(default=.8, gt=0, le=1)
    mean_pressure_deviation_percent: float = Field(default=5, ge=0)
    maximum_pressure_deviation_percent: float = Field(default=10, ge=0)
    maximum_power_deviation_percent: float = Field(default=10, ge=0)
    efficiency_difference_points: float = Field(default=5, ge=0)
    operating_flow_deviation_percent: float = Field(default=10, ge=0)
    operating_pressure_deviation_percent: float = Field(default=10, ge=0)


class SwitchPolicy(FactModel):
    maximum_switch_interval_minutes: float = Field(default=10, ge=0)
    window_minutes: int = Field(default=30, ge=1)
    transition_minutes: int = Field(default=5, ge=0)
    minimum_completeness: float = Field(default=.9, gt=0, le=1)
    sample_interval_seconds: int = Field(default=60, ge=1)
    maximum_cv: float = Field(default=.05, ge=0)
    flow_deviation_percent: float = Field(default=10, ge=0)
    pressure_deviation_percent: float = Field(default=10, ge=0)
    minimum_duration_minutes: int = Field(default=10, ge=1)
    maximum_recovery_minutes: int = Field(default=30, ge=1)


class SeverityPolicy(FactModel):
    # 严重阈值由业务另行配置，不把刚超过正常线直接当成严重异常。
    abnormal_switch_interval_minutes: float | None = Field(default=None, gt=0)
    abnormal_flow_deviation_percent: float | None = Field(default=None, gt=0)
    abnormal_parameter_fields: list[str] = Field(default_factory=list)


class RulePolicy(FactModel):
    rule_version: str = Field(min_length=1)
    threshold_version: str = Field(min_length=1)
    curve: CurvePolicy = Field(default_factory=CurvePolicy)
    switch: SwitchPolicy = Field(default_factory=SwitchPolicy)
    severity: SeverityPolicy = Field(default_factory=SeverityPolicy)

    @model_validator(mode="after")
    def separate_severe_limits(self):
        severe = self.severity
        if (severe.abnormal_switch_interval_minutes is not None
                and severe.abnormal_switch_interval_minutes <= self.switch.maximum_switch_interval_minutes):
            raise ValueError("严重切换间隔阈值必须高于正常切换间隔上限")
        if (severe.abnormal_flow_deviation_percent is not None
                and severe.abnormal_flow_deviation_percent <= self.switch.flow_deviation_percent):
            raise ValueError("严重风量差异阈值必须高于正常风量差异上限")
        return self


def model_key(value):
    value = unicodedata.normalize("NFKC", value).casefold()
    return re.sub(r"[\s\-_.]", "", value.replace("№", "no").replace("nº", "no"))


def rated_power(report):
    """2×185 表达的是两台185kW电动机，受控计算总功率并保留组成。"""
    raw = unicodedata.normalize("NFKC", report.rated_power_raw).lower()
    raw = re.sub(r"\s|kw|千瓦", "", raw).replace("×", "*").replace("x", "*")
    parts = None
    if re.fullmatch(r"[1-9]\d?\*\d+(?:\.\d+)?", raw):
        count, value = raw.split("*")
        parts = [float(value)] * int(count)
    elif re.fullmatch(r"\d+(?:\.\d+)?(?:\+\d+(?:\.\d+)?)+", raw):
        parts = [float(value) for value in raw.split("+")]
    elif re.fullmatch(r"\d+(?:\.\d+)?", raw):
        parts = [float(raw)]
    elif not raw and report.rated_power_kw:
        parts = [report.rated_power_kw]
    if not parts or any(p <= 0 for p in parts):
        return None
    total = sum(parts)
    if report.rated_power_kw is not None and abs(total-report.rated_power_kw) > .001:
        return None
    return {"raw": report.rated_power_raw or f"{total:g} kW", "components_kw": parts, "total_kw": total}


def evidence_id(evidence):
    content = json.dumps(evidence.model_dump(mode="json"), sort_keys=True, ensure_ascii=False)
    return "e-" + hashlib.sha256(content.encode()).hexdigest()[:16]


def envelope(value, policy, evidence=()):
    return {**value, "evidence_ids": sorted({evidence_id(e) for e in evidence if e.source or e.excerpt}),
            "rule_version": policy.rule_version, "confidence": None, "warnings": value.get("warnings", [])}


@dataclass
class Dataset:
    data: ScreeningInput
    fans: list[dict]
    policy: RulePolicy
    warnings: list[str]


def normalize_and_validate_fan_data(data, policy, fan_names=None):
    fan_names = dict(fan_names or {})
    fan_names.update({d.device_code: d.name for d in data.devices if same_mine(d, data)})
    if data.fan_id and data.fan_name:
        fan_names[data.fan_id] = data.fan_name
    now = local_time(data.evaluation_time or datetime.now(LOCAL_ZONE))
    groups = defaultdict(list)
    for report in data.reports:
        if same_mine(report, data) and (not report.inspected_on or report.inspected_on <= now.date()):
            key = tuple(sorted(fan_numbers(report.fan_name)))
            if len(key) == 1:
                groups[key].append(report)
    selected = []
    warnings = []
    if data.fan_pair:
        if len(data.fan_pair) != 2 or len(set(data.fan_pair)) != 2:
            raise ValueError("fan_pair必须是两个不同的风机标识")
        for fan_id in data.fan_pair:
            key = tuple(sorted(fan_numbers(fan_names.get(fan_id, ""))))
            candidates = groups.get(key, [])
            if candidates:
                selected.append((fan_id, max(candidates, key=lambda r: r.inspected_on.isoformat() if r.inspected_on else "")))
    elif len(groups) == 2:
        for key in sorted(groups):
            report = max(groups[key], key=lambda r: r.inspected_on.isoformat() if r.inspected_on else "")
            known = [code for code, name in fan_names.items() if tuple(sorted(fan_numbers(name))) == key]
            fan_id = known[0] if len(known) == 1 else "report:" + report.report_id
            selected.append((fan_id, report))
    else:
        key = tuple(sorted(fan_numbers(data.fan_name)))
        candidates = groups.get(key, [])
        if candidates:
            selected = [(data.fan_id or "report:"+candidates[0].report_id,
                         max(candidates, key=lambda r: r.inspected_on.isoformat() if r.inspected_on else ""))]
        if len(groups) > 2:
            warnings.append("本矿井有多组风机资料，未自动选择任意两台进行比较。")
    if len({r.report_id for _, r in selected}) != len(selected):
        raise ValueError("两台风机身份指向同一份报告")
    fans = [{"fan_id": code, "fan_name": report.fan_name, "report": report, "rated_power": rated_power(report)} for code, report in selected]
    # 没有性能报告时，只有实际设备记录/明确配置的名称才构造身份，仍不猜主备角色。
    if not fans:
        ids = data.fan_pair or ([data.fan_id] if data.fan_id else [])
        fans = [{"fan_id": code, "fan_name": fan_names[code], "report": None, "rated_power": None}
                for code in ids if code in fan_names]
    return Dataset(data, fans, policy, warnings)


def compare_basic_parameters(dataset):
    fans, data, policy = dataset.fans, dataset.data, dataset.policy
    rows = []
    fields = [("model", "风机型号", True), ("motor_model", "电动机型号", False),
              ("rated_speed_rpm", "额定转速", False), ("rated_voltage_raw", "额定电压", False),
              ("impeller_diameter_m", "叶轮直径", False), ("manufacturer", "风机制造商", False),
              ("structure_type", "结构类型", False), ("ventilation_mode", "通风方式", False),
              ("rated_flow_range_m3_min", "额定风量范围", False), ("rated_pressure_range_pa", "额定风压范围", False)]
    evidence = []
    if len(fans) == 2 and all(f["report"] for f in fans):
        reports = [f["report"] for f in fans]
        for field, label, core in fields:
            values = [getattr(r, field) for r in reports]
            if not all(v is not None and v != "" and v != [] for v in values):
                continue
            match = model_key(values[0]) == model_key(values[1]) if isinstance(values[0], str) else values[0] == values[1]
            used = [r.metadata_evidence.get(field, r.evidence) for r in reports]
            evidence.extend(used)
            rows.append({"field": field, "label": label, "values": values, "status": "match" if match else "mismatch",
                         "core": core, "evidence_ids": [evidence_id(e) for e in used]})
        powers = [f["rated_power"] for f in fans]
        if all(powers):
            used = [r.metadata_evidence.get("rated_power_raw", r.metadata_evidence.get("rated_power_kw", r.evidence)) for r in reports]
            evidence.extend(used)
            rows.append({"field": "rated_power", "label": "电动机额定功率", "values": powers,
                         "status": "match" if powers[0]["total_kw"] == powers[1]["total_kw"] else "mismatch",
                         "core": True, "evidence_ids": [evidence_id(e) for e in used]})
    devices = [d for d in data.devices if same_mine(d, data) and d.device_code in {f['fan_id'] for f in fans}]
    standby = [d for d in devices if d.fan_role == "standby"]
    availability = None
    if len(standby) == 1 and standby[0].availability_status != "unknown" and standby[0].observed_at:
        age = (local_time(data.evaluation_time or datetime.now(LOCAL_ZONE))-local_time(standby[0].observed_at)).total_seconds()
        if -120 <= age <= 600:
            availability = {"fan_id": standby[0].device_code, "status": standby[0].availability_status,
                            "evidence_ids": [evidence_id(standby[0].evidence)]}
            evidence.append(standby[0].evidence)
    core = [r for r in rows if r["core"]]
    status = "fail" if any(r["status"] == "mismatch" for r in rows) else "pass" if len(core) == 2 else "insufficient"
    return envelope({"status": status, "rows": rows, "backup_availability": availability}, policy, evidence)


def _curve_points(rows):
    by_flow = {}
    for row in rows:
        value = (row.static_pressure_pa, row.shaft_power_kw,
                 row.reported_efficiency_pct if row.reported_efficiency_pct is not None else row.flow_m3_s*row.static_pressure_pa/(10*row.shaft_power_kw))
        if row.flow_m3_s in by_flow and by_flow[row.flow_m3_s] != value:
            raise ValueError("同一风量存在不同性能值，不能任意覆盖")
        by_flow[row.flow_m3_s] = value
    return sorted(by_flow.items())


def _interpolate(points, q):
    if q < points[0][0] or q > points[-1][0]:
        raise ValueError("禁止超出原始风量范围外推")
    if q == points[-1][0]:
        return points[-1][1]
    index = max(0, bisect_right([p[0] for p in points], q)-1)
    a, b = points[index], points[index+1]
    weight = (q-a[0])/(b[0]-a[0])
    return tuple(x+(y-x)*weight for x, y in zip(a[1], b[1]))


def _relative(a, b):
    return abs(b-a)/abs(a)*100 if a else (0.0 if b == 0 else None)


def analyze_fan_performance_curve(dataset):
    fans, policy = dataset.fans, dataset.policy
    results, warnings, used = [], [], []
    if len(fans) != 2 or not all(f["report"] for f in fans):
        return envelope({"status": "insufficient", "results": []}, policy)
    groups = []
    for fan in fans:
        current = defaultdict(list)
        report = fan["report"]
        for point in report.points:
            if point.condition == "measured" and point.frequency_hz:
                current[(point.frequency_hz, point.blade_angle_deg if point.blade_angle_deg is not None else report.blade_angle_deg)].append(point)
        groups.append(current)
    for key in sorted(set(groups[0]) & set(groups[1]), key=lambda k: (k[0], k[1] is None, k[1] or 0)):
        rows = [g[key] for g in groups]
        try:
            a, b = [_curve_points(values) for values in rows]
        except ValueError as exc:
            warnings.append(f"{key[0]:g}Hz：{exc}")
            continue
        if min(len(a), len(b)) < policy.curve.minimum_points:
            continue
        low, high = max(a[0][0], b[0][0]), min(a[-1][0], b[-1][0])
        spans = [points[-1][0]-points[0][0] for points in (a, b)]
        coverage = min((high-low)/span for span in spans) if high > low and all(spans) else 0
        if coverage < policy.curve.minimum_common_flow_coverage:
            warnings.append(f"{key[0]:g}Hz共同风量区间覆盖不足。")
            continue
        grid = sorted({low, high, *(q for points in (a, b) for q, _ in points if low < q < high)})
        samples = [(q, _interpolate(a, q), _interpolate(b, q)) for q in grid]
        if any(left[0] <= 0 or left[1] <= 0 for _, left, _ in samples):
            warnings.append(f"{key[0]:g}Hz基准静压或功率不能用于相对偏差。")
            continue
        pressure_pct = [_relative(left[0], right[0]) for _, left, right in samples]
        power_pct = [_relative(left[1], right[1]) for _, left, right in samples]
        integrated_pct = 0
        for i in range(len(samples)-1):
            q0,left0,right0=samples[i]
            q1,left1,right1=samples[i+1]
            d0,d1=right0[0]-left0[0],right1[0]-left1[0]
            # |线性差值/线性基准|在零交叉处拆分后解析积分，避免网格密度影响均值。
            segments=[(left0[0],left1[0],d0,d1,q1-q0)]
            if d0*d1<0:
                weight=-d0/(d1-d0)
                pressure=left0[0]+(left1[0]-left0[0])*weight
                segments=[(left0[0],pressure,d0,0,(q1-q0)*weight),(pressure,left1[0],0,d1,(q1-q0)*(1-weight))]
            for p0,p1,delta0,delta1,span in segments:
                slope=p1-p0
                if abs(slope)<1e-9:
                    integral=(delta0+delta1)/(2*p0)
                else:
                    ratio=(delta1-delta0)/slope
                    integral=ratio+(delta0-ratio*p0)/slope*math.log(p1/p0)
                integrated_pct += abs(integral)*span*100
        integrated_pct /= high-low
        integrated_square = 0
        for i in range(len(samples)-1):
            d0, d1 = samples[i][2][0]-samples[i][1][0], samples[i+1][2][0]-samples[i+1][1][0]
            integrated_square += (samples[i+1][0]-samples[i][0])*(d0*d0+d0*d1+d1*d1)/3
        metrics = {"mean_pressure_deviation_percent": integrated_pct,
                   "maximum_pressure_deviation_percent": max(pressure_pct),
                   "pressure_rmse_pa": math.sqrt(integrated_square/(high-low)),
                   "maximum_power_deviation_percent": max(power_pct),
                   "efficiency_difference_points": max(abs(left[2]-right[2]) for _, left, right in samples)}
        operating = [[r for r in values if "运行工况" in r.label] for values in rows]
        operating_result = None
        if all(len(values) == 1 for values in operating):
            left, right = [values[0] for values in operating]
            metrics.update(operating_flow_deviation_percent=_relative(left.flow_m3_s, right.flow_m3_s),
                           operating_pressure_deviation_percent=_relative(left.static_pressure_pa, right.static_pressure_pa))
            operating_result = [{"flow_m3_min": r.flow_m3_s*60, "pressure_pa": r.static_pressure_pa,
                                 "shaft_power_kw": r.shaft_power_kw, "input_power_kw": r.input_power_kw,
                                 "efficiency_pct": r.reported_efficiency_pct if r.reported_efficiency_pct is not None else r.flow_m3_s*r.static_pressure_pa/(10*r.shaft_power_kw)} for r in (left, right)]
        thresholds = policy.curve.model_dump()
        exceeded = [name for name, value in metrics.items() if value is not None and name in thresholds and value > thresholds[name]]
        comparable = key[1] is not None
        evidence = [r.evidence for values in rows for r in values]
        used.extend(evidence)
        results.append({"frequency_hz": key[0], "blade_angle_deg": key[1],
            "comparability": "same_angle_frequency" if comparable else "frequency_only",
            "status": ("fail" if exceeded else "pass") if comparable else "informational",
            "common_flow_range_m3_min": [low*60, high*60], "common_flow_coverage": coverage,
            "metrics": metrics, "thresholds": thresholds, "exceeded": exceeded,
            "operating_points": operating_result, "fit_method": "piecewise_linear", "extrapolation_used": False,
            "curve_samples": [{"flow_m3_min": q*60, "fan_1": list(left), "fan_2": list(right)} for q, left, right in samples],
            "evidence_ids": sorted({evidence_id(e) for e in evidence})})
    formal = [r for r in results if r["comparability"] == "same_angle_frequency"]
    status = "fail" if any(r["status"] == "fail" for r in formal) else "pass" if formal else "insufficient"
    return envelope({"status": status, "results": results, "warnings": warnings}, policy, used)


def _state_groups(dataset):
    data, ids = dataset.data, {f["fan_id"] for f in dataset.fans}
    latest = local_time(data.evaluation_time or datetime.now(LOCAL_ZONE))
    groups = defaultdict(dict)
    conflicting = set()
    for record in data.fan_state_records:
        if same_mine(record, data) and record.fan_id in ids and record.complete_fan_state and record.event_type in {"state", "start_stop"}:
            at = local_time(record.observed_at)
            if at > latest:
                continue
            prior = groups[at].get(record.fan_id)
            if prior and prior.state != record.state:
                conflicting.add(at)
            groups[at][record.fan_id] = record
    return [(at, records if at not in conflicting and set(records) == ids and
             all(r.state != 'unknown' for r in records.values()) else None)
            for at, records in sorted(groups.items())]


def _events(dataset):
    """整机同步状态允许经过双停过渡；缺测、冲突或未知状态切断事件链。"""
    data, ids = dataset.data, {f["fan_id"] for f in dataset.fans}
    if len(ids) != 2:
        return []
    latest = local_time(data.evaluation_time or datetime.now(LOCAL_ZONE))
    events = [e for e in data.switch_records if same_mine(e, data) and local_time(e.switched_at) <= latest and
              {e.from_fan_id, e.to_fan_id} == ids and e.from_fan_id != e.to_fan_id]
    previous, origin, stopped_at, started_at = None, None, None, None
    maximum_gap = dataset.policy.switch.sample_interval_seconds * 2
    groups = _state_groups(dataset)
    for at, records in groups:
        if records is None:
            previous, origin, stopped_at, started_at = None, None, None, None
            continue
        if previous and (at-previous[0]).total_seconds() > maximum_gap:
            previous, origin, stopped_at, started_at = None, None, None, None
        running = [r.fan_id for r in records.values() if r.state == "running"]
        if origin and previous:
            destination = next(iter(ids - {origin}))
            prior = previous[1]
            if prior[origin].state == 'running' and records[origin].state == 'stopped':
                stopped_at = at
            if prior[destination].state == 'stopped' and records[destination].state == 'running':
                started_at = at
            if running == [destination] and stopped_at is not None and started_at is not None:
                begin, complete = min(stopped_at, started_at), max(stopped_at, started_at)
                matching = [e for e in events if e.from_fan_id == origin and e.to_fan_id == destination and
                    begin.timestamp()-maximum_gap <= local_time(e.switched_at).timestamp() <= complete.timestamp()+maximum_gap]
                if not matching:
                    events.append(SwitchRecord(mine_code=data.mine_code, mine_name=data.mine_name,
                        event_id="state-"+hashlib.sha256(str(at).encode()).hexdigest()[:12], switched_at=complete,
                        from_fan_id=origin, to_fan_id=destination, from_stopped_at=stopped_at,
                        to_started_at=started_at, evidence=records[destination].evidence))
                elif len(matching) == 1 and matching[0].from_stopped_at is None and matching[0].to_started_at is None:
                    index = events.index(matching[0])
                    events[index] = matching[0].model_copy(update={'switched_at':complete,
                        'from_stopped_at':stopped_at,'to_started_at':started_at})
        if len(running) == 1:
            origin, stopped_at, started_at = running[0], None, None
        previous = (at, records)
    return sorted(events, key=lambda e: local_time(e.switched_at))


def _window(series, start, end, policy, latest=math.inf):
    slots = defaultdict(lambda:defaultdict(list))
    for sample in series.samples:
        at = local_time(sample.observed_at).timestamp()
        if start <= at < end and at <= latest and sample.status in {"normal", "正常", "online", "在线"} and sample.value >= 0:
            slot = int((at-start)//policy.sample_interval_seconds)
            slots[slot][at].append(sample.value*UNITS[series.metric][series.unit])
    expected = max(1, math.ceil((end-start)/policy.sample_interval_seconds))
    # 同一时刻冲突值无效；不能用重复/密集回补样本虚增完整率。
    values = [mean(values[0] for values in samples.values()) for samples in slots.values()
              if all(len(set(values))==1 for values in samples.values())]
    completeness = len(values)/expected
    if not values or completeness < policy.minimum_completeness:
        return None
    average = mean(values)
    cv = pstdev(values)/abs(average) if average else (0 if all(v == 0 for v in values) else math.inf)
    if cv > policy.maximum_cv:
        return None
    return {"start": datetime.fromtimestamp(start, LOCAL_ZONE).isoformat(), "end": datetime.fromtimestamp(end, LOCAL_ZONE).isoformat(),
            "mean": average, "std": pstdev(values), "min": min(values), "max": max(values),
            "sample_count": len(values), "completeness": completeness}


def _over_limit(value, limit):
    return value > limit and not math.isclose(value, limit, rel_tol=0, abs_tol=1e-9)


def _switch_timing(event, dataset):
    limit = dataset.policy.switch.maximum_switch_interval_minutes
    result = {'status':'insufficient','from_stopped_at':None,'to_started_at':None,
              'interval_minutes':None,'overlap_minutes':None,'limit_minutes':limit}
    if event.from_stopped_at is None or event.to_started_at is None:
        return result
    stop, start = local_time(event.from_stopped_at), local_time(event.to_started_at)
    latest = local_time(dataset.data.evaluation_time or datetime.now(LOCAL_ZONE))
    if max(stop, start) > latest:
        return result
    # 明示切换记录与同一时刻的整机状态冲突时，不用任一来源静默覆盖另一来源。
    for fan_id, at, expected in ((event.from_fan_id, stop, 'stopped'), (event.to_fan_id, start, 'running')):
        states = {r.state for r in dataset.data.fan_state_records if same_mine(r, dataset.data) and
            r.fan_id == fan_id and r.complete_fan_state and r.event_type in {'state','start_stop'} and local_time(r.observed_at) == at}
        if states and states != {expected}:
            return result
    interval = max(0., (start-stop).total_seconds()/60)
    return {**result,'status':'fail' if _over_limit(interval, limit) else 'pass',
            'from_stopped_at':stop.isoformat(),'to_started_at':start.isoformat(),
            'interval_minutes':interval,'overlap_minutes':max(0.,(stop-start).total_seconds()/60)}


def _switch_chart_data(dataset, event, series, start, end):
    ids = (event.from_fan_id, event.to_fan_id)
    state_curves = {fan_id:[] for fan_id in ids}
    for at, records in _state_groups(dataset):
        if records and start <= at.timestamp() < end:
            for fan_id in ids:
                state_curves[fan_id].append({'observed_at':at.isoformat(),'value':1 if records[fan_id].state == 'running' else 0})
    flow_curves = []
    latest = local_time(dataset.data.evaluation_time or datetime.now(LOCAL_ZONE)).timestamp()
    for fan_id in ids:
        candidates = [s for s in series if s.fan_id == fan_id and s.metric == 'flow']
        if len(candidates) != 1:
            continue
        source, samples = candidates[0], defaultdict(list)
        for sample in source.samples:
            at = local_time(sample.observed_at)
            if start <= at.timestamp() < end and at.timestamp() <= latest and sample.value >= 0 and sample.status in {'normal','正常','online','在线'}:
                samples[at].append(sample.value*UNITS['flow'][source.unit]*60)
        flow_curves.append({'fan_id':fan_id,'unit':'m3/min','samples':[{'observed_at':at.isoformat(),'value':values[0]}
            for at, values in sorted(samples.items()) if len(set(values)) == 1]})
    return {'state_curves':[{'fan_id':fan_id,'samples':samples} for fan_id,samples in state_curves.items() if samples],
            'flow_curves':flow_curves,'sample_interval_seconds':dataset.policy.switch.sample_interval_seconds,
            'synthetic':event.evidence.method == 'synthetic' or
                any(s.evidence.method == 'synthetic' for s in series if s.fan_id in ids)}


def analyze_fan_switch_event(dataset):
    data, policy = dataset.data, dataset.policy
    events, results, warnings, used = _events(dataset), [], [], []
    series = [s for s in data.operating_series if same_mine(s, data) and s.mapping_confirmed and
              s.metric in {"flow", "static_pressure", "pressure"} and s.unit in UNITS.get(s.metric, {})]
    width, margin = policy.switch.window_minutes*60, policy.switch.transition_minutes*60
    latest = local_time(data.evaluation_time or datetime.now(LOCAL_ZONE)).timestamp()
    for event in events:
        t = local_time(event.switched_at).timestamp()
        timing = _switch_timing(event, dataset)
        if timing['interval_minutes'] is not None:
            endpoints = [local_time(event.from_stopped_at).timestamp(),local_time(event.to_started_at).timestamp()]
            begin, complete = min(endpoints),max(endpoints)
        else:
            begin = complete = t
        before, after = {}, {}
        inputs = {}
        for metric in ("flow", "static_pressure", "pressure"):
            chosen = [[s for s in series if s.fan_id == fan_id and s.metric == metric] for fan_id in (event.from_fan_id, event.to_fan_id)]
            if not all(len(items) == 1 for items in chosen):
                continue
            a, b = [items[0] for items in chosen]
            wa = _window(a, begin-margin-width, begin-margin, policy.switch, latest)
            wb = _window(b, complete+margin, complete+margin+width, policy.switch, latest)
            if wa and wb:
                before[metric], after[metric], inputs[metric] = wa, wb, (a, b)
        # 有调角/调频/故障或另一切换的窗口不做等价比较。
        interfering = [r for r in data.fan_state_records if same_mine(r, data) and
            r.fan_id in {event.from_fan_id, event.to_fan_id} and r.event_type in {"frequency_change", "angle_change", "fault"} and
            begin-margin-width <= local_time(r.observed_at).timestamp() < complete+margin+width]
        nearby_switches = [other for other in events if other is not event and
            begin-margin-width <= local_time(other.switched_at).timestamp() < complete+margin+width]
        if not before or interfering or nearby_switches:
            warnings.append(f"切换{event.event_id}未形成可比较的稳定窗口。")
            before, after, inputs = {}, {}, {}
        metrics, changes = {}, {}
        recovery = None
        exceeded = []
        for metric, left in before.items():
            right = after[metric]
            if left['mean'] <= 0:
                continue
            deviation = _relative(left["mean"], right["mean"])
            if deviation is None:
                continue
            changes[metric] = {"before": left["mean"], "after": right["mean"],
                "absolute_change": right["mean"]-left["mean"], "relative_change_percent": (right["mean"]-left["mean"])/abs(left["mean"])*100 if left["mean"] else 0}
            threshold = policy.switch.flow_deviation_percent if metric == "flow" else policy.switch.pressure_deviation_percent
            if _over_limit(deviation, threshold):
                exceeded.append(metric)
            name = "flow" if metric == "flow" else metric
            metrics[name+"_deviation_percent"] = deviation
        if not changes and timing['interval_minutes'] is None:
            continue
        # 从过渡期后开始寻找连续完整且回到基线带内的窗口，不能以一个点宣称恢复。
        recovery_width = policy.switch.minimum_duration_minutes*60
        for offset in range(margin, (policy.switch.maximum_recovery_minutes+policy.switch.window_minutes)*60+1, policy.switch.sample_interval_seconds) if changes else ():
            valid = True
            for metric, (a, b) in inputs.items():
                window = _window(b, complete+offset, complete+offset+recovery_width, policy.switch, latest)
                threshold = policy.switch.flow_deviation_percent if metric == "flow" else policy.switch.pressure_deviation_percent
                deviation = _relative(before[metric]["mean"], window["mean"]) if window else None
                if deviation is None or deviation > threshold:
                    valid = False
                    break
            if valid:
                recovery = offset/60
                break
        metrics["recovery_time_minutes"] = recovery
        metrics['switch_interval_minutes'] = timing['interval_minutes']
        for a, b in inputs.values():
            used.extend([a.evidence, b.evidence])
        used.append(event.evidence)
        state_evidence = [r.evidence for r in data.fan_state_records if same_mine(r,data) and
            r.fan_id in {event.from_fan_id,event.to_fan_id} and r.complete_fan_state and
            begin-policy.switch.sample_interval_seconds <= local_time(r.observed_at).timestamp() <= complete]
        used.extend(state_evidence)
        flow_status = 'fail' if 'flow' in exceeded else 'pass' if 'flow' in changes else 'insufficient'
        checks = {'switch_interval':timing['status'],'flow_change':flow_status}
        status = 'fail' if exceeded or timing['status'] == 'fail' else 'pass' if all(v == 'pass' for v in checks.values()) else 'insufficient'
        results.append({"event_id": event.event_id, "switched_at": datetime.fromtimestamp(complete,LOCAL_ZONE).isoformat(),
            "from_fan_id": event.from_fan_id, "to_fan_id": event.to_fan_id,
            "before_window": before, "after_window": after, "changes": changes, "metrics": metrics,
            'timing':timing,'checks':checks,'limits':{'flow_deviation_percent':policy.switch.flow_deviation_percent},
            'chart_data':_switch_chart_data(dataset,event,series,begin-margin-width,complete+margin+width),
            "status":status, "exceeded": exceeded + (['switch_interval'] if timing['status'] == 'fail' else []),
            "assessed_metrics": sorted(changes), "evidence_ids": sorted({evidence_id(e) for e in [event.evidence,*state_evidence, *(s.evidence for pair in inputs.values() for s in pair)]})})
    fully_assessed=bool(results) and len(results) == len(events) and all(r['status'] == 'pass' for r in results)
    return envelope({"status": "fail" if any(r["status"] == "fail" for r in results) else "pass" if fully_assessed else "insufficient",
                     "events": results, "candidate_event_count": len(events),
                     'assessment_scope':['switch_interval','flow_change'],"warnings": warnings}, policy, used)


def three_check_statuses(basic, switches):
    """只汇总基础参数、切换间隔、稳定风量三项；补充指标不参与否决。"""
    events = switches.get("events", [])

    def switch_check(name):
        statuses = [event.get("checks", {}).get(name, "insufficient") for event in events]
        if "fail" in statuses:
            return "fail"
        complete = len(events) >= switches.get("candidate_event_count", len(events))
        return "pass" if statuses and complete and all(status == "pass" for status in statuses) else "insufficient"

    return {"parameters": basic.get("status", "insufficient"),
            "switch_interval": switch_check("switch_interval"),
            "flow_change": switch_check("flow_change")}


def _severe_findings(dataset, basic, switches):
    """仅针对已经形成明确不通过结果的核心项检查单独配置的严重线。"""
    policy = dataset.policy.severity
    findings = []
    for row in basic.get("rows", []):
        if (row["status"] == "mismatch" and row["field"] in policy.abnormal_parameter_fields
                and len(set(row.get("evidence_ids", []))) == 2):
            findings.append({"item": "parameters", "field": row["field"], "label": row["label"],
                             "evidence_ids": row["evidence_ids"]})
    checks = (("switch_interval", "switch_interval_minutes", policy.abnormal_switch_interval_minutes),
              ("flow_change", "flow_deviation_percent", policy.abnormal_flow_deviation_percent))
    for event in switches.get("events", []):
        for item, metric, limit in checks:
            value = event.get("metrics", {}).get(metric)
            if (limit is not None and value is not None
                    and (value >= limit or math.isclose(value, limit, rel_tol=0, abs_tol=1e-9))
                    and event.get("checks", {}).get(item) == "fail" and event.get("evidence_ids")):
                findings.append({"item": item, "event_id": event["event_id"], "value": value,
                                 "severe_limit": limit, "evidence_ids": event["evidence_ids"]})
    return findings


def match_fan_rules_and_knowledge(dataset, basic, curves, switches):
    return envelope({"status": "completed", "threshold_version": dataset.policy.threshold_version,
        "rules": three_check_statuses(basic, switches),
        "supplementary": {"performance_curves": curves["status"]},
        "thresholds": dataset.policy.model_dump(), "exception_assessment": None}, dataset.policy)


def assess_fan_equivalent_capability(dataset, basic, curves, switches, rules):
    items = three_check_statuses(basic, switches)
    severe = _severe_findings(dataset, basic, switches)
    code = ("NORMAL" if all(status == "pass" for status in items.values()) else
            "ABNORMAL" if severe else "WARNING" if "fail" in items.values() else "UNDETERMINED")
    overall = "pass" if code == "NORMAL" else "fail" if code in {"WARNING", "ABNORMAL"} else "insufficient_evidence"
    ids = {e for result in (basic, curves, switches) for e in result["evidence_ids"]}
    return {"overall_status": overall, "result_code": code, "items": items,
            "required_items": ["parameters", "switch_interval", "flow_change"], "severe_findings": severe,
            "evidence_ids": sorted(ids), "rule_version": dataset.policy.rule_version,
            "threshold_version": dataset.policy.threshold_version, "confidence": None, "warnings": dataset.warnings,
            "reported_scope": [key for key, available in (("report_parameters", bool(basic["rows"])),
                ("report_performance", bool(curves["results"])), ("switch_performance", bool(switches["events"]))) if available]}


def trace_fan_evidence_and_review(dataset, basic, curves, switches):
    index = {}
    for fan in dataset.fans:
        report = fan["report"]
        if report:
            for evidence in [report.evidence, *report.metadata_evidence.values(), *(r.evidence for r in [*report.points, *report.curves])]:
                if evidence.source or evidence.excerpt:
                    index[evidence_id(evidence)] = evidence.model_dump(mode="json")
    for item in [*dataset.data.operating_series, *dataset.data.switch_records, *dataset.data.fan_state_records, *dataset.data.devices]:
        if same_mine(item, dataset.data):
            index[evidence_id(item.evidence)] = item.evidence.model_dump(mode="json")
    chains = [{"item": key, "evidence_ids": result["evidence_ids"], "rule_version": dataset.policy.rule_version}
              for key, result in (("basic_parameters", basic), ("performance_data", curves), ("switch_performance", switches))]
    return {"evidence_index": index, "evidence_chains": chains, "tool_version": TOOL_VERSION, "skill_version": SKILL_VERSION,
            "rule_version": dataset.policy.rule_version, "threshold_version": dataset.policy.threshold_version,
            "confidence": None, "warnings": dataset.warnings}