Add parallel processing for individual batch builds

README.md

@@ -281,6 +281,8 @@ Output:
 
 This is intended for quickly converting a large batch of independent single-plate 3MF files into multi-copy SwapMod packs.
 
+The GUI builds these independent outputs in parallel, capped by CPU count and a small safety limit.
+
 It does **not** combine all inputs into one file.
 
 ### Input handling

a1_swap_mod_packer/batch.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import os
+from concurrent.futures import ProcessPoolExecutor, as_completed
+from dataclasses import dataclass
+from typing import Sequence
+
+from .builder import build_packed_3mf
+from .models import BuildOptions, BuildResult, PlateJob
+
+INDIVIDUAL_BATCH_MAX_WORKERS = 8
+
+
+@dataclass(frozen=True)
+class IndividualBuildTask:
+    job: PlateJob
+    options: BuildOptions
+
+
+@dataclass(frozen=True)
+class IndividualBuildFailure:
+    index: int
+    job: PlateJob
+    error: str
+
+
+@dataclass(frozen=True)
+class IndividualBatchBuildResult:
+    results: tuple[BuildResult, ...]
+    failures: tuple[IndividualBuildFailure, ...]
+    worker_count: int
+
+
+def individual_batch_worker_count(task_count: int, max_workers: int | None = None) -> int:
+    if task_count <= 0:
+        return 0
+    if max_workers is not None:
+        return min(task_count, max(1, int(max_workers)))
+    detected_cpu_count = os.cpu_count() or 1
+    return min(task_count, max(1, detected_cpu_count), INDIVIDUAL_BATCH_MAX_WORKERS)
+
+
+def _build_individual_task(task: IndividualBuildTask) -> BuildResult:
+    return build_packed_3mf([task.job], task.options)
+
+
+def _run_individual_batch_serial(
+    tasks: Sequence[IndividualBuildTask],
+    worker_count: int,
+) -> IndividualBatchBuildResult:
+    results: list[BuildResult] = []
+    failures: list[IndividualBuildFailure] = []
+    for index, task in enumerate(tasks):
+        try:
+            results.append(_build_individual_task(task))
+        except Exception as exc:
+            failures.append(IndividualBuildFailure(index, task.job, str(exc)))
+    return IndividualBatchBuildResult(tuple(results), tuple(failures), worker_count)
+
+
+def run_individual_batch_builds(
+    tasks: Sequence[IndividualBuildTask],
+    max_workers: int | None = None,
+) -> IndividualBatchBuildResult:
+    task_list = list(tasks)
+    worker_count = individual_batch_worker_count(len(task_list), max_workers)
+    if worker_count <= 1:
+        return _run_individual_batch_serial(task_list, worker_count)
+
+    ordered_results: list[BuildResult | None] = [None] * len(task_list)
+    failures: list[IndividualBuildFailure] = []
+    with ProcessPoolExecutor(max_workers=worker_count) as executor:
+        future_to_index = {
+            executor.submit(_build_individual_task, task): index
+            for index, task in enumerate(task_list)
+        }
+        for future in as_completed(future_to_index):
+            index = future_to_index[future]
+            task = task_list[index]
+            try:
+                ordered_results[index] = future.result()
+            except Exception as exc:
+                failures.append(IndividualBuildFailure(index, task.job, str(exc)))
+
+    results = tuple(result for result in ordered_results if result is not None)
+    return IndividualBatchBuildResult(
+        tuple(results),
+        tuple(sorted(failures, key=lambda item: item.index)),
+        worker_count,
+    )

a1_swap_mod_packer/gui.py

@@ -6,6 +6,7 @@ import re
 import subprocess
 import sys
 import zipfile
+from multiprocessing import freeze_support
 from pathlib import Path
 from typing import Any, Callable
 
@@ -44,6 +45,7 @@ except ImportError as exc:  # pragma: no cover
     raise SystemExit("PySide6 is required to run the GUI. Install it with: pip install PySide6") from exc
 
 from . import APP_NAME, APP_TITLE
+from .batch import IndividualBuildTask, individual_batch_worker_count, run_individual_batch_builds
 from .core import (
     BuildOptions,
     DEFAULT_ZIP_COMPRESS_LEVEL,
@@ -1122,6 +1124,7 @@ class MainWindow(QMainWindow):
         used_paths: set[Path] = set()
         success_count = 0
         errors: list[str] = []
+        tasks: list[IndividualBuildTask] = []
         for job in jobs:
             try:
                 output_path = make_unique_for_run(
@@ -1129,12 +1132,23 @@ class MainWindow(QMainWindow):
                     used_paths,
                 )
                 options = self.build_options_for_output(output_path)
-                result = build_packed_3mf([job], options)
-                self.log_build_result(result)
-                success_count += 1
             except Exception as exc:
                 errors.append(f"{job.source_3mf.name}: {exc}")
                 self.log.append(f"Error building {job.source_3mf}: {exc}")
+                continue
+            tasks.append(IndividualBuildTask(job, options))
+
+        if tasks:
+            worker_count = individual_batch_worker_count(len(tasks))
+            self.log.append(f"Building {len(tasks)} individual output(s) with {worker_count} worker(s).")
+            batch_result = run_individual_batch_builds(tasks, max_workers=worker_count)
+            for result in batch_result.results:
+                self.log_build_result(result)
+            success_count = len(batch_result.results)
+            for failure in batch_result.failures:
+                errors.append(f"{failure.job.source_3mf.name}: {failure.error}")
+                self.log.append(f"Error building {failure.job.source_3mf}: {failure.error}")
+
         if errors:
             QMessageBox.warning(
                 self,
@@ -1166,6 +1180,7 @@ class MainWindow(QMainWindow):
 
 
 def main() -> int:
+    freeze_support()
     app = QApplication(sys.argv)
     app.setFont(QFont("Segoe UI", 10))
     window = MainWindow()

tests/test_batch.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import a1_swap_mod_packer.batch as batch
+from a1_swap_mod_packer.models import BuildOptions, BuildResult, PlateJob
+
+
+class IndividualBatchBuildTest(unittest.TestCase):
+    def test_worker_count_uses_task_count_and_caps_cpu_count(self) -> None:
+        self.assertEqual(batch.individual_batch_worker_count(0), 0)
+        self.assertEqual(batch.individual_batch_worker_count(3, max_workers=99), 3)
+        self.assertEqual(batch.individual_batch_worker_count(3, max_workers=0), 1)
+        with patch.object(batch.os, "cpu_count", return_value=32):
+            self.assertEqual(batch.individual_batch_worker_count(20), batch.INDIVIDUAL_BATCH_MAX_WORKERS)
+        with patch.object(batch.os, "cpu_count", return_value=None):
+            self.assertEqual(batch.individual_batch_worker_count(20), 1)
+
+    def test_serial_runner_keeps_success_order_and_collects_failures(self) -> None:
+        tasks = [
+            self.task("one.3mf", "one.out.3mf", copies=2),
+            self.task("fail.3mf", "fail.out.3mf", copies=3),
+            self.task("two.3mf", "two.out.3mf", copies=4),
+        ]
+
+        def fake_build(jobs: list[PlateJob], options: BuildOptions) -> BuildResult:
+            job = jobs[0]
+            if job.source_3mf.name == "fail.3mf":
+                raise ValueError("planned failure")
+            return BuildResult(
+                output_3mf=options.output_3mf,
+                plate_count=job.copies,
+                total_prediction_seconds=None,
+                total_weight_grams=None,
+                gcode_md5=job.source_3mf.stem,
+            )
+
+        with patch.object(batch, "build_packed_3mf", side_effect=fake_build):
+            result = batch.run_individual_batch_builds(tasks, max_workers=1)
+
+        self.assertEqual(result.worker_count, 1)
+        self.assertEqual(
+            [item.output_3mf.name for item in result.results],
+            ["one.out.3mf", "two.out.3mf"],
+        )
+        self.assertEqual([item.plate_count for item in result.results], [2, 4])
+        self.assertEqual(len(result.failures), 1)
+        self.assertEqual(result.failures[0].index, 1)
+        self.assertEqual(result.failures[0].job.source_3mf.name, "fail.3mf")
+        self.assertIn("planned failure", result.failures[0].error)
+
+    @staticmethod
+    def task(source_name: str, output_name: str, copies: int) -> batch.IndividualBuildTask:
+        return batch.IndividualBuildTask(
+            PlateJob(Path(source_name), copies),
+            BuildOptions(swap_gcode=Path("swap.gcode"), output_3mf=Path(output_name)),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()