PKS：设备号输入、全量/实时运行与 compressor_alarm 落库，含定时服务脚本

models/README.md

@@ -15,16 +15,60 @@
 | `datas/` | 中间数据与训练数据：`hourly/`（小时中间表）、`pressure/`、`vibration/`（窗口样本）、`artifacts/`（训练产物）、`reference/`（golden 基准） |
 | `results/` | 算法扫描结果与模型测试结果：`common/`、`pressure/`、`vibration/`、`verification/` |
 | `markdown/` | 文档：压力算法、压力模型、振动算法、振动模型、模型辅助确认方案 |
-| `scan_all.py` | 一次运行压力+振动扫描并缓存小时中间表 |
+| `service/` | 实时服务定时脚本：`mac/`（launchd）、`linux/`（systemd）、`windows/`（schtasks） |
+| `migrations/` | 数据库结构变更 SQL（如 `alarm_des` 扩到 4000） |
+| `scan_all.py` | 全量扫描（压力+振动），按设备号输入，结果可存本地或数据库 |
+| `realtime_service.py` | 实时扫描（`--once` / `--loop` / `--simulate`），结果合并写入 `compressor_alarm` |
 
-## 复现（3 小时）
+## 设备号输入
+
+算法与扫描统一使用 `--device 7# 8# 9#`（默认 `7# 8# 9#`），设备通过
+`compressor_data_import.directory_path` 解析到 PKS 导入批次。`4#/5#/6#` 暂不纳入。
+算法内部仍按 `批次` 分组，因此本地全量结果与历史 CSV 逐字节一致。
+
+## 全量运行
 
 ```bash
-# 1) 算法扫描结果（复用已缓存小时中间表，逐字节对齐 golden）
+# 本地：自动查询起止时间（floor(最小)..floor(最大)+1h），先清理旧中间/结果再生成
+python models/scan_all.py --device 7# 8# 9# --store local
+
+# 数据库：本地结果与中间文件照常生成，并额外把最终段写入 compressor_alarm
+python models/scan_all.py --device 7# 8# 9# --store db
+
+# 也可显式指定时间，或复用已缓存小时中间表
+python models/scan_all.py --start "2025-01-01 00:00:00" --end "2026-09-01 00:00:00"
 python models/scan_all.py --hourly-csv models/datas/hourly/pks_hourly_full.csv
+```
+
+数据库存储映射（表 `compressor_alarm`）：
+
+| 结果 | device_code | device_part | device_point | alarm_type | alarm_level |
+|---|---|---|---|---|---|
+| 压力段 | `7#` | `润滑油压力` | `YSJ_5` | `润滑油压力故障` | 压力阶段 |
+| 振动段 | `7#` | 按 `最强异常侧` 取 `联轴器端振动` / `链轮端振动` | 对应 `YSJ_10` / `YSJ_11` | `振动故障` | 振动阶段 |
+
+`alarm_des=阶段触发依据`（≤4000）、`alarm_info_json`=对应 CSV 行 JSON、`status=1`。
+全量写库会先删除所选设备的历史压力/振动告警再插入。
 
-# 或从数据库全量重跑并缓存小时中间表
-python models/scan_all.py --save-hourly
+## 实时运行
+
+```bash
+python models/realtime_service.py --once                 # 单次
+python models/realtime_service.py --loop --interval 300  # 每 5 分钟（命令行定时）
+```
+
+- 窗口：`end = min(floor(now), floor(最新数据)+1h)`，`start = end - 45 天`。
+- 合并：同 `(设备, 类型, 等级)` 且时间重叠或间隔 ≤6h 的段合并为一条，
+  同一等级连续段不会因跨小时或“最强侧”切换被拆开；等级变化保留为不同行，
+  合并时 `device_part/device_point` 更新为最新 `最强异常侧`。
+- 定时服务安装/卸载见 `models/service/README.md`。
+- 冒烟测试：`python models/realtime_service.py --simulate "2026-02-03 20:00:00,2026-02-04 00:00:00"`（默认结束后回滚）。
+
+## 复现（3 小时）
+
+```bash
+# 1) 算法扫描结果
+python models/scan_all.py --device 7# 8# 9# --store local
 
 # 2) 压力模型 3h：build -> split -> train -> evaluate -> infer
 python models/generalized/run_pipeline.py --window-hours 3
@@ -32,20 +76,20 @@ python models/generalized/run_pipeline.py --window-hours 3
 # 3) 振动模型 3h
 python models/vibration/run_pipeline.py --window-hours 3
 
-# 4) 自动校验（算法 CSV、数据库全量重跑、模型 3h 指标）
-python models/common/verify.py
+# 4) 自动校验（含 compressor_alarm 对应关系）
+python models/common/verify.py --with-db
 ```
 
 `--window-hours 6` 的参数化代码路径保留，但本轮不重新生成 6h 产物。
 
 ## 校验基线
 
-`python models/common/verify.py` 会比对：
+`python models/common/verify.py --with-db` 会比对：
 
 - 5 个扫描结果 CSV 与 `models/datas/reference/` 逐字节一致；
-- `models/results/verification/db_parity.json` 显示数据库全量重跑与缓存一致；
-- 压力 3h：阈值 `0.862553`、异常 F1 `96.59%`、事件召回 `10/10`、分机组正常误报率 `0.32%/1.13%/0.00%`；
-- 振动 3h：阈值 `0.916630`、异常 F1 `71.43%`、事件召回 `4/4`、分机组正常误报率 `0.00%/0.71%/0.13%`。
+- 数据库全量重跑与缓存一致；
+- 压力/振动 3h 模型指标与文档一致（阈值 `0.862553`/`0.916630`，异常 F1 `96.59%`/`71.43%`，事件召回 `10/10`、`4/4`）；
+- `compressor_alarm` 行数与参考 CSV 对应（压力 72、振动 25，共 97）。
 
 报告写入 `models/results/verification/verification_report.json`。
 

models/common/alarm_store.py

@@ -0,0 +1,268 @@
+"""Persist pressure / vibration stage segments into ``compressor_alarm``.
+
+Two write modes:
+
+* ``write_full``  – full scan: delete previous pressure/vibration alarms for the
+  selected devices, then insert every segment (vibration writes one row per
+  side: YSJ_10 and YSJ_11).
+* ``merge_realtime`` – realtime tick: a growing segment is merged into the
+  existing row with the same identity (device + part + point + type + level)
+  when their time ranges overlap or are within ``MERGE_GAP_HOURS``.  Different
+  levels are kept as separate rows, matching the stage CSV rows, so a continuous
+  same-level fault is never split while a level transition is preserved.
+"""
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta
+from typing import Any
+
+from .db import connect
+
+OIL_ALARM_TYPE = "润滑油压力故障"
+VIB_ALARM_TYPE = "振动故障"
+OIL_PART = "润滑油压力"
+OIL_POINT = "YSJ_5"
+COUPLING_PART = "联轴器端振动"
+COUPLING_POINT = "YSJ_10"
+CHAIN_PART = "链轮端振动"
+CHAIN_POINT = "YSJ_11"
+ALARM_STATUS = 1
+MERGE_GAP_HOURS = 6
+DES_MAX = 4000
+
+INSERT_SQL = """
+INSERT INTO compressor_alarm
+    (device_code, device_part, device_point, alarm_type, alarm_des,
+     alarm_time_start, alarm_time_end, status, alarm_level,
+     scan_start_time, scan_end_time, alarm_info_json)
+VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+"""
+
+
+def _as_dt(value) -> datetime:
+    if isinstance(value, datetime):
+        return value
+    return datetime.strptime(str(value), "%Y-%m-%d %H:%M:%S")
+
+
+def _json(row: dict) -> str:
+    return json.dumps(row, ensure_ascii=False, separators=(",", ":"), default=str)
+
+
+def _targets(device_code: str, kind: str, row: dict) -> list[dict]:
+    level = row.get("压力阶段") if kind == "pressure" else row.get("振动阶段")
+    start, end = row.get("开始小时"), row.get("结束小时")
+    des = str(row.get("阶段触发依据") or "")[:DES_MAX]
+    info = _json(row)
+    if kind == "pressure":
+        parts = [(OIL_PART, OIL_POINT, OIL_ALARM_TYPE)]
+    else:
+        # One row per segment, at the strongest side (matches 振动异常段.csv).
+        side = str(row.get("最强异常侧") or "")
+        if side == "联轴器端":
+            parts = [(COUPLING_PART, COUPLING_POINT, VIB_ALARM_TYPE)]
+        else:
+            parts = [(CHAIN_PART, CHAIN_POINT, VIB_ALARM_TYPE)]
+    result = []
+    for part, point, alarm_type in parts:
+        result.append({
+            "device_code": device_code,
+            "device_part": part,
+            "device_point": point,
+            "alarm_type": alarm_type,
+            "alarm_level": level,
+            "alarm_des": des,
+            "alarm_time_start": _as_dt(start),
+            "alarm_time_end": _as_dt(end),
+            "alarm_info_json": info,
+        })
+    return result
+
+
+def build_alarms(pressure_rows: list[dict], vibration_rows: list[dict], batch_to_device: dict[int, str]) -> list[dict]:
+    alarms = []
+    for row in pressure_rows:
+        device = batch_to_device.get(int(row["批次"]))
+        if device:
+            alarms.extend(_targets(device, "pressure", row))
+    for row in vibration_rows:
+        device = batch_to_device.get(int(row["批次"]))
+        if device:
+            alarms.extend(_targets(device, "vibration", row))
+    return alarms
+
+
+def _insert(cursor, alarm: dict) -> int:
+    cursor.execute(INSERT_SQL, (
+        alarm["device_code"], alarm["device_part"], alarm["device_point"], alarm["alarm_type"],
+        alarm["alarm_des"], alarm["alarm_time_start"], alarm["alarm_time_end"], ALARM_STATUS,
+        alarm["alarm_level"], alarm["scan_start_time"], alarm["scan_end_time"], alarm["alarm_info_json"],
+    ))
+    return int(cursor.lastrowid)
+
+
+def write_full(alarms: list[dict], devices: list[str], scan_start: datetime, scan_end: datetime, dry_run: bool = False) -> dict:
+    for alarm in alarms:
+        alarm["scan_start_time"] = scan_start
+        alarm["scan_end_time"] = scan_end
+    if dry_run:
+        return {"mode": "full", "dry_run": True, "delete_scope": [devices, OIL_ALARM_TYPE, VIB_ALARM_TYPE], "inserted": len(alarms)}
+    connection = connect()
+    deleted = 0
+    inserted_ids = []
+    try:
+        with connection.cursor() as cursor:
+            placeholders = ",".join(["%s"] * len(devices))
+            cursor.execute(
+                f"DELETE FROM compressor_alarm WHERE device_code IN ({placeholders}) AND alarm_type IN (%s, %s)",
+                (*devices, OIL_ALARM_TYPE, VIB_ALARM_TYPE),
+            )
+            deleted = int(cursor.rowcount)
+            for alarm in alarms:
+                inserted_ids.append(_insert(cursor, alarm))
+    finally:
+        connection.close()
+    return {"mode": "full", "deleted": deleted, "inserted": len(inserted_ids), "ids": inserted_ids}
+
+
+def merge_realtime(alarms: list[dict], scan_start: datetime, scan_end: datetime, journal: list | None = None, dry_run: bool = False) -> dict:
+    """Merge each segment into the matching existing row (same identity+level)."""
+    inserted = updated = merged_extra = 0
+    if dry_run:
+        return {"mode": "realtime", "dry_run": True, "segments": len(alarms)}
+    connection = connect()
+    gap = timedelta(hours=MERGE_GAP_HOURS)
+    try:
+        with connection.cursor() as cursor:
+            for alarm in alarms:
+                start, end = alarm["alarm_time_start"], alarm["alarm_time_end"]
+                cursor.execute(
+                    # Identity ignores part/point so a segment whose strongest side
+                    # flips (coupling <-> chain) still merges into one row instead
+                    # of being split.  part/point are refreshed to the latest side.
+                    """
+                    SELECT id, alarm_time_start, alarm_time_end, alarm_des, alarm_info_json,
+                           scan_start_time, scan_end_time, status
+                    FROM compressor_alarm
+                    WHERE device_code = %s AND alarm_type = %s AND alarm_level <=> %s
+                      AND alarm_time_start <= %s AND alarm_time_end >= %s
+                    ORDER BY alarm_time_start ASC, id ASC
+                    """,
+                    (
+                        alarm["device_code"], alarm["alarm_type"],
+                        alarm["alarm_level"], end + gap, start - gap,
+                    ),
+                )
+                existing = cursor.fetchall()
+                if not existing:
+                    new_id = _insert(cursor, {**alarm, "scan_start_time": scan_start, "scan_end_time": scan_end})
+                    inserted += 1
+                    if journal is not None:
+                        journal.append({"op": "insert", "id": new_id})
+                    continue
+                keep = existing[0]
+                new_start = min(start, *(row["alarm_time_start"] for row in existing))
+                new_end = max(end, *(row["alarm_time_end"] for row in existing))
+                new_scan_start = min([scan_start] + [row["scan_start_time"] for row in existing if row["scan_start_time"]])
+                new_scan_end = max([scan_end] + [row["scan_end_time"] for row in existing if row["scan_end_time"]])
+                if journal is not None:
+                    journal.append({
+                        "op": "update", "id": int(keep["id"]),
+                        "alarm_time_start": keep["alarm_time_start"], "alarm_time_end": keep["alarm_time_end"],
+                        "alarm_des": keep["alarm_des"], "alarm_info_json": keep["alarm_info_json"],
+                        "scan_start_time": keep["scan_start_time"], "scan_end_time": keep["scan_end_time"],
+                        "status": keep["status"],
+                    })
+                if len(existing) > 1:
+                    extra_ids = [int(row["id"]) for row in existing[1:]]
+                    placeholders = ",".join(["%s"] * len(extra_ids))
+                    cursor.execute(f"DELETE FROM compressor_alarm WHERE id IN ({placeholders})", extra_ids)
+                    merged_extra += len(extra_ids)
+                    if journal is not None:
+                        for row in existing[1:]:
+                            journal.append({"op": "delete_restore", "row": dict(row)})
+                cursor.execute(
+                    """
+                    UPDATE compressor_alarm
+                    SET device_part = %s, device_point = %s,
+                        alarm_time_start = %s, alarm_time_end = %s, alarm_des = %s,
+                        scan_start_time = %s, scan_end_time = %s, status = %s,
+                        alarm_info_json = %s
+                    WHERE id = %s
+                    """,
+                    (alarm["device_part"], alarm["device_point"],
+                     new_start, new_end, alarm["alarm_des"], new_scan_start, new_scan_end,
+                     ALARM_STATUS, alarm["alarm_info_json"], keep["id"]),
+                )
+                updated += 1
+    finally:
+        connection.close()
+    return {"mode": "realtime", "inserted": inserted, "updated": updated, "merged_extra": merged_extra}
+
+
+def rollback(journal: list) -> dict:
+    """Undo a realtime smoke run using the journal produced by ``merge_realtime``."""
+    if not journal:
+        return {"undone": 0}
+    connection = connect()
+    undone = 0
+    try:
+        with connection.cursor() as cursor:
+            for entry in reversed(journal):
+                if entry["op"] == "insert":
+                    cursor.execute("DELETE FROM compressor_alarm WHERE id = %s", (entry["id"],))
+                    undone += 1
+                elif entry["op"] == "update":
+                    cursor.execute(
+                        """
+                        UPDATE compressor_alarm
+                        SET alarm_time_start = %s, alarm_time_end = %s, alarm_des = %s,
+                            alarm_info_json = %s, scan_start_time = %s, scan_end_time = %s, status = %s
+                        WHERE id = %s
+                        """,
+                        (entry["alarm_time_start"], entry["alarm_time_end"], entry["alarm_des"],
+                         entry["alarm_info_json"], entry["scan_start_time"], entry["scan_end_time"],
+                         entry["status"], entry["id"]),
+                    )
+                    undone += 1
+                elif entry["op"] == "delete_restore":
+                    row = entry["row"]
+                    cursor.execute(INSERT_SQL, (
+                        row["device_code"], row["device_part"], row["device_point"], row["alarm_type"],
+                        row["alarm_des"], row["alarm_time_start"], row["alarm_time_end"], row["status"],
+                        row["alarm_level"] if "alarm_level" in row else None,
+                        row["scan_start_time"], row["scan_end_time"], row["alarm_info_json"],
+                    ))
+                    undone += 1
+    finally:
+        connection.close()
+    return {"undone": undone}
+
+
+def fetch_alarms(devices: list[str]) -> list[dict]:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            placeholders = ",".join(["%s"] * len(devices))
+            cursor.execute(
+                f"""
+                SELECT id, device_code, device_part, device_point, alarm_type, alarm_level,
+                       alarm_time_start, alarm_time_end, status, alarm_des, alarm_info_json
+                FROM compressor_alarm
+                WHERE device_code IN ({placeholders}) AND alarm_type IN (%s, %s)
+                ORDER BY device_code, alarm_time_start, device_part
+                """,
+                (*devices, OIL_ALARM_TYPE, VIB_ALARM_TYPE),
+            )
+            return [dict(row) for row in cursor.fetchall()]
+    finally:
+        connection.close()
+
+
+__all__ = [
+    "OIL_ALARM_TYPE", "VIB_ALARM_TYPE", "OIL_PART", "OIL_POINT",
+    "COUPLING_PART", "COUPLING_POINT", "CHAIN_PART", "CHAIN_POINT",
+    "MERGE_GAP_HOURS", "build_alarms", "write_full", "merge_realtime",
+    "rollback", "fetch_alarms",
+]

models/common/devices.py

@@ -0,0 +1,123 @@
+"""Device-code resolution for PKS scans.
+
+``--device`` accepts compressor codes such as ``7#`` / ``8#`` / ``9#`` (also
+tolerates ``7`` or ``7号机``).  A device maps to one or more PKS import batches
+through ``compressor_data_import.directory_path``; the batch ids are then used
+for the (index-friendly) hourly aggregation so results stay identical to the
+original batch-based scanner output.
+
+4#/5#/6# exist in the database but are intentionally not part of the default
+scan set.
+"""
+from __future__ import annotations
+
+import re
+from datetime import datetime, timedelta
+
+from .db import connect
+
+DEVICE_CODES = ("7#", "8#", "9#")
+DEVICE_TO_UNIT = {"7#": "7号机", "8#": "8号机", "9#": "9号机"}
+
+
+def normalize_device(value) -> str:
+    text = str(value).strip()
+    if text in DEVICE_TO_UNIT:
+        return text
+    digits = re.sub(r"\D", "", text)
+    candidate = f"{digits}#" if digits else ""
+    if candidate in DEVICE_TO_UNIT:
+        return candidate
+    raise ValueError(f"无法识别的压缩机编号：{value!r}（应为 7# / 8# / 9#）")
+
+
+def normalize_devices(values=None) -> list[str]:
+    devices = []
+    for value in (values or DEVICE_CODES):
+        device = normalize_device(value)
+        if device not in devices:
+            devices.append(device)
+    return devices
+
+
+def load_device_batches(devices=None) -> dict[str, list[int]]:
+    """Resolve ``device -> [import_batch_id, ...]`` from the import metadata."""
+    selected = normalize_devices(devices)
+    placeholders = ",".join(["%s"] * len(selected))
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"""
+                SELECT id, directory_path
+                FROM compressor_data_import
+                WHERE data_type = 'PKS'
+                  AND directory_path IN ({placeholders})
+                ORDER BY id
+                """,
+                selected,
+            )
+            result: dict[str, list[int]] = {device: [] for device in selected}
+            for row in cursor.fetchall():
+                device = str(row["directory_path"]).strip()
+                if device in result:
+                    result[device].append(int(row["id"]))
+    finally:
+        connection.close()
+    missing = [device for device, batches in result.items() if not batches]
+    if missing:
+        raise RuntimeError(f"以下设备没有找到 PKS 导入批次：{missing}")
+    return result
+
+
+def batch_to_device(device_batches: dict[str, list[int]]) -> dict[int, str]:
+    return {batch: device for device, batches in device_batches.items() for batch in batches}
+
+
+def batch_to_unit(device_batches: dict[str, list[int]]) -> dict[int, str]:
+    return {batch: DEVICE_TO_UNIT[device] for device, batches in device_batches.items() for batch in batches}
+
+
+def floor_hour(moment: datetime) -> datetime:
+    return moment.replace(minute=0, second=0, microsecond=0)
+
+
+def device_data_range(device_batches: dict[str, list[int]]) -> dict[str, tuple[datetime, datetime]]:
+    """Return ``device -> (min_sample_time, max_sample_time)`` using the batch index."""
+    ranges: dict[str, tuple[datetime, datetime]] = {}
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            for device, batches in device_batches.items():
+                placeholders = ",".join(["%s"] * len(batches))
+                cursor.execute(
+                    f"""
+                    SELECT MIN(sample_time) AS mn, MAX(sample_time) AS mx
+                    FROM pks_long_sample
+                    WHERE import_batch_id IN ({placeholders})
+                    """,
+                    batches,
+                )
+                row = cursor.fetchone()
+                if row and row["mn"] and row["mx"]:
+                    ranges[device] = (row["mn"], row["mx"])
+    finally:
+        connection.close()
+    if not ranges:
+        raise RuntimeError("所选设备没有可用样本时间范围")
+    return ranges
+
+
+def auto_range(device_batches: dict[str, list[int]]) -> tuple[datetime, datetime]:
+    """Settled scan range: floor(min) .. floor(max)+1h (last complete hour included)."""
+    ranges = device_data_range(device_batches)
+    start = min(floor_hour(mn) for mn, _ in ranges.values())
+    end = max(floor_hour(mx) + timedelta(hours=1) for _, mx in ranges.values())
+    return start, end
+
+
+__all__ = [
+    "DEVICE_CODES", "DEVICE_TO_UNIT", "normalize_device", "normalize_devices",
+    "load_device_batches", "batch_to_device", "batch_to_unit", "floor_hour",
+    "device_data_range", "auto_range",
+]

models/common/verify.py

@@ -13,14 +13,16 @@ non-zero if any check fails.
 """
 from __future__ import annotations
 
+import argparse
 import csv
 import json
 import sys
+from collections import Counter
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
 
-from models.common import paths
+from models.common import alarm_store, devices, paths
 
 ALGORITHM_FILES = {
     "pressure": ["润滑油压力小时等级.csv", "润滑油压力小时阶段段.csv", "润滑油压力演化周期.csv"],
@@ -133,16 +135,71 @@ def check_db_parity():
     return {"passed": bool(report.get("identical")), "report": report}
 
 
+def _reference_counts(path: Path, batch_to_device: dict[int, str]) -> Counter:
+    counts: Counter = Counter()
+    if not path.is_file():
+        return counts
+    with path.open(encoding="utf-8-sig", newline="") as handle:
+        for row in csv.DictReader(handle):
+            counts[batch_to_device[int(row["批次"])]] += 1
+    return counts
+
+
+def check_alarm_store():
+    """Compare compressor_alarm rows to the reference stage CSVs (1 row per segment)."""
+    device_list = list(devices.DEVICE_CODES)
+    device_batches = devices.load_device_batches(device_list)
+    batch_device = devices.batch_to_device(device_batches)
+    pressure_ref = _reference_counts(paths.REFERENCE / "润滑油压力小时阶段段.csv", batch_device)
+    vibration_ref = _reference_counts(paths.REFERENCE / "振动异常段.csv", batch_device)
+    expected: Counter = Counter()
+    for device in device_list:
+        expected[(device, "pressure")] = pressure_ref.get(device, 0)
+        expected[(device, "vibration")] = vibration_ref.get(device, 0)
+
+    rows = alarm_store.fetch_alarms(device_list)
+    actual: Counter = Counter()
+    for row in rows:
+        kind = "pressure" if row["alarm_type"] == alarm_store.OIL_ALARM_TYPE else "vibration"
+        actual[(row["device_code"], kind)] += 1
+
+    checks = []
+    for key in sorted(set(expected) | set(actual), key=str):
+        checks.append({"name": f"{key[0]}_{key[1]}", "actual": actual.get(key, 0), "expected": expected.get(key, 0),
+                       "passed": actual.get(key, 0) == expected.get(key, 0)})
+    status_ok = all(int(row["status"]) == 1 for row in rows)
+    json_ok = True
+    des_ok = True
+    for row in rows:
+        try:
+            json.loads(row["alarm_info_json"])
+        except (TypeError, ValueError):
+            json_ok = False
+        if len(row["alarm_des"] or "") > 4000:
+            des_ok = False
+    checks.append({"name": "status_all_1", "actual": status_ok, "expected": True, "passed": status_ok})
+    checks.append({"name": "json_valid", "actual": json_ok, "expected": True, "passed": json_ok})
+    checks.append({"name": "des_within_4000", "actual": des_ok, "expected": True, "passed": des_ok})
+    return {"passed": all(item["passed"] for item in checks), "checks": checks, "rows": len(rows)}
+
+
 def main() -> None:
+    parser = argparse.ArgumentParser(description="PKS 重构算法/模型自动校验")
+    parser.add_argument("--with-db", action="store_true", help="额外校验 compressor_alarm 与参考 CSV 的对应关系")
+    args = parser.parse_args()
+
     report = {
         "algorithm": check_algorithm(),
         "db_parity": check_db_parity(),
         "models": {mt: check_model(mt) for mt in ("pressure", "vibration")},
     }
+    if args.with_db:
+        report["alarm_store"] = check_alarm_store()
     report["passed"] = (
         report["algorithm"]["passed"]
         and report["db_parity"]["passed"]
         and all(item["passed"] for item in report["models"].values())
+        and report.get("alarm_store", {"passed": True})["passed"]
     )
     output = paths.RESULTS / "verification"
     output.mkdir(parents=True, exist_ok=True)
@@ -153,6 +210,10 @@ def main() -> None:
     for model_type, result in report["models"].items():
         failed = [item["name"] for item in result["checks"] if not item["passed"]]
         print(f"{model_type} 模型复现：{'通过' if result['passed'] else '失败'}" + (f" 差异={failed}" if failed else ""))
+    if "alarm_store" in report:
+        failed = [item["name"] for item in report["alarm_store"]["checks"] if not item["passed"]]
+        print(f"compressor_alarm 对应：{'通过' if report['alarm_store']['passed'] else '失败'}"
+              + (f" 差异={failed}" if failed else "") + f"（{report['alarm_store']['rows']} 行）")
     print(f"总体：{'全部通过' if report['passed'] else '存在失败项'}")
     print(f"报告：{output / 'verification_report.json'}")
     if not report["passed"]:

models/generalized/pressure_algorithm.py

@@ -27,7 +27,7 @@ from pathlib import Path
 if __package__ in (None, ""):
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
 
-from models.common import db, io, paths, queries
+from models.common import db, devices, io, paths, queries
 from models.common.constants import (
     HOURLY_BASELINE_HOURS,
     MIN_HOUR_COVERAGE,
@@ -315,34 +315,32 @@ def write_results(output_dir: Path, hourly_levels, hour_segments, cycles):
     io.write_csv(output_dir / "润滑油压力演化周期.csv", io.clean_rows(cycles), CYCLE_FIELDS, {field: field for field in CYCLE_FIELDS})
 
 
-def load_hourly(args):
+def load_hourly(args, selected_batches, unit_by_batch, configs):
     if args.hourly_csv:
-        return queries.load_hourly_csv(Path(args.hourly_csv))
-    batch_to_unit = db.load_pks_batches()
-    selected = args.batch or sorted(batch_to_unit)
-    configs = db.load_alarm_config({batch: batch_to_unit[batch] for batch in selected})
+        selected_set = set(selected_batches)
+        return [row for row in queries.load_hourly_csv(Path(args.hourly_csv)) if row["批次"] in selected_set]
     start, end = queries.parse_dt(args.start), queries.parse_dt(args.end)
-    return queries.fetch_all_hourly(start, end, args.chunk_days, batch_to_unit, configs, selected)
+    return queries.fetch_all_hourly(start, end, args.chunk_days, unit_by_batch, configs, selected_batches)
 
 
 def main() -> None:
     parser = argparse.ArgumentParser(description="PKS 润滑油压力演化算法（独立于振动）")
-    parser.add_argument("--batch", nargs="+", type=int)
+    parser.add_argument("--device", nargs="+", default=list(devices.DEVICE_CODES), help="压缩机编号，如 7# 8# 9#")
     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("--hourly-csv", default=None)
     parser.add_argument("--output-dir", default=str(paths.PRESSURE_RESULTS))
     args = parser.parse_args()
-    batch_to_unit = db.load_pks_batches()
-    selected = args.batch or sorted(batch_to_unit)
-    configs = db.load_alarm_config({batch: batch_to_unit[batch] for batch in selected})
-    rows = load_hourly(args)
-    if args.batch:
-        selected_set = set(selected)
-        rows = [row for row in rows if row["批次"] in selected_set]
+    devices_selected = devices.normalize_devices(args.device)
+    device_batches = devices.load_device_batches(devices_selected)
+    unit_by_batch = devices.batch_to_unit(device_batches)
+    configs = db.load_alarm_config(unit_by_batch)
+    selected_batches = sorted(unit_by_batch)
+    rows = load_hourly(args, selected_batches, unit_by_batch, configs)
     hourly_levels, hour_segments, cycles = run(rows, configs)
     write_results(Path(args.output_dir), hourly_levels, hour_segments, cycles)
+    print(f"设备：{devices_selected}")
     print(f"小时特征：{len(rows):,} 行")
     print(f"润滑油压力小时等级：{len(hourly_levels):,} 行")
     print(f"润滑油压力小时阶段段：{len(hour_segments):,} 段")

models/migrations/alter_compressor_alarm_alarm_des_4000.sql

@@ -0,0 +1,3 @@
+-- 阶段触发依据（润滑油压力 / 振动）可能较长，扩大到 4000 以避免截断。
+ALTER TABLE compressor_alarm
+    MODIFY alarm_des VARCHAR(4000) NULL;

models/realtime_service.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""Realtime PKS alarm scanner.
+
+Runs the same pressure/vibration algorithm as the full scan over a rolling
+window ending at the current (settled) hour, then merges the resulting stage
+segments into ``compressor_alarm``.  A growing same-level segment is merged into
+one row (never split), while different levels remain separate rows.
+
+Modes
+-----
+* ``--once``                      run a single tick at the current local time.
+* ``--loop --interval 300``       run a tick every ``interval`` seconds (CLI timer).
+* ``--simulate "t1,t2,t3"``       smoke test: run ticks at the given times and
+                                  (by default) roll the database back afterwards.
+
+Examples
+--------
+    python models/realtime_service.py --once
+    python models/realtime_service.py --loop --interval 300
+    python models/realtime_service.py --simulate "2026-02-03 20:00:00,2026-02-04 00:00:00"
+"""
+from __future__ import annotations
+
+import argparse
+import sys
+import time
+from collections import Counter
+from datetime import datetime, timedelta
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from models.common import alarm_store, db, devices, io, queries
+from models.generalized import pressure_algorithm
+from models.vibration import vibration_algorithm
+
+REALTIME_CHUNK_DAYS = 7
+
+
+def run_tick(now: datetime, device_list: list[str], lookback_days: int, write: bool, journal: list | None) -> dict:
+    device_batches = devices.load_device_batches(device_list)
+    unit_by_batch = devices.batch_to_unit(device_batches)
+    batch_device = devices.batch_to_device(device_batches)
+    configs = db.load_alarm_config(unit_by_batch)
+
+    ranges = devices.device_data_range(device_batches)
+    latest = max(mx for _, mx in ranges.values())
+    end = min(devices.floor_hour(now), devices.floor_hour(latest) + timedelta(hours=1))
+    start = end - timedelta(days=lookback_days)
+    if start >= end:
+        return {"tick": now, "end": end, "action": "no_data", "inserted": 0, "updated": 0}
+
+    rows = queries.fetch_all_hourly(start, end, REALTIME_CHUNK_DAYS, unit_by_batch, configs, sorted(unit_by_batch))
+    hourly_levels, hour_segments, cycles = pressure_algorithm.run(rows, configs)
+    events = vibration_algorithm.run(rows, configs)
+    alarms = alarm_store.build_alarms(io.clean_rows(hour_segments), io.clean_rows(events), batch_device)
+    result = alarm_store.merge_realtime(alarms, start, end, journal=journal if write else None, dry_run=not write)
+    return {
+        "tick": now, "window": (start, end), "hourly_rows": len(rows),
+        "pressure_segments": len(hour_segments), "vibration_events": len(events),
+        "segments_to_store": len(alarms), **result,
+    }
+
+
+def format_tick(summary: dict) -> str:
+    window = summary.get("window")
+    window_text = f"{window[0]}..{window[1]}" if window else "-"
+    return (
+        f"[{summary['tick']}] window={window_text} 压力段={summary.get('pressure_segments')} "
+        f"振动段={summary.get('vibration_events')} 写入段={summary.get('segments_to_store')} "
+        f"insert={summary.get('inserted')} update={summary.get('updated')} merged_extra={summary.get('merged_extra')}"
+    )
+
+
+def snapshot(device_list: list[str], limit: int = 8) -> str:
+    rows = alarm_store.fetch_alarms(device_list)
+    counter = Counter((r["device_code"], r["alarm_type"], r["device_point"], r["alarm_level"]) for r in rows)
+    return "; ".join(f"{k}={v}" for k, v in sorted(counter.items(), key=lambda item: str(item[0]))[:limit])
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="PKS 实时告警扫描（结果写入 compressor_alarm）")
+    parser.add_argument("--device", nargs="+", default=list(devices.DEVICE_CODES))
+    parser.add_argument("--lookback-days", type=int, default=45)
+    parser.add_argument("--once", action="store_true", help="运行一次（默认）")
+    parser.add_argument("--loop", action="store_true", help="按 --interval 循环运行")
+    parser.add_argument("--interval", type=int, default=300, help="循环间隔秒，默认 300")
+    parser.add_argument("--simulate", default=None, help="逗号分隔的时间列表，冒烟测试")
+    parser.add_argument("--no-rollback", action="store_true", help="冒烟测试后不回滚数据库")
+    args = parser.parse_args()
+    device_list = devices.normalize_devices(args.device)
+    if args.lookback_days < 1:
+        parser.error("--lookback-days 必须大于 0")
+
+    if args.simulate:
+        times = [queries.parse_dt(item.strip()) for item in args.simulate.split(",") if item.strip()]
+        journal: list = []
+        for moment in times:
+            summary = run_tick(moment, device_list, args.lookback_days, write=True, journal=journal)
+            print(format_tick(summary), flush=True)
+            print("  当前告警分布：" + snapshot(device_list), flush=True)
+        if not args.no_rollback:
+            undone = alarm_store.rollback(journal)
+            print(f"冒烟测试回滚：{undone}", flush=True)
+        return
+
+    if args.loop:
+        print(f"实时服务启动，每 {args.interval} 秒运行一次；Ctrl+C 退出。", flush=True)
+        while True:
+            summary = run_tick(datetime.now(), device_list, args.lookback_days, write=True, journal=None)
+            print(format_tick(summary), flush=True)
+            time.sleep(args.interval)
+
+    summary = run_tick(datetime.now(), device_list, args.lookback_days, write=True, journal=None)
+    print(format_tick(summary))
+    print("当前告警分布：" + snapshot(device_list))
+
+
+if __name__ == "__main__":
+    main()

models/results/verification/verification_report.json

@@ -373,5 +373,65 @@
       ]
     }
   },
+  "alarm_store": {
+    "passed": true,
+    "checks": [
+      {
+        "name": "7#_pressure",
+        "actual": 7,
+        "expected": 7,
+        "passed": true
+      },
+      {
+        "name": "7#_vibration",
+        "actual": 8,
+        "expected": 8,
+        "passed": true
+      },
+      {
+        "name": "8#_pressure",
+        "actual": 63,
+        "expected": 63,
+        "passed": true
+      },
+      {
+        "name": "8#_vibration",
+        "actual": 8,
+        "expected": 8,
+        "passed": true
+      },
+      {
+        "name": "9#_pressure",
+        "actual": 2,
+        "expected": 2,
+        "passed": true
+      },
+      {
+        "name": "9#_vibration",
+        "actual": 9,
+        "expected": 9,
+        "passed": true
+      },
+      {
+        "name": "status_all_1",
+        "actual": true,
+        "expected": true,
+        "passed": true
+      },
+      {
+        "name": "json_valid",
+        "actual": true,
+        "expected": true,
+        "passed": true
+      },
+      {
+        "name": "des_within_4000",
+        "actual": true,
+        "expected": true,
+        "passed": true
+      }
+    ],
+    "rows": 97
+  },
   "passed": true
 }

models/scan_all.py

@@ -1,15 +1,25 @@
 #!/usr/bin/env python3
-"""Run both reconstructed scanners and cache the shared hourly intermediate.
+"""Full PKS scan (pressure + vibration) with device-code input.
 
-Usage
+Modes
 -----
-::
-
-    # Full DB scan (authoritative reproduction), cache hourly for later offline runs
-    python models/scan_all.py --save-hourly
-
-    # Offline rerun from a cached hourly table
-    python models/scan_all.py --hourly-csv models/datas/hourly/pks_hourly_full.csv
+* ``--store local`` (default): delete previous intermediate/result files, then
+  rebuild them under ``models/datas`` / ``models/results``.  Output is byte
+  compatible with the reference CSVs.
+* ``--store db``: rebuild intermediate files locally, then replace the
+  ``compressor_alarm`` rows (pressure + vibration) for the selected devices.
+
+Device input
+------------
+``--device 7# 8# 9#`` (default).  Each device resolves to its PKS import
+batches; hourly aggregation keeps the original batch grouping so the result
+matches the existing ``润滑油压力小时阶段段.csv`` / ``振动异常段.csv``.
+
+Examples
+--------
+    python models/scan_all.py --device 7# 8# 9#
+    python models/scan_all.py --auto-range --store db
+    python models/scan_all.py --start "2025-01-01 00:00:00" --end "2026-09-01 00:00:00"
 """
 from __future__ import annotations
 
@@ -19,50 +29,100 @@ from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
-from models.common import db, io, paths, queries
+from models.common import alarm_store, db, devices, io, paths, queries
 from models.generalized import pressure_algorithm
 from models.vibration import vibration_algorithm
 
 
+SCAN_OUTPUT_FILES = {
+    paths.COMMON_RESULTS: ["报警阈值配置.csv"],
+    paths.PRESSURE_RESULTS: ["润滑油压力小时等级.csv", "润滑油压力小时阶段段.csv", "润滑油压力演化周期.csv"],
+    paths.VIBRATION_RESULTS: ["振动异常段.csv"],
+}
+
+
+def clear_dir_files(directory: Path, names: list[str] | None = None) -> int:
+    """Delete files in a directory; if ``names`` given, only those file names."""
+    directory = Path(directory)
+    if not directory.exists():
+        return 0
+    removed = 0
+    targets = [directory / name for name in names] if names is not None else [entry for entry in directory.iterdir() if entry.is_file()]
+    for entry in targets:
+        if entry.is_file():
+            entry.unlink()
+            removed += 1
+    return removed
+
+
+def clean_intermediates() -> dict:
+    """Remove previous scan intermediate/result files only (keep model outputs)."""
+    removed = {"hourly": clear_dir_files(paths.HOURLY_DIR)}
+    for directory, names in SCAN_OUTPUT_FILES.items():
+        removed[directory.name] = clear_dir_files(directory, names)
+    return removed
+
+
 def main() -> None:
-    parser = argparse.ArgumentParser(description="PKS 压力+振动重构扫描器")
-    parser.add_argument("--batch", nargs="+", type=int)
-    parser.add_argument("--start", default="2025-01-01 00:00:00")
-    parser.add_argument("--end", default="2026-09-01 00:00:00")
+    parser = argparse.ArgumentParser(description="PKS 压力+振动全量扫描（按设备号）")
+    parser.add_argument("--device", nargs="+", default=list(devices.DEVICE_CODES), help="压缩机编号，如 7# 8# 9#")
+    parser.add_argument("--start", default=None)
+    parser.add_argument("--end", default=None)
+    parser.add_argument("--auto-range", action="store_true", help="按数据自动确定起止时间（默认在未指定 --start/--end 时启用）")
     parser.add_argument("--chunk-days", type=int, default=14)
     parser.add_argument("--hourly-csv", default=None, help="复用已落盘的小时中间表，不查询数据库")
-    parser.add_argument("--save-hourly", action="store_true", help="扫描后将小时中间表写入 models/datas/hourly/pks_hourly_full.csv")
+    parser.add_argument("--store", choices=("local", "db"), default="local")
+    parser.add_argument("--dry-run", action="store_true", help="只统计不写库（store=db 时）")
     args = parser.parse_args()
 
-    batch_to_unit = db.load_pks_batches()
-    selected = args.batch or sorted(batch_to_unit)
-    configs = db.load_alarm_config({batch: batch_to_unit[batch] for batch in selected})
+    devices_selected = devices.normalize_devices(args.device)
+    device_batches = devices.load_device_batches(devices_selected)
+    unit_by_batch = devices.batch_to_unit(device_batches)
+    batch_device = devices.batch_to_device(device_batches)
+    configs = db.load_alarm_config(unit_by_batch)
+    print(f"设备：{devices_selected}")
+    print(f"批次映射：{ {d: bs for d, bs in device_batches.items()} }")
 
+    rows = None
     if args.hourly_csv:
         print(f"复用小时中间表：{args.hourly_csv}")
         rows = queries.load_hourly_csv(Path(args.hourly_csv))
-    else:
+
+    removed = clean_intermediates()
+    print(f"已清理中间/结果文件：{removed}")
+
+    if args.start and args.end:
         start, end = queries.parse_dt(args.start), queries.parse_dt(args.end)
-        print(f"只读数据库扫描 {start} .. {end}，批次 {selected}")
-        rows = queries.fetch_all_hourly(start, end, args.chunk_days, batch_to_unit, configs, selected)
-        if args.save_hourly:
-            queries.write_hourly_csv(paths.HOURLY_FULL, rows)
-            print(f"小时中间表已缓存：{paths.HOURLY_FULL}")
+    else:
+        start, end = devices.auto_range(device_batches)
+        print(f"自动时间范围：{start} .. {end}")
+
+    if rows is None:
+        rows = queries.fetch_all_hourly(start, end, args.chunk_days, unit_by_batch, configs, sorted(unit_by_batch))
+    queries.write_hourly_csv(paths.HOURLY_FULL, rows)
+    print(f"小时中间表已缓存：{paths.HOURLY_FULL}")
 
     io.write_thresholds(paths.COMMON_RESULTS / "报警阈值配置.csv", configs)
 
     hourly_levels, hour_segments, cycles = pressure_algorithm.run(rows, configs)
-    pressure_algorithm.write_results(paths.PRESSURE_RESULTS, hourly_levels, hour_segments, cycles)
-
     events = vibration_algorithm.run(rows, configs)
-    vibration_algorithm.write_results(paths.VIBRATION_RESULTS, events)
-
     print(f"小时特征：{len(rows):,} 行")
     print(f"润滑油压力小时等级：{len(hourly_levels):,} 行")
     print(f"润滑油压力小时阶段段：{len(hour_segments):,} 段")
     print(f"润滑油压力周期：{len(cycles):,} 个")
     print(f"振动异常段：{len(events):,} 段")
 
+    pressure_algorithm.write_results(paths.PRESSURE_RESULTS, hourly_levels, hour_segments, cycles)
+    vibration_algorithm.write_results(paths.VIBRATION_RESULTS, events)
+    print(f"结果已写入本地：{paths.RESULTS}")
+
+    if args.store == "db":
+        clean_segments = io.clean_rows(hour_segments)
+        clean_events = io.clean_rows(events)
+        alarms = alarm_store.build_alarms(clean_segments, clean_events, batch_device)
+        result = alarm_store.write_full(alarms, devices_selected, start, end, dry_run=args.dry_run)
+        print(f"数据库写入：{result}")
+
 
 if __name__ == "__main__":
     main()

models/service/README.md

@@ -0,0 +1,63 @@
+# 实时告警定时服务
+
+所有平台都是每隔 5 分钟执行一次：
+
+```bash
+python models/realtime_service.py --once
+```
+
+扫描结果按“重叠或间隔不超过 6 小时”的规则合并进 `compressor_alarm`，
+同一等级连续段不会因为跨小时被拆成多条。
+
+## 命令行定时（三平台通用）
+
+```bash
+# 前台循环，每 300 秒一次，Ctrl+C 退出
+python models/realtime_service.py --loop --interval 300
+```
+
+## macOS（launchd）
+
+```bash
+bash models/service/mac/install.sh      # 安装并立即加载
+bash models/service/mac/uninstall.sh    # 卸载
+```
+
+- 安装后 plist 位于 `~/Library/LaunchAgents/com.compressor.alarm.plist`。
+- 日志：`models/service/realtime.out.log` / `realtime.err.log`。
+- 可用 `PYTHON=/path/to/python bash models/service/mac/install.sh` 指定解释器。
+
+## Linux（systemd，需要 sudo）
+
+```bash
+sudo bash models/service/linux/install.sh     # 安装并启用 timer
+sudo bash models/service/linux/uninstall.sh   # 卸载
+```
+
+- 查看：`systemctl status compressor-alarm.timer`、`journalctl -u compressor-alarm.service`。
+
+crontab 备选：
+
+```cron
+*/5 * * * * cd /path/to/Compressor && /usr/bin/python3 models/realtime_service.py --once >> models/service/realtime.log 2>&1
+```
+
+## Windows（计划任务）
+
+```bat
+models\service\windows\install_service.bat
+models\service\windows\uninstall_service.bat
+```
+
+- 任务名 `CompressorAlarm`，每 5 分钟运行一次。
+- 手动触发：`schtasks /Run /TN CompressorAlarm`。
+- 可用 `set PYTHON=C:\path\python.exe` 后再运行安装脚本指定解释器。
+
+## 冒烟测试
+
+```bash
+# 顺序模拟多个时刻运行，结束后自动回滚数据库
+python models/realtime_service.py --simulate "2026-02-03 20:00:00,2026-02-04 00:00:00,2026-02-05 00:00:00"
+```
+
+加 `--no-rollback` 可保留冒烟写入（默认回滚）。

models/service/linux/compressor-alarm.service

@@ -0,0 +1,7 @@
+[Unit]
+Description=PKS realtime compressor alarm scan (once)
+
+[Service]
+Type=oneshot
+WorkingDirectory=__REPO__
+ExecStart=__PYTHON__ __REPO__/models/realtime_service.py --once

models/service/linux/compressor-alarm.timer

@@ -0,0 +1,11 @@
+[Unit]
+Description=Run PKS realtime compressor alarm scan every 5 minutes
+
+[Timer]
+OnBootSec=1min
+OnUnitActiveSec=5min
+AccuracySec=30s
+Unit=compressor-alarm.service
+
+[Install]
+WantedBy=timers.target

models/service/linux/install.sh

@@ -0,0 +1,15 @@
+#!/usr/bin/env bash
+# 安装 Linux systemd 定时服务（需要 sudo）。
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO="$(cd "$SCRIPT_DIR/../../.." && pwd)"
+PYTHON="${PYTHON:-$(command -v python3)}"
+SYSTEMD_DIR="/etc/systemd/system"
+
+sudo sed -e "s|__PYTHON__|$PYTHON|g" -e "s|__REPO__|$REPO|g" \
+    "$SCRIPT_DIR/compressor-alarm.service" | sudo tee "$SYSTEMD_DIR/compressor-alarm.service" >/dev/null
+sudo cp "$SCRIPT_DIR/compressor-alarm.timer" "$SYSTEMD_DIR/compressor-alarm.timer"
+sudo systemctl daemon-reload
+sudo systemctl enable --now compressor-alarm.timer
+echo "已安装 compressor-alarm.timer（每 5 分钟运行）。查看：systemctl list-timers compressor-alarm.timer"

models/service/linux/uninstall.sh

@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+# 卸载 Linux systemd 定时服务（需要 sudo）。
+set -euo pipefail
+sudo systemctl disable --now compressor-alarm.timer 2>/dev/null || true
+sudo rm -f /etc/systemd/system/compressor-alarm.timer /etc/systemd/system/compressor-alarm.service
+sudo systemctl daemon-reload
+echo "已卸载 compressor-alarm.timer"

models/service/mac/com.compressor.alarm.plist

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<!-- 模板：安装脚本会把 __PYTHON__ / __REPO__ 替换为实际路径 -->
+<plist version="1.0">
+<dict>
+    <key>Label</key>
+    <string>com.compressor.alarm</string>
+    <key>ProgramArguments</key>
+    <array>
+        <string>__PYTHON__</string>
+        <string>__REPO__/models/realtime_service.py</string>
+        <string>--once</string>
+    </array>
+    <key>WorkingDirectory</key>
+    <string>__REPO__</string>
+    <key>StartInterval</key>
+    <integer>300</integer>
+    <key>RunAtLoad</key>
+    <true/>
+    <key>StandardOutPath</key>
+    <string>__REPO__/models/service/realtime.out.log</string>
+    <key>StandardErrorPath</key>
+    <string>__REPO__/models/service/realtime.err.log</string>
+</dict>
+</plist>

models/service/mac/install.sh

@@ -0,0 +1,17 @@
+#!/usr/bin/env bash
+# 安装 macOS launchd 定时服务：每 5 分钟运行一次实时扫描。
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO="$(cd "$SCRIPT_DIR/../../.." && pwd)"
+PYTHON="${PYTHON:-$(command -v python3)}"
+LABEL="com.compressor.alarm"
+TARGET="$HOME/Library/LaunchAgents/$LABEL.plist"
+
+mkdir -p "$HOME/Library/LaunchAgents"
+sed -e "s|__PYTHON__|$PYTHON|g" -e "s|__REPO__|$REPO|g" "$SCRIPT_DIR/$LABEL.plist" > "$TARGET"
+
+launchctl unload "$TARGET" 2>/dev/null || true
+launchctl load "$TARGET"
+echo "已安装：$TARGET（每 300 秒运行一次）"
+echo "查看日志：tail -f \"$REPO/models/service/realtime.out.log\""

models/service/mac/uninstall.sh

@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+# 卸载 macOS launchd 定时服务。
+set -euo pipefail
+LABEL="com.compressor.alarm"
+TARGET="$HOME/Library/LaunchAgents/$LABEL.plist"
+launchctl unload "$TARGET" 2>/dev/null || true
+rm -f "$TARGET"
+echo "已卸载：$TARGET"

models/service/windows/install_service.bat

@@ -0,0 +1,12 @@
+@echo off
+rem 安装 Windows 计划任务：每 5 分钟运行一次实时扫描。
+setlocal
+set "SCRIPT_DIR=%~dp0"
+for %%I in ("%SCRIPT_DIR%..\..\..") do set "REPO=%%~fI"
+if "%PYTHON%"=="" set "PYTHON=python"
+set "TASK=CompressorAlarm"
+set "RUN=%PYTHON% %REPO%\models\realtime_service.py --once"
+schtasks /Create /TN "%TASK%" /SC MINUTE /MO 5 /F /TR "%RUN%"
+echo 已安装计划任务 %TASK%（每 5 分钟运行一次）
+echo 手动运行：schtasks /Run /TN %TASK%
+endlocal

models/service/windows/uninstall_service.bat

@@ -0,0 +1,5 @@
+@echo off
+rem 卸载 Windows 计划任务。
+set "TASK=CompressorAlarm"
+schtasks /Delete /TN "%TASK%" /F
+echo 已卸载计划任务 %TASK%

models/vibration/vibration_algorithm.py

@@ -26,7 +26,7 @@ from statistics import median
 if __package__ in (None, ""):
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
 
-from models.common import db, io, paths, queries
+from models.common import db, devices, io, paths, queries
 from models.common.constants import (
     MIN_HOUR_COVERAGE,
     VIBRATION_BASELINE_DAYS,
@@ -317,34 +317,32 @@ def write_results(output_dir: Path, events):
     io.write_csv(output_dir / "振动异常段.csv", io.clean_rows(events), VIBRATION_FIELDS, {field: field for field in VIBRATION_FIELDS})
 
 
-def load_hourly(args):
+def load_hourly(args, selected_batches, unit_by_batch, configs):
     if args.hourly_csv:
-        return queries.load_hourly_csv(Path(args.hourly_csv))
-    batch_to_unit = db.load_pks_batches()
-    selected = args.batch or sorted(batch_to_unit)
-    configs = db.load_alarm_config({batch: batch_to_unit[batch] for batch in selected})
+        selected_set = set(selected_batches)
+        return [row for row in queries.load_hourly_csv(Path(args.hourly_csv)) if row["批次"] in selected_set]
     start, end = queries.parse_dt(args.start), queries.parse_dt(args.end)
-    return queries.fetch_all_hourly(start, end, args.chunk_days, batch_to_unit, configs, selected)
+    return queries.fetch_all_hourly(start, end, args.chunk_days, unit_by_batch, configs, selected_batches)
 
 
 def main() -> None:
     parser = argparse.ArgumentParser(description="PKS 振动异常段算法（独立于压力）")
-    parser.add_argument("--batch", nargs="+", type=int)
+    parser.add_argument("--device", nargs="+", default=list(devices.DEVICE_CODES), help="压缩机编号，如 7# 8# 9#")
     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("--hourly-csv", default=None)
     parser.add_argument("--output-dir", default=str(paths.VIBRATION_RESULTS))
     args = parser.parse_args()
-    batch_to_unit = db.load_pks_batches()
-    selected = args.batch or sorted(batch_to_unit)
-    configs = db.load_alarm_config({batch: batch_to_unit[batch] for batch in selected})
-    rows = load_hourly(args)
-    if args.batch:
-        selected_set = set(selected)
-        rows = [row for row in rows if row["批次"] in selected_set]
+    devices_selected = devices.normalize_devices(args.device)
+    device_batches = devices.load_device_batches(devices_selected)
+    unit_by_batch = devices.batch_to_unit(device_batches)
+    configs = db.load_alarm_config(unit_by_batch)
+    selected_batches = sorted(unit_by_batch)
+    rows = load_hourly(args, selected_batches, unit_by_batch, configs)
     events = run(rows, configs)
     write_results(Path(args.output_dir), events)
+    print(f"设备：{devices_selected}")
     print(f"小时特征：{len(rows):,} 行")
     print(f"振动异常段：{len(events):,} 段")
     print(f"输出目录：{args.output_dir}")