Compressor/backend/app/services/data_service.py

from __future__ import annotations

import re
import json
from bisect import bisect_left
from collections import OrderedDict
from datetime import datetime, timedelta
from functools import lru_cache
from time import monotonic
from typing import Any, Callable

import numpy as np

from ..algorithms.cycles import (
    DetectedCycle,
    PULSES_PER_REVOLUTION,
    build_angle_vector,
    detect_cycles,
    downsample_indices,
)
from ..config import settings
from ..db import get_connection


MEASUREMENT_TYPES = ("位移", "加速度", "压力")
MEASUREMENT_COLORS = {
    "压力": "#e4572e",
    "位移": "#1f7a8c",
    "加速度": "#7b61a8",
}
DEVICE_POINTS = ("压力盖侧", "压力轴侧", "活塞杆沉降", "十字头振动", "自由端振动", "驱动端振动")
PRESSURE_SIDES = ("压力盖侧", "压力轴侧")
DEVICE_POINT_TO_TYPE = {
    "压力盖侧": "压力",
    "压力轴侧": "压力",
    "活塞杆沉降": "位移",
    "十字头振动": "加速度",
    "自由端振动": "加速度",
    "驱动端振动": "加速度",
}
PRIMARY_DEVICE_POINT = "压力盖侧"
PHASES = (
    ("排气", "#7b1fa2", 0.0, 120.0),
    ("压缩", "#c62828", 120.0, 195.0),
    ("膨胀", "#1565c0", 195.0, 270.0),
    ("进气", "#2e7d32", 270.0, 330.0),
)
ANNOTATION_LABELS = ("正常", "异常")

OIL_PRESSURE_ALARM_TYPE = "润滑油压力低"
CRUCIFORM_FAULT_ALARM_TYPE = "十字头故障"
VALVE_FAULT_ALARM_TYPE = "气阀故障"
OIL_PRESSURE_DEVICE_PART = "润滑油"
OIL_PRESSURE_DEVICE_POINT = "压力"
VIBRATION_DEVICE_POINT = "振动"
OIL_PRESSURE_SCAN_DAYS = 14

# PKS 全场点位：机组号 -> compressor_data_import 中 directory_path 为 "{号}#"
# 且 data_type='PKS' 的所有导入批次（同一机组可能有多个时间段批次）。
UNIT_PKS_DIRECTORY_SUFFIX = "#"
# 全场点位(PKS)每个时间点的曲线窗口：以采样时刻为中心的前后各 7.5 分钟。
PKS_WINDOW_SECONDS = 15 * 60
# 仅 PKS 时间点返回给时间条的最大锚点数：超过后按等步长均匀采样（首尾必保）。
PKS_STRIP_MAX_ANCHORS = 5000
_SITE_POINT_PATTERN = re.compile(r"^YSJ([4-9])_([1-9]|[1-3][0-9]|4[0-4])$")
_SITE_INDEX_MAX = 44
# 不作为全场点位(PKS)下拉候选的序号（各机组不同：备用/累计运行等信号）。
_EXCLUDED_SITE_INDEXES: dict[str, frozenset[int]] = {
    "4": frozenset({31, 32, 33, 34, 35, 36}),
    "5": frozenset(),
    "6": frozenset(),
    "7": frozenset({33, 34, 35, 41}),
    "8": frozenset({33, 34, 35, 41}),
    "9": frozenset({33, 34, 35, 41}),
}
# 仅这些机组的 YSJ_41 是转速/运行信号，可用于停机判断；4/5/6 的 YSJ_41 非转速，
# 无法据此判断停机，一律按运转处理。
_RUNNING_SIGNAL_UNITS = frozenset({"7", "8", "9"})

# Extra samples fetched past cycle_end so detect_cycles can see the zero marker
# that closes the first cycle (cycle_end is exclusive, the marker sits at it).
FIRST_CYCLE_PAD = 64

CYLINDER_BORE_MM = {
    "一缸": 360.0,
    "二缸": 490.0,
    "三缸": 390.0,
    "四缸": 490.0,
    "五缸": 390.0,
    "六缸": 490.0,
}
PISTON_STROKE_MM = 148.0
CONNECTING_ROD_LENGTH_MM = 460.0
CLEARANCE_VOLUME_L_BY_BORE = {
    490.0: 1.51,
    390.0: 0.74,
    360.0: 0.62,
}

DEMO_POINT_NAME = "7号机组一缸压力盖侧"
DEMO_DEVICE_PART = "7号机组一缸"
DEMO_START = datetime(2026, 4, 12, 8, 0, 0)
DEMO_POINT_COUNT = 72
DEMO_SAMPLE_COUNT = 32768
DEMO_REVOLUTION_SAMPLES = 800
DEMO_ID_BY_TYPE = {name: 100000 + index * 1000 for index, name in enumerate(MEASUREMENT_TYPES)}


def _time_string(value: Any) -> str:
    if isinstance(value, datetime):
        return value.strftime("%Y-%m-%d %H:%M:%S")
    return str(value)


def _parse_time(value: str | None) -> datetime | None:
    if not value:
        return None
    return datetime.fromisoformat(value.replace("Z", "+00:00").replace("T", " "))


def _safe_float(value: Any) -> float | None:
    if value is None:
        return None
    number = float(value)
    return number if np.isfinite(number) else None


def _stored_first_cycle(samples: np.ndarray) -> DetectedCycle | None:
    """Build a cycle from wave_sample_one, which contains one cycle only."""
    if samples.ndim != 2 or samples.shape[1] < 3 or len(samples) < 2:
        return None
    start = 0
    end = len(samples)
    angle = np.linspace(0.0, 360.0, end - start, endpoint=False)
    trigger_offsets = tuple(
        int(offset)
        for offset in np.flatnonzero(
            np.nan_to_num(samples[:, 2], nan=-np.inf) >= 30.0,
        )
        if offset == 0 or samples[offset - 1, 2] < 30.0
    )
    return DetectedCycle(
        number=1,
        start_offset=start,
        end_offset=end,
        angle=angle,
        signal=samples[start:end, 1].copy(),
        trigger_offsets=trigger_offsets,
    )


def _sample_first_cycle_360(samples: np.ndarray) -> list[tuple[int, float]]:
    """Resample the stored first cycle into 360 integer-angle points.

    Each integer angle 0..359 keeps the first raw sample whose crank angle is at
    or after that integer value (the "取第一个值" rule from showPV).
    """
    if samples.ndim != 2 or samples.shape[1] < 2 or len(samples) < 2:
        return []
    count = len(samples)
    angle = np.linspace(0.0, 360.0, count, endpoint=False)
    indices = np.searchsorted(angle, np.arange(360, dtype=float), side="left")
    indices = np.clip(indices, 0, count - 1)
    return [
        (int(deg), float(samples[int(idx), 1]))
        for deg, idx in enumerate(indices)
    ]


def _detect_source_cycles(
    samples: np.ndarray,
    single_cycle: bool,
) -> tuple[list[DetectedCycle], dict[str, Any]]:
    """Detect cycles: wave_sample_one holds exactly one stored cycle already."""
    if single_cycle:
        stored = _stored_first_cycle(samples)
        if stored is None:
            return [], {"completeCycleCount": 0}
        return [stored], {"completeCycleCount": 1, "storedFirstCycle": 1}
    return detect_cycles(samples)


def _series_extent(items: list[dict[str, Any]]) -> tuple[float, float]:
    """Whole-window min/max over a series' finite raw values."""
    if not items:
        return (0.0, 1.0)
    values = np.fromiter((item["rawValue"] for item in items), dtype=float, count=len(items))
    values = values[np.isfinite(values)]
    if values.size == 0:
        return (0.0, 1.0)
    return (float(values.min()), float(values.max()))


def _annotation_dict(row: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": int(row["id"]),
        "waveFileId": int(row["wave_file_id"]),
        "label": row["label"],
        "periodStart": int(row["period_start"]),
        "periodEnd": int(row["period_end"]),
        "sampleIndexStart": int(row["sample_index_start"]),
        "sampleIndexEnd": int(row["sample_index_end"]),
    }


def _alarm_dict(row: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": int(row["id"]),
        "deviceCode": row["device_code"] or "",
        "devicePart": row["device_part"] or "",
        "devicePoint": row["device_point"] or "",
        "alarmType": row["alarm_type"] or "",
        "alarmDes": row["alarm_des"] or "",
        "status": int(row.get("status") or 0),
        "alarmLevel": row.get("alarm_level"),
        "scanStartTime": _time_string(row["scan_start_time"]) if row.get("scan_start_time") else "",
        "scanEndTime": _time_string(row["scan_end_time"]) if row.get("scan_end_time") else "",
        "alarmInfoJson": row.get("alarm_info_json") or "",
        "alarmTimeStart": _time_string(row["alarm_time_start"]),
        "alarmTimeEnd": _time_string(row["alarm_time_end"]),
    }


def _validate_device_points(
    values: list[str] | tuple[str, ...] | None,
    *,
    allow_empty: bool = False,
) -> list[str]:
    """校验波形点位。None 表示未传参 → 回退为全部点位；显式空列表仅在 allow_empty 时放行。"""
    selected = list(DEVICE_POINTS) if values is None else list(values)
    invalid = [value for value in selected if value not in DEVICE_POINTS]
    if invalid:
        raise ValueError(f"不支持的测试点位：{'、'.join(invalid)}")
    result = [value for value in DEVICE_POINTS if value in selected]
    if not result and not allow_empty:
        raise ValueError("请至少选择一个测试点位")
    return result


def _primary_device_point(points: list[str]) -> str:
    """Prefer pressure cap, then pressure shaft, then any selected point."""
    for point in ("压力盖侧", "压力轴侧"):
        if point in points:
            return point
    return points[0]


def _unit_number(device_part: str) -> str | None:
    """从 机组与部位 提取机组号（4~9），非 4~9 机组返回 None。"""
    match = re.match(r"^([4-9])号机组", device_part.strip())
    return match.group(1) if match else None


def _site_point_column(item_name: str, unit: str | None) -> str | None:
    """校验全场点位名称并映射到 pks_long_sample 的列名（如 YSJ7_3 -> YSJ_3）。"""
    match = _SITE_POINT_PATTERN.match(item_name)
    if not match or match.group(1) != unit:
        return None
    return f"YSJ_{match.group(2)}"


class DataService:
    # 数据库失败后的重试冷却时间（秒）。超过该时间后自动重连数据库，
    # 避免一次网络抖动就把服务永久锁死在演示数据模式。
    DB_RETRY_COOLDOWN = 30.0

    def __init__(self) -> None:
        self._db_failed = settings.demo_mode == "always"
        self._db_failed_at = monotonic() if self._db_failed else 0.0
        self._last_db_error = ""
        self._demo_annotations: dict[int, dict[str, Any]] = {}
        self._demo_annotation_seq = 1
        # 机组号 -> pks_long_sample 该机组所有批次 [min_time, max_time]，进程内只查一次。
        self._pks_batch_bounds: dict[str, tuple[datetime, datetime] | None] = {}
        # 机组号 -> compressor_data_import 中该机组 PKS 批次 id 列表，进程内只查一次。
        self._pks_batch_ids_cache: dict[str, list[int]] = {}

    @property
    def source(self) -> str:
        return "demo" if self._db_failed else "database"

    @property
    def last_db_error(self) -> str:
        return self._last_db_error

    def _run_with_fallback(
        self,
        database_function: Callable[[], Any],
        demo_function: Callable[[], Any],
    ) -> tuple[Any, str]:
        if self._db_failed:
            if settings.demo_mode == "always":
                return demo_function(), "demo"
            if monotonic() - self._db_failed_at < self.DB_RETRY_COOLDOWN:
                return demo_function(), "demo"
            # 冷却结束，重新尝试数据库，数据库恢复后可自动切回真实数据。
        try:
            result = database_function()
            self._db_failed = False
            self._last_db_error = ""
            return result, "database"
        except Exception as error:
            if settings.demo_mode == "never":
                raise
            self._db_failed = True
            self._db_failed_at = monotonic()
            self._last_db_error = str(error)
            return demo_function(), "demo"

    def query_options(self) -> dict[str, Any]:
        def database_query():
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        """
                        SELECT device_part, device_point, measurement_type,
                               MAX(sample_time) AS max_time,
                               MIN(sample_time) AS min_time,
                               COUNT(*) AS file_count
                        FROM wave_file
                        WHERE rpm > 0 AND device_part <> '' AND device_point <> ''
                        GROUP BY device_part, device_point, measurement_type
                        ORDER BY device_part, device_point
                        """,
                    )
                    rows = cursor.fetchall()
            return [
                {
                    "devicePart": row["device_part"],
                    "devicePoint": row["device_point"],
                    "measurementType": row["measurement_type"],
                    "minTime": _time_string(row["min_time"]),
                    "maxTime": _time_string(row["max_time"]),
                    "fileCount": int(row["file_count"]),
                }
                for row in rows
            ]

        rows, source = self._run_with_fallback(database_query, self._demo_options)
        device_parts = list(dict.fromkeys(row["devicePart"] for row in rows))
        if source == "database":
            device_parts.extend(self._pks_only_device_parts(device_parts))
        return {
            "source": source,
            "measurementTypes": list(MEASUREMENT_TYPES),
            "deviceParts": device_parts,
            "devicePoints": list(DEVICE_POINTS),
            "devicePointToType": dict(DEVICE_POINT_TO_TYPE),
            "options": rows,
            "notice": self._source_notice(source),
        }

    def _pks_only_device_parts(self, existing: list[str]) -> list[str]:
        """有 PKS 数据但 wave_file 无部位的机组，补一个纯 PKS 的"{号}号机组"入口。"""
        existing_units = {unit for part in existing if (unit := _unit_number(part)) is not None}
        try:
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        "SELECT DISTINCT directory_path FROM compressor_data_import "
                        "WHERE data_type = 'PKS' AND directory_path REGEXP '^[4-9]#$'",
                    )
                    directories = [str(row["directory_path"]) for row in cursor.fetchall()]
        except Exception:
            return []
        parts: list[str] = []
        for directory in directories:
            unit = directory.rstrip(UNIT_PKS_DIRECTORY_SUFFIX).strip()
            if unit.isdigit() and unit not in existing_units:
                parts.append(f"{unit}号机组")
        return parts

    def abnormal_counts(self) -> dict[str, int]:
        """每个 point_name 的异常文件数量（tspluse_status > 0）。"""

        def database_query():
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        """
                        SELECT point_name, COUNT(*) AS cnt
                        FROM wave_file
                        WHERE tspluse_status > 0 AND point_name <> ''
                        GROUP BY point_name
                        """,
                    )
                    rows = cursor.fetchall()
            return {row["point_name"]: int(row["cnt"]) for row in rows}

        def demo_query():
            return {}

        result, _source = self._run_with_fallback(database_query, demo_query)
        return result

    def tspluse_ruler(self) -> dict[str, Any]:
        """压力部位 tspluse_status 的全局标尺，进入页面时只查询一次。"""

        def database_query():
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        """
                        SELECT MIN(tspluse_status) AS min_status,
                               MAX(tspluse_status) AS max_status
                        FROM wave_file
                        WHERE rpm > 0 AND measurement_type = '压力'
                        """,
                    )
                    row = cursor.fetchone()
            return {
                "min": int(row["min_status"]) if row and row["min_status"] is not None else 0,
                "max": int(row["max_status"]) if row and row["max_status"] is not None else 0,
            }

        def demo_query():
            return {"min": 0, "max": 0}

        result, source = self._run_with_fallback(database_query, demo_query)
        result["source"] = source
        return result

    def _pks_batch_ids(self, unit: str) -> list[int]:
        """机组 PKS 导入批次 id 列表（directory_path = "{号}#" 且 data_type='PKS'）。"""
        cached = self._pks_batch_ids_cache.get(unit)
        if cached is not None:
            return cached
        with get_connection() as connection:
            with connection.cursor() as cursor:
                cursor.execute(
                    "SELECT id FROM compressor_data_import "
                    "WHERE data_type = 'PKS' AND directory_path = %s ORDER BY id",
                    (f"{unit}{UNIT_PKS_DIRECTORY_SUFFIX}",),
                )
                batches = [int(row["id"]) for row in cursor.fetchall()]
        self._pks_batch_ids_cache[unit] = batches
        return batches

    @staticmethod
    def _in_clause(values: list[Any]) -> str:
        return ", ".join(["%s"] * len(values))

    def _pks_batch_range(self, unit: str) -> tuple[datetime, datetime] | None:
        """机组 pks 所有批次的时间覆盖范围（仅查一次并缓存）。"""
        cached = self._pks_batch_bounds.get(unit)
        if cached is not None:
            return cached
        batches = self._pks_batch_ids(unit)
        if not batches:
            self._pks_batch_bounds[unit] = None
            return None
        with get_connection() as connection:
            with connection.cursor() as cursor:
                cursor.execute(
                    "SELECT MIN(sample_time) AS lo, MAX(sample_time) AS hi "
                    f"FROM pks_long_sample WHERE import_batch_id IN ({self._in_clause(batches)})",
                    tuple(batches),
                )
                row = cursor.fetchone()
        bounds = (row["lo"], row["hi"]) if row and row["lo"] is not None else None
        self._pks_batch_bounds[unit] = bounds
        return bounds

    def site_points(self, device_part: str) -> dict[str, Any]:
        """机组(4~9)的全场点位记录（site_point 表中 YSJ{机组号}_*）。

        附带该机组 pks 批次的时间覆盖范围，供“仅 PKS 点位”组合自动赋值
        开始/结束时间。
        """
        unit = _unit_number(device_part)

        def database_query():
            if unit is None:
                return {"items": [], "minTime": None, "maxTime": None}
            allowed = {
                f"YSJ{unit}_{n}"
                for n in range(1, _SITE_INDEX_MAX + 1)
                if n not in _EXCLUDED_SITE_INDEXES.get(unit, frozenset())
            }
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        "SELECT ItemName, ItemDescription FROM site_point WHERE ItemName LIKE %s",
                        (f"YSJ{unit}\\_%",),
                    )
                    rows = cursor.fetchall()
            items = [
                {"itemName": row["ItemName"], "itemDescription": row["ItemDescription"] or ""}
                for row in rows
                if row["ItemName"] in allowed
            ]
            items.sort(key=lambda item: int(item["itemName"].rsplit("_", 1)[1]))
            bounds = self._pks_batch_range(unit)
            return {
                "items": items,
                "minTime": _time_string(bounds[0]) if bounds else None,
                "maxTime": _time_string(bounds[1]) if bounds else None,
            }

        def demo_query():
            return {"items": [], "minTime": None, "maxTime": None}

        result, source = self._run_with_fallback(database_query, demo_query)
        return {
            "source": source,
            "unit": unit,
            "items": result["items"],
            "minTime": result["minTime"],
            "maxTime": result["maxTime"],
            "notice": self._source_notice(source),
        }

    def _attach_site_values(self, points: list[dict[str, Any]], unit: str, site_points: list[str]) -> None:
        """把 pks_long_sample 的最近邻值挂到每个时间点的 siteValues 上。

        pks 数据 5 秒一条、wave_file 15 分钟一条。按用户口径做分钟/5秒级对齐：
        把 wave 采样时刻四舍五入到最近的 5 秒格点，再用一次 ``sample_time IN (...)``
        精确取数（结果行数 = 时间点数），避免把整段 pks 拉出来。
        """
        columns = [
            (item_name, column)
            for item_name in site_points
            if (column := _site_point_column(item_name, unit)) is not None
        ]
        batches = self._pks_batch_ids(unit)
        if not columns or not points or not batches:
            for point in points:
                point.setdefault("siteValues", {})
            return
        rounded: list[datetime] = []
        for point in points:
            timestamp = datetime.strptime(point["sampleTime"], "%Y-%m-%d %H:%M:%S")
            rounded.append(datetime.fromtimestamp(round(timestamp.timestamp() / 5.0) * 5))
        chunk_size = 1000
        select_expr = ", ".join(f"`{column}`" for _item, column in columns)
        for index in range(0, len(rounded), chunk_size):
            chunk = rounded[index:index + chunk_size]
            placeholders = ", ".join(["%s"] * len(chunk))
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        f"SELECT sample_time, {select_expr} FROM pks_long_sample "
                        f"WHERE import_batch_id IN ({self._in_clause(batches)}) "
                        f"AND sample_time IN ({placeholders})",
                        (*batches, *chunk),
                    )
                    rows = cursor.fetchall()
            by_time: dict[datetime, dict[str, Any]] = {
                row["sample_time"]: row for row in rows
            }
            for point_index in range(index, min(index + chunk_size, len(rounded))):
                target_time = rounded[point_index]
                row = by_time.get(target_time)
                if row is None:
                    continue
                site_values = points[point_index].setdefault("siteValues", {})
                for item_name, column in columns:
                    value = row[column]
                    site_values[item_name] = float(value) if value is not None else None

    def _build_site_series(
        self,
        site_points: list[str],
        unit: str | None,
        points: list[dict[str, Any]],
    ) -> dict[str, Any]:
        """把每个时间点扩展为前后各 7.5 分钟的 pks 5s 曲线段。

        每个选中点位(PKS)在该窗口内的每个时间点不再只画单值，而是以该
        时间点采样时刻为中心，取 [t-7.5min, t+7.5min) 的原始 5s 数据映射
        到该时间点在 x 轴占据的格子（索引 slot ~ slot+1）内连成一小段曲线。
        相邻时间点若恰好间隔 15 分钟，则相邻窗口首尾衔接、无重叠。
        数据按列做一次整段范围查询，再按时间点二分切段。
        """
        columns = [
            (item_name, column)
            for item_name in site_points
            if (column := _site_point_column(item_name, unit)) is not None
        ]
        centers: list[tuple[int, datetime]] = []
        for index, point in enumerate(points):
            try:
                timestamp = datetime.strptime(point["sampleTime"], "%Y-%m-%d %H:%M:%S")
            except (TypeError, ValueError):
                continue
            centers.append((index, timestamp))
        batches = self._pks_batch_ids(unit) if unit else []
        if not columns or not centers or not batches:
            return {"points": site_points, "series": []}
        half = timedelta(seconds=PKS_WINDOW_SECONDS // 2)
        query_start = min(timestamp for _index, timestamp in centers) - half
        query_end = max(timestamp for _index, timestamp in centers) + half
        select_expr = ", ".join(f"`{column}`" for _item, column in columns)
        with get_connection() as connection:
            with connection.cursor() as cursor:
                cursor.execute(
                    f"SELECT sample_time, {select_expr} FROM pks_long_sample "
                    f"WHERE import_batch_id IN ({self._in_clause(batches)}) "
                    "AND sample_time >= %s AND sample_time < %s "
                    "ORDER BY sample_time",
                    (*batches, query_start, query_end),
                )
                rows = cursor.fetchall()
        # 同一机组可能有多个时间段批次，按 sample_time 去重，避免边界重叠重复绘制。
        dedup: dict[datetime, dict[str, Any]] = {}
        for row in rows:
            dedup.setdefault(row["sample_time"], row)
        ordered = [dedup[timestamp] for timestamp in sorted(dedup)]
        sample_times = [row["sample_time"] for row in ordered]
        series = []
        for item_name, column in columns:
            values = [row[column] for row in ordered]
            data: list[dict[str, Any]] = []
            for index, center in centers:
                low = bisect_left(sample_times, center - half)
                high = bisect_left(sample_times, center + half)
                for pos in range(low, high):
                    raw_value = values[pos]
                    if raw_value is None:
                        continue
                    value = float(raw_value)
                    if not np.isfinite(value):
                        continue
                    x = index + 0.5 + (sample_times[pos] - center).total_seconds() / PKS_WINDOW_SECONDS
                    data.append(
                        {
                            "value": [x, value],
                            "x": x,
                            "rawValue": value,
                            "sampleTime": _time_string(sample_times[pos]),
                        },
                    )
            series.append({"itemName": item_name, "data": data})
        return {"points": site_points, "series": series}

    def time_points(
        self,
        device_part: str,
        device_points: list[str] | None,
        min_time: str | None,
        max_time: str | None,
        include_stopped: bool = False,
        min_status: int | None = None,
        status_filter: list[str] | None = None,
        site_points: list[str] | None = None,
    ) -> dict[str, Any]:
        if not device_part.strip():
            raise ValueError("机组与部位不能为空")
        status_filters = {value for value in (status_filter or [])}
        unknown = status_filters - {"abnormal", "no_cycle"}
        if unknown:
            raise ValueError(f"不支持的状态筛选：{'、'.join(sorted(unknown))}")
        start = _parse_time(min_time)
        end = _parse_time(max_time)
        if start and end and start > end:
            raise ValueError("开始时间不能晚于结束时间")
        unit = _unit_number(device_part)
        site_list = list(site_points or [])
        site_columns: list[tuple[str, str]] = []
        if unit:
            site_columns = [
                (item, column)
                for item in site_list
                if (column := _site_point_column(item, unit)) is not None
            ]
        if device_points is None or device_points == []:
            selected_points = (
                [] if (unit and site_columns) else _validate_device_points(None)
            )
        else:
            selected_points = _validate_device_points(device_points, allow_empty=True)
        pks_only = bool(unit and site_columns and not selected_points)
        if not selected_points and not pks_only:
            raise ValueError("请至少选择一个测试点位")
        point_names = [f"{device_part}{point}" for point in selected_points]

        if pks_only:
            points, source = self._run_with_fallback(
                lambda: self._pks_time_points(unit, site_columns, start, end, include_stopped),
                lambda: [],
            )
            return {
                "source": source,
                "devicePart": device_part,
                "devicePoints": [],
                "total": len(points),
                "points": points,
                "referencePoints": [],
                "notice": self._source_notice(source),
            }

        def database_query():
            point_placeholders = ", ".join(["%s"] * len(point_names))
            clauses = [
                f"point_name IN ({point_placeholders})",
            ]
            params: list[Any] = list(point_names)
            if not include_stopped:
                clauses.append("rpm > 0")
            if min_status is not None and min_status > 0 and "no_cycle" not in status_filters:
                clauses.append("tspluse_status >= %s")
                params.append(min_status)
            if "abnormal" in status_filters and "no_cycle" in status_filters:
                clauses.append("(tspluse_status > 0 OR tspluse_status = -1)")
            elif "abnormal" in status_filters:
                clauses.append("tspluse_status > 0")
            elif "no_cycle" in status_filters:
                clauses.append("tspluse_status = -1")
            if start:
                clauses.append("sample_time >= %s")
                params.append(start)
            if end:
                clauses.append("sample_time <= %s")
                params.append(end)
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        f"""
                        SELECT id, point_name, device_point, measurement_type,
                               sample_time, sample_count, sample_frequency_hz,
                               rpm, tspluse_status
                        FROM wave_file
                        WHERE {' AND '.join(clauses)}
                        ORDER BY sample_time ASC, id ASC
                        """,
                        params,
                    )
                    rows = cursor.fetchall()
            reference_rows: list[dict[str, Any]] = []
            if status_filters and rows:
                primary = _primary_device_point(selected_points)
                reference_where = [
                    "point_name = %s",
                    "measurement_type = '压力'",
                    "tspluse_status = 0",
                ]
                reference_params: list[Any] = [f"{device_part}{primary}"]
                if not include_stopped:
                    reference_where.append("rpm > 0")
                if start:
                    reference_where.append("sample_time >= %s")
                    reference_params.append(start)
                if end:
                    reference_where.append("sample_time <= %s")
                    reference_params.append(end)
                reference_params.append(rows[0]["sample_time"])
                with get_connection() as connection:
                    with connection.cursor() as cursor:
                        cursor.execute(
                            f"""
                            SELECT id, point_name, device_point, measurement_type,
                                   sample_time, sample_count, sample_frequency_hz,
                                   rpm, tspluse_status
                            FROM wave_file
                            WHERE {' AND '.join(reference_where)}
                            ORDER BY ABS(TIMESTAMPDIFF(SECOND, sample_time, %s)) ASC
                            LIMIT 1
                            """,
                            reference_params,
                        )
                        reference_rows = cursor.fetchall()
            return self._group_time_points(rows), self._group_time_points(reference_rows)

        def demo_query():
            return self._demo_time_points(device_part, selected_points, start, end), []

        result, source = self._run_with_fallback(database_query, demo_query)
        points, reference = result
        # 全场点位(PKS)数据是辅助层：任意失败都静默跳过，不影响主查询。
        try:
            unit = _unit_number(device_part)
            if unit and site_points:
                self._attach_site_values(points, unit, site_points)
        except Exception:
            pass
        for point in points:
            point.setdefault("siteValues", {})
        return {
            "source": source,
            "devicePart": device_part,
            "devicePoints": selected_points,
            "total": len(points),
            "points": points,
            "referencePoints": reference,
            "notice": self._source_notice(source),
        }

    def faults(
        self,
        device_part: str,
        min_time: str | None,
        max_time: str | None,
    ) -> dict[str, Any]:
        """Return compressor_fault rows for the unit of ``device_part``.

        ``compressor_fault.unit_name`` ("7号机组") is matched against the unit
        prefix of ``wave_file.device_part`` ("7号机组一缸"). Only faults whose
        ``fault_date`` falls inside the requested time range are returned.
        """
        unit = _unit_number(device_part)
        start = _parse_time(min_time)
        end = _parse_time(max_time)

        def database_query():
            if unit is None:
                return []
            clauses = ["(unit_name = %s OR unit_name LIKE %s)"]
            params: list[Any] = [f"{unit}号机组", f"{unit}号机组%"]
            if start:
                clauses.append("fault_date >= DATE(%s)")
                params.append(start)
            if end:
                clauses.append("fault_date <= DATE(%s)")
                params.append(end)
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        f"""
                        SELECT id, unit_name, fault_date, fault_category
                        FROM compressor_fault
                        WHERE {' AND '.join(clauses)}
                        ORDER BY fault_date ASC, id ASC
                        """,
                        params,
                    )
                    rows = cursor.fetchall()
            return [
                {
                    "id": int(row["id"]),
                    "unitName": row["unit_name"] or "",
                    "faultDate": _time_string(row["fault_date"])[:10],
                    "faultCategory": row["fault_category"] or "",
                }
                for row in rows
            ]

        result, source = self._run_with_fallback(database_query, lambda: [])
        return {
            "source": source,
            "unit": unit,
            "faults": result,
            "notice": self._source_notice(source),
        }

    def list_alarms(self, current_time: str | None = None) -> dict[str, Any]:
        """列出在给定时刻命中的告警。

        命中条件为 ``alarm_time_start <= 当前时间 <= alarm_time_end``，
        即告警时间段覆盖“告警当前时间”。
        """
        moment = _parse_time(current_time) or datetime.now()

        def database_query():
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        """
                        SELECT id, 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
                        FROM compressor_alarm
                        WHERE alarm_time_start <= %s AND alarm_time_end >= %s
                        ORDER BY alarm_time_start DESC, id DESC
                        """,
                        (moment, moment),
                    )
                    rows = cursor.fetchall()
            return [_alarm_dict(row) for row in rows]

        result, source = self._run_with_fallback(database_query, lambda: [])
        return {
            "source": source,
            "currentTime": _time_string(moment),
            "alarms": result,
            "notice": self._source_notice(source),
        }

    def scan_alarm(self, payload: dict[str, Any]) -> dict[str, Any]:
        """执行告警算法并按原有协议写入 compressor_alarm。"""
        device_code = str(payload.get("device_code") or "").strip()
        alarm_type = str(payload.get("alarm_type") or "").strip()
        start = _parse_time(payload.get("forecast_time"))
        try:
            hours = float(payload.get("hours") or 0)
        except (TypeError, ValueError):
            raise ValueError("告警时长必须是数字") from None
        if not device_code:
            raise ValueError("压缩机不能为空")
        if not alarm_type:
            raise ValueError("算法不能为空")
        if start is None:
            raise ValueError("预警时间不能为空")
        if hours <= 0:
            raise ValueError("告警时长必须大于 0")
        if alarm_type == VALVE_FAULT_ALARM_TYPE:
            return self._scan_valve_fault(device_code, start, hours)

        if alarm_type == OIL_PRESSURE_ALARM_TYPE:
            # `hours` controls the alarm validity window, not the analysis window.
            # Oil pressure analysis always uses the previous 14 days below.
            analysis = self._analyze_oil_pressure_alarm(device_code, start)
            if analysis is None:
                return {
                    "source": "database",
                    "notice": None,
                    "action": "no_alarm",
                    "id": 0,
                }
            device_part = analysis["device_part"]
            device_point = OIL_PRESSURE_DEVICE_POINT
            alarm_des = analysis["basis"]
            status = 1
            alarm_level = analysis["stage"]
            scan_start = analysis["scan_start"]
            scan_end = analysis["scan_end"]
            alarm_info_json = json.dumps(analysis["info"], ensure_ascii=False, separators=(",", ":"))
        elif alarm_type == CRUCIFORM_FAULT_ALARM_TYPE:
            analysis = self._analyze_cruciform_vibration_alarm(device_code, start)
            if analysis is None:
                return {
                    "source": "database",
                    "notice": None,
                    "action": "no_alarm",
                    "id": 0,
                }
            device_part = analysis["device_part"]
            device_point = VIBRATION_DEVICE_POINT
            alarm_des = analysis["basis"]
            status = 1
            alarm_level = analysis["stage"]
            scan_start = analysis["scan_start"]
            scan_end = analysis["scan_end"]
            alarm_info_json = json.dumps(analysis["info"], ensure_ascii=False, separators=(",", ":"))
        else:
            # Keep the existing endpoint contract for algorithms not yet implemented.
            device_part = "测试组件"
            device_point = "测试组件"
            alarm_des = "测试数据"
            status = 0
            alarm_level = None
            scan_start = None
            scan_end = None
            alarm_info_json = None
        end = start + timedelta(hours=hours)

        def database_query():
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        """
                        SELECT id FROM compressor_alarm
                        WHERE device_code = %s AND device_part = %s
                          AND device_point = %s AND alarm_type = %s
                          AND alarm_time_start <= %s AND alarm_time_end >= %s
                        ORDER BY id ASC
                        LIMIT 1
                        """,
                        (device_code, device_part, device_point, alarm_type, start, start),
                    )
                    row = cursor.fetchone()
                    if row is not None:
                        cursor.execute(
                            """
                            UPDATE compressor_alarm
                            SET alarm_time_end = %s, alarm_des = %s, status = %s,
                                alarm_level = %s, scan_start_time = %s, scan_end_time = %s,
                                alarm_info_json = %s
                            WHERE id = %s
                            """,
                            (end, alarm_des, status, alarm_level, scan_start, scan_end, alarm_info_json, row["id"]),
                        )
                        return {"action": "updated", "id": int(row["id"])}
                    cursor.execute(
                        """
                        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)
                        """,
                            (device_code, device_part, device_point, alarm_type, alarm_des, start, end, status,
                             alarm_level, scan_start, scan_end, alarm_info_json),
                    )
                    return {"action": "inserted", "id": int(cursor.lastrowid)}

        result, source = self._run_with_fallback(
            database_query,
            lambda: {"action": "demo", "id": 0},
        )
        return {"source": source, "notice": self._source_notice(source), **result}

    @staticmethod
    def _valve_segment_values(values: list[float]) -> list[float]:
        if not values:
            return []
        raw_average = sum(values) / len(values)
        return [value for value in values if value >= raw_average * 0.5]

    @classmethod
    def _valve_segment_average(cls, values: list[float]) -> float | None:
        filtered = cls._valve_segment_values(values)
        return sum(filtered) / len(filtered) if filtered else None

    @staticmethod
    def _split_valve_segments(rows: list[dict[str, Any]]) -> list[list[dict[str, Any]]]:
        segments: list[list[dict[str, Any]]] = []
        current: list[dict[str, Any]] = []
        for row in rows:
            if current and row["sample_time"] - current[-1]["sample_time"] > timedelta(days=1):
                segments.append(current)
                current = []
            current.append(row)
        if current:
            segments.append(current)
        return segments

    @staticmethod
    def _previous_valve_segment(
        cursor: Any, part: str, point: str, before: datetime, current_first: datetime
    ) -> tuple[datetime, datetime] | None:
        """Find the nearest preceding running segment lasting at least 24 hours."""
        boundary = before
        boundary_id = 1 << 63
        newer = None
        previous_end = None
        previous_start = None
        while True:
            cursor.execute(
                """
                SELECT id, sample_time FROM wave_file
                WHERE device_part = %s AND device_point = %s
                  AND measurement_type = %s AND rpm > 0
                  AND (sample_time < %s OR (sample_time = %s AND id < %s))
                ORDER BY sample_time DESC, id DESC LIMIT 1000
                """,
                (part, point, "压力", boundary, boundary, boundary_id),
            )
            rows = cursor.fetchall()
            if not rows:
                break
            for row in rows:
                stamp = row["sample_time"]
                if newer is None:
                    previous_end = stamp
                    previous_start = stamp
                elif newer - stamp > timedelta(days=1):
                    if previous_end - previous_start >= timedelta(hours=24):
                        return previous_start, previous_end
                    previous_end = stamp
                    previous_start = stamp
                else:
                    previous_start = stamp
                newer = stamp
            boundary = rows[-1]["sample_time"]
            boundary_id = int(rows[-1]["id"])
            if len(rows) < 1000:
                break
        if (
            previous_start is not None
            and previous_end is not None
            and previous_end - previous_start >= timedelta(hours=24)
        ):
            return previous_start, previous_end
        return None

    def _scan_valve_fault(self, device_code: str, forecast_time: datetime, hours: float) -> dict[str, Any]:
        unit_name = device_code if device_code.endswith("号机组") else f"{device_code.rstrip('#')}号机组"
        scan_start = forecast_time - timedelta(days=14)
        scan_end = forecast_time + timedelta(days=1)

        def database_query():
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        """
                        SELECT f.device_part, f.device_point, f.sample_time, pa.pressure
                        FROM wave_file f
                        LEFT JOIN statistic_pressure_angle pa
                          ON f.id = pa.wave_file_id AND pa.angle = 230
                        WHERE SUBSTRING(f.device_part, 1, 4) = %s
                          AND f.rpm > 0
                          AND f.measurement_type = %s
                          AND f.sample_time >= %s AND f.sample_time < %s
                        ORDER BY f.device_part, f.device_point, f.sample_time
                        """,
                        (unit_name, "压力", scan_start, scan_end),
                    )
                    current_rows = cursor.fetchall()
                    grouped: dict[tuple[str, str], list[dict[str, Any]]] = {}
                    for row in current_rows:
                        grouped.setdefault((row["device_part"], row["device_point"]), []).append(row)
                    results = []
                    for (part, point), running_rows in grouped.items():
                        segments = self._split_valve_segments(running_rows)
                        if not segments:
                            continue
                        current = segments[-1]
                        current_scan_start = current[0]["sample_time"]
                        current_values = self._valve_segment_values(
                            [float(row["pressure"]) for row in current if row["pressure"] is not None]
                        )
                        current_avg = sum(current_values) / len(current_values) if current_values else None
                        if current_avg is None or current_avg == 0:
                            continue
                        previous_range = None
                        previous_values: list[float] = []
                        eligible_previous = [
                            segment for segment in segments[:-1]
                            if segment[-1]["sample_time"] - segment[0]["sample_time"] >= timedelta(hours=24)
                        ]
                        if eligible_previous:
                            previous = eligible_previous[-1]
                            previous_start = previous[0]["sample_time"]
                            previous_end = previous[-1]["sample_time"]
                            previous_values = self._valve_segment_values(
                                [float(row["pressure"]) for row in previous if row["pressure"] is not None]
                            )
                            previous_range = (previous_start, previous_end)
                        else:
                            previous_range = self._previous_valve_segment(
                                cursor, part, point, current_scan_start, current_scan_start
                            )
                        previous_start = previous_end = None
                        previous_avg = None
                        if previous_range is not None:
                            previous_start, previous_end = previous_range
                            if not previous_values:
                                cursor.execute(
                                    """
                                    SELECT pa.pressure FROM wave_file f
                                    INNER JOIN statistic_pressure_angle pa
                                      ON f.id = pa.wave_file_id AND pa.angle = 230
                                    WHERE f.device_part = %s AND f.device_point = %s
                                      AND f.measurement_type = %s AND f.rpm > 0
                                      AND f.sample_time >= %s AND f.sample_time <= %s
                                    """,
                                    (part, point, "压力", previous_start, previous_end),
                                )
                                previous_values = self._valve_segment_values(
                                    [float(row["pressure"]) for row in cursor.fetchall() if row["pressure"] is not None]
                                )
                            previous_avg = sum(previous_values) / len(previous_values) if previous_values else None
                        else:
                            previous_values = []
                        is_fault = previous_avg is not None and previous_avg > current_avg * 1.05
                        if not is_fault:
                            continue
                        info = json.dumps(
                            {
                                "当前段平均值": current_avg,
                                "上一段平均值": previous_avg,
                                "当前段最高值": max(current_values) if current_values else None,
                                "当前段最低值": min(current_values) if current_values else None,
                                "上一段最高值": max(previous_values) if previous_values else None,
                                "上一段最低值": min(previous_values) if previous_values else None,
                                "上一段开始时间": _time_string(previous_start) if previous_start else None,
                                "上一段结束时间": _time_string(previous_end) if previous_end else None,
                            },
                            ensure_ascii=False,
                        )
                        description = (
                            f"230°压力上一段均值{previous_avg:.4f}高于当前段均值"
                            f"{current_avg:.4f}的105%"
                        )
                        status = 1
                        alarm_level = "1"
                        cursor.execute(
                            """
                            SELECT id FROM compressor_alarm
                            WHERE device_code = %s AND device_part = %s AND device_point = %s
                              AND alarm_type = %s AND alarm_time_start = %s
                            LIMIT 1
                            """,
                            (device_code, part, point, VALVE_FAULT_ALARM_TYPE, forecast_time),
                        )
                        existing = cursor.fetchone()
                        if existing:
                            cursor.execute(
                                """
                                UPDATE compressor_alarm SET alarm_time_end = %s, alarm_des = %s,
                                  alarm_level = %s, alarm_info_json = %s, status = %s,
                                  scan_start_time = %s, scan_end_time = %s
                                WHERE id = %s
                                """,
                                (forecast_time + timedelta(hours=hours), description, alarm_level, info,
                                 status, current_scan_start, forecast_time, existing["id"]),
                            )
                            results.append(("updated", int(existing["id"])))
                        else:
                            cursor.execute(
                                """
                                INSERT INTO compressor_alarm
                                  (device_code, device_part, device_point, alarm_type, alarm_des,
                                   alarm_level, alarm_info_json, alarm_time_start, alarm_time_end,
                                   scan_start_time, scan_end_time, status)
                                  VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                                """,
                                (device_code, part, point, VALVE_FAULT_ALARM_TYPE, description, alarm_level, info,
                                 forecast_time, forecast_time + timedelta(hours=hours), current_scan_start, forecast_time, status),
                            )
                            results.append(("inserted", int(cursor.lastrowid)))
                    return {
                        "action": results[0][0] if results else "no_alarm",
                        "id": results[0][1] if results else 0,
                        "inserted": sum(action == "inserted" for action, _ in results),
                        "updated": sum(action == "updated" for action, _ in results),
                    }

        result, source = self._run_with_fallback(database_query, lambda: {"action": "demo", "id": 0})
        return {"source": source, "notice": self._source_notice(source), **result}

    @staticmethod
    def _alarm_limit(item: dict[str, Any], alarm_type: str) -> float | None:
        for index in range(1, 5):
            if str(item.get(f"AlarmType{index}") or "").strip() == alarm_type:
                return _safe_float(item.get(f"AlarmLimit{index}"))
        return None

    def _analyze_oil_pressure_alarm(self, device_code: str, end: datetime) -> dict[str, Any] | None:
        """Analyze oil-pressure stages in the 14 days before ``end``."""
        unit = device_code.rstrip("#").strip()
        if not device_code.endswith("#") or not device_code[:-1].isdigit():
            raise ValueError("压缩机编号格式必须为数字+#")
        start = end - timedelta(days=OIL_PRESSURE_SCAN_DAYS)

        with get_connection() as connection:
            with connection.cursor() as cursor:
                cursor.execute(
                    """
                    SELECT ItemName, ItemDescription, AlarmType1, AlarmType2, AlarmType3, AlarmType4,
                           AlarmLimit1, AlarmLimit2, AlarmLimit3, AlarmLimit4
                    FROM site_point
                    WHERE ItemName = %s
                    """,
                    (f"YSJ{unit}_5",),
                )
                config = cursor.fetchone()
                if config is None:
                    raise ValueError(f"未找到 {device_code} 的润滑油压力报警配置")
                low = self._alarm_limit(config, "PVLow")
                low_low = self._alarm_limit(config, "PVLowLow")
                if low is None or low_low is None:
                    raise ValueError(f"{device_code} 缺少润滑油压力低压报警阈值")

                point_descriptions = {}
                for point in (5, 41):
                    cursor.execute(
                        "SELECT ItemName, ItemDescription FROM site_point WHERE ItemName = %s",
                        (f"YSJ{unit}_{point}",),
                    )
                    item = cursor.fetchone()
                    if item:
                        point_descriptions[f"YSJ_{point}"] = item.get("ItemDescription") or f"YSJ_{point}"

                cursor.execute(
                    """
                    SELECT 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,
                           AVG(CASE WHEN YSJ_41 > 0 THEN YSJ_5 END) AS oil_avg,
                           GROUP_CONCAT(DISTINCT import_batch_id ORDER BY import_batch_id) AS batch_ids
                    FROM pks_long_sample
                    WHERE device_code = %s
                      AND sample_time >= %s AND sample_time < %s
                    GROUP BY FROM_UNIXTIME((UNIX_TIMESTAMP(sample_time) DIV 3600) * 3600)
                    ORDER BY hour_start
                    """,
                    (device_code, start, end),
                )
                rows = []
                for row in cursor.fetchall():
                    running = int(row["running_samples"] or 0)
                    value = _safe_float(row["oil_avg"])
                    if running / 720 >= 0.80 and value is not None:
                        rows.append({
                            "time": row["hour_start"], "value": value, "running": running,
                            "batch_ids": [int(value) for value in str(row["batch_ids"] or "").split(",") if value],
                        })

        if not rows:
            return None
        baseline = float(np.median([row["value"] for row in rows]))
        candidates = []
        history = []
        for row in rows:
            recent = history[-5:] + [row]
            trend_values = [item["value"] for item in recent]
            trend = float(np.polyfit(range(len(trend_values)), trend_values, 1)[0]) if len(trend_values) >= 3 else None
            decline = 1
            newer = row
            for older in reversed(history):
                if newer["time"] - older["time"] != timedelta(hours=1):
                    break
                if newer["value"] > older["value"]:
                    break
                decline += 1
                newer = older
            current = row["value"]
            deviation = (baseline - current) / baseline if baseline else 0.0
            downward = decline >= 3 or (trend is not None and trend < -0.0005)
            if current <= low_low or (deviation >= 0.20 and downward):
                phase = "严重异常"
            elif current <= low or (deviation >= 0.10 and downward):
                phase = "异常"
            elif deviation >= 0.05 and downward:
                phase = "轻微"
            else:
                phase = "正常"
            candidates.append({"row": recent[-1], "phase": phase, "trend": trend, "decline": decline, "deviation": deviation})
            history.append(recent[-1])

        abnormal = [item for item in candidates if item["phase"] != "正常"]
        if not abnormal:
            return None
        stage_runs = []
        current_run = []
        for item in candidates:
            if current_run and (
                item["phase"] != current_run[-1]["phase"]
                or item["row"]["time"] - current_run[-1]["row"]["time"] != timedelta(hours=1)
            ):
                if current_run[0]["phase"] != "正常":
                    stage_runs.append(current_run)
                current_run = []
            current_run.append(item)
        if current_run and current_run[0]["phase"] != "正常":
            stage_runs.append(current_run)
        stage_rows = max(
            stage_runs,
            key=lambda run: (
                {"轻微": 1, "异常": 2, "严重异常": 3}[run[0]["phase"]],
                len(run),
            ),
        )
        stage = stage_rows[0]["phase"]
        first = stage_rows[0]
        last = stage_rows[-1]
        stage_values = [item["row"]["value"] for item in stage_rows]
        basis = (
            f"油压{first['row']['value']:.3f}～{last['row']['value']:.3f}MPa，"
            f"最低{min(stage_values):.3f}MPa，连续{len(stage_rows)}个有效运行小时；"
            f"周期基线油压{baseline:.3f}MPa，相对基线变化{last['deviation']:+.1%}"
        )
        batch_ids = sorted({batch for item in stage_rows for batch in item["row"]["batch_ids"]})
        info = {
            "机组": f"{unit}号机",
            "批次": batch_ids[0] if len(batch_ids) == 1 else batch_ids,
            "周期编号": f"{device_code}-P14",
            "开始小时": first["row"]["time"].strftime("%Y-%m-%d %H:%M:%S"),
            "结束小时": last["row"]["time"].strftime("%Y-%m-%d %H:%M:%S"),
            "持续自然小时数": int((last["row"]["time"] - first["row"]["time"]).total_seconds() / 3600) + 1,
            "有效运行小时数": len(stage_rows),
            "周期基线油压": baseline,
            "正常波动带": "",
            "阶段开始小时油压": first["row"]["value"],
            "阶段结束小时油压": last["row"]["value"],
            "阶段最低小时油压": min(stage_values),
            "阶段开始相对周期基线变化": first["deviation"],
            "阶段结束相对周期基线变化": last["deviation"],
            "阶段最大连续下降有效小时数": max(item["decline"] for item in stage_rows),
            "阶段低报警小时数": sum(item["row"]["value"] <= low for item in stage_rows),
            "阶段低低报警小时数": sum(item["row"]["value"] <= low_low for item in stage_rows),
            "是否形成确认等级": "是" if len(stage_rows) >= 3 else "否",
        }
        return {
            "stage": stage,
            "basis": basis,
            "device_part": "-".join(dict.fromkeys(point_descriptions.values())),
            "scan_start": start,
            "scan_end": end,
            "info": info,
        }

    def _analyze_cruciform_vibration_alarm(self, device_code: str, end: datetime) -> dict[str, Any] | None:
        """Build the strongest 14-day vibration stage for the crosshead alarm."""
        unit = device_code.rstrip("#").strip()
        if not device_code.endswith("#") or not device_code[:-1].isdigit():
            raise ValueError("压缩机编号格式必须为数字+#")
        start = end - timedelta(days=OIL_PRESSURE_SCAN_DAYS)

        with get_connection() as connection:
            with connection.cursor() as cursor:
                configs = {}
                descriptions = []
                for point, side in ((10, "联轴器端"), (11, "链轮端"), (41, "运行状态")):
                    cursor.execute(
                        """
                        SELECT ItemDescription, AlarmType1, AlarmType2, AlarmType3, AlarmType4,
                               AlarmLimit1, AlarmLimit2, AlarmLimit3, AlarmLimit4
                        FROM site_point WHERE ItemName = %s
                        """,
                        (f"YSJ{unit}_{point}",),
                    )
                    item = cursor.fetchone()
                    if item is None:
                        raise ValueError(f"未找到 {device_code} 的 YSJ_{point} 点位配置")
                    descriptions.append(item.get("ItemDescription") or f"YSJ_{point}")
                    if point != 41:
                        high = self._alarm_limit(item, "PVHigh")
                        high_high = self._alarm_limit(item, "PVHighHigh")
                        if high is None or high_high is None:
                            raise ValueError(f"{device_code} YSJ_{point} 缺少振动报警阈值")
                        configs[side] = {"high": high, "high_high": high_high}

                cursor.execute(
                    """
                    SELECT FROM_UNIXTIME((UNIX_TIMESTAMP(sample_time) DIV 3600) * 3600) AS hour_start,
                           SUM(CASE WHEN YSJ_41 > 0 THEN 1 ELSE 0 END) AS running_samples,
                           AVG(CASE WHEN YSJ_41 > 0 THEN YSJ_10 END) AS coupling_avg,
                           MAX(CASE WHEN YSJ_41 > 0 THEN YSJ_10 END) AS coupling_max,
                           AVG(CASE WHEN YSJ_41 > 0 THEN YSJ_11 END) AS chain_avg,
                           MAX(CASE WHEN YSJ_41 > 0 THEN YSJ_11 END) AS chain_max,
                           GROUP_CONCAT(DISTINCT import_batch_id ORDER BY import_batch_id) AS batch_ids
                    FROM pks_long_sample
                    WHERE device_code = %s
                      AND sample_time >= %s AND sample_time < %s
                    GROUP BY FROM_UNIXTIME((UNIX_TIMESTAMP(sample_time) DIV 3600) * 3600)
                    ORDER BY hour_start
                    """,
                    (device_code, start, end),
                )
                rows = []
                for row in cursor.fetchall():
                    running = int(row["running_samples"] or 0)
                    if running / 720 < 0.80:
                        continue
                    item = {
                        "time": row["hour_start"], "running": running,
                        "batch_ids": [int(value) for value in str(row["batch_ids"] or "").split(",") if value],
                    }
                    for key in ("coupling_avg", "coupling_max", "chain_avg", "chain_max"):
                        item[key] = _safe_float(row[key])
                    if item["coupling_max"] is not None or item["chain_max"] is not None:
                        rows.append(item)

        if len(rows) < 3:
            return None
        signals = []
        for index, row in enumerate(rows):
            prior = rows[max(0, index - 336):index]
            if len(prior) < 24:
                continue
            side_signals = []
            for side, prefix in (("联轴器端", "coupling"), ("链轮端", "chain")):
                current = row[f"{prefix}_max"]
                values = [item[f"{prefix}_max"] for item in prior if item[f"{prefix}_max"] is not None]
                averages = [item[f"{prefix}_avg"] for item in prior if item[f"{prefix}_avg"] is not None]
                if current is None or len(values) < 24:
                    continue
                baseline = float(np.median(values))
                scale = max(float(np.median(np.abs(np.asarray(values) - baseline))) * 1.4826, abs(baseline) * 0.05, 1e-6)
                peak_z = (current - baseline) / scale
                average = row[f"{prefix}_avg"]
                average_base = float(np.median(averages)) if averages else None
                average_change = average / average_base - 1 if average is not None and average_base else 0.0
                limit = configs[side]
                severe = current >= limit["high_high"] or peak_z >= 8
                abnormal = severe or current >= limit["high"] or peak_z >= 5 or average_change >= 0.10
                mild = abnormal or peak_z >= 4 or average_change >= 0.05
                if mild:
                    side_signals.append({
                        "side": side, "value": current, "baseline": baseline, "scale": scale,
                        "peak_z": peak_z, "average_change": average_change,
                        "level": "严重异常" if severe else "异常" if abnormal else "轻微",
                    })
            if side_signals:
                signals.append({"row": row, "signals": side_signals})

        if not signals:
            return None
        runs = []
        current_run = []
        for signal in signals:
            if current_run and signal["row"]["time"] - current_run[-1]["row"]["time"] > timedelta(hours=6):
                runs.append(current_run)
                current_run = []
            current_run.append(signal)
        if current_run:
            runs.append(current_run)
        run = max(
            runs,
            key=lambda values: (
                max({"轻微": 1, "异常": 2, "严重异常": 3}[item["level"]] for signal in values for item in signal["signals"]),
                len(values),
            ),
        )
        all_signals = [item for signal in run for item in signal["signals"]]
        strongest = max(all_signals, key=lambda item: ({"轻微": 1, "异常": 2, "严重异常": 3}[item["level"]], item["peak_z"]))
        stage = strongest["level"]
        first, last = run[0], run[-1]
        batch_ids = sorted({batch for signal in run for batch in signal["row"]["batch_ids"]})
        max_value = max(item["value"] for item in all_signals)
        basis = (
            f"{strongest['side']}振动峰值或均值偏离近期基线，阶段内{len(run)}个有效运行小时触发；"
            f"最大振动值{max_value:.3f}mm/s，峰值高于近期基线{strongest['peak_z']:.2f}个稳健尺度"
            f"（峰值/基线{max_value / max(strongest['baseline'], 1e-6):.2f}倍）"
        )
        info = {
            "机组": f"{unit}号机", "批次": batch_ids[0] if len(batch_ids) == 1 else batch_ids,
            "周期编号": f"{device_code}-V14", "开始小时": first["row"]["time"].strftime("%Y-%m-%d %H:%M:%S"),
            "结束小时": last["row"]["time"].strftime("%Y-%m-%d %H:%M:%S"),
            "持续自然小时数": int((last["row"]["time"] - first["row"]["time"]).total_seconds() / 3600) + 1,
            "有效运行小时数": len(run), "近期基线振动": strongest["baseline"],
            "正常波动带": f"{strongest['baseline'] - 2 * strongest['scale']:.3f}～{strongest['baseline'] + 2 * strongest['scale']:.3f} mm/s",
            "阶段开始小时振动": max(item["value"] for item in first["signals"]),
            "阶段结束小时振动": max(item["value"] for item in last["signals"]),
            "阶段最大振动值": max_value,
            "阶段开始相对近期基线变化": max(item["value"] / item["baseline"] - 1 for item in first["signals"]),
            "阶段结束相对近期基线变化": max(item["value"] / item["baseline"] - 1 for item in last["signals"]),
            "阶段最大连续异常有效小时数": len(run),
            "阶段高报警小时数": sum(any(item["value"] >= configs[item["side"]]["high"] for item in signal["signals"]) for signal in run),
            "阶段高高报警小时数": sum(any(item["value"] >= configs[item["side"]]["high_high"] for item in signal["signals"]) for signal in run),
            "是否形成确认等级": "是" if stage != "轻微" else "否",
            "最强异常侧": strongest["side"], "最强侧近期峰值基线": strongest["baseline"],
            "最强侧近期稳健尺度": strongest["scale"], "阶段最大峰值/基线比例": max_value / max(strongest["baseline"], 1e-6),
        }
        return {
            "stage": stage, "basis": basis, "device_part": "-".join(dict.fromkeys(descriptions)),
            "scan_start": start, "scan_end": end, "info": info,
        }

    def sync_pressure_angle(self, device_part: str, device_points: list[str] | None = None) -> dict[str, Any]:
        """按机组与部位批量生成 statistic_pressure_angle 采样数据。

        对每个压力 wave_file，若 statistic_pressure_angle 已存在至少一条记录则跳过，
        否则读 wave_sample_one（第一个周期）按“每个整数角度取第一个值”生成 360 个角度点。
        """
        points = device_points or []

        def database_query():
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    clauses = ["measurement_type = %s", "device_part = %s"]
                    params: list[Any] = ["压力", device_part]
                    if points:
                        placeholders = ", ".join(["%s"] * len(points))
                        clauses.append(f"device_point IN ({placeholders})")
                        params.extend(points)
                    cursor.execute(
                        f"SELECT id FROM wave_file WHERE {' AND '.join(clauses)} ORDER BY id",
                        params,
                    )
                    file_ids = [int(row["id"]) for row in cursor.fetchall()]
                    if not file_ids:
                        return {"total": 0, "created": 0, "skipped": 0}
                    placeholders = ", ".join(["%s"] * len(file_ids))
                    cursor.execute(
                        f"SELECT DISTINCT wave_file_id FROM statistic_pressure_angle WHERE wave_file_id IN ({placeholders})",
                        file_ids,
                    )
                    existing = {int(row["wave_file_id"]) for row in cursor.fetchall()}
                    created = 0
                    skipped = 0
                    for file_id in file_ids:
                        if file_id in existing:
                            skipped += 1
                            continue
                        cursor.execute(
                            "SELECT sample_index, signal_value, second_value FROM wave_sample_one "
                            "WHERE wave_file_id = %s ORDER BY sample_index",
                            (file_id,),
                        )
                        rows = cursor.fetchall()
                        if not rows:
                            skipped += 1
                            continue
                        samples = np.asarray(
                            [
                                (float(row["sample_index"]), float(row["signal_value"]),
                                 float(row["second_value"]) if row["second_value"] is not None else 0.0)
                                for row in rows
                            ],
                            dtype=float,
                        )
                        sampled = _sample_first_cycle_360(samples)
                        if not sampled:
                            skipped += 1
                            continue
                        cursor.executemany(
                            "INSERT INTO statistic_pressure_angle (wave_file_id, period, angle, pressure) "
                            "VALUES (%s, %s, %s, %s)",
                            [(file_id, 1, angle, pressure) for angle, pressure in sampled],
                        )
                        created += 1
                    return {"total": len(file_ids), "created": created, "skipped": skipped}

        result, source = self._run_with_fallback(
            database_query,
            lambda: {"total": 0, "created": 0, "skipped": 0},
        )
        return {"source": source, "notice": self._source_notice(source), **result}

    def query_pressure_angle(
        self,
        device_part: str,
        device_points: list[str] | None,
        min_time: str | None,
        max_time: str | None,
        angles: list[int],
    ) -> dict[str, Any]:
        """按时间范围和角度列表查询 statistic_pressure_angle 曲线数据。"""
        if not angles:
            raise ValueError("请至少输入一个角度")
        angles = [int(angle) for angle in angles if 0 <= int(angle) <= 360]

        def database_query():
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    clauses = ["wf.measurement_type = %s", "wf.device_part = %s"]
                    params: list[Any] = ["压力", device_part]
                    if device_points:
                        placeholders = ", ".join(["%s"] * len(device_points))
                        clauses.append(f"wf.device_point IN ({placeholders})")
                        params.extend(device_points)
                    start = _parse_time(min_time) if min_time else None
                    end = _parse_time(max_time) if max_time else None
                    if start:
                        clauses.append("wf.sample_time >= %s")
                        params.append(start)
                    if end:
                        clauses.append("wf.sample_time <= %s")
                        params.append(end)
                    angle_placeholders = ", ".join(["%s"] * len(angles))
                    clauses.append(f"spa.angle IN ({angle_placeholders})")
                    params.extend(angles)
                    cursor.execute(
                        f"""
                        SELECT wf.sample_time, spa.angle, spa.pressure
                        FROM statistic_pressure_angle spa
                        INNER JOIN wave_file wf ON wf.id = spa.wave_file_id
                        WHERE {' AND '.join(clauses)}
                        ORDER BY wf.sample_time, spa.angle
                        """,
                        params,
                    )
                    rows = cursor.fetchall()

                    stop_days = self._query_stop_days(
                        cursor, device_part, device_points, start, end,
                    )
            series: dict[int, list[dict[str, Any]]] = {}
            for row in rows:
                angle = int(round(float(row["angle"])))
                series.setdefault(angle, []).append({
                    "time": _time_string(row["sample_time"]),
                    "pressure": float(row["pressure"]),
                })
            for day in sorted(stop_days):
                day_time = f"{day} 00:00:00"
                for angle in angles:
                    series.setdefault(angle, []).append({
                        "time": day_time,
                        "pressure": 0.0,
                    })
            for angle in angles:
                series.setdefault(angle, []).sort(key=lambda point: point["time"])
            return {
                "angles": angles,
                "series": [
                    {"angle": angle, "points": series.get(angle, [])}
                    for angle in angles
                ],
            }

        result, source = self._run_with_fallback(
            database_query,
            lambda: {"angles": angles, "series": [{"angle": angle, "points": []} for angle in angles]},
        )
        return {"source": source, "notice": self._source_notice(source), **result}

    @staticmethod
    def _query_stop_days(
        cursor: Any,
        device_part: str,
        device_points: list[str] | None,
        start: datetime | None,
        end: datetime | None,
    ) -> set[str]:
        """返回停机日集合：该机组在时间范围内 rpm=0 且当天无 rpm>0 记录的日期。"""
        clauses = ["device_part = %s", "rpm = 0"]
        params: list[Any] = [device_part]
        if device_points:
            placeholders = ", ".join(["%s"] * len(device_points))
            clauses.append(f"device_point IN ({placeholders})")
            params.extend(device_points)
        if start:
            clauses.append("sample_time >= %s")
            params.append(start)
        if end:
            clauses.append("sample_time <= %s")
            params.append(end)
        cursor.execute(
            f"""
            SELECT DISTINCT DATE(sample_time) AS d
            FROM wave_file
            WHERE {' AND '.join(clauses)}
            """,
            params,
        )
        stop_days = {str(row["d"]) for row in cursor.fetchall()}

        run_clauses = ["device_part = %s", "rpm > 0"]
        run_params: list[Any] = [device_part]
        if device_points:
            placeholders = ", ".join(["%s"] * len(device_points))
            run_clauses.append(f"device_point IN ({placeholders})")
            run_params.extend(device_points)
        if start:
            run_clauses.append("sample_time >= %s")
            run_params.append(start)
        if end:
            run_clauses.append("sample_time <= %s")
            run_params.append(end)
        cursor.execute(
            f"""
            SELECT DISTINCT DATE(sample_time) AS d
            FROM wave_file
            WHERE {' AND '.join(run_clauses)}
            """,
            run_params,
        )
        run_days = {str(row["d"]) for row in cursor.fetchall()}
        return stop_days - run_days

    @staticmethod
    def _group_time_points(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
        grouped: OrderedDict[Any, dict[str, Any]] = OrderedDict()
        for row in rows:
            # Channels from one acquisition batch can be stamped a few seconds
            # apart (e.g. 10:30:02 vs 10:30:05), yet the acquisition cadence is
            # 15 minutes. Align on the minute so one batch is one time point.
            sample_time = row["sample_time"]
            key = str(sample_time)[:16]
            point = grouped.setdefault(
                key,
                {
                    "sampleTime": _time_string(sample_time),
                    "files": {},
                },
            )
            device_point = row["device_point"]
            point["files"].setdefault(device_point, {
                "id": int(row["id"]),
                "devicePoint": device_point,
                "measurementType": row["measurement_type"],
                "sampleCount": int(row["sample_count"] or 0),
                "sampleFrequencyHz": int(row["sample_frequency_hz"] or 0),
                "rpm": float(row["rpm"] or 0),
                "status": int(row.get("tspluse_status") or 0),
            })
        return [
            {"index": index, **point}
            for index, point in enumerate(grouped.values())
        ]

    def _pks_time_points(
        self,
        unit: str,
        columns: list[tuple[str, str]],
        start: datetime | None,
        end: datetime | None,
        include_stopped: bool,
    ) -> list[dict[str, Any]]:
        """PKS-only 时间点：按 15 分钟槽从 pks_long_sample 生成骨架。

        columns 为 [(item_name, pks 列名), ...]，siteValues 直挂所选各列
        原值。7/8/9 机组的停机/运转由该机组批次内的 YSJ_41（转速）决定
        （>0=运转、=0=停机）；4/5/6 无转速信号，按运转处理。未勾选停机时
        仅保留运转锚点；勾选停机则停机锚点一并显示。
        """
        batches = self._pks_batch_ids(unit)
        if not batches:
            return []
        batch_clause = f"import_batch_id IN ({self._in_clause(batches)})"
        clauses = [batch_clause]
        params: list[Any] = list(batches)
        if start:
            clauses.append("sample_time >= %s")
            params.append(start)
        if end:
            clauses.append("sample_time <= %s")
            params.append(end)
        with get_connection() as connection:
            with connection.cursor() as cursor:
                cursor.execute(
                    f"""
                    SELECT MIN(sample_time) AS anchor
                    FROM pks_long_sample
                    WHERE {' AND '.join(clauses)}
                    GROUP BY FLOOR(UNIX_TIMESTAMP(sample_time) / 900)
                    ORDER BY anchor
                    """,
                    params,
                )
                anchors = [row["anchor"] for row in cursor.fetchall()]
        if not anchors:
            return []
        if len(anchors) > PKS_STRIP_MAX_ANCHORS:
            step = (len(anchors) + PKS_STRIP_MAX_ANCHORS - 1) // PKS_STRIP_MAX_ANCHORS
            sampled = anchors[::step]
            if sampled[-1] != anchors[-1]:
                sampled.append(anchors[-1])
            anchors = sampled
        chunk_size = 1000
        select_expr = ", ".join(f"`{column}`" for _item, column in columns)
        use_running_signal = unit in _RUNNING_SIGNAL_UNITS
        values_by_time: dict[datetime, dict[str, float | None]] = {anchor: {} for anchor in anchors}
        running_by_time: dict[datetime, bool] = {}
        for index in range(0, len(anchors), chunk_size):
            chunk = anchors[index:index + chunk_size]
            placeholders = ", ".join(["%s"] * len(chunk))
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        f"SELECT sample_time, {select_expr}, `YSJ_41` FROM pks_long_sample "
                        f"WHERE {batch_clause} AND sample_time IN ({placeholders})",
                        (*batches, *chunk),
                    )
                    rows = cursor.fetchall()
            for row in rows:
                anchor_time = row["sample_time"]
                values = values_by_time[anchor_time]
                for item_name, column in columns:
                    value = row[column]
                    values[item_name] = float(value) if value is not None else None
                if use_running_signal:
                    running_value = row["YSJ_41"]
                    running_by_time[anchor_time] = bool(
                        running_value is not None and running_value > 0
                    )
                else:
                    running_by_time.setdefault(anchor_time, True)
        points: list[dict[str, Any]] = []
        for anchor in anchors:
            running = running_by_time.get(anchor, False)
            if not include_stopped and not running:
                continue
            points.append(
                {
                    "sampleTime": _time_string(anchor),
                    "files": {},
                    "siteValues": values_by_time[anchor],
                    "machineRunning": running,
                },
            )
        for index, point in enumerate(points):
            point["index"] = index
        return points

    def _pks_only_wave_window(
        self,
        device_part: str,
        points: list[dict[str, Any]],
        site_points: list[str],
        unit: str,
    ) -> dict[str, Any]:
        """PKS-only 波形窗口：不取 wave_file，仅构建所选 PKS 点位的微曲线。"""

        def database_query():
            return {
                "points": points,
                "siteSeries": self._build_site_series(site_points, unit, points),
            }

        def demo_query():
            return {"points": points, "siteSeries": {"points": [], "series": []}}

        result, source = self._run_with_fallback(database_query, demo_query)
        return {
            "devicePart": device_part,
            "devicePoints": [],
            "primaryPoint": None,
            "points": points,
            "xMin": 0,
            "xMax": len(points),
            "series": [],
            "secondSeries": {
                "name": "周期数据",
                "color": "#f56c6c",
                "sourceMeasurementType": None,
                "sourceDevicePoint": None,
                "data": [],
                "finiteCount": 0,
                "nonZeroCount": 0,
                "min": None,
                "max": None,
            },
            "angleSeries": {"color": "#d59b2b", "data": []},
            "volumeSeries": {"color": "#4d9e6f", "data": [], "info": None},
            "cycles": [],
            "triggerXs": [],
            "files": [],
            "diagnostics": [],
            "siteSeries": result["siteSeries"],
            "source": source,
            "notice": self._source_notice(source),
        }

    def alarm_compressor_options(self) -> dict[str, Any]:
        """Return running compressor groups using device_part's machine prefix."""
        def database_query():
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        """
                        SELECT SUBSTRING(device_part, 1, 4) AS device_name,
                               COUNT(1) AS count_no
                        FROM wave_file
                        WHERE rpm > 0 AND device_part <> ''
                        GROUP BY SUBSTRING(device_part, 1, 4)
                        ORDER BY device_name
                        """,
                    )
                    return [
                        {
                            "deviceName": row["device_name"],
                            "deviceCode": f"{str(row['device_name'])[0]}#",
                            "count": int(row["count_no"]),
                        }
                        for row in cursor.fetchall()
                        if row["device_name"]
                    ]

        result, source = self._run_with_fallback(database_query, lambda: [])
        return {"source": source, "notice": self._source_notice(source), "items": result}

    def wave_window(
        self,
        device_part: str,
        device_points: list[str],
        points: list[dict[str, Any]],
        max_points: int,
        no_sampling: bool = False,
        first_cycle_only: bool = False,
        site_points: list[str] | None = None,
    ) -> dict[str, Any]:
        selected_points = _validate_device_points(device_points, allow_empty=True)
        if not points:
            raise ValueError("至少选择一个时间点")
        if len(points) > 200:
            raise ValueError("单次最多预览 200 个时间点，请缩小时间窗口")
        max_points = min(max(int(max_points), 256), 200000)
        unit = _unit_number(device_part)
        site_list = list(site_points or [])
        site_columns: list[tuple[str, str]] = []
        if unit:
            site_columns = [
                (item, column)
                for item in site_list
                if (column := _site_point_column(item, unit)) is not None
            ]
        pks_only = bool(unit and site_columns and not selected_points)
        if not selected_points and not pks_only:
            raise ValueError("请至少选择一个测试点位")

        if pks_only:
            return self._pks_only_wave_window(device_part, points, site_list, unit)

        def database_query():
            if first_cycle_only:
                return self._build_first_cycle_window(
                    device_part,
                    selected_points,
                    points,
                    self._load_db_wave,
                )
            return self._build_wave_window(
                device_part,
                selected_points,
                points,
                max_points,
                self._load_db_wave,
                no_sampling,
            )

        def demo_query():
            if first_cycle_only:
                return self._build_first_cycle_window(
                    device_part,
                    selected_points,
                    points,
                    self._load_demo_wave,
                )
            return self._build_wave_window(
                device_part,
                selected_points,
                points,
                max_points,
                self._load_demo_wave,
                no_sampling,
            )

        result, source = self._run_with_fallback(database_query, demo_query)
        # 全场点位(PKS)系列：每周期(PKS 5s)一小段、跨时间点连续的 15 分钟窗口曲线。
        if first_cycle_only and site_points:
            try:
                unit = _unit_number(device_part)
                result["siteSeries"] = (
                    self._build_site_series(site_points, unit, result["points"])
                    if unit is not None
                    else {"points": [], "series": []}
                )
            except Exception:
                result["siteSeries"] = {"points": [], "series": []}
        else:
            result["siteSeries"] = {"points": [], "series": []}
        result["source"] = source
        result["notice"] = self._source_notice(source)
        return result

    def period_detail(self, wave_file_id: int, period_number: int) -> dict[str, Any]:
        if wave_file_id <= 0 or period_number <= 0:
            raise ValueError("wave_file_id 和周期编号必须为正整数")
        if period_number != 1:
            raise ValueError("当前数据仅保留第一个周期")

        def database_query():
            metadata, samples = self._load_db_wave(wave_file_id)
            details = [self._build_period_detail(metadata, samples, period_number)]
            notice: str | None = None
            companion = self._find_opposite_pressure_metadata(metadata)
            if companion is None:
                notice = "未找到同一时刻的对侧压力数据，仅显示当前侧。"
            else:
                try:
                    companion_meta, companion_samples = self._load_db_wave(int(companion["id"]))
                    details.append(
                        self._build_period_detail(companion_meta, companion_samples, period_number),
                    )
                except Exception as exc:
                    notice = f"对侧压力数据读取失败，仅显示当前侧：{exc}"
            return self._assemble_period_detail(metadata, details, notice)

        def demo_query():
            metadata, samples = self._load_demo_wave(
                wave_file_id,
                "压力",
                DEMO_POINT_NAME,
                DEMO_START,
            )
            details = [self._build_period_detail(metadata, samples, period_number)]
            return self._assemble_period_detail(
                metadata,
                details,
                "演示模式仅提供单侧压力数据。",
            )

        result, source = self._run_with_fallback(database_query, demo_query)
        result["source"] = source
        result["notice"] = self._source_notice(source)
        return result

    def _find_opposite_pressure_metadata(self, metadata: dict[str, Any]) -> dict[str, Any] | None:
        point_name = str(metadata.get("point_name") or "")
        side = metadata.get("device_point")
        if side not in PRESSURE_SIDES:
            side = next((item for item in PRESSURE_SIDES if point_name.endswith(item)), None)
        if side not in PRESSURE_SIDES:
            return None
        other_side = PRESSURE_SIDES[1] if side == PRESSURE_SIDES[0] else PRESSURE_SIDES[0]
        device_part = metadata.get("device_part")
        sample_time = metadata.get("sample_time")
        if not device_part or sample_time is None:
            return None
        try:
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        """
                        SELECT id
                        FROM wave_file
                        WHERE measurement_type = %s
                          AND device_part = %s
                          AND sample_time = %s
                          AND device_point = %s
                        ORDER BY id ASC
                        LIMIT 1
                        """,
                        ("压力", device_part, sample_time, other_side),
                    )
                    row = cursor.fetchone()
            return row
        except Exception:
            return None

    @staticmethod
    def _assemble_period_detail(
        metadata: dict[str, Any],
        details: list[dict[str, Any]],
        side_notice: str | None,
    ) -> dict[str, Any]:
        order = {side: index for index, side in enumerate(PRESSURE_SIDES)}
        details.sort(key=lambda item: order.get(str(item.get("side")), len(order)))
        point_name = str(metadata.get("point_name") or "")
        device_part = metadata.get("device_part")
        sample_time = metadata.get("sample_time")
        primary_side = metadata.get("device_point")
        if primary_side not in PRESSURE_SIDES:
            primary_side = next(
                (item for item in PRESSURE_SIDES if point_name.endswith(item)),
                details[0].get("side") if details else None,
            )
        if not device_part and primary_side and point_name.endswith(str(primary_side)):
            device_part = point_name[: -len(str(primary_side))]
        return {
            "devicePart": device_part or "",
            "sampleTime": _time_string(sample_time) if sample_time is not None else "",
            "primarySide": primary_side,
            "sideNotice": side_notice,
            "sides": details,
        }

    def annotation_config(self) -> dict[str, Any]:
        return {
            "source": self.source,
            "annotationWidth": settings.annotation_width,
            "notice": self._source_notice(self.source),
        }

    def list_annotations(self, wave_file_ids: list[int]) -> dict[str, Any]:
        ids = sorted({int(value) for value in wave_file_ids if value})
        if not ids:
            return {
                "source": self.source,
                "annotations": [],
                "notice": self._source_notice(self.source),
            }

        def database_query():
            placeholders = ", ".join(["%s"] * len(ids))
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        f"""
                        SELECT id, wave_file_id, label, period_start, period_end,
                               sample_index_start, sample_index_end
                        FROM wave_annotation
                        WHERE wave_file_id IN ({placeholders})
                        ORDER BY id ASC
                        """,
                        ids,
                    )
                    rows = cursor.fetchall()
            return [_annotation_dict(row) for row in rows]

        def demo_query():
            return [
                _annotation_dict(annotation)
                for annotation in self._demo_annotations.values()
                if annotation["wave_file_id"] in ids
            ]

        annotations, source = self._run_with_fallback(database_query, demo_query)
        return {
            "source": source,
            "annotations": annotations,
            "notice": self._source_notice(source),
        }

    def create_annotation(self, payload: dict[str, Any]) -> dict[str, Any]:
        self._validate_annotation(payload)
        wave_file_id = int(payload["wave_file_id"])
        label = payload["label"]
        period_start = int(payload["period_start"])
        period_end = int(payload["period_end"])
        sample_index_start = int(payload["sample_index_start"])
        sample_index_end = int(payload["sample_index_end"])

        def database_query():
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        """
                        INSERT INTO wave_annotation
                            (wave_file_id, label, period_start, period_end,
                             sample_index_start, sample_index_end)
                        VALUES (%s, %s, %s, %s, %s, %s)
                        """,
                        (
                            wave_file_id,
                            label,
                            period_start,
                            period_end,
                            sample_index_start,
                            sample_index_end,
                        ),
                    )
                    annotation_id = cursor.lastrowid
                    cursor.execute(
                        """
                        SELECT id, wave_file_id, label, period_start, period_end,
                               sample_index_start, sample_index_end
                        FROM wave_annotation
                        WHERE id = %s
                        """,
                        (annotation_id,),
                    )
                    return _annotation_dict(cursor.fetchone())

        def demo_query():
            annotation_id = self._demo_annotation_seq
            self._demo_annotation_seq += 1
            annotation = {
                "id": annotation_id,
                "wave_file_id": wave_file_id,
                "label": label,
                "period_start": period_start,
                "period_end": period_end,
                "sample_index_start": sample_index_start,
                "sample_index_end": sample_index_end,
            }
            self._demo_annotations[annotation_id] = annotation
            return _annotation_dict(annotation)

        result, source = self._run_with_fallback(database_query, demo_query)
        result["source"] = source
        result["notice"] = self._source_notice(source)
        return result

    def delete_annotation(self, annotation_id: int) -> dict[str, Any]:
        if annotation_id <= 0:
            raise ValueError("标注 id 必须为正整数")

        def database_query():
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        "DELETE FROM wave_annotation WHERE id = %s",
                        (annotation_id,),
                    )
                    return int(cursor.rowcount)

        def demo_query():
            if annotation_id not in self._demo_annotations:
                return 0
            del self._demo_annotations[annotation_id]
            return 1

        deleted, source = self._run_with_fallback(database_query, demo_query)
        if not deleted:
            raise ValueError(f"标注 id={annotation_id} 不存在")
        return {
            "deleted": annotation_id,
            "source": source,
            "notice": self._source_notice(source),
        }

    @staticmethod
    def _validate_annotation(payload: dict[str, Any]) -> None:
        label = payload.get("label")
        if label not in ANNOTATION_LABELS:
            raise ValueError("样本类型只能是 正常 或 异常")
        for field in ("wave_file_id", "period_start", "period_end", "sample_index_start", "sample_index_end"):
            if payload.get(field) is None:
                raise ValueError(f"{field} 不能为空")
        if int(payload["wave_file_id"]) <= 0:
            raise ValueError("wave_file_id 必须为正整数")
        if int(payload["period_start"]) <= 0 or int(payload["period_end"]) <= 0:
            raise ValueError("周期编号必须为正整数")
        if int(payload["period_start"]) > int(payload["period_end"]):
            raise ValueError("起始周期不能大于结束周期")
        if int(payload["sample_index_start"]) < 0 or int(payload["sample_index_end"]) < 0:
            raise ValueError("采样点索引不能为负")
        if int(payload["sample_index_start"]) > int(payload["sample_index_end"]):
            raise ValueError("起始采样点不能大于结束采样点")

    def health(self) -> dict[str, Any]:
        return {
            "status": "ok",
            "source": self.source,
            "databaseError": self._last_db_error or None,
        }

    def _source_notice(self, source: str) -> str | None:
        if source == "demo":
            if self._last_db_error:
                return f"当前为演示数据：数据库暂不可用（{self._last_db_error}）"
            return "当前为演示数据：可设置 DEMO_MODE=never 强制使用数据库"
        return "已连接 MySQL 数据库"

    def _build_wave_window(
        self,
        device_part: str,
        device_points: list[str],
        points: list[dict[str, Any]],
        max_points: int,
        loader: Callable[..., tuple[dict[str, Any], np.ndarray]],
        no_sampling: bool = False,
    ) -> dict[str, Any]:
        series_data: dict[str, list[dict[str, Any]]] = {point: [] for point in device_points}
        angle_data: list[dict[str, Any]] = []
        volume_data: list[dict[str, Any]] = []
        volume_info: dict[str, Any] | None = None
        cycles: list[dict[str, Any]] = []
        triggers: list[float] = []
        files: list[dict[str, Any]] = []
        diagnostics: list[dict[str, Any]] = []
        second_series_data: list[dict[str, Any]] = []
        second_finite_count = 0
        second_non_zero_count = 0
        second_min: float | None = None
        second_max: float | None = None
        primary = _primary_device_point(device_points)
        primary_type = DEVICE_POINT_TO_TYPE[primary]
        pressure_points = [point for point in device_points if DEVICE_POINT_TO_TYPE[point] == "压力"]
        load_cache: dict[int, tuple[dict[str, Any], np.ndarray]] = {}
        single_cycle = loader.__name__ != "_load_demo_wave"

        for slot, point in enumerate(points):
            target_per_file = max(256, int(np.ceil(max_points / max(len(points), 1))))

            slot_files: dict[str, dict[str, Any]] = {}
            for device_point in device_points:
                file_info = point.get("files", {}).get(device_point)
                if file_info:
                    slot_files[device_point] = file_info

            # The per-slot "period source" supplies the 周期数据/体积/角度 series
            # and the background cycle bands. Primary point is preferred; when a
            # selected point's timestamp differs (e.g. 10:30:02 vs 10:30:05), a
            # slot may only contain another device point, which is used instead
            # so the period data and cycle bands do not disappear there.
            source_point = primary if primary in slot_files else (next(iter(slot_files)) if slot_files else None)

            source_samples: np.ndarray | None = None
            source_detected: list[Any] = []
            source_angle_full: np.ndarray | None = None
            source_volume: np.ndarray | None = None
            source_required: set[int] | None = None
            source_id: int | None = None
            source_type = ""
            if source_point is not None:
                source_info = slot_files[source_point]
                source_id = int(source_info["id"] if isinstance(source_info, dict) else source_info)
                source_type = DEVICE_POINT_TO_TYPE[source_point]
                try:
                    source_meta, source_samples = self._load_for_window(
                        loader,
                        source_id,
                        source_type,
                        device_part + source_point,
                        point["sampleTime"],
                        load_cache,
                    )
                except ValueError:
                    source_samples = None
                if source_samples is not None and len(source_samples):
                    source_detected, source_diagnostic = _detect_source_cycles(source_samples, single_cycle)
                    source_angle_vector = build_angle_vector(len(source_samples), source_detected)
                    source_angle_full = np.full(len(source_samples), np.nan, dtype=float)
                    for detected_cycle in source_detected:
                        source_angle_full[detected_cycle.start_offset:detected_cycle.end_offset] = detected_cycle.angle
                    source_volume, current_volume_info = self._build_volume_vector(
                        len(source_samples),
                        source_detected,
                        device_part + primary,
                    )
                    if current_volume_info is not None:
                        volume_info = current_volume_info
                    source_indices = source_samples[:, 0].astype(np.int64)
                    source_span = max(len(source_samples), 1)
                    finite_second = source_samples[:, 2][np.isfinite(source_samples[:, 2])]
                    if len(finite_second):
                        second_finite_count += int(len(finite_second))
                        second_non_zero_count += int(np.count_nonzero(finite_second != 0))
                        current_min = float(np.min(finite_second))
                        current_max = float(np.max(finite_second))
                        second_min = current_min if second_min is None else min(second_min, current_min)
                        second_max = current_max if second_max is None else max(second_max, current_max)
                    source_required = {0, len(source_samples) - 1}
                    for cycle in source_detected:
                        source_required.update(
                            {
                                cycle.start_offset,
                                max(cycle.end_offset - 1, cycle.start_offset),
                                *cycle.trigger_offsets,
                            },
                        )
                        start_x = slot + cycle.start_offset / source_span
                        end_x = slot + cycle.end_offset / source_span
                        cycles.append(
                            {
                                "id": f"{source_id}:{cycle.number}",
                                "waveFileId": source_id,
                                "periodNo": cycle.number,
                                "pointIndex": slot,
                                "sampleTime": point["sampleTime"],
                                "startX": start_x,
                                "endX": end_x,
                                "startSampleIndex": int(source_indices[cycle.start_offset]),
                                "endSampleIndex": int(
                                    source_indices[max(cycle.end_offset - 1, cycle.start_offset)],
                                ),
                                "sourceType": source_type,
                                "devicePoint": source_point,
                                "background": True,
                            },
                        )
                    for trigger_offset in sorted(source_required):
                        if any(
                            trigger_offset == run_offset
                            for cycle in source_detected
                            for run_offset in cycle.trigger_offsets
                        ):
                            triggers.append(slot + trigger_offset / source_span)
                    diagnostics.append(
                        {
                            "waveFileId": source_id,
                            "measurementType": source_type,
                            "devicePoint": source_point,
                            "sampleTime": point["sampleTime"],
                            **source_diagnostic,
                        },
                    )
                    source_base_index = int(source_indices[0])
                    second_chosen = downsample_indices(source_samples[:, 2], target_per_file, source_required)
                    for offset in second_chosen:
                        second_value = _safe_float(source_samples[offset, 2])
                        if second_value is None:
                            continue
                        x = slot + (int(source_indices[offset]) - source_base_index) / source_span
                        second_series_data.append(
                            {
                                "value": [x, second_value],
                                "x": x,
                                "rawValue": second_value,
                                "sampleIndex": int(source_indices[offset]),
                                "waveFileId": source_id,
                                "sampleTime": point["sampleTime"],
                            },
                        )
                    angle_chosen = downsample_indices(source_angle_vector, target_per_file, source_required)
                    for offset in angle_chosen:
                        value = _safe_float(source_angle_vector[offset])
                        if value is None:
                            continue
                        x = slot + (int(source_indices[offset]) - source_base_index) / source_span
                        angle_data.append(
                            {
                                "value": [x, value],
                                "x": x,
                                "angle": value,
                                "sampleIndex": int(source_indices[offset]),
                                "waveFileId": source_id,
                                "sampleTime": point["sampleTime"],
                            },
                        )
                    volume_chosen = downsample_indices(source_volume, target_per_file, source_required)
                    for offset in volume_chosen:
                        value = _safe_float(source_volume[offset])
                        if value is None:
                            continue
                        x = slot + (int(source_indices[offset]) - source_base_index) / source_span
                        volume_data.append(
                            {
                                "value": [x, value],
                                "x": x,
                                "volume": value,
                                "sampleIndex": int(source_indices[offset]),
                                "waveFileId": source_id,
                                "sampleTime": point["sampleTime"],
                            },
                        )

            for device_point in device_points:
                file_info = slot_files.get(device_point)
                if not file_info:
                    continue
                file_id = int(file_info["id"] if isinstance(file_info, dict) else file_info)
                measurement_type = DEVICE_POINT_TO_TYPE[device_point]
                try:
                    metadata, samples = self._load_for_window(
                        loader,
                        file_id,
                        measurement_type,
                        device_part + device_point,
                        point["sampleTime"],
                        load_cache,
                    )
                except ValueError:
                    continue
                sample_count = len(samples)
                if not sample_count:
                    continue
                sample_indices_for_file = samples[:, 0].astype(np.int64)
                file_base_index = int(sample_indices_for_file[0])
                file_span = max(sample_count, 1)
                file_required = {0, sample_count - 1}
                own_volume: np.ndarray | None = None
                own_angle: np.ndarray | None = None
                if device_point in pressure_points:
                    if source_point == device_point and source_samples is not None:
                        own_volume = source_volume
                        own_angle = source_angle_full
                        if source_required is not None:
                            file_required.update(source_required)
                    else:
                        detected_own, _ = _detect_source_cycles(samples, single_cycle)
                        for cycle in detected_own:
                            file_required.update(
                                {
                                    cycle.start_offset,
                                    max(cycle.end_offset - 1, cycle.start_offset),
                                    *cycle.trigger_offsets,
                                },
                            )
                            cycles.append(
                                {
                                    "id": f"{file_id}:{cycle.number}",
                                    "waveFileId": file_id,
                                    "periodNo": cycle.number,
                                    "pointIndex": slot,
                                    "sampleTime": point["sampleTime"],
                                    "startX": slot + cycle.start_offset / file_span,
                                    "endX": slot + cycle.end_offset / file_span,
                                    "startSampleIndex": int(sample_indices_for_file[cycle.start_offset]),
                                    "endSampleIndex": int(
                                        sample_indices_for_file[max(cycle.end_offset - 1, cycle.start_offset)],
                                    ),
                                    "sourceType": measurement_type,
                                    "devicePoint": device_point,
                                    "background": False,
                                },
                            )
                        if detected_own:
                            angle_own = build_angle_vector(len(samples), detected_own)
                            full_own = np.full(len(samples), np.nan, dtype=float)
                            for detected_cycle in detected_own:
                                full_own[detected_cycle.start_offset:detected_cycle.end_offset] = detected_cycle.angle
                            own_volume, _ = self._build_volume_vector(
                                len(samples),
                                detected_own,
                                device_part + device_point,
                            )
                            own_angle = full_own
                if no_sampling:
                    target_per_file_own = sample_count
                else:
                    target_per_file_own = target_per_file
                chosen = downsample_indices(samples[:, 1], target_per_file_own, file_required)
                for offset in chosen:
                    x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span
                    raw_value = float(samples[offset, 1])
                    series_data[device_point].append(
                        {
                            "value": [x, raw_value],
                            "x": x,
                            "rawValue": raw_value,
                            "sampleIndex": int(sample_indices_for_file[offset]),
                            "waveFileId": file_id,
                            "sampleTime": point["sampleTime"],
                            "secondValue": _safe_float(samples[offset, 2]),
                            "volume": _safe_float(own_volume[offset]) if own_volume is not None else None,
                            "angle360": _safe_float(own_angle[offset]) if own_angle is not None else None,
                        },
                    )
                files.append(
                    {
                        "id": file_id,
                        "pointIndex": slot,
                        "sampleTime": point["sampleTime"],
                        "devicePoint": device_point,
                        "measurementType": measurement_type,
                        "sampleCount": int(metadata.get("sample_count") or sample_count),
                        "sampleFrequencyHz": int(metadata.get("sample_frequency_hz") or 0),
                        "pointName": str(metadata.get("point_name") or device_part + device_point),
                        "rpm": float(metadata.get("rpm") or 0),
                        "status": int(metadata.get("tspluse_status") or 0),
                        "fileName": str(metadata.get("file_name") or ""),
                    },
                )
            load_cache.clear()

        for device_point in series_data:
            series_data[device_point].sort(key=lambda item: item["x"])
        second_series_data.sort(key=lambda item: item["x"])
        angle_data.sort(key=lambda item: item["x"])
        volume_data.sort(key=lambda item: item["x"])
        extents = {device_point: _series_extent(series_data[device_point]) for device_point in device_points}
        return {
            "devicePart": device_part,
            "devicePoints": device_points,
            "primaryPoint": primary,
            "points": points,
            "xMin": 0,
            "xMax": len(points),
            "series": [
                {
                    "devicePoint": device_point,
                    "measurementType": DEVICE_POINT_TO_TYPE[device_point],
                    "color": MEASUREMENT_COLORS[DEVICE_POINT_TO_TYPE[device_point]],
                    "data": series_data[device_point],
                    "min": extents[device_point][0],
                    "max": extents[device_point][1],
                }
                for device_point in device_points
            ],
            "secondSeries": {
                "name": "周期数据",
                "color": "#f56c6c",
                "sourceMeasurementType": primary_type,
                "sourceDevicePoint": primary,
                "data": second_series_data,
                "finiteCount": second_finite_count,
                "nonZeroCount": second_non_zero_count,
                "min": second_min,
                "max": second_max,
            },
            "angleSeries": {
                "color": "#d59b2b",
                "data": angle_data,
            },
            "volumeSeries": {
                "color": "#4d9e6f",
                "data": volume_data,
                "info": volume_info,
            },
            "cycles": cycles,
            "triggerXs": sorted(set(triggers)),
            "files": files,
            "diagnostics": diagnostics,
        }

    def _build_first_cycle_window(
        self,
        device_part: str,
        device_points: list[str],
        points: list[dict[str, Any]],
        loader: Callable[..., tuple[dict[str, Any], np.ndarray]],
    ) -> dict[str, Any]:
        """Build a slot-layout window where each file shows only its first cycle.

        The primary point is selected in pressure-cap, pressure-shaft, then
        selected-point order. It is the point used for the first-cycle slice;
        files without recorded bounds still appear in the file list.
        The slice for every primary file is fetched in a single JOIN query (range
        scan on the ``(wave_file_id, sample_index)`` primary key). Other selected
        device points contribute no curve but still appear in the file list. The
        curve is continuous: all cycle samples are kept and each file spans its
        own x slot.
        """
        primary = _primary_device_point(device_points)
        primary_type = DEVICE_POINT_TO_TYPE[primary]
        series_data: dict[str, list[dict[str, Any]]] = {point: [] for point in device_points}
        second_series_data: list[dict[str, Any]] = []
        angle_data: list[dict[str, Any]] = []
        volume_data: list[dict[str, Any]] = []
        volume_info: dict[str, Any] | None = None
        cycles: list[dict[str, Any]] = []
        triggers: list[float] = []
        files: list[dict[str, Any]] = []
        diagnostics: list[dict[str, Any]] = []
        second_finite_count = 0
        second_non_zero_count = 0
        second_min: float | None = None
        second_max: float | None = None

        slot_sources: list[tuple[int, dict[str, Any], str, int]] = []
        all_file_ids: set[int] = set()
        for slot, point in enumerate(points):
            for device_point in device_points:
                file_info = point.get("files", {}).get(device_point)
                if not file_info:
                    continue
                file_id = int(file_info["id"] if isinstance(file_info, dict) else file_info)
                slot_sources.append((slot, point, device_point, file_id))
                all_file_ids.add(file_id)

        if not slot_sources:
            return self._assemble_first_cycle_window(
                device_part, device_points, primary, points, series_data, second_series_data,
                angle_data, volume_data, volume_info,
                second_finite_count, second_non_zero_count, second_min, second_max,
                cycles, triggers, files, diagnostics, 0,
            )

        is_demo = loader.__name__ == "_load_demo_wave"
        metas: dict[int, dict[str, Any]] = {}
        # file_id -> (cycle_start, cycle_end, padded_slice[offset, signal, second])
        slices: dict[int, tuple[int, int, np.ndarray]] = {}
        if is_demo:
            for _slot, point, device_point, source_id in slot_sources:
                metadata, samples = self._load_for_window(
                    loader, source_id, DEVICE_POINT_TO_TYPE[device_point],
                    device_part + device_point, point["sampleTime"], {},
                )
                metas[source_id] = metadata
                detected, _ = detect_cycles(samples)
                if detected:
                    cycle = detected[0]
                    start_si = int(samples[cycle.start_offset, 0])
                    end_si = int(samples[cycle.end_offset, 0])
                    pad_end = min(cycle.end_offset + FIRST_CYCLE_PAD, len(samples))
                    slices[source_id] = (start_si, end_si, samples[cycle.start_offset:pad_end])
        else:
            source_ids = [source_id for _, _, _, source_id in slot_sources]
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    placeholders = ", ".join(["%s"] * len(all_file_ids))
                    cursor.execute(
                        f"""
                        SELECT id, point_name, measurement_type, sample_time,
                               sample_count, sample_frequency_hz, rpm, tspluse_status,
                               file_name, cycle_start, cycle_end
                        FROM wave_file
                        WHERE id IN ({placeholders})
                        """,
                        tuple(all_file_ids),
                    )
                    for row in cursor.fetchall():
                        metas[int(row["id"])] = row
                    source_placeholders = ", ".join(["%s"] * len(source_ids))
                    cursor.execute(
                        f"""
                        SELECT ws.wave_file_id,
                               ws.sample_index,
                               CAST(ws.signal_value AS FLOAT) AS sig,
                               CAST(ws.second_value AS FLOAT) AS sec
                        FROM wave_sample_one ws
                        WHERE ws.wave_file_id IN ({source_placeholders})
                        ORDER BY ws.wave_file_id, ws.sample_index
                        """,
                        tuple(source_ids),
                    )
                    raw: dict[int, list[tuple[float, float, float]]] = {}
                    for row in cursor.fetchall():
                        raw.setdefault(int(row["wave_file_id"]), []).append(
                            (
                                float(row["sample_index"]),
                                float(row["sig"]),
                                float(row["sec"]) if row["sec"] is not None else float("nan"),
                            ),
                        )
            for file_id, rows in raw.items():
                meta = metas.get(file_id)
                if meta is None or not rows:
                    continue
                # wave_sample_one 只保留第一个周期，返回的数组本身就是该周期。
                slices[file_id] = (
                    int(rows[0][0]),
                    int(rows[-1][0]) + 1,
                    np.asarray(rows, dtype=float),
                )

        # Per-slot period source: primary preferred, else the first selected
        # point with a file at that slot (timestamps may differ by seconds).
        source_by_slot: dict[int, str] = {}
        for _slot, _point, device_point, _file_id in slot_sources:
            if _slot not in source_by_slot:
                source_by_slot[_slot] = device_point
        for _slot, _point, device_point, _file_id in slot_sources:
            if device_point == primary:
                source_by_slot[_slot] = primary

        for slot, point, device_point, source_id in slot_sources:
            sample_time = point["sampleTime"]
            is_slot_source = device_point == source_by_slot.get(slot, device_point)
            cycle_bounds = slices.get(source_id)
            cycle_start = 0
            cycle_end = 0
            slice_arr: np.ndarray | None = None
            if cycle_bounds is not None:
                cycle_start, cycle_end, slice_arr = cycle_bounds
            cycle_len = len(slice_arr) if slice_arr is not None else 0
            has_cycle = slice_arr is not None and 1 < cycle_len
            if has_cycle:
                angle: np.ndarray | None = None
                volume: np.ndarray | None = None
                point_volume_info: dict[str, Any] | None = None
                stored_cycle = _stored_first_cycle(slice_arr)
                detected = [stored_cycle] if stored_cycle is not None else []
                if detected:
                    cycle = detected[0]
                    angle = cycle.angle
                    volume, point_volume_info = self._build_volume_vector(
                        len(slice_arr),
                        [cycle],
                        device_part + primary,
                    )
                if is_slot_source and point_volume_info is not None:
                    volume_info = point_volume_info
                for offset in range(cycle_len):
                    sample_index = int(slice_arr[offset, 0])
                    raw_value = float(slice_arr[offset, 1])
                    second = _safe_float(slice_arr[offset, 2])
                    angle360 = _safe_float(angle[offset]) if angle is not None and offset < len(angle) else None
                    volume_value = _safe_float(volume[offset]) if volume is not None and offset < len(volume) else None
                    x = slot + (sample_index - cycle_start) / cycle_len
                    series_data[device_point].append(
                        {
                            "value": [x, raw_value],
                            "x": x,
                            "rawValue": raw_value,
                            "sampleIndex": sample_index,
                            "waveFileId": source_id,
                            "sampleTime": sample_time,
                            "secondValue": second,
                            "volume": volume_value,
                            "angle360": angle360,
                        },
                    )
                    if is_slot_source:
                        if volume_value is not None:
                            volume_data.append(
                                {
                                    "value": [x, volume_value],
                                    "x": x,
                                    "volume": volume_value,
                                    "sampleIndex": sample_index,
                                    "waveFileId": source_id,
                                    "sampleTime": sample_time,
                                },
                            )
                        if angle360 is not None:
                            display_angle = angle360 if angle360 <= 180.0 else 360.0 - angle360
                            angle_data.append(
                                {
                                    "value": [x, display_angle],
                                    "x": x,
                                    "angle": display_angle,
                                    "sampleIndex": sample_index,
                                    "waveFileId": source_id,
                                    "sampleTime": sample_time,
                                },
                            )
                        if second is not None:
                            second_finite_count += 1
                            if second != 0:
                                second_non_zero_count += 1
                            second_min = second if second_min is None else min(second_min, second)
                            second_max = second if second_max is None else max(second_max, second)
                            second_series_data.append(
                                {
                                    "value": [x, second],
                                    "x": x,
                                    "rawValue": second,
                                    "sampleIndex": sample_index,
                                    "waveFileId": source_id,
                                    "sampleTime": sample_time,
                                },
                            )
                        if offset > 0:
                            prev = _safe_float(slice_arr[offset - 1, 2])
                            if second is not None and (prev is None or prev < 30) and second >= 30:
                                triggers.append(x)
                cycles.append(
                    {
                        "id": f"{source_id}:1",
                        "waveFileId": source_id,
                        "periodNo": 1,
                        "pointIndex": slot,
                        "sampleTime": sample_time,
                        "startX": float(slot),
                        "endX": float(slot + 1),
                        "startSampleIndex": cycle_start,
                        "endSampleIndex": cycle_end - 1,
                        "sourceType": DEVICE_POINT_TO_TYPE[device_point],
                        "devicePoint": device_point,
                        "background": is_slot_source,
                    },
                )
                diagnostics.append(
                    {
                        "waveFileId": source_id,
                        "measurementType": DEVICE_POINT_TO_TYPE[device_point],
                        "devicePoint": device_point,
                        "sampleTime": sample_time,
                        "cycleStart": cycle_start,
                        "cycleEnd": cycle_end,
                        "cycleSampleCount": cycle_len,
                    },
                )

        for slot, point in enumerate(points):
            for device_point in device_points:
                file_info = point.get("files", {}).get(device_point)
                if not file_info:
                    continue
                file_id = int(file_info["id"] if isinstance(file_info, dict) else file_info)
                metadata = metas.get(file_id)
                cycle_start = metadata.get("cycle_start") if metadata else None
                cycle_end = metadata.get("cycle_end") if metadata else None
                files.append(
                    {
                        "id": file_id,
                        "pointIndex": slot,
                        "sampleTime": point["sampleTime"],
                        "devicePoint": device_point,
                        "measurementType": DEVICE_POINT_TO_TYPE[device_point],
                        "sampleCount": int(
                            (metadata.get("sample_count") if metadata else file_info.get("sampleCount") or 0) or 0,
                        ),
                        "sampleFrequencyHz": int(
                            (metadata.get("sample_frequency_hz") if metadata else file_info.get("sampleFrequencyHz") or 0) or 0,
                        ),
                        "pointName": str(
                            (metadata.get("point_name") if metadata else device_part + device_point)
                            or device_part + device_point,
                        ),
                        "rpm": float((metadata.get("rpm") if metadata else file_info.get("rpm") or 0) or 0),
                        "status": int(
                            (metadata.get("tspluse_status") if metadata else file_info.get("status") or 0) or 0,
                        ),
                        "fileName": str((metadata.get("file_name") if metadata else "") or ""),
                        "cycleStart": int(cycle_start) if cycle_start is not None else None,
                        "cycleEnd": int(cycle_end) if cycle_end is not None else None,
                    },
                )

        return self._assemble_first_cycle_window(
            device_part, device_points, primary, points, series_data, second_series_data,
            angle_data, volume_data, volume_info,
            second_finite_count, second_non_zero_count, second_min, second_max,
            cycles, triggers, files, diagnostics,
            max(len(slot_sources) - len(cycles), 0),
        )

    @staticmethod
    def _assemble_first_cycle_window(
        device_part: str,
        device_points: list[str],
        primary: str,
        points: list[dict[str, Any]],
        series_data: dict[str, list[dict[str, Any]]],
        second_series_data: list[dict[str, Any]],
        angle_data: list[dict[str, Any]],
        volume_data: list[dict[str, Any]],
        volume_info: dict[str, Any] | None,
        second_finite_count: int,
        second_non_zero_count: int,
        second_min: float | None,
        second_max: float | None,
        cycles: list[dict[str, Any]],
        triggers: list[float],
        files: list[dict[str, Any]],
        diagnostics: list[dict[str, Any]],
        missing_cycles: int = 0,
    ) -> dict[str, Any]:
        for device_point in series_data:
            series_data[device_point].sort(key=lambda item: item["x"])
        second_series_data.sort(key=lambda item: item["x"])
        angle_data.sort(key=lambda item: item["x"])
        volume_data.sort(key=lambda item: item["x"])
        extents = {device_point: _series_extent(series_data[device_point]) for device_point in device_points}
        return {
            "devicePart": device_part,
            "devicePoints": device_points,
            "primaryPoint": primary,
            "points": points,
            "xMin": 0,
            "xMax": len(points),
            "series": [
                {
                    "devicePoint": device_point,
                    "measurementType": DEVICE_POINT_TO_TYPE[device_point],
                    "color": MEASUREMENT_COLORS[DEVICE_POINT_TO_TYPE[device_point]],
                    "data": series_data[device_point],
                    "min": extents[device_point][0],
                    "max": extents[device_point][1],
                }
                for device_point in device_points
            ],
            "secondSeries": {
                "name": "周期数据",
                "color": "#f56c6c",
                "sourceMeasurementType": DEVICE_POINT_TO_TYPE[primary],
                "sourceDevicePoint": primary,
                "data": second_series_data,
                "finiteCount": second_finite_count,
                "nonZeroCount": second_non_zero_count,
                "min": second_min,
                "max": second_max,
            },
            "angleSeries": {
                "color": "#d59b2b",
                "data": angle_data,
            },
            "volumeSeries": {
                "color": "#4d9e6f",
                "data": volume_data,
                "info": volume_info,
            },
            "cycles": cycles,
            "triggerXs": sorted(set(triggers)),
            "files": files,
            "diagnostics": diagnostics,
            "firstCycleMode": True,
            "firstCycleNotice": (
                f"窗口内 {missing_cycles} 个文件没有可用的单周期采样数据（wave_sample_one），"
                "多为停机（rpm=0）文件，无法绘制首周期曲线。"
                if missing_cycles > 0
                else None
            ),
        }

    @staticmethod
    def _build_volume_vector(
        sample_count: int,
        cycles: list[DetectedCycle],
        point_name: str,
    ) -> tuple[np.ndarray, dict[str, Any] | None]:
        volume = np.full(sample_count, np.nan, dtype=float)
        cylinder_name = next(
            (name for name in CYLINDER_BORE_MM if name in point_name),
            None,
        )
        if cylinder_name is None or not cycles:
            return volume, None
        bore_mm = CYLINDER_BORE_MM[cylinder_name]
        clearance = CLEARANCE_VOLUME_L_BY_BORE[bore_mm]
        crank_radius = PISTON_STROKE_MM / 2.0
        piston_area = np.pi * (bore_mm / 2.0) ** 2
        for cycle in cycles:
            angle_rad = np.deg2rad(cycle.angle)
            travel = (
                crank_radius * (1.0 - np.cos(angle_rad))
                + CONNECTING_ROD_LENGTH_MM
                - np.sqrt(
                    CONNECTING_ROD_LENGTH_MM**2
                    - (crank_radius * np.sin(angle_rad)) ** 2,
                )
            )
            volume[cycle.start_offset : cycle.end_offset] = (
                clearance + piston_area * travel / 1_000_000.0
            )
        finite = volume[np.isfinite(volume)]
        return volume, {
            "cylinder": cylinder_name,
            "boreMm": bore_mm,
            "clearanceVolumeL": clearance,
            "minVolumeL": float(np.min(finite)) if len(finite) else None,
            "maxVolumeL": float(np.max(finite)) if len(finite) else None,
        }

    @staticmethod
    def _build_period_detail(
        metadata: dict[str, Any],
        samples: np.ndarray,
        period_number: int,
    ) -> dict[str, Any]:
        stored_cycle = _stored_first_cycle(samples) if metadata.get("single_cycle") else None
        if stored_cycle is not None:
            detected = [stored_cycle]
            diagnostics = {"completeCycleCount": 1, "storedFirstCycle": 1}
        else:
            detected, diagnostics = detect_cycles(samples)
        cycle = next((item for item in detected if item.number == period_number), None)
        if cycle is None:
            raise ValueError(f"没有找到周期 {period_number}")

        # 保留 wave_sample_one 的完整周期原始采样点，不再插值成固定 360 点。
        angles360 = np.asarray(cycle.angle, dtype=float)
        pressure = np.asarray(cycle.signal, dtype=float)
        display_angles = np.where(angles360 <= 180.0, angles360, 360.0 - angles360)
        point_name = str(metadata.get("point_name") or "")
        side = metadata.get("device_point")
        if side not in PRESSURE_SIDES:
            side = next((item for item in PRESSURE_SIDES if point_name.endswith(item)), None)
        device_part = metadata.get("device_part")
        if not device_part and side and point_name.endswith(str(side)):
            device_part = point_name[: -len(str(side))]
        cylinder_name = next(
            (name for name in CYLINDER_BORE_MM if name in point_name),
            None,
        )
        volume: np.ndarray | None = None
        volume_info: dict[str, Any] | None = None
        if cylinder_name:
            bore_mm = CYLINDER_BORE_MM[cylinder_name]
            clearance_volume = CLEARANCE_VOLUME_L_BY_BORE[bore_mm]
            angle_rad = np.deg2rad(angles360)
            crank_radius = PISTON_STROKE_MM / 2.0
            piston_travel = (
                crank_radius * (1.0 - np.cos(angle_rad))
                + CONNECTING_ROD_LENGTH_MM
                - np.sqrt(
                    CONNECTING_ROD_LENGTH_MM**2
                    - (crank_radius * np.sin(angle_rad)) ** 2,
                )
            )
            piston_area = np.pi * (bore_mm / 2.0) ** 2
            volume = clearance_volume + piston_area * piston_travel / 1_000_000.0
            volume_info = {
                "cylinder": cylinder_name,
                "boreMm": bore_mm,
                "clearanceVolumeL": clearance_volume,
                "minVolumeL": float(np.min(volume)),
                "maxVolumeL": float(np.max(volume)),
            }

        start_index = int(samples[cycle.start_offset, 0])
        end_offset = min(cycle.end_offset, len(samples) - 1)
        end_index = int(samples[max(cycle.end_offset - 1, cycle.start_offset), 0])
        return {
            "side": side,
            "devicePart": device_part or "",
            "sampleTime": _time_string(metadata.get("sample_time")),
            "waveFile": {
                "id": int(metadata["id"]),
                "devicePart": device_part or "",
                "devicePoint": side,
                "pointName": point_name,
                "measurementType": metadata.get("measurement_type"),
                "sampleTime": _time_string(metadata.get("sample_time")),
                "sampleFrequencyHz": int(metadata.get("sample_frequency_hz") or 0),
                "sampleCount": int(metadata.get("sample_count") or len(samples)),
            },
            "period": {
                "periodNo": cycle.number,
                "startSampleIndex": start_index,
                "endSampleIndex": end_index,
                "sampleCount": int(cycle.end_offset - cycle.start_offset),
                "triggerSampleIndices": [
                    int(samples[offset, 0])
                    for offset in cycle.trigger_offsets
                    if 0 <= offset < len(samples)
                ],
            },
            "angles": display_angles.tolist(),
            "angles360": angles360.tolist(),
            "pressure": pressure.tolist(),
            "volume": volume.tolist() if volume is not None else None,
            "volumeInfo": volume_info,
            "phases": [
                {
                    "name": name,
                    "color": color,
                    "start": start,
                    "end": end,
                }
                for name, color, start, end in PHASES
            ],
            "diagnostics": diagnostics,
        }

    @staticmethod
    def _load_for_window(
        loader: Callable[..., tuple[dict[str, Any], np.ndarray]],
        file_id: int,
        measurement_type: str,
        point_name: str,
        sample_time: str,
        load_cache: dict[int, tuple[dict[str, Any], np.ndarray]],
    ) -> tuple[dict[str, Any], np.ndarray]:
        if file_id not in load_cache:
            if loader.__name__ == "_load_demo_wave":
                load_cache[file_id] = loader(file_id, measurement_type, point_name, sample_time)
            else:
                load_cache[file_id] = loader(file_id)
        return load_cache[file_id]

    def _load_db_wave(self, file_id: int) -> tuple[dict[str, Any], np.ndarray]:
        with get_connection() as connection:
            with connection.cursor() as cursor:
                cursor.execute(
                    """
                    SELECT id, device_part, device_point, point_name, measurement_type,
                           sample_frequency_hz, sample_count, sample_time, rpm, file_name,
                           tspluse_status, cycle_start, cycle_end
                    FROM wave_file
                    WHERE id = %s
                    """,
                    (file_id,),
                )
                metadata = cursor.fetchone()
                if metadata is None:
                    raise ValueError(f"wave_file.id={file_id} 不存在")
                # wave_sample_one 只保留第一个周期，直接读取该文件全部样本。
                cursor.execute(
                    """
                    SELECT sample_index, signal_value, second_value
                    FROM wave_sample_one
                    WHERE wave_file_id = %s
                    ORDER BY sample_index ASC
                    """,
                    (file_id,),
                )
                rows = cursor.fetchall()
        if not rows:
            raise ValueError(f"wave_file.id={file_id} 没有采样数据")
        metadata["single_cycle"] = True
        samples = np.asarray(
            [
                (
                    float(row["sample_index"]),
                    float(row["signal_value"]),
                    float(row["second_value"]) if row["second_value"] is not None else np.nan,
                )
                for row in rows
            ],
            dtype=float,
        )
        return metadata, samples

    @staticmethod
    @lru_cache(maxsize=24)
    def _demo_samples(file_id: int, measurement_type: str) -> np.ndarray:
        count = DEMO_SAMPLE_COUNT
        index = np.arange(count, dtype=float)
        revolution = DEMO_REVOLUTION_SAMPLES
        phase = (index % revolution) / revolution * 2 * np.pi
        second = np.zeros(count, dtype=float)
        for revolution_start in range(0, count, revolution):
            for pulse in range(PULSES_PER_REVOLUTION):
                pulse_start = revolution_start + int(round(pulse * revolution / PULSES_PER_REVOLUTION))
                width = 22 if pulse == 0 else 8
                pulse_end = min(count, pulse_start + width)
                second[pulse_start:pulse_end] = 40.0

        variation = (file_id % 17) / 17.0
        if measurement_type == "压力":
            signal = (
                4.2
                + 1.8 * np.sin(phase - 0.4)
                + 0.55 * np.sin(2 * phase + variation)
                + 0.22 * np.sin(7 * phase)
            )
            signal += 0.2 * np.maximum(np.sin(phase - 0.2), 0) ** 5
        elif measurement_type == "位移":
            signal = 0.5 + 0.18 * np.cos(phase) + 0.035 * np.sin(3 * phase + variation)
        else:
            signal = 0.15 * np.sin(phase * 2 + variation) + 0.04 * np.sin(11 * phase)
            signal += 0.018 * np.cos(index / 37.0)
        return np.column_stack((index, signal, second))

    def _load_demo_wave(
        self,
        file_id: int,
        measurement_type: str,
        point_name: str,
        sample_time: str,
    ) -> tuple[dict[str, Any], np.ndarray]:
        samples = self._demo_samples(file_id, measurement_type)
        metadata = {
            "id": file_id,
            "point_name": point_name,
            "measurement_type": measurement_type,
            "sample_frequency_hz": 25600,
            "sample_count": len(samples),
            "sample_time": sample_time,
            "rpm": 998.0,
            "file_name": f"demo-{file_id}.dat",
        }
        return metadata, samples

    @staticmethod
    def _demo_options() -> list[dict[str, Any]]:
        end = DEMO_START + timedelta(minutes=5 * (DEMO_POINT_COUNT - 1))
        return [
            {
                "devicePart": DEMO_DEVICE_PART,
                "devicePoint": device_point,
                "measurementType": DEVICE_POINT_TO_TYPE[device_point],
                "minTime": _time_string(DEMO_START),
                "maxTime": _time_string(end),
                "fileCount": DEMO_POINT_COUNT,
            }
            for device_point in DEVICE_POINTS
        ]

    @staticmethod
    def _demo_time_points(
        device_part: str,
        device_points: list[str],
        start: datetime | None,
        end: datetime | None,
    ) -> list[dict[str, Any]]:
        points = []
        for index in range(DEMO_POINT_COUNT):
            timestamp = DEMO_START + timedelta(minutes=5 * index)
            if start and timestamp < start:
                continue
            if end and timestamp > end:
                continue
            files = {}
            for device_point in device_points:
                measurement_type = DEVICE_POINT_TO_TYPE[device_point]
                files[device_point] = {
                    "id": DEMO_ID_BY_TYPE[measurement_type] + index,
                    "devicePoint": device_point,
                    "measurementType": measurement_type,
                    "sampleCount": DEMO_SAMPLE_COUNT,
                    "sampleFrequencyHz": 25600,
                    "rpm": 998.0,
                }
            points.append(
                {
                    "index": len(points),
                    "sampleTime": _time_string(timestamp),
                    "files": files,
                },
            )
        return points


data_service = DataService()