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):
    """Return {file_id: (median_span, has_samples)} from wave_sample_one."""
    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"]
                span = float(mx) - float(mn) if mn is not None and mx is not None else 0.0
                result[int(row["wave_file_id"])] = (span, True)
    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"])))
                all_days = sorted(by_day)
                has_days = [d for d in all_days if any(v is not None for v in by_day[d])]
                valid_days = [d for d in all_days
                              if np.median([v[0] for v in by_day[d] if v is not None] or [0.0]) >= MIN_SPAN]
                print(f"{point}: 文件 {len(files)}，抽样日 {len(all_days)}，"
                      f"有 wave_sample_one 的日 {len(has_days)}，其中有效周期日 {len(valid_days)}")
                if has_days:
                    print(f"   有样本数据段: {has_days[0]} ~ {has_days[-1]}  共 {len(has_days)} 天")
    finally:
        connection.close()


if __name__ == "__main__":
    main()