Use adaptive sample windows in predict and add predict_re for mislabelled no-cycle files

backend/LabelingPreTraining/predict.py

@@ -1,11 +1,13 @@
 """Step 3: long-running TSPulse anomaly detection poller.
 
 Polls the wave_file table for rows matching ``rpm > 0 AND measurement_type =
-'压力'``, reads only the first 3000 wave_sample rows per file, locates the
-first complete crankshaft cycle, and encodes it with the frozen TSPulse model
-into a 128-dim fingerprint. The fingerprint is compared against the part's
-centroid and radius from the baseline table and ``tspluse_status`` is written
-back as a proportional integer of the centroid distance:
+'压力'``, reads each file's samples with an adaptive multi-stage window
+(``STAGE_LIMITS`` = 3500 -> 7000 -> 14000 rows), escalating until a complete
+crankshaft cycle is found or the file's data is exhausted, and encodes the
+first complete cycle with the frozen TSPulse model into a 128-dim fingerprint.
+The fingerprint is compared against the part's centroid and radius from the
+baseline table and ``tspluse_status`` is written back as a proportional
+integer of the centroid distance:
 
 * 无周期 (no complete cycle found) -> ``-1``
 * 正常 (distance <= 2 x radius)   -> ``0``
@@ -15,18 +17,21 @@ Every processed file is written back, so a full re-run from an empty watermark
 cleanly replaces any previously stored levels.
 
 The samples of a whole batch of files are read with one ``IN (...) AND
-sample_index < 3000`` query and all candidate cycles are encoded in a single
-forward pass, so the network round trips and MPS kernel launches are amortised
-across the batch. Files are processed in ``(sample_time, id)`` ascending order
-and a tuple watermark on those two columns prevents re-processing files that
-have already been seen, so the poller first drains the backlog and then keeps
-watching for newly inserted rows.
+sample_index < {stage_limit}`` query per stage and all candidate cycles are
+encoded in a single forward pass, so the network round trips and MPS kernel
+launches are amortised across the batch. Only files that still report no cycle
+at the current stage are re-fetched at the next, larger window; a file is
+finally marked 无周期 only when no window yields a cycle. Files are processed
+in ``(sample_time, id)`` ascending order and a tuple watermark on those two
+columns prevents re-processing files that have already been seen, so the
+poller first drains the backlog and then keeps watching for newly inserted
+rows.
 
 Usage:
     conda activate tspulse
     python predict.py [--start-time ""] [--start-id 0] [--interval 60]
                       [--batch 100] [--limit 0] [--report-every 100]
-                      [--write-batch 10] [--dry-run]
+                      [--write-batch 100] [--dry-run]
 """
 
 import argparse
@@ -50,7 +55,7 @@ HERE = Path(__file__).resolve().parent
 CHECKPOINT_PATH = HERE / "checkpoints" / "tspulse_frozen.pt"
 BASELINE_PATH = HERE / "baseline" / "baseline.npz"
 
-SAMPLE_LIMIT = 3000
+STAGE_LIMITS = (3500, 7000, 14000)
 MEASUREMENT_TYPE = "压力"
 
 ANOMALY_FACTOR = 2.0
@@ -74,8 +79,8 @@ def load_baseline() -> tuple[dict[str, int], np.ndarray, np.ndarray]:
     return part_index, data["centroids"], data["radii"]
 
 
-def _fetch_chunk(connection, file_ids: list[int]) -> dict[int, np.ndarray]:
-    """Fetch the first SAMPLE_LIMIT rows of a chunk of files in one query."""
+def _fetch_chunk(connection, file_ids: list[int], limit: int) -> dict[int, np.ndarray]:
+    """Fetch the first ``limit`` rows of a chunk of files in one query."""
     if not file_ids:
         return {}
     placeholders = ",".join(["%s"] * len(file_ids))
@@ -89,7 +94,7 @@ def _fetch_chunk(connection, file_ids: list[int]) -> dict[int, np.ndarray]:
             WHERE wave_file_id IN ({placeholders}) AND sample_index < %s
             ORDER BY wave_file_id, sample_index
             """,
-            (*file_ids, SAMPLE_LIMIT),
+            (*file_ids, limit),
         )
         rows = cursor.fetchall()
     grouped: dict[int, tuple[list[float], list[float]]] = {}
@@ -110,8 +115,9 @@ def read_batch_samples(
     reader_connections: list,
     file_ids: list[int],
     workers: int,
+    limit: int,
 ) -> dict[int, np.ndarray]:
-    """Fetch the first SAMPLE_LIMIT rows of every file in one round-trip.
+    """Fetch the first ``limit`` rows of every file in one round-trip.
 
     The batch is split into ``workers`` chunks fetched on parallel persistent
     connections. Returns {file_id: (n, 3) array} with columns
@@ -120,11 +126,11 @@ def read_batch_samples(
     if not file_ids:
         return {}
     if workers <= 1 or len(file_ids) <= workers:
-        return _fetch_chunk(reader_connections[0], file_ids)
+        return _fetch_chunk(reader_connections[0], file_ids, limit)
     chunks = np.array_split(file_ids, min(workers, len(file_ids)))
     with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
         futures = [
-            executor.submit(_fetch_chunk, reader_connections[i], chunk.tolist())
+            executor.submit(_fetch_chunk, reader_connections[i], chunk.tolist(), limit)
             for i, chunk in enumerate(chunks)
         ]
         merged: dict[int, np.ndarray] = {}
@@ -168,7 +174,7 @@ def fetch_batch(
         if watermark_time is None:
             cursor.execute(
                 """
-                SELECT id, point_name, measurement_type, sample_time
+                SELECT id, point_name, measurement_type, sample_time, sample_count
                 FROM wave_file
                 WHERE rpm > 0 AND measurement_type = %s
                 ORDER BY sample_time ASC, id ASC
@@ -179,7 +185,7 @@ def fetch_batch(
         else:
             cursor.execute(
                 """
-                SELECT id, point_name, measurement_type, sample_time
+                SELECT id, point_name, measurement_type, sample_time, sample_count
                 FROM wave_file
                 WHERE rpm > 0 AND measurement_type = %s
                   AND (sample_time > %s OR (sample_time = %s AND id > %s))
@@ -191,6 +197,55 @@ def fetch_batch(
         return cursor.fetchall()
 
 
+def resolve_batch_cycles(
+    rows: list[dict],
+    part_of: dict[int, str],
+    part_index: dict[str, int],
+    reader_connections: list,
+    workers: int,
+    first_stage: int = 0,
+) -> tuple[dict[int, tuple[str, np.ndarray | None]], int]:
+    """Adaptively detect one cycle per file with progressively larger windows.
+
+    The whole batch is first fetched at ``STAGE_LIMITS[first_stage]``. Files
+    that still report 无周期 and have more rows than the current window are
+    re-fetched at the next stage, up to ``STAGE_LIMITS[-1]``; a file is only
+    finally 无周期 when no window yields a cycle. ``first_stage`` lets callers
+    (predict_re) skip cheap windows for files already known to need more data.
+
+    Returns (``{file_id: (status, matrix)}``, error_count). Status is "ok" or a
+    skip reason, mirroring ``prepare_cycle``.
+    """
+    errors = 0
+    resolved: dict[int, tuple[str, np.ndarray | None]] = {}
+    stage_files: dict[int, dict] = {int(row["id"]): row for row in rows}
+    for stage in range(first_stage, len(STAGE_LIMITS)):
+        if not stage_files:
+            break
+        limit = STAGE_LIMITS[stage]
+        samples_map = read_batch_samples(reader_connections, list(stage_files), workers, limit)
+        next_pending: dict[int, dict] = {}
+        for file_id, row in stage_files.items():
+            samples = samples_map.get(file_id)
+            if samples is None or len(samples) == 0:
+                resolved[file_id] = ("无样本", None)
+                continue
+            try:
+                status, matrix = prepare_cycle(samples, part_of[file_id], part_index)
+            except Exception:
+                errors += 1
+                resolved[file_id] = ("错误", None)
+                continue
+            if status == "无周期" and int(row.get("sample_count") or 0) > limit:
+                next_pending[file_id] = row
+            else:
+                resolved[file_id] = (status, matrix)
+        stage_files = next_pending
+    for file_id in stage_files:
+        resolved[file_id] = ("无周期", None)
+    return resolved, errors
+
+
 def compute_status(distance: float, radius: float) -> int:
     """Map a fingerprint distance to a proportional integer status.
 
@@ -327,26 +382,18 @@ def main() -> int:
 
                 file_ids = [int(row["id"]) for row in rows]
                 part_of = {int(row["id"]): f"{row['point_name']}_{row['measurement_type']}" for row in rows}
-                sample_map = read_batch_samples(reader_connections, file_ids, read_workers)
+                resolved, batch_errors = resolve_batch_cycles(
+                    rows, part_of, part_index, reader_connections, read_workers,
+                )
+                errors += batch_errors
 
                 candidates: list[tuple[int, str, np.ndarray]] = []
                 for row in rows:
                     file_id = int(row["id"])
-                    part = part_of[file_id]
-                    samples = sample_map.get(file_id)
-                    if samples is None or len(samples) == 0:
-                        status, matrix = "无样本", None
-                    else:
-                        try:
-                            status, matrix = prepare_cycle(samples, part, part_index)
-                        except Exception:
-                            errors += 1
-                            status, matrix = "错误", None
-
                     total += 1
-
+                    status, matrix = resolved[file_id]
                     if status == "ok":
-                        candidates.append((file_id, part, matrix))
+                        candidates.append((file_id, part_of[file_id], matrix))
                     elif status == "无周期":
                         stats["无周期"] += 1
                         write_buffer.append((file_id, -1))

backend/LabelingPreTraining/predict_re.py

@@ -0,0 +1,226 @@
+"""Re-run TSPulse detection for files currently marked 无周期 (tspluse_status = -1).
+
+The original full re-run used a fixed 3000-sample prefix per file. Files that
+start mid-revolution have too few trigger pulses inside that prefix for the
+zero-marker segmentation to find a complete cycle, so they were mislabelled -1
+even though the client and the training generator (which read the whole file)
+detect them normally.
+
+This script re-processes ONLY the rows still holding ``tspluse_status = -1``,
+using predict.py's adaptive multi-stage window but starting at the 7000-sample
+stage (these files already failed the smaller windows). Files that still yield
+no cycle stay -1; the rest are written their corrected status (0 or the
+proportional anomaly value).
+
+The query filters on ``tspluse_status = -1``, so the ``(sample_time, id)``
+watermark is naturally idempotent: a file already fixed is no longer selected
+on resume. Run it once after the full re-run finishes; no column reset needed.
+
+Usage:
+    conda activate tspulse
+    python predict_re.py [--start-time ""] [--start-id 0] [--batch 100]
+                         [--limit 0] [--report-every 100] [--write-batch 100]
+                         [--dry-run]
+"""
+
+import argparse
+import sys
+
+import numpy as np
+import torch
+
+import predict
+from predict import (
+    MEASUREMENT_TYPE,
+    STAGE_LIMITS,
+    compute_status,
+    encode_batch,
+    flush_status,
+    load_baseline,
+    load_model,
+    resolve_batch_cycles,
+)
+
+
+def fetch_batch(
+    connection,
+    watermark_time: object | None,
+    watermark_id: int,
+    batch: int,
+) -> list[dict]:
+    with connection.cursor() as cursor:
+        if watermark_time is None:
+            cursor.execute(
+                """
+                SELECT id, point_name, measurement_type, sample_time, sample_count
+                FROM wave_file
+                WHERE rpm > 0 AND measurement_type = %s AND tspluse_status = -1
+                ORDER BY sample_time ASC, id ASC
+                LIMIT %s
+                """,
+                (MEASUREMENT_TYPE, batch),
+            )
+        else:
+            cursor.execute(
+                """
+                SELECT id, point_name, measurement_type, sample_time, sample_count
+                FROM wave_file
+                WHERE rpm > 0 AND measurement_type = %s AND tspluse_status = -1
+                  AND (sample_time > %s OR (sample_time = %s AND id > %s))
+                ORDER BY sample_time ASC, id ASC
+                LIMIT %s
+                """,
+                (MEASUREMENT_TYPE, watermark_time, watermark_time, watermark_id, batch),
+            )
+        return cursor.fetchall()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="重新处理 tspluse_status = -1 的文件")
+    parser.add_argument("--start-time", type=str, default="",
+                        help="起始水位时间(含)，留空从头开始")
+    parser.add_argument("--start-id", type=int, default=0,
+                        help="起始水位 id(含)")
+    parser.add_argument("--batch", type=int, default=100, help="每轮读取的文件数")
+    parser.add_argument("--read-workers", type=int, default=4, help="并行读样本的连接数")
+    parser.add_argument("--limit", type=int, default=0, help="最多处理文件数，0 表示不限")
+    parser.add_argument("--report-every", type=int, default=100, help="每 N 个文件打印一次汇总")
+    parser.add_argument("--write-batch", type=int, default=100,
+                        help="回写攒满 N 条才批量写一次")
+    parser.add_argument("--dry-run", action="store_true", help="只预测不回写数据库")
+    args = parser.parse_args()
+
+    device = predict.get_device()
+    print(f"设备: {predict.describe(device)}")
+    model, mean, std = load_model(device)
+    part_index, centroids, radii = load_baseline()
+    print(f"基准表: {len(part_index)} 个部位, 重新处理 tspluse_status = -1 的文件, "
+          f"自适应窗口: {STAGE_LIMITS[1:]} (跳过 {STAGE_LIMITS[0]})")
+
+    start_time = args.start_time.strip()
+    watermark_time: object | None = start_time or None
+    watermark_id = args.start_id
+    total = 0
+    stats = {"正常": 0, "异常": 0, "无周期": 0}
+    skipped = {"无基准": 0, "坏数据": 0, "周期过长": 0, "无样本": 0}
+    errors = 0
+    last_reported = 0
+    write_buffer: list[tuple[int, int]] = []
+
+    read_workers = max(1, min(args.read_workers, args.batch))
+    reader_connections = [predict.get_connection() for _ in range(read_workers)]
+
+    def flush(connection) -> None:
+        if args.dry_run:
+            write_buffer.clear()
+        else:
+            flush_status(connection, write_buffer)
+
+    def summary_text() -> str:
+        return (
+            f"已处理 {total}, 正常 {stats['正常']}, 异常 {stats['异常']}, "
+            f"无周期 {stats['无周期']}, 跳过 {sum(skipped.values())}({dict(skipped)}), "
+            f"错误 {errors}, 水位 sample_time={watermark_time} id={watermark_id}"
+        )
+
+    print(f"水位起点: sample_time={start_time or '从头'}, id={args.start_id}, "
+          f"批大小: {args.batch}, 读并行: {read_workers} 连接"
+          + (", 干跑(不回写)" if args.dry_run else ""))
+
+    try:
+        while True:
+            connection = predict.get_connection()
+            try:
+                rows = fetch_batch(connection, watermark_time, watermark_id, args.batch)
+                if not rows:
+                    flush(connection)
+                    connection.close()
+                    print(f"无待处理的 -1 文件，处理完成 [{summary_text()}]")
+                    return 0
+
+                file_ids = [int(row["id"]) for row in rows]
+                part_of = {int(row["id"]): f"{row['point_name']}_{row['measurement_type']}" for row in rows}
+                resolved, batch_errors = resolve_batch_cycles(
+                    rows, part_of, part_index, reader_connections, read_workers,
+                    first_stage=1,
+                )
+                errors += batch_errors
+
+                candidates: list[tuple[int, str, np.ndarray]] = []
+                for row in rows:
+                    file_id = int(row["id"])
+                    total += 1
+                    status, matrix = resolved[file_id]
+                    if status == "ok":
+                        candidates.append((file_id, part_of[file_id], matrix))
+                    elif status == "无周期":
+                        stats["无周期"] += 1
+                        write_buffer.append((file_id, -1))
+                    else:
+                        if status in skipped:
+                            skipped[status] += 1
+                        write_buffer.append((file_id, 0))
+
+                if candidates:
+                    fingerprints = encode_batch(
+                        model, mean, std, device, [item[2] for item in candidates]
+                    )
+                    for (file_id, part, _matrix), fingerprint in zip(candidates, fingerprints):
+                        index = part_index[part]
+                        distance = float(np.linalg.norm(fingerprint - centroids[index]))
+                        radius = float(radii[index])
+                        status = compute_status(distance, radius)
+                        stats["异常" if status > 0 else "正常"] += 1
+                        write_buffer.append((file_id, status))
+                        if len(write_buffer) >= args.write_batch:
+                            flush(connection)
+
+                if total // args.report_every > last_reported:
+                    last_reported = total // args.report_every
+                    print(f"  汇总: {summary_text()}", flush=True)
+
+                if args.limit and total >= args.limit:
+                    flush(connection)
+                    watermark_time = rows[-1]["sample_time"]
+                    watermark_id = int(rows[-1]["id"])
+                    connection.close()
+                    print(f"已达 --limit={args.limit}，退出")
+                    print(f"  汇总: {summary_text()}")
+                    return 0
+
+                flush(connection)
+                watermark_time = rows[-1]["sample_time"]
+                watermark_id = int(rows[-1]["id"])
+            finally:
+                try:
+                    connection.close()
+                except Exception:
+                    pass
+
+    except KeyboardInterrupt:
+        if write_buffer and not args.dry_run:
+            try:
+                with predict.get_connection() as connection:
+                    flush_status(connection, write_buffer)
+            except Exception:
+                pass
+        for connection in reader_connections:
+            try:
+                connection.close()
+            except Exception:
+                pass
+        print(f"\n已中断。{summary_text()}")
+        if watermark_time is not None:
+            print(f"续跑命令: python predict_re.py --start-time \"{watermark_time}\" --start-id {watermark_id}")
+        return 0
+
+    for connection in reader_connections:
+        try:
+            connection.close()
+        except Exception:
+            pass
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())