feat: implement parallel processing for batch feature extraction to improve performance

app/training/extract_features_batch.py

@@ -208,41 +208,64 @@ def extract_batch(
         for row in reader:
             samples.append(row)
 
-    print(f"Extracting features from {len(samples)} samples...")
+    # Parallel processing via multiprocessing.Pool
+    import multiprocessing as mp
+    import os as _os
+    import time as _time
+
+    n_workers = max(1, (_os.cpu_count() or 4) - 1)
+    print(f"Extracting features from {len(samples)} samples using {n_workers} workers...", flush=True)
 
     out_columns = ["file_path", "label_int"] + FEATURE_COLUMNS
     success = 0
     failed = 0
+    t_start = _time.time()
+
+    tasks = [(s["file_path"], int(s["label_int"])) for s in samples]
+
+    def _worker(args):
+        audio_path, label_int = args
+        features = extract_sample_features(audio_path)
+        if features is None:
+            return None
+        features["file_path"] = audio_path
+        features["label_int"] = label_int
+        return features
 
     with open(output_path, "w", newline="", encoding="utf-8") as f:
         writer = csv.DictWriter(f, fieldnames=out_columns)
         writer.writeheader()
-
-        for i, sample in enumerate(samples):
-            audio_path = sample["file_path"]
-            label_int = int(sample["label_int"])
-
-            features = extract_sample_features(audio_path)
-            if features is None:
-                failed += 1
-                continue
-
-            features["file_path"] = audio_path
-            features["label_int"] = label_int
-            writer.writerow(features)
-            success += 1
-
-            if (i + 1) % 50 == 0:
-                print(
-                    f"  [{i + 1}/{len(samples)}] "
-                    f"success={success}, failed={failed}"
-                )
-
+        f.flush()
+
+        with mp.Pool(processes=n_workers) as pool:
+            for i, result in enumerate(
+                pool.imap_unordered(_worker, tasks, chunksize=4), 1
+            ):
+                if result is None:
+                    failed += 1
+                    continue
+                writer.writerow(result)
+                success += 1
+
+                if i % 25 == 0:
+                    f.flush()
+                    elapsed = _time.time() - t_start
+                    rate = i / elapsed if elapsed > 0 else 0
+                    eta = (len(samples) - i) / rate if rate > 0 else 0
+                    print(
+                        f"  [{i}/{len(samples)}] "
+                        f"ok={success} fail={failed} "
+                        f"rate={rate:.1f}/s eta={eta / 60:.1f}m",
+                        flush=True,
+                    )
+
+    elapsed = _time.time() - t_start
     print(
         f"\nDone: {success} extracted, "
-        f"{failed} failed"
+        f"{failed} failed in {elapsed / 60:.1f}m",
+        flush=True,
     )
-    print(f"Output: {output_path}")
+    print(f"Output: {output_path}", flush=True)
 
     return output_path