"""主通风机双机报告参数、性能和切换状态比较；所有指标由受控函数计算。""" 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}