From 7dabdc676ae70b3f60262f9b6509f16698038590 Mon Sep 17 00:00:00 2001 From: hyyz17200 Date: Fri, 8 May 2026 11:29:09 +0800 Subject: [PATCH] Introduce caching for prepared G-code --- a1_swap_mod_packer/builder.py | 37 +++++++++++- tests/test_gcode_prepare_cache.py | 94 +++++++++++++++++++++++++++++++ 2 files changed, 130 insertions(+), 1 deletion(-) create mode 100644 tests/test_gcode_prepare_cache.py diff --git a/a1_swap_mod_packer/builder.py b/a1_swap_mod_packer/builder.py index 1c5caeb..f9417d3 100644 --- a/a1_swap_mod_packer/builder.py +++ b/a1_swap_mod_packer/builder.py @@ -120,9 +120,25 @@ def process_plate_gcode( swap_gcode_text: str, patch_config: GcodePatchConfig, ) -> str: + text = _prepare_plate_gcode(source, options, patch_config) + return _process_prepared_plate_gcode(text, plate_number, total_plates, options, swap_gcode_text, patch_config) + + +def _prepare_plate_gcode(source: PlateSource, options: BuildOptions, patch_config: GcodePatchConfig) -> str: text = normalize_newlines(source.gcode_text) if options.apply_gcode_patches: text = apply_gcode_patches(text, patch_config) + return text + + +def _process_prepared_plate_gcode( + text: str, + plate_number: int, + total_plates: int, + options: BuildOptions, + swap_gcode_text: str, + patch_config: GcodePatchConfig, +) -> str: if options.show_plate_number: text = apply_plate_number_offset(text, plate_number) should_swap = options.swap_after_final or plate_number < total_plates @@ -132,6 +148,10 @@ def process_plate_gcode( return text.rstrip("\n") + "\n" +def _plate_gcode_cache_key(source: PlateSource) -> tuple[Path, str]: + return (source.source_3mf.resolve(strict=False), source.member_name) + + def save_png_bytes(image: object) -> bytes: import io @@ -353,8 +373,23 @@ def build_packed_3mf(jobs: list[PlateJob], options: BuildOptions) -> BuildResult swap_gcode_text = read_swap_gcode_file(swap_gcode_path) patch_config = parse_patch_config() if options.apply_gcode_patches else GcodePatchConfig(rules=()) processed: list[str] = [] + prepared_gcode_cache: dict[tuple[Path, str], str] = {} for plate_number, source in enumerate(sources, start=1): - processed.append(process_plate_gcode(source, plate_number, len(sources), options, swap_gcode_text, patch_config)) + cache_key = _plate_gcode_cache_key(source) + prepared_text = prepared_gcode_cache.get(cache_key) + if prepared_text is None: + prepared_text = _prepare_plate_gcode(source, options, patch_config) + prepared_gcode_cache[cache_key] = prepared_text + processed.append( + _process_prepared_plate_gcode( + prepared_text, + plate_number, + len(sources), + options, + swap_gcode_text, + patch_config, + ) + ) final_gcode = "\n".join(item.rstrip("\n") for item in processed) + "\n" gcode_bytes = apply_line_ending(final_gcode, options.line_ending) output_3mf = options.output_3mf diff --git a/tests/test_gcode_prepare_cache.py b/tests/test_gcode_prepare_cache.py new file mode 100644 index 0000000..3ccef35 --- /dev/null +++ b/tests/test_gcode_prepare_cache.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +import hashlib +import tempfile +import unittest +import zipfile +from pathlib import Path +from unittest.mock import patch + +import a1_swap_mod_packer.builder as builder +from a1_swap_mod_packer.models import BuildOptions, GcodePatchConfig, GcodePatchRule, PlateJob + + +class GcodePrepareCacheTest(unittest.TestCase): + def test_repeated_copies_prepare_gcode_once_per_member(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source_3mf = root / "source.3mf" + output_3mf = root / "out.3mf" + swap_gcode = root / "swap.gcode" + self.write_archive(source_3mf) + swap_gcode.write_text("; swap\nG1 X5\n", encoding="utf-8") + + patch_config = GcodePatchConfig( + rules=( + GcodePatchRule( + name="test", + find="G0 Y254 F3000", + replace="G0 Y250 F3000 ;Patched", + ), + ) + ) + patch_calls = 0 + captured_gcode = "" + + def count_patch_calls(text: str, config: GcodePatchConfig) -> str: + nonlocal patch_calls + patch_calls += 1 + return text.replace("G0 Y254 F3000", "G0 Y250 F3000 ;Patched", 1) + + def capture_gcode( + base_3mf: Path, + output_3mf: Path, + gcode_bytes: bytes, + sources: list[builder.PlateSource], + options: BuildOptions, + ) -> str: + nonlocal captured_gcode + captured_gcode = gcode_bytes.decode("utf-8") + return hashlib.md5(gcode_bytes).hexdigest() + + options = BuildOptions( + swap_gcode=swap_gcode, + output_3mf=output_3mf, + apply_gcode_patches=True, + show_plate_number=False, + swap_after_final=False, + line_ending="lf", + add_preview_label=False, + ) + + with patch.object(builder, "parse_patch_config", return_value=patch_config), patch.object( + builder, + "apply_gcode_patches", + side_effect=count_patch_calls, + ), patch.object(builder, "write_output_3mf", side_effect=capture_gcode): + result = builder.build_packed_3mf([PlateJob(source_3mf, 3)], options) + + self.assertEqual(result.plate_count, 3) + self.assertEqual(patch_calls, 1) + self.assertEqual(captured_gcode.count("G0 Y250 F3000 ;Patched"), 3) + self.assertNotIn("G0 Y254 F3000", captured_gcode) + + @staticmethod + def write_archive(path: Path) -> None: + slice_info = """ + + + + + +""" + gcode = """G0 X128 F30000 +G0 Y254 F3000 +M73 P0 R10 +;=====printer finish sound========= +""" + with zipfile.ZipFile(path, "w") as archive: + archive.writestr("Metadata/slice_info.config", slice_info) + archive.writestr("Metadata/plate_1.gcode", gcode) + + +if __name__ == "__main__": + unittest.main()