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

"""PKS 原始采样曲线：按设备与时间窗口读取 YSJ_5 / YSJ_10 / YSJ_11。

只取运行样本（``YSJ_41 > 0``）；停机造成的长时间空档插入空点，让曲线断开。
"""

from datetime import timedelta
from typing import Any

from db import connect

GAP = timedelta(seconds=60)


def _num(value: Any) -> float | None:
    """0（传感器掉点/停机残留）视为无效，置空以断开曲线。"""
    try:
        number = float(value) if value is not None else None
    except (TypeError, ValueError):
        return None
    if number is None or number <= 0:
        return None
    return number


def query_recent_series(device_code: str, hours: int = 6) -> dict[str, Any]:
    """数据库中最新的 ``hours`` 小时实时数据（以该设备最新采样时间为终点）。"""
    connection = connect()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT MAX(sample_time) AS latest FROM pks_long_sample WHERE device_code = %s",
                (device_code,),
            )
            row = cursor.fetchone()
    finally:
        connection.close()

    latest = row["latest"] if row else None
    if latest is None:
        return {
            "device_code": device_code, "start": None, "end": None, "count": 0,
            "times": [], "oil": [], "coupling": [], "chain": [],
        }
    fmt = "%Y-%m-%d %H:%M:%S"
    start = (latest - timedelta(hours=hours)).strftime(fmt)
    return query_series(device_code, start, latest.strftime(fmt))


def query_series(device_code: str, start: str, end: str) -> dict[str, Any]:
    connection = connect()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                SELECT sample_time, YSJ_5, YSJ_10, YSJ_11
                FROM pks_long_sample
                WHERE device_code = %s AND sample_time >= %s AND sample_time <= %s
                  AND YSJ_41 > 0
                ORDER BY sample_time
                """,
                (device_code, start, end),
            )
            rows = cursor.fetchall()
    finally:
        connection.close()

    times: list[str] = []
    oil: list[float | None] = []
    coupling: list[float | None] = []
    chain: list[float | None] = []
    previous = None
    for row in rows:
        moment = row["sample_time"]
        if previous is not None and (moment - previous) > GAP:
            middle = previous + (moment - previous) / 2
            times.append(middle.strftime("%Y-%m-%d %H:%M:%S"))
            oil.append(None)
            coupling.append(None)
            chain.append(None)
        times.append(moment.strftime("%Y-%m-%d %H:%M:%S"))
        oil.append(_num(row["YSJ_5"]))
        coupling.append(_num(row["YSJ_10"]))
        chain.append(_num(row["YSJ_11"]))
        previous = moment

    return {
        "device_code": device_code,
        "start": start,
        "end": end,
        "count": len(times),
        "times": times,
        "oil": oil,
        "coupling": coupling,
        "chain": chain,
    }