Compressor/backend/app/services/data_service.py

from __future__ import annotations

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",
}
PHASES = (
    ("排气", "#7b1fa2", 0.0, 120.0),
    ("压缩", "#c62828", 120.0, 195.0),
    ("膨胀", "#1565c0", 195.0, 270.0),
    ("进气", "#2e7d32", 270.0, 330.0),
)
ANNOTATION_LABELS = ("正常", "异常")

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_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 _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_measurement_types(values: list[str] | tuple[str, ...] | None) -> list[str]:
    selected = list(values or MEASUREMENT_TYPES)
    invalid = [value for value in selected if value not in MEASUREMENT_TYPES]
    if invalid:
        raise ValueError(f"不支持的数据名称：{'、'.join(invalid)}")
    return [value for value in MEASUREMENT_TYPES if value in selected]


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

    @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():
            placeholders = ", ".join(["%s"] * len(MEASUREMENT_TYPES))
            with get_connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute(
                        f"""
                        SELECT point_name, measurement_type,
                               MAX(sample_time) AS max_time,
                               MIN(sample_time) AS min_time,
                               COUNT(*) AS file_count
                        FROM wave_file
                        WHERE measurement_type IN ({placeholders})
                        GROUP BY point_name, measurement_type
                        ORDER BY point_name, measurement_type
                        """,
                        MEASUREMENT_TYPES,
                    )
                    rows = cursor.fetchall()
            return [
                {
                    "pointName": row["point_name"],
                    "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)
        point_names = list(dict.fromkeys(row["pointName"] for row in rows))
        return {
            "source": source,
            "measurementTypes": list(MEASUREMENT_TYPES),
            "pointNames": point_names,
            "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
                        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 time_points(
        self,
        point_name: str,
        measurement_types: 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,
    ) -> dict[str, Any]:
        if not point_name.strip():
            raise ValueError("机组与部位不能为空")
        types = _validate_measurement_types(measurement_types)
        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("开始时间不能晚于结束时间")

        def database_query():
            type_placeholders = ", ".join(["%s"] * len(types))
            clauses = [
                "point_name = %s",
                f"measurement_type IN ({type_placeholders})",
            ]
            params: list[Any] = [point_name, *types]
            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, 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:
                reference_where = [
                    "point_name = %s",
                    "measurement_type = '压力'",
                    "tspluse_status = 0",
                ]
                reference_params: list[Any] = [point_name]
                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, 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(point_name, types, start, end), []

        result, source = self._run_with_fallback(database_query, demo_query)
        points, reference = result
        return {
            "source": source,
            "pointName": point_name,
            "measurementTypes": types,
            "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:
            key = row["sample_time"]
            point = grouped.setdefault(
                key,
                {
                    "sampleTime": _time_string(key),
                    "files": {},
                },
            )
            point["files"][row["measurement_type"]] = {
                "id": int(row["id"]),
                "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 wave_window(
        self,
        point_name: str,
        measurement_types: list[str],
        points: list[dict[str, Any]],
        max_points: int,
        no_sampling: bool = False,
    ) -> dict[str, Any]:
        types = _validate_measurement_types(measurement_types)
        if not points:
            raise ValueError("至少选择一个时间点")
        if len(points) > 200:
            raise ValueError("单次最多预览 200 个时间点，请缩小时间窗口")
        max_points = min(max(int(max_points), 256), 200000)

        def database_query():
            return self._build_wave_window(
                point_name,
                types,
                points,
                max_points,
                self._load_db_wave,
                no_sampling,
            )

        def demo_query():
            return self._build_wave_window(
                point_name,
                types,
                points,
                max_points,
                self._load_demo_wave,
                no_sampling,
            )

        result, source = self._run_with_fallback(database_query, demo_query)
        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,
        point_name: str,
        types: 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]]] = {measurement_type: [] for measurement_type in types}
        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
        source_type = "压力" if "压力" in types else types[0]
        load_cache: dict[int, tuple[dict[str, Any], np.ndarray]] = {}

        for slot, point in enumerate(points):
            source_measurement_type = source_type
            source_file = point.get("files", {}).get(source_measurement_type)
            if not source_file:
                available = [
                    (measurement_type, file_info)
                    for measurement_type, file_info in point.get("files", {}).items()
                    if measurement_type in types and file_info
                ]
                if available:
                    source_measurement_type, source_file = available[0]
                else:
                    continue
            source_id = int(source_file["id"] if isinstance(source_file, dict) else source_file)
            try:
                source_metadata, source_samples = self._load_for_window(
                    loader,
                    source_id,
                    source_measurement_type,
                    point_name,
                    point["sampleTime"],
                    load_cache,
                )
            except ValueError:
                # A selected file may legitimately have no samples. A database
                # connection error must escape and activate the demo fallback.
                continue

            detected, diagnostic = detect_cycles(source_samples)
            angle_vector = build_angle_vector(len(source_samples), detected)
            full_angle_vector = np.full(len(source_samples), np.nan, dtype=float)
            for detected_cycle in detected:
                full_angle_vector[detected_cycle.start_offset:detected_cycle.end_offset] = detected_cycle.angle
            current_volume, current_volume_info = self._build_volume_vector(
                len(source_samples),
                detected,
                point_name,
            )
            if current_volume_info is not None:
                volume_info = current_volume_info
            sample_indices = source_samples[:, 0].astype(np.int64)
            sample_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)
            required = {0, len(source_samples) - 1}
            for cycle in detected:
                required.update(
                    {
                        cycle.start_offset,
                        max(cycle.end_offset - 1, cycle.start_offset),
                        *cycle.trigger_offsets,
                    },
                )
                start_x = slot + cycle.start_offset / sample_span
                end_x = slot + cycle.end_offset / sample_span
                start_sample_index = int(sample_indices[cycle.start_offset])
                end_sample_index = int(sample_indices[max(cycle.end_offset - 1, cycle.start_offset)])
                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": start_sample_index,
                        "endSampleIndex": end_sample_index,
                        "sourceType": source_measurement_type,
                    },
                )
            for trigger_offset in sorted(required):
                if trigger_offset in required and any(
                    trigger_offset == run_offset
                    for cycle in detected
                    for run_offset in cycle.trigger_offsets
                ):
                    triggers.append(slot + trigger_offset / sample_span)

            diagnostics.append(
                {
                    "waveFileId": source_id,
                    "measurementType": source_measurement_type,
                    "sampleTime": point["sampleTime"],
                    **diagnostic,
                },
            )

            for measurement_type in types:
                file_info = point.get("files", {}).get(measurement_type)
                if not file_info:
                    continue
                file_id = int(file_info["id"] if isinstance(file_info, dict) else file_info)
                if file_id == source_id:
                    metadata, samples = source_metadata, source_samples
                else:
                    try:
                        metadata, samples = self._load_for_window(
                            loader,
                            file_id,
                            measurement_type,
                            point_name,
                            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}
                if file_id == source_id:
                    file_required.update(required)
                if no_sampling:
                    target_per_file = sample_count
                else:
                    target_per_file = max(256, int(np.ceil(max_points / max(len(points), 1))))
                chosen = downsample_indices(samples[:, 1], target_per_file, file_required)
                for offset in chosen:
                    x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span
                    second = _safe_float(samples[offset, 2])
                    raw_value = float(samples[offset, 1])
                    series_data[measurement_type].append(
                        {
                            "value": [x, raw_value],
                            "x": x,
                            "rawValue": raw_value,
                            "sampleIndex": int(sample_indices_for_file[offset]),
                            "waveFileId": file_id,
                            "sampleTime": point["sampleTime"],
                            "secondValue": second,
                            "volume": (
                                _safe_float(current_volume[offset])
                                if file_id == source_id
                                else None
                            ),
                            "angle360": (
                                _safe_float(full_angle_vector[offset])
                                if file_id == source_id
                                else None
                            ),
                        },
                    )
                if file_id == source_id:
                    second_chosen = downsample_indices(samples[:, 2], target_per_file, file_required)
                    for offset in second_chosen:
                        second_value = _safe_float(samples[offset, 2])
                        if second_value is None:
                            continue
                        x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span
                        second_series_data.append(
                            {
                                "value": [x, second_value],
                                "x": x,
                                "rawValue": second_value,
                                "sampleIndex": int(sample_indices_for_file[offset]),
                                "waveFileId": file_id,
                                "sampleTime": point["sampleTime"],
                            },
                        )
                    angle_chosen = downsample_indices(angle_vector, target_per_file, file_required)
                    for offset in angle_chosen:
                        value = _safe_float(angle_vector[offset])
                        if value is None:
                            continue
                        x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span
                        angle_data.append(
                            {
                                "value": [x, value],
                                "x": x,
                                "angle": value,
                                "sampleIndex": int(sample_indices_for_file[offset]),
                                "waveFileId": file_id,
                                "sampleTime": point["sampleTime"],
                            },
                        )
                    volume_chosen = downsample_indices(current_volume, target_per_file, file_required)
                    for offset in volume_chosen:
                        value = _safe_float(current_volume[offset])
                        if value is None:
                            continue
                        x = slot + (int(sample_indices_for_file[offset]) - file_base_index) / file_span
                        volume_data.append(
                            {
                                "value": [x, value],
                                "x": x,
                                "volume": value,
                                "sampleIndex": int(sample_indices_for_file[offset]),
                                "waveFileId": file_id,
                                "sampleTime": point["sampleTime"],
                            },
                        )
                files.append(
                    {
                        "id": file_id,
                        "pointIndex": slot,
                        "sampleTime": point["sampleTime"],
                        "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 point_name),
                        "rpm": float(metadata.get("rpm") or 0),
                        "status": int(metadata.get("tspluse_status") or 0),
                        "fileName": str(metadata.get("file_name") or ""),
                    },
                )
            # Keep at most the current time point's raw arrays resident. The
            # response retains only downsampled values and period metadata.
            load_cache.clear()

        for measurement_type in series_data:
            series_data[measurement_type].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"])
        return {
            "pointName": point_name,
            "measurementTypes": types,
            "points": points,
            "xMin": 0,
            "xMax": len(points),
            "series": [
                {
                    "measurementType": measurement_type,
                    "color": MEASUREMENT_COLORS[measurement_type],
                    "data": series_data[measurement_type],
                }
                for measurement_type in types
            ],
            "secondSeries": {
                "name": "周期数据",
                "color": "#f56c6c",
                "sourceMeasurementType": source_type,
                "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,
        }

    @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 [
            {
                "pointName": DEMO_POINT_NAME,
                "measurementType": measurement_type,
                "minTime": _time_string(DEMO_START),
                "maxTime": _time_string(end),
                "fileCount": DEMO_POINT_COUNT,
            }
            for measurement_type in MEASUREMENT_TYPES
        ]

    @staticmethod
    def _demo_time_points(
        point_name: str,
        types: 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 = {
                measurement_type: {
                    "id": DEMO_ID_BY_TYPE[measurement_type] + index,
                    "sampleCount": DEMO_SAMPLE_COUNT,
                    "sampleFrequencyHz": 25600,
                    "rpm": 998.0,
                }
                for measurement_type in types
            }
            points.append(
                {
                    "index": len(points),
                    "sampleTime": _time_string(timestamp),
                    "files": files,
                },
            )
        return points


data_service = DataService()