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