Optimize G-code processing time by using M73OffsetTemplate for plate number offsets

a1_swap_mod_packer/builder.py

@@ -7,15 +7,17 @@ import shutil
 import tempfile
 import zipfile
 from contextlib import contextmanager
+from dataclasses import dataclass
 from pathlib import Path
 
 from .archive import GCODE_MEMBER_RE, MD5_MEMBER_RE, list_gcode_members
 from .gcode import (
+    M73OffsetTemplate,
     apply_line_ending,
-    apply_plate_number_offset,
     build_swap_block,
     insert_swap_block,
     normalize_newlines,
+    prepare_m73_offset_template,
     read_swap_gcode_file,
     resolve_swap_gcode_path,
 )
@@ -42,6 +44,12 @@ MIN_ZIP_COMPRESS_LEVEL = 1
 MAX_ZIP_COMPRESS_LEVEL = 9
 
 
+@dataclass(frozen=True)
+class _PreparedPlateGcode:
+    text: str
+    m73_template: M73OffsetTemplate | None = None
+
+
 def normalized_zip_compress_level(level: int | None) -> int:
     try:
         value = int(level if level is not None else DEFAULT_ZIP_COMPRESS_LEVEL)
@@ -120,27 +128,40 @@ def process_plate_gcode(
     swap_gcode_text: str,
     patch_config: GcodePatchConfig,
 ) -> str:
-    text = _prepare_plate_gcode(source, options, patch_config)
+    prepared = _prepare_plate_gcode(source, options, patch_config)
-    return _process_prepared_plate_gcode(text, plate_number, total_plates, options, swap_gcode_text, patch_config)
+    return _process_prepared_plate_gcode(
+        prepared,
+        plate_number,
+        total_plates,
+        options,
+        swap_gcode_text,
+        patch_config,
+    )
 
 
-def _prepare_plate_gcode(source: PlateSource, options: BuildOptions, patch_config: GcodePatchConfig) -> str:
+def _prepare_plate_gcode(
+    source: PlateSource,
+    options: BuildOptions,
+    patch_config: GcodePatchConfig,
+) -> _PreparedPlateGcode:
     text = normalize_newlines(source.gcode_text)
     if options.apply_gcode_patches:
         text = apply_gcode_patches(text, patch_config)
-    return text
+    m73_template = prepare_m73_offset_template(text) if options.show_plate_number else None
+    return _PreparedPlateGcode(text, m73_template)
 
 
 def _process_prepared_plate_gcode(
-    text: str,
+    prepared: _PreparedPlateGcode,
     plate_number: int,
     total_plates: int,
     options: BuildOptions,
     swap_gcode_text: str,
     patch_config: GcodePatchConfig,
 ) -> str:
-    if options.show_plate_number:
+    text = prepared.text
-        text = apply_plate_number_offset(text, plate_number)
+    if options.show_plate_number and prepared.m73_template is not None:
+        text = prepared.m73_template.apply(plate_number)
     should_swap = options.swap_after_final or plate_number < total_plates
     if should_swap:
         swap_block = build_swap_block(swap_gcode_text, options.cool_bed_temp, options.wait_after_eject_seconds)
@@ -373,7 +394,7 @@ 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] = {}
+    prepared_gcode_cache: dict[tuple[Path, str], _PreparedPlateGcode] = {}
     for plate_number, source in enumerate(sources, start=1):
         cache_key = _plate_gcode_cache_key(source)
         prepared_text = prepared_gcode_cache.get(cache_key)

a1_swap_mod_packer/gcode.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import re
+from dataclasses import dataclass
 from pathlib import Path
 
 from .models import DEFAULT_INSERT_BEFORE_MARKER, LineEnding
@@ -73,13 +74,37 @@ def format_filament(weight_grams: float | None, used_m: float | None = None) ->
     return " / ".join(parts)
 
 
+@dataclass(frozen=True)
+class M73OffsetTemplate:
+    segments: tuple[str, ...]
+    base_remaining_minutes: tuple[int, ...]
+
+    def apply(self, plate_number: int) -> str:
+        if not self.base_remaining_minutes:
+            return self.segments[0]
+        minute_offset = plate_number * 100 * 60
+        parts: list[str] = []
+        for index, remaining_minutes in enumerate(self.base_remaining_minutes):
+            parts.append(self.segments[index])
+            parts.append(str(remaining_minutes + minute_offset))
+        parts.append(self.segments[-1])
+        return "".join(parts)
+
+
+def prepare_m73_offset_template(gcode: str) -> M73OffsetTemplate:
+    segments: list[str] = []
+    base_remaining_minutes: list[int] = []
+    last_index = 0
+    for match in M73_RE.finditer(gcode):
+        segments.append(gcode[last_index : match.start(2)])
+        base_remaining_minutes.append(int(match.group(2)))
+        last_index = match.end(2)
+    segments.append(gcode[last_index:])
+    return M73OffsetTemplate(tuple(segments), tuple(base_remaining_minutes))
+
+
 def apply_plate_number_offset(gcode: str, plate_number: int) -> str:
-    minute_offset = plate_number * 100 * 60
+    return prepare_m73_offset_template(gcode).apply(plate_number)
-
-    def replace(match: re.Match[str]) -> str:
-        return f"{match.group(1)}{int(match.group(2)) + minute_offset}{match.group(3)}"
-
-    return M73_RE.sub(replace, gcode)
 
 
 def build_swap_block(swap_gcode: str, cool_bed_temp: int | None, wait_seconds: int) -> str:

tests/test_gcode_prepare_cache.py

@@ -71,6 +71,61 @@ class GcodePrepareCacheTest(unittest.TestCase):
             self.assertEqual(captured_gcode.count("G0 Y250 F3000 ;Patched"), 3)
             self.assertNotIn("G0 Y254 F3000", captured_gcode)
 
+    def test_repeated_copies_prepare_m73_template_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")
+
+            m73_template_calls = 0
+            captured_gcode = ""
+            original_prepare_m73 = builder.prepare_m73_offset_template
+
+            def count_m73_template_calls(text: str):
+                nonlocal m73_template_calls
+                m73_template_calls += 1
+                return original_prepare_m73(text)
+
+            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=False,
+                show_plate_number=True,
+                swap_after_final=False,
+                line_ending="lf",
+                add_preview_label=False,
+            )
+
+            with (
+                patch.object(builder, "prepare_m73_offset_template", side_effect=count_m73_template_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(m73_template_calls, 1)
+            self.assertIn("M73 P0 R6010", captured_gcode)
+            self.assertIn("M73 P0 R12010", captured_gcode)
+            self.assertIn("M73 P0 R18010", captured_gcode)
+
     @staticmethod
     def write_archive(path: Path) -> None:
         slice_info = """<?xml version="1.0" encoding="UTF-8"?>

tests/test_m73_offset_template.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import unittest
+
+from a1_swap_mod_packer.gcode import M73_RE, apply_plate_number_offset, prepare_m73_offset_template
+
+
+class M73OffsetTemplateTest(unittest.TestCase):
+    def test_template_matches_previous_regex_replacement_semantics(self) -> None:
+        gcode = (
+            "M73 P0 R10\n"
+            "; M73 P0 R20\n"
+            " M73 P0 R30\n"
+            "M73 P50 R5999 ; comment\n"
+            "G1 X1\n"
+            "M73 P100 R0\n"
+        )
+
+        def previous_apply(text: str, plate_number: int) -> str:
+            minute_offset = plate_number * 100 * 60
+
+            def replace(match) -> str:
+                prefix = match.group(1)
+                remaining_minutes = int(match.group(2)) + minute_offset
+                suffix = match.group(3)
+                return f"{prefix}{remaining_minutes}{suffix}"
+
+            return M73_RE.sub(replace, text)
+
+        template = prepare_m73_offset_template(gcode)
+        for plate_number in (1, 2, 10):
+            self.assertEqual(template.apply(plate_number), previous_apply(gcode, plate_number))
+            self.assertEqual(apply_plate_number_offset(gcode, plate_number), previous_apply(gcode, plate_number))
+
+    def test_template_without_m73_matches_returns_original_text(self) -> None:
+        gcode = "G1 X1\n; M73 P0 R10\n"
+        template = prepare_m73_offset_template(gcode)
+        self.assertEqual(template.apply(3), gcode)
+
+
+if __name__ == "__main__":
+    unittest.main()