Compressor/backend/app/services/data_service.py

from __future__ import annotations

import re
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 = ("压力盖侧", "压力轴侧", "活塞杆沉降", "十字头振动", "自由端振动", "驱动端振动")
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 = ("正常", "异常")

# PKS 全场点位：机组号 -> pks_long_sample.import_batch_id
# 7号机=30、8号机=31、9号机=32。
UNIT_BATCH = {"7": 30, "8": 31, "9": 32}
# 全场点位(PKS)每个时间点的曲线窗口：以采样时刻为中心的前后各 7.5 分钟。
PKS_WINDOW_SECONDS = 15 * 60
# 仅 PKS 时间点返回给时间条的最大锚点数：超过后按等步长均匀采样（首尾必保）。
PKS_STRIP_MAX_ANCHORS = 5000
_SITE_POINT_PATTERN = re.compile(r"^YSJ([789])_([1-9]|1[0-9]|2[0-9]|3[0-9]|4[0-1])$")
# 不作为全场点位(PKS)下拉候选的序号（对应机组自己的运行状态/转速信号）。
_EXCLUDED_SITE_INDEXES = frozenset({33, 34, 35, 41})

# 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 _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 _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:
    """从 机组与部位 提取机组号（7/8/9），非 7/8/9 机组返回 None。"""
    match = re.match(r"^([789])号机组", 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]] = {}

    @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))
        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 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_range(self, unit: str) -> tuple[datetime, datetime] | None:
        """机组 pks 批次的时间覆盖范围（仅查一次并缓存）。"""
        cached = self._pks_batch_bounds.get(unit)
        if cached is not None:
            return cached
        batch = UNIT_BATCH[unit]
        with get_connection() as connection:
            with connection.cursor() as cursor:
                cursor.execute(
                    "SELECT MIN(sample_time) AS lo, MAX(sample_time) AS hi "
                    "FROM pks_long_sample WHERE import_batch_id = %s",
                    (batch,),
                )
                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]:
        """机组(7/8/9)的全场点位记录（site_point 表中 YSJ{机组号}_1..41）。

        附带该机组 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, 42)
                if n not in _EXCLUDED_SITE_INDEXES
            }
            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),
        }

    @staticmethod
    def _attach_site_values(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
        ]
        if not columns or not points:
            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 = %s AND sample_time IN ({placeholders})",
                        (UNIT_BATCH[unit], *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

    @staticmethod
    def _build_site_series(
        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))
        if not columns or not centers:
            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 "
                    "WHERE import_batch_id = %s AND sample_time >= %s AND sample_time < %s "
                    "ORDER BY sample_time",
                    (UNIT_BATCH[unit], query_start, query_end),
                )
                rows = cursor.fetchall()
        sample_times = [row["sample_time"] for row in rows]
        series = []
        for item_name, column in columns:
            values = [row[column] for row in rows]
            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),
        }

    @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 直挂所选各列
        原值。停机/运转只由该机组批次内的 YSJ_41 决定（>0=运转、=0=停机），
        未勾选停机时仅保留运转锚点；勾选停机则停机锚点一并显示。
        """
        batch = UNIT_BATCH[unit]
        clauses = ["import_batch_id = %s"]
        params: list[Any] = [batch]
        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)
        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 import_batch_id = %s AND sample_time IN ({placeholders})",
                        (batch, *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
                running_value = row["YSJ_41"]
                running_by_time[anchor_time] = bool(
                    running_value is not None and running_value > 0
                )
        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 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 和周期编号必须为正整数")

        def database_query():
            metadata, samples = self._load_db_wave(wave_file_id)
            return self._build_period_detail(metadata, samples, period_number)

        def demo_query():
            metadata, samples = self._load_demo_wave(
                wave_file_id,
                "压力",
                DEMO_POINT_NAME,
                DEMO_START,
            )
            return self._build_period_detail(metadata, samples, period_number)

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

    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]] = {}

        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_cycles(source_samples)
                    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_cycles(samples)
                        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 ws
                        JOIN wave_file wf ON wf.id = ws.wave_file_id
                        WHERE ws.wave_file_id IN ({source_placeholders})
                          AND wf.cycle_start >= 0
                          AND ws.sample_index >= wf.cycle_start
                          AND ws.sample_index < wf.cycle_end + %s
                        ORDER BY ws.wave_file_id, ws.sample_index
                        """,
                        (*tuple(source_ids), FIRST_CYCLE_PAD),
                    )
                    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 meta.get("cycle_start") is None:
                    continue
                slices[file_id] = (
                    int(meta["cycle_start"]),
                    int(meta["cycle_end"]),
                    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 = max(cycle_end - cycle_start, 0)
            has_cycle = slice_arr is not None and 1 < cycle_len <= len(slice_arr)
            if has_cycle:
                angle: np.ndarray | None = None
                volume: np.ndarray | None = None
                point_volume_info: dict[str, Any] | None = None
                detected, _ = detect_cycles(slice_arr)
                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} 个文件尚未回写首个周期索引（cycle_start），"
                "请先运行 detect_cycle_index.py 回写后再查看。"
                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]:
        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}")

        angles360 = np.linspace(0.0, 359.0, 360)
        pressure = np.interp(angles360, cycle.angle, cycle.signal)
        display_angles = np.where(angles360 <= 180.0, angles360, 360.0 - angles360)
        point_name = str(metadata.get("point_name") or "")
        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 {
            "waveFile": {
                "id": int(metadata["id"]),
                "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, point_name, measurement_type, sample_frequency_hz,
                           sample_count, sample_time, rpm, file_name, tspluse_status
                    FROM wave_file
                    WHERE id = %s
                    """,
                    (file_id,),
                )
                metadata = cursor.fetchone()
                if metadata is None:
                    raise ValueError(f"wave_file.id={file_id} 不存在")
                cursor.execute(
                    """
                    SELECT sample_index, signal_value, second_value
                    FROM wave_sample
                    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} 没有采样数据")
        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()