Compressor/viewer/backend/services/indicator_service.py

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

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

首周期本身即一个完整冲程。7/8/9 号机按
``月度报告/沁水压缩机参数20250102.docx`` 的相位角把共享键相角旋转到各缸自身
0°（显示角 = 共享角 − 相位角）；4/5/6 号机暂无资料，保持原逻辑。
轴侧容积扣除活塞杆直径 78mm，盖侧用全缸径。
"""

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
PISTON_ROD_DIAMETER_MM = 78.0
CLEARANCE_VOLUME_L_BY_BORE = {490.0: 1.51, 390.0: 0.74, 360.0: 0.62}
# 7/8/9 号机各缸相位角（月度报告/沁水压缩机参数20250102.docx）。
CYLINDER_PHASE_DEGREE = {"一": 0.0, "二": 0.0, "三": 120.0, "四": 240.0, "五": 240.0, "六": 120.0}
CYLINDER_PHASE_OVERRIDE = {"二": 24.0, "三": 146.0, "四": 147.0, "五": 265.0, "六": 265.0}
PHASE_ROTATION_UNITS = frozenset({"7", "8", "9"})
PULSES_PER_REVOLUTION = 8
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 piecewise_angle(trigger: np.ndarray, count: int) -> np.ndarray:
    """Crank angle from the 8 keyphase pulses; falls back to uniform sampling."""
    high = np.nan_to_num(np.asarray(trigger, dtype=float), nan=-np.inf) >= 30.0
    changes = np.diff(np.r_[False, high, False].astype(np.int8))
    offsets = [int(offset) for offset in np.flatnonzero(changes == 1)]
    if len(offsets) < PULSES_PER_REVOLUTION or offsets[0] != 0:
        return np.linspace(0.0, 360.0, count, endpoint=False)
    offsets = offsets[:PULSES_PER_REVOLUTION]
    bounds = offsets + [count]
    angle = np.full(count, np.nan, dtype=float)
    for pulse in range(PULSES_PER_REVOLUTION):
        start, end = bounds[pulse], min(bounds[pulse + 1], count)
        if end <= start:
            continue
        angle[start:end] = np.linspace(
            pulse * 45.0, (pulse + 1) * 45.0, end - start, endpoint=False,
        )
    if np.isnan(angle).any():
        good = np.flatnonzero(~np.isnan(angle))
        if not len(good):
            return np.linspace(0.0, 360.0, count, endpoint=False)
        angle = np.interp(np.arange(count), good, angle[good])
    return angle


def keyphase_angle(keyphase_info) -> float | None:
    if not keyphase_info:
        return None
    parts = str(keyphase_info).strip().split("-")
    try:
        return float(parts[-1])
    except (TypeError, ValueError):
        return None


def phase_degrees(unit: str, cylinder: str, keyphase_info=None) -> float:
    if unit not in PHASE_ROTATION_UNITS:
        return 0.0
    if cylinder in CYLINDER_PHASE_OVERRIDE:
        return CYLINDER_PHASE_OVERRIDE[cylinder]
    angle = keyphase_angle(keyphase_info)
    if angle is not None:
        return angle
    return CYLINDER_PHASE_DEGREE[cylinder]


def volume_from_angle(
    cylinder: str,
    side: str,
    unit: str,
    angle_degree: np.ndarray,
) -> np.ndarray:
    bore = CYLINDER_BORE_MM[cylinder]
    clearance = CLEARANCE_VOLUME_L_BY_BORE[bore]
    crank_radius = PISTON_STROKE_MM / 2.0
    rod_length = CONNECTING_ROD_LENGTH_MM
    rod_diameter = (
        PISTON_ROD_DIAMETER_MM
        if side == "轴侧" and unit in PHASE_ROTATION_UNITS
        else 0.0
    )
    theta = np.deg2rad(np.asarray(angle_degree, dtype=float))
    travel = (
        crank_radius * (1.0 - np.cos(theta))
        + rod_length
        - np.sqrt(rod_length**2 - (crank_radius * np.sin(theta)) ** 2)
    )
    piston_area = np.pi * ((bore / 2.0) ** 2 - (rod_diameter / 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, keyphase_info 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, second_value FROM wave_sample_one "
                "WHERE wave_file_id = %s ORDER BY sample_index",
                (file_id,),
            )
            samples = [
                (float(item["signal_value"]), item["second_value"])
                for item in cursor.fetchall()
                if item["signal_value"] is not None
            ]
    finally:
        connection.close()

    if len(samples) < MIN_SAMPLES:
        return None
    pressure = np.asarray([item[0] for item in samples], dtype=float)
    trigger = np.asarray(
        [float(item[1]) if item[1] is not None else np.nan for item in samples],
        dtype=float,
    )
    angle = piecewise_angle(trigger, len(pressure))
    phase = phase_degrees(unit, cylinder, row.get("keyphase_info"))
    if phase:
        angle = np.mod(angle - phase, 360.0)
    volume = volume_from_angle(cylinder, side, unit, angle)
    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"]