Compressor/backend/indicator_features/availability.py

#!/usr/bin/env python3
"""Scan where indicator-diagram cycles actually exist (read-only).

For each point it samples a few files per day *without* an rpm filter (rpm is
unreliable for wave_file) and checks whether a valid first cycle exists in
wave_sample_one with a real pressure span.  This finds the last data segment
before a fault, which may be weeks before the repair record.
"""
from __future__ import annotations

import argparse
import sys
from collections import defaultdict
from pathlib import Path

import numpy as np
import pymysql

BACKEND = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND))

from app.config import settings  # noqa: E402

CYLINDERS = ("一", "二", "三", "四", "五", "六")
SIDES = ("盖侧", "轴侧")
MIN_SPAN = 0.3


def connect():
    return pymysql.connect(
        host=settings.db_host, port=settings.db_port, user=settings.db_user,
        password=settings.db_password, database=settings.db_name, charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor, connect_timeout=8, read_timeout=300,
    )


def point_name(unit, cylinder, side):
    return f"{unit}号机组{cylinder}缸压力{side}"


def sample_per_day(rows, per_day):
    by_day = defaultdict(list)
    for row in rows:
        by_day[row["sample_time"].date()].append(row)
    picked = []
    for day in sorted(by_day):
        day_rows = by_day[day]
        if len(day_rows) <= per_day:
            picked.extend(day_rows)
        else:
            idx = np.linspace(0, len(day_rows) - 1, per_day).round().astype(int)
            picked.extend(day_rows[int(i)] for i in idx)
    return picked


def spans(connection, file_ids):
    """Median pressure span of wave_sample_one per file_id."""
    result = {}
    with connection.cursor() as cursor:
        for start in range(0, len(file_ids), 200):
            chunk = file_ids[start:start + 200]
            ph = ", ".join(["%s"] * len(chunk))
            cursor.execute(
                f"SELECT wave_file_id, COUNT(*) n, MIN(signal_value) mn, MAX(signal_value) mx "
                f"FROM wave_sample_one WHERE wave_file_id IN ({ph}) GROUP BY wave_file_id",
                tuple(chunk),
            )
            for row in cursor.fetchall():
                mn, mx = row["mn"], row["mx"]
                result[int(row["wave_file_id"])] = float(mx) - float(mn) if mn is not None and mx is not None else 0.0
    return result


def main():
    parser = argparse.ArgumentParser(description="扫描功图周期实际存在的日期")
    parser.add_argument("--unit", required=True, choices=("7", "8", "9"))
    parser.add_argument("--cylinders", nargs="+", default=list(CYLINDERS), choices=CYLINDERS)
    parser.add_argument("--sides", nargs="+", default=["盖侧"], choices=SIDES)
    parser.add_argument("--start", required=True)
    parser.add_argument("--end", required=True)
    parser.add_argument("--per-day", type=int, default=2)
    args = parser.parse_args()

    connection = connect()
    try:
        for cylinder in args.cylinders:
            for side in args.sides:
                point = point_name(args.unit, cylinder, side)
                with connection.cursor() as cursor:
                    cursor.execute(
                        "SELECT id, sample_time FROM wave_file WHERE point_name=%s AND measurement_type='压力' "
                        "AND sample_time >= %s AND sample_time < %s ORDER BY sample_time",
                        (point, args.start, args.end),
                    )
                    files = list(cursor.fetchall())
                picked = sample_per_day(files, args.per_day)
                span_by_id = spans(connection, [int(r["id"]) for r in picked])
                by_day = defaultdict(list)
                for row in picked:
                    by_day[row["sample_time"].date()].append(span_by_id.get(int(row["id"]), 0.0))
                valid_days = [d for d, v in sorted(by_day.items()) if np.median(v) >= MIN_SPAN and len(span_by_id)]
                all_days = sorted(by_day)
                print(f"{point}: 文件 {len(files)}，抽样日 {len(all_days)}，其中有效日 {len(valid_days)}")
                if valid_days:
                    print(f"   有效数据段: {valid_days[0]} ~ {valid_days[-1]}  共 {len(valid_days)} 天")
                    print("   有效日:", ", ".join(str(d) for d in valid_days))
    finally:
        connection.close()


if __name__ == "__main__":
    main()