Compressor/backend/LabelingPreTraining/validate_samples.py

"""Sample validation for the generated per-cycle pretraining samples.

Cross-checks a sampled subset of the generated .npy files against the database:

1. 周期数核对: npy 文件数与 manifest 中的 cycle_count 一致。
2. 顺序核对: 每个 npy 内的信号列与 DB 中 [start, end) 区间的原始行完全一致,
   且 sample_index 逐行递增（时间顺序未打乱）。
3. 边界核对: 周期起点/终点处的 second_value >= 30（键相触发），
   第 3、4 列全为 1，信号列为有限值。
4. 命名核对: 文件名中的 sample_index 起止与 DB 实测起止一致。

Usage: python3 validate_samples.py [--all] [--max-files N]
"""

import argparse
import csv
import re
import sys
from pathlib import Path

import numpy as np
from tqdm import tqdm

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

from app.algorithms.cycles import detect_cycles  # noqa: E402
from app.db import get_connection  # noqa: E402

HERE = Path(__file__).resolve().parent
OUTPUT_DIR = HERE / "inputdatas"
MANIFEST_PATH = HERE / "manifest.csv"

NAME_RE = re.compile(r"^(\d+)_(\d+)_(\d+)_(\d+)\.npy$")


def load_manifest() -> list[dict[str, str]]:
    with MANIFEST_PATH.open(encoding="utf-8-sig") as handle:
        return list(csv.DictReader(handle))


def list_npy_files(folder: Path, file_id: int | None = None) -> list[Path]:
    pattern = f"{file_id}_*.npy" if file_id is not None else "*.npy"
    return sorted(folder.glob(pattern))


def read_wave_samples(connection, file_id: int) -> np.ndarray:
    with connection.cursor() as cursor:
        cursor.execute(
            """
            SELECT sample_index, signal_value, second_value
            FROM wave_sample
            WHERE wave_file_id = %s
            ORDER BY wave_file_id, sample_index
            """,
            (file_id,),
        )
        rows = cursor.fetchall()
    return np.asarray(
        [
            (
                float(row["sample_index"]),
                float(row["signal_value"]),
                float(row["second_value"]) if row["second_value"] is not None else np.nan,
            )
            for row in rows
        ],
        dtype=float,
    )


def validate_file(connection, manifest_row: dict[str, str]) -> list[str]:
    problems: list[str] = []
    file_id = int(manifest_row["wave_file_id"])
    folder = OUTPUT_DIR / manifest_row["group"]
    files = sorted(folder.glob(f"{file_id}_*.npy"))
    expected_count = int(manifest_row["cycle_count"] or 0)
    if len(files) != expected_count:
        problems.append(
            f"周期数不符: manifest={expected_count}, 实际 npy 文件={len(files)}"
        )
        return problems

    samples = read_wave_samples(connection, file_id)
    cycles, _ = detect_cycles(samples)
    if len(cycles) != expected_count:
        problems.append(f"重算周期数不符: detect_cycles={len(cycles)}, manifest={expected_count}")

    by_number: dict[int, Path] = {}
    parsed: list[tuple[int, int, int]] = []
    for path in files:
        match = NAME_RE.match(path.name)
        if not match:
            problems.append(f"文件名不符合规范: {path.name}")
            continue
        file_id_name, cycle_number, start_name, end_name = (int(part) for part in match.groups())
        if file_id_name != file_id:
            problems.append(f"文件名 id 不符: {path.name}")
        by_number[cycle_number] = path
        parsed.append((cycle_number, start_name, end_name))

    for cycle in cycles:
        start_sample_index, end_exclusive = (
            int(samples[cycle.start_offset, 0]),
            int(samples[cycle.end_offset, 0]) if cycle.end_offset < len(samples) else int(samples[-1, 0]) + 1,
        )
        path = by_number.get(cycle.number)
        if path is None:
            problems.append(f"周期 {cycle.number} 缺少 npy 文件")
            continue
        expected_name = f"{file_id}_{cycle.number}_{start_sample_index}_{end_exclusive}.npy"
        if path.name != expected_name:
            problems.append(
                f"周期 {cycle.number} 文件名不符: 应为 {expected_name}, 实际 {path.name}"
            )

        matrix = np.load(path)
        if matrix.ndim != 2 or matrix.shape[1] != 4:
            problems.append(f"周期 {cycle.number} 矩阵形状异常: {matrix.shape}")
            continue
        if not np.all(np.isfinite(matrix[:, :2])):
            problems.append(f"周期 {cycle.number} 信号列存在非有限值")

        signal = samples[cycle.start_offset : cycle.end_offset, 1]
        second = samples[cycle.start_offset : cycle.end_offset, 2]
        if matrix.shape[0] != len(signal):
            problems.append(
                f"周期 {cycle.number} 行数不符: npy={matrix.shape[0]}, DB={len(signal)}"
            )
        if not np.array_equal(matrix[:, 0], signal):
            problems.append(f"周期 {cycle.number} signal_value 与 DB 不一致")
        if not np.array_equal(matrix[:, 1], second):
            problems.append(f"周期 {cycle.number} second_value 与 DB 不一致")
        if not np.all(matrix[:, 2] == 1) or not np.all(matrix[:, 3] == 1):
            problems.append(f"周期 {cycle.number} 扩展列不为 1")

        indices = samples[cycle.start_offset : cycle.end_offset, 0]
        if len(indices) > 1 and np.any(np.diff(indices) <= 0):
            problems.append(f"周期 {cycle.number} sample_index 未严格递增")

        boundary_start = samples[cycle.start_offset, 2]
        boundary_end = samples[cycle.end_offset, 2]
        if boundary_start < 30 or boundary_end < 30:
            problems.append(
                f"周期 {cycle.number} 零标记边界未达触发阈值 30 (start={boundary_start:.3f}, end={boundary_end:.3f})"
            )

    return problems


def main() -> int:
    parser = argparse.ArgumentParser(description="抽样验证预训练样本数据")
    parser.add_argument("--all", action="store_true", help="验证全部文件而非抽样")
    parser.add_argument("--max-files", type=int, default=20, help="抽样文件数上限（默认 20）")
    args = parser.parse_args()

    manifest = [row for row in load_manifest() if not row.get("error")]
    if not manifest:
        print("manifest 中没有成功记录，请先运行 generate_samples.py")
        return 1

    if args.all:
        sampled = manifest
    else:
        step = max(1, len(manifest) // args.max_files)
        sampled = manifest[::step][: args.max_files]
        if len(sampled) < len(manifest):
            sampled.append(manifest[-1])

    print(f"抽样验证 {len(sampled)}/{len(manifest)} 个文件: " + ", ".join(item["wave_file_id"] for item in sampled))

    problems_by_id: dict[str, list[str]] = {}
    connection = get_connection()
    try:
        for row in tqdm(sampled, desc="验证样本", unit="文件"):
            problems = validate_file(connection, row)
            if problems:
                problems_by_id[row["wave_file_id"]] = problems
                print(f"  id={row['wave_file_id']} ({row['group']}): {len(problems)} 个问题")
                for problem in problems[:8]:
                    print(f"    - {problem}")
    finally:
        connection.close()

    checked = sum(1 for item in sampled if item["wave_file_id"] not in problems_by_id)
    print(
        f"\n结果: {checked}/{len(sampled)} 个文件完全通过"
        if not problems_by_id
        else f"\n结果: {checked}/{len(sampled)} 通过, {len(problems_by_id)} 个文件存在问题"
    )
    return 1 if problems_by_id else 0


if __name__ == "__main__":
    sys.exit(main())