"""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())