Compressor/3d/backend/services/indicator_service.py

"""压缩机示功图（P-V）。

读取波库中该测点**最近一次运行**（``rpm>0``）的 ``wave_file`` 首个周期
（``wave_sample_one``），按缸径/行程/连杆/余隙把曲柄转角换算为容积（L），
返回盖侧 / 轴侧两副压力功图（P-V）。

首周期本身即一个完整冲程，样本按时间顺序对应曲柄角 0→360°，直接成闭环，
无需额外相位旋转（与 backend/app/services/data_service.py 的示功图一致）。
"""

from __future__ import annotations

import re
from typing import Any

import numpy as np

from db import connect

CYLINDERS = ("一", "二", "三", "四", "五", "六")
SIDES = ("盖侧", "轴侧")
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}
MIN_SAMPLES = 60


def unit_number(device_code: str) -> str | None:
    match = re.search(r"([789])", device_code or "")
    return match.group(1) if match else None


def volume_from_angle(cylinder: str, count: int) -> np.ndarray:
    bore = CYLINDER_BORE_MM[cylinder]
    clearance = CLEARANCE_VOLUME_L_BY_BORE[bore]
    crank_radius = PISTON_STROKE_MM / 2.0
    rod = CONNECTING_ROD_LENGTH_MM
    theta = np.linspace(0.0, 2.0 * np.pi, count, endpoint=False)
    travel = (
        crank_radius * (1.0 - np.cos(theta))
        + rod
        - np.sqrt(rod**2 - (crank_radius * np.sin(theta)) ** 2)
    )
    piston_area = np.pi * (bore / 2.0) ** 2
    return clearance + piston_area * travel / 1_000_000.0


def _side(unit: str, cylinder: str, side: str) -> dict[str, Any] | None:
    point = f"{unit}号机组{cylinder}缸压力{side}"
    connection = connect()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                SELECT id, sample_time, rpm FROM wave_file
                WHERE point_name = %s AND measurement_type = '压力' AND rpm > 0
                ORDER BY sample_time DESC
                LIMIT 1
                """,
                (point,),
            )
            row = cursor.fetchone()
            if not row:
                return None
            file_id = int(row["id"])
            cursor.execute(
                "SELECT signal_value FROM wave_sample_one WHERE wave_file_id = %s ORDER BY sample_index",
                (file_id,),
            )
            values = [
                float(item["signal_value"])
                for item in cursor.fetchall()
                if item["signal_value"] is not None
            ]
    finally:
        connection.close()

    if len(values) < MIN_SAMPLES:
        return None
    pressure = np.asarray(values, dtype=float)
    volume = volume_from_angle(cylinder, len(pressure))
    angle = np.linspace(0.0, 360.0, len(pressure), endpoint=False)
    sample_time = row["sample_time"]
    return {
        "side": side,
        "point_name": point,
        "wave_file_id": file_id,
        "sample_time": sample_time.strftime("%Y-%m-%d %H:%M:%S") if sample_time else None,
        "rpm": float(row["rpm"]) if row["rpm"] is not None else None,
        "pressure": [round(float(value), 4) for value in pressure.tolist()],
        "volume": [round(float(value), 5) for value in volume.tolist()],
        "angle": [round(float(value), 2) for value in angle.tolist()],
    }


def query_indicator(device_code: str, cylinder: str) -> dict[str, Any]:
    unit = unit_number(device_code)
    if unit is None:
        raise ValueError(f"不支持的机组：{device_code}")
    if cylinder not in CYLINDERS:
        raise ValueError(f"不支持的缸号：{cylinder}")
    sides = [item for item in (_side(unit, cylinder, side) for side in SIDES) if item]
    return {"device_code": device_code, "unit": unit, "cylinder": cylinder, "sides": sides}


__all__ = ["CYLINDERS", "SIDES", "query_indicator"]