Introduce caching for prepared G-code
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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 = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<config>
|
||||
<plate>
|
||||
<metadata key="index" value="1"/>
|
||||
</plate>
|
||||
</config>
|
||||
"""
|
||||
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()
|
||||
Reference in New Issue
Block a user