From 4ba654ca3fb2c0e49c02290db3ad25a61d88defd Mon Sep 17 00:00:00 2001 From: hyyz17200 Date: Fri, 8 May 2026 12:17:27 +0800 Subject: [PATCH] Add parallel processing for individual batch builds --- README.md | 2 + a1_swap_mod_packer/batch.py | 90 +++++++++++++++++++++++++++++++++++++ a1_swap_mod_packer/gui.py | 21 +++++++-- tests/test_batch.py | 63 ++++++++++++++++++++++++++ 4 files changed, 173 insertions(+), 3 deletions(-) create mode 100644 a1_swap_mod_packer/batch.py create mode 100644 tests/test_batch.py diff --git a/README.md b/README.md index 14cbc50..2079d11 100644 --- a/README.md +++ b/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 diff --git a/a1_swap_mod_packer/batch.py b/a1_swap_mod_packer/batch.py new file mode 100644 index 0000000..ac85196 --- /dev/null +++ b/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, + ) diff --git a/a1_swap_mod_packer/gui.py b/a1_swap_mod_packer/gui.py index 30b7671..0ae3c5c 100644 --- a/a1_swap_mod_packer/gui.py +++ b/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() diff --git a/tests/test_batch.py b/tests/test_batch.py new file mode 100644 index 0000000..3675cbc --- /dev/null +++ b/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()