周期回填提速：增量窗口读取+并行检测；split_device_point 去掉 rpm 限制全表补齐 device_part/device_point

backend/LabelingPreTraining/detect_cycle_index.py

@@ -63,8 +63,10 @@ import predict  # noqa: E402
 from predict import STAGE_LIMITS  # noqa: E402
 
 
-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.
+def _fetch_chunk(
+    connection, file_ids: list[int], limit: int, lo: int = 0
+) -> dict[int, np.ndarray]:
+    """Fetch rows in ``[lo, limit)`` of a chunk of files in one query.
 
     Returns {file_id: (n, 4) array} with columns
     [array_offset(0..n-1), signal_value, second_value, sample_index].
@@ -80,10 +82,11 @@ def _fetch_chunk(connection, file_ids: list[int], limit: int) -> dict[int, np.nd
                    CAST(second_value AS FLOAT) AS sec,
                    sample_index
             FROM wave_sample
-            WHERE wave_file_id IN ({placeholders}) AND sample_index < %s
+            WHERE wave_file_id IN ({placeholders})
+              AND sample_index >= %s AND sample_index < %s
             ORDER BY wave_file_id, sample_index
             """,
-            (*file_ids, limit),
+            (*file_ids, lo, limit),
         )
         rows = cursor.fetchall()
     grouped: dict[int, tuple[list[float], list[float], list[int]]] = {}
@@ -111,16 +114,17 @@ def read_batch_samples(
     file_ids: list[int],
     workers: int,
     limit: int,
+    lo: int = 0,
 ) -> dict[int, np.ndarray]:
-    """Fetch the first ``limit`` rows of every file in one round-trip."""
+    """Fetch rows in ``[lo, limit)`` of every file in one round-trip."""
     if not file_ids:
         return {}
     if workers <= 1 or len(file_ids) <= workers:
-        return _fetch_chunk(reader_connections[0], file_ids, limit)
+        return _fetch_chunk(reader_connections[0], file_ids, limit, lo)
     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(), limit)
+            executor.submit(_fetch_chunk, reader_connections[i], chunk.tolist(), limit, lo)
             for i, chunk in enumerate(chunks)
         ]
         merged: dict[int, np.ndarray] = {}
@@ -196,35 +200,69 @@ def resolve_batch_bounds(
 ) -> tuple[dict[int, tuple[int, int] | None], int]:
     """Adaptively detect the first cycle per file, escalating window sizes.
 
+    Each escalation re-feeds the same growing prefix to ``detect_cycles`` as the
+    original implementation (the zero-marker threshold is computed over the runs
+    visible in that prefix), but only the rows that were not seen before are
+    downloaded again: stage deltas are appended to an in-memory prefix instead of
+    re-fetching ``[0, limit)`` on every stage. Detection itself runs in parallel
+    across the worker connections.
+
     Returns (``{file_id: (start, end) | None}``, error_count). None means no
     complete cycle was found at any window (-> -1).
     """
     errors = 0
     resolved: dict[int, tuple[int, int] | None] = {}
-    stage_files: dict[int, dict] = {int(row["id"]): row for row in rows}
+    pending: dict[int, dict] = {int(row["id"]): row for row in rows}
+    prefixes: dict[int, np.ndarray] = {}
+
     for stage in range(first_stage, len(STAGE_LIMITS)):
-        if not stage_files:
+        if not pending:
             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
+        lo = 0 if stage == first_stage else STAGE_LIMITS[stage - 1]
+        deltas = read_batch_samples(reader_connections, list(pending), workers, limit, lo=lo)
+        detect_workers = min(workers, len(pending))
+
+        def work(item: tuple[int, dict]) -> tuple[int, str | tuple[int, int]]:
+            file_id, row = item
+            delta = deltas.get(file_id)
+            if delta is None or len(delta) == 0:
+                return file_id, "empty"
+            prefix = prefixes.get(file_id)
+            prefix = delta if prefix is None else np.vstack([prefix, delta])
+            prefixes[file_id] = prefix
             try:
-                bounds = first_cycle_bounds(samples)
+                bounds = first_cycle_bounds(prefix)
             except Exception:
+                return file_id, "error"
+            if bounds is not None:
+                return file_id, bounds
+            if int(row.get("sample_count") or 0) > limit:
+                return file_id, "pending"
+            return file_id, "none"
+
+        results: dict[int, str | tuple[int, int]] = {}
+        with ThreadPoolExecutor(max_workers=detect_workers) as executor:
+            for file_id, result in executor.map(work, pending.items()):
+                results[file_id] = result
+
+        next_pending: dict[int, dict] = {}
+        for file_id, row in pending.items():
+            result = results[file_id]
+            if isinstance(result, tuple):
+                resolved[file_id] = result
+            elif result == "pending":
+                next_pending[file_id] = row
+                continue
+            elif result == "error":
                 errors += 1
                 resolved[file_id] = None
-                continue
-            if bounds is None and int(row.get("sample_count") or 0) > limit:
-                next_pending[file_id] = row
             else:
-                resolved[file_id] = bounds
-        stage_files = next_pending
-    for file_id in stage_files:
+                resolved[file_id] = None
+            prefixes.pop(file_id, None)
+        pending = next_pending
+
+    for file_id in pending:
         resolved[file_id] = None
     return resolved, errors
 

backend/LabelingPreTraining/split_device_point.py

@@ -9,9 +9,10 @@ is reported and the write is aborted, so no data is ever lost or mangled.
 
 The mapping is computed locally from the distinct point_names, then written in
 a single ``UPDATE ... CASE point_name WHEN ... END`` (one pass over the table),
-limited to ``rpm > 0`` rows per the confirmed scope. ``device_point`` is added
-with an ``ALTER TABLE`` if it does not exist yet. Run without ``--commit`` to
-only preview the 78 mappings.
+covering every row whose point_name can be split (no rpm filter, so rpm<=0/NULL
+rows are backfilled too; already-correct rows are idempotently rewritten with
+the same values). ``device_point`` is added with an ``ALTER TABLE`` if it does
+not exist yet. Run without ``--commit`` to only preview the mappings.
 
 Usage:
     conda activate tspulse
@@ -113,7 +114,7 @@ def main() -> int:
             UPDATE wave_file
             SET device_part = CASE point_name {case_part} END,
                 device_point = CASE point_name {case_point} END
-            WHERE rpm > 0 AND point_name IN ({placeholders})
+            WHERE point_name IN ({placeholders})
         """
         params: list[str] = []
         for name in names:
@@ -124,7 +125,7 @@ def main() -> int:
 
         with connection.cursor() as cursor:
             affected = cursor.execute(sql, tuple(params))
-        print(f"已回写 {affected} 行（rpm>0）。")
+        print(f"已回写 {affected} 行（point_name 可拆的全部行）。")
         return 0