优化PKS多批次故障演化扫描

backend/scan_pks_fault_evolution.py

@@ -12,6 +12,7 @@ import argparse
 import csv
 import math
 import os
+import re
 import sys
 from collections import defaultdict
 from datetime import datetime, timedelta
@@ -23,12 +24,12 @@ import pymysql
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 from app.config import settings  # noqa: E402
 
-BATCH_TO_UNIT = {30: "7号机", 31: "8号机", 32: "9号机"}
 EXPECTED_PER_HOUR = 720
 MIN_HOUR_COVERAGE = 0.80
 MIN_DAILY_HOURS = 4
 MIN_HISTORY_HOURS = 24
 BASELINE_DAYS = 3
+VIBRATION_BASELINE_DAYS = 7
 RESET_GAP_HOURS = 12
 RESET_RISE_ABS = 0.04
 RESET_RISE_REL = 0.10
@@ -50,6 +51,7 @@ SITE_POINT_COLUMNS = (
 
 HOURLY_SQL = """
 SELECT
+    import_batch_id,
     FROM_UNIXTIME((UNIX_TIMESTAMP(sample_time) DIV 3600) * 3600) AS hour_start,
     COUNT(*) AS samples,
     SUM(CASE WHEN YSJ_41 > 0 THEN 1 ELSE 0 END) AS running_samples,
@@ -70,8 +72,8 @@ FROM pks_long_sample
 WHERE import_batch_id = %s
   AND sample_time >= %s
   AND sample_time < %s
-GROUP BY FROM_UNIXTIME((UNIX_TIMESTAMP(sample_time) DIV 3600) * 3600)
-ORDER BY hour_start
+GROUP BY import_batch_id, FROM_UNIXTIME((UNIX_TIMESTAMP(sample_time) DIV 3600) * 3600)
+ORDER BY hour_start, import_batch_id
 """
 
 NUMERIC_HOURLY_FIELDS = (
@@ -120,6 +122,33 @@ def number(value):
     return value if math.isfinite(value) else None
 
 
+def load_pks_batches() -> dict[int, str]:
+    """Resolve PKS import batches to units from the import metadata table."""
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT id, directory_path
+                FROM compressor_data_import
+                WHERE data_type = %s
+                  AND directory_path IS NOT NULL
+                ORDER BY id
+                """,
+                ("PKS",),
+            )
+            result = {}
+            for row in cursor.fetchall():
+                match = re.search(r"(?<!\d)([789])\s*#", str(row["directory_path"]))
+                if match:
+                    result[int(row["id"])] = f"{match.group(1)}号机"
+    finally:
+        connection.close()
+    if not result:
+        raise RuntimeError("compressor_data_import 中没有可识别 7#/8#/9# 的 PKS 导入批次")
+    return result
+
+
 def alarm_entry(item: dict, alarm_type: str) -> tuple[float | None, str]:
     for index in range(1, 5):
         configured_type = item.get(f"AlarmType{index}")
@@ -128,13 +157,14 @@ def alarm_entry(item: dict, alarm_type: str) -> tuple[float | None, str]:
     return None, ""
 
 
-def load_alarm_config() -> dict[int, dict]:
+def load_alarm_config(batch_to_unit: dict[int, str]) -> dict[int, dict]:
     """Read alarm limits by AlarmType, because AlarmLimit positions differ."""
     result = {}
     connection = connect()
     try:
         with connection.cursor() as cursor:
-            names = [f"YSJ{unit}_{point}" for unit in (7, 8, 9) for point in (5, 10, 11)]
+            unit_numbers = sorted({unit[0] for unit in batch_to_unit.values()})
+            names = [f"YSJ{unit}_{point}" for unit in unit_numbers for point in (5, 10, 11)]
             placeholders = ",".join(["%s"] * len(names))
             cursor.execute(
                 f"SELECT {SITE_POINT_COLUMNS} FROM site_point WHERE ItemName IN ({placeholders})",
@@ -144,7 +174,8 @@ def load_alarm_config() -> dict[int, dict]:
     finally:
         connection.close()
 
-    for batch, unit in BATCH_TO_UNIT.items():
+    unit_configs = {}
+    for unit in sorted(set(batch_to_unit.values())):
         unit_number = unit[0]
         oil = items.get(f"YSJ{unit_number}_5")
         coupling = items.get(f"YSJ{unit_number}_10")
@@ -172,9 +203,8 @@ def load_alarm_config() -> dict[int, dict]:
             raise RuntimeError(f"{unit} YSJ_5 缺少完整低压报警阈值")
         if any(value is None for value in (*coupling_limits.values(), *chain_limits.values())):
             raise RuntimeError(f"{unit} YSJ_10/11 缺少完整振动报警阈值")
-        result[batch] = {
+        unit_configs[unit] = {
             "unit": unit,
-            "batch": batch,
             "oil": oil_limits,
             "coupling": coupling_limits,
             "chain": chain_limits,
@@ -192,38 +222,69 @@ def load_alarm_config() -> dict[int, dict]:
                 "chain_units": chain.get("Units") or "",
             },
         }
+    for batch, unit in batch_to_unit.items():
+        result[batch] = {**unit_configs[unit], "batch": batch}
     return result
 
 
-def query_params(batch: int, start: datetime, end: datetime, config: dict) -> tuple:
+def load_batch_ranges(batches: list[int], start: datetime, end: datetime) -> dict[int, tuple[datetime, datetime]]:
+    placeholders = ",".join(["%s"] * len(batches))
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"""
+                SELECT import_batch_id, MIN(sample_time) AS data_start, MAX(sample_time) AS data_end
+                FROM pks_long_sample
+                WHERE import_batch_id IN ({placeholders})
+                  AND sample_time >= %s AND sample_time < %s
+                GROUP BY import_batch_id
+                """,
+                (*batches, start, end),
+            )
+            return {
+                int(row["import_batch_id"]): (
+                    max(row["data_start"], start),
+                    min(row["data_end"] + timedelta(seconds=5), end),
+                )
+                for row in cursor.fetchall()
+            }
+    finally:
+        connection.close()
+
+
+def query_params(batches: list[int], start: datetime, end: datetime, config: dict) -> tuple:
     return (
         config["oil"]["low"], config["oil"]["low_low"],
         config["coupling"]["high"], config["coupling"]["high_high"],
         config["chain"]["high"], config["chain"]["high_high"],
-        batch, start, end,
+        *batches, start, end,
     )
 
 
-def fetch_hourly(batch: int, start: datetime, end: datetime, chunk_days: int, config: dict):
+def fetch_hourly(unit: str, batches: list[int], start: datetime, end: datetime, chunk_days: int, config: dict):
     rows = []
+    placeholders = ",".join(["%s"] * len(batches))
+    sql = HOURLY_SQL.replace("import_batch_id = %s", f"import_batch_id IN ({placeholders})")
     connection = connect()
     try:
         with connection.cursor() as cursor:
             for index, (chunk_start, chunk_end) in enumerate(chunks(start, end, chunk_days), 1):
-                params = query_params(batch, chunk_start, chunk_end, config)
-                cursor.execute("EXPLAIN " + HOURLY_SQL, params)
+                params = query_params(batches, chunk_start, chunk_end, config)
+                cursor.execute("EXPLAIN " + sql, params)
                 plan = cursor.fetchone()
                 if not plan or plan.get("key") not in ("PRIMARY",):
                     raise RuntimeError(
-                        f"安全检查失败：batch={batch} chunk={chunk_start} "
+                        f"安全检查失败：unit={unit} batches={batches} chunk={chunk_start} "
                         f"EXPLAIN 未使用 PRIMARY: {plan}"
                     )
-                cursor.execute(HOURLY_SQL, params)
+                cursor.execute(sql, params)
                 chunk_count = 0
                 for raw in cursor.fetchall():
+                    batch = int(raw["import_batch_id"])
                     row = {
                         "批次": batch,
-                        "机组": config["unit"],
+                        "机组": unit,
                         "小时": raw["hour_start"].strftime("%Y-%m-%d %H:%M:%S"),
                         "样本数": int(raw["samples"] or 0),
                         "运行样本数": int(raw["running_samples"] or 0),
@@ -235,7 +296,7 @@ def fetch_hourly(batch: int, start: datetime, end: datetime, chunk_days: int, co
                     rows.append(row)
                     chunk_count += 1
                 print(
-                    f"batch={batch} chunk={index} {chunk_start}..{chunk_end}: "
+                    f"unit={unit} batches={batches} chunk={index} {chunk_start}..{chunk_end}: "
                     f"{chunk_count:,} hourly rows"
                 )
     finally:
@@ -1206,7 +1267,8 @@ def vibration_hour_signals(rows: list[dict], configs: dict[int, dict]):
                 # robust reference, but do not evaluate this transition hour.
                 last_valid_dt = row["_dt"]
                 continue
-            prior = valid[-336:]
+            baseline_start = row["_dt"] - timedelta(days=VIBRATION_BASELINE_DAYS)
+            prior = [item for item in valid if item["_dt"] >= baseline_start]
             side_signals = []
             for side in ("coupling", "chain"):
                 avg_center, avg_scale = robust_baseline([item[f"{side}_avg"] for item in prior])
@@ -1295,8 +1357,11 @@ def merge_vibration_signals(signals: list[dict], configs: dict[int, dict]):
         if current:
             events.append(build_vibration_event(current, configs[batch]))
 
-    for index, event in enumerate(events, 1):
-        event["周期编号"] = f'{event["机组"]}-V{index:02d}'
+    events.sort(key=lambda row: (_unit_sort_key(row["机组"]), row["开始小时"], row["批次"]))
+    unit_indexes = defaultdict(int)
+    for event in events:
+        unit_indexes[event["机组"]] += 1
+        event["周期编号"] = f'{event["机组"]}-V{unit_indexes[event["机组"]]:02d}'
 
     filtered = []
     for event in events:
@@ -1304,7 +1369,7 @@ def merge_vibration_signals(signals: list[dict], configs: dict[int, dict]):
         # absolute severe signals remain eligible on their own.
         if event["_严重信号小时数"] > 0 or event["_触发小时数"] >= 2:
             filtered.append(event)
-    return sorted(filtered, key=lambda row: (row["批次"], row["开始小时"]))
+    return sorted(filtered, key=_output_sort_key)
 
 
 def vibration_score(duration_hours: int, valid_hours: int, avg_change: float, max_value: float, peak_z: float, high_hours: int, high_high_hours: int, config: dict, side: str):
@@ -1441,8 +1506,29 @@ def build_vibration_event(signals: list[dict], config: dict):
     }
 
 
+def _unit_sort_key(value):
+    match = re.search(r"[789]", str(value or ""))
+    return int(match.group()) if match else 999
+
+
+def _output_sort_key(row: dict):
+    """Keep exported records grouped by unit, then ordered by their start time."""
+    time_field = next(
+        (field for field in ("小时", "开始小时", "周期开始小时", "开始日期", "日期") if row.get(field)),
+        None,
+    )
+    time_value = str(row.get(time_field, "")) if time_field else ""
+    return (
+        _unit_sort_key(row.get("机组")),
+        time_value,
+        int(row.get("批次") or 0),
+        str(row.get("周期编号", "")),
+    )
+
+
 def write_csv(path: Path, rows: list[dict], fields: list[str], headers: dict[str, str]):
     path.parent.mkdir(parents=True, exist_ok=True)
+    rows = sorted(rows, key=_output_sort_key)
     with path.open("w", encoding="utf-8-sig", newline="") as handle:
         writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore")
         writer.writerow({field: headers.get(field, field) for field in fields})
@@ -1512,9 +1598,9 @@ def write_thresholds(path: Path, configs: dict[int, dict]):
 
 def main():
     parser = argparse.ArgumentParser(description="只读生成 PKS 润滑油压力长期演化和振动短期异常段")
-    parser.add_argument("--batch", nargs="+", type=int, choices=sorted(BATCH_TO_UNIT), default=list(BATCH_TO_UNIT))
-    parser.add_argument("--start", default="2025-04-01 00:00:00")
-    parser.add_argument("--end", default="2026-05-01 00:00:00")
+    parser.add_argument("--batch", nargs="+", type=int, help="PKS 导入批次 ID，默认扫描 compressor_data_import 中可识别的全部批次")
+    parser.add_argument("--start", default="2025-01-01 00:00:00")
+    parser.add_argument("--end", default="2026-09-01 00:00:00")
     parser.add_argument("--chunk-days", type=int, default=14)
     parser.add_argument("--output-dir", default=str(Path(__file__).resolve().parents[1] / "cache" / "pks_scanner" / "evolution"))
     args = parser.parse_args()
@@ -1523,12 +1609,26 @@ def main():
         parser.error("时间范围或 chunk-days 无效")
 
     print("只读模式：不修改数据库；只分析 YSJ_5、YSJ_10、YSJ_11、YSJ_41")
-    configs = load_alarm_config()
-    print("已按 AlarmType 读取每台机器的 site_point 阈值")
+    batch_to_unit = load_pks_batches()
+    selected_batches = args.batch or sorted(batch_to_unit)
+    unknown_batches = sorted(set(selected_batches) - set(batch_to_unit))
+    if unknown_batches:
+        parser.error(f"以下批次不是 compressor_data_import 中可识别的 PKS 批次：{unknown_batches}")
+    configs = load_alarm_config({batch: batch_to_unit[batch] for batch in selected_batches})
+    print(f"已从 compressor_data_import 解析 {len(selected_batches)} 个 PKS 批次的机组：{batch_to_unit}")
+    print(f"振动基线最多使用最近 {VIBRATION_BASELINE_DAYS} 天的有效运行小时")
+    batch_ranges = load_batch_ranges(selected_batches, start, end)
+    grouped_batches = defaultdict(list)
+    for batch in selected_batches:
+        if batch in batch_ranges:
+            grouped_batches[batch_to_unit[batch]].append(batch)
     all_rows = []
-    for batch in args.batch:
-        all_rows.extend(fetch_hourly(batch, start, end, args.chunk_days, configs[batch]))
-    all_rows.sort(key=lambda row: (row["批次"], row["_dt"]))
+    for unit, batches in sorted(grouped_batches.items(), key=lambda item: _unit_sort_key(item[0])):
+        unit_start = min(batch_ranges[batch][0] for batch in batches)
+        unit_end = max(batch_ranges[batch][1] for batch in batches)
+        config = configs[batches[0]]
+        all_rows.extend(fetch_hourly(unit, sorted(batches), unit_start, unit_end, args.chunk_days, config))
+    all_rows.sort(key=lambda row: (_unit_sort_key(row["机组"]), row["_dt"], row["批次"]))
 
     oil_hourly_levels = classify_oil_hourly_segments(all_rows, configs)
     oil_hour_stage_segments = build_oil_hour_stage_segments(oil_hourly_levels)