From 13c3c653b853f54a7ecc0de3969a01eeb921b809 Mon Sep 17 00:00:00 2001 From: Frank Schwenk Date: Sun, 12 Jul 2026 11:15:23 +0200 Subject: [PATCH] feat: add xcf_stack module for GIMP layer export Collect prior pipeline step outputs per image and stack them into XCF files via headless GIMP Script-Fu. Co-authored-by: Cursor --- imagepipeline/core/context.py | 1 + imagepipeline/core/runner.py | 1 + imagepipeline/modules/__init__.py | 1 + imagepipeline/modules/xcf_stack.py | 89 +++++++++++++ imagepipeline/utils/files.py | 9 ++ imagepipeline/utils/gimp.py | 111 ++++++++++++++++ tests/test_xcf_stack.py | 200 +++++++++++++++++++++++++++++ 7 files changed, 412 insertions(+) create mode 100644 imagepipeline/modules/xcf_stack.py create mode 100644 imagepipeline/utils/gimp.py create mode 100644 tests/test_xcf_stack.py diff --git a/imagepipeline/core/context.py b/imagepipeline/core/context.py index a4db3cd..f90e389 100644 --- a/imagepipeline/core/context.py +++ b/imagepipeline/core/context.py @@ -18,6 +18,7 @@ class ModuleContext: pipeline_output_root: Path step_id: str matched_groups: list[list[Path]] = field(default_factory=list) + prior_steps: list[tuple[str, Path]] = field(default_factory=list) logger: PipelineLogger | None = None @property diff --git a/imagepipeline/core/runner.py b/imagepipeline/core/runner.py index d7d5fd7..f3f07e5 100644 --- a/imagepipeline/core/runner.py +++ b/imagepipeline/core/runner.py @@ -220,6 +220,7 @@ class PipelineRunner: pipeline_output_root=self.output_root, step_id=step.step_id, matched_groups=matched_groups, + prior_steps=[(sid, res.output_dir) for sid, res in self._results.items()], logger=self.logger, ) diff --git a/imagepipeline/modules/__init__.py b/imagepipeline/modules/__init__.py index 5b10917..fdb9aa5 100644 --- a/imagepipeline/modules/__init__.py +++ b/imagepipeline/modules/__init__.py @@ -15,3 +15,4 @@ import imagepipeline.modules.imagemagick_resize # noqa: F401 import imagepipeline.modules.imagemagick_scale_crop # noqa: F401 import imagepipeline.modules.openrouter_edit # noqa: F401 import imagepipeline.modules.rembg # noqa: F401 +import imagepipeline.modules.xcf_stack # noqa: F401 diff --git a/imagepipeline/modules/xcf_stack.py b/imagepipeline/modules/xcf_stack.py new file mode 100644 index 0000000..ba8dff5 --- /dev/null +++ b/imagepipeline/modules/xcf_stack.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +from pathlib import Path + +from imagepipeline.core.context import ModuleContext +from imagepipeline.core.params import Param +from imagepipeline.modules.base import SubprocessModule +from imagepipeline.modules.registry import register +from imagepipeline.utils.files import find_image_by_stem +from imagepipeline.utils.gimp import stack_images_to_xcf + + +@register +class XcfStackModule(SubprocessModule): + name = "xcf_stack" + description = ( + "Stack all prior pipeline step outputs as GIMP layers into one XCF per image" + ) + command_candidates = ("gimp",) + default_timeout = 600.0 + + @classmethod + def expected_output_filenames( + cls, + *, + matched_groups: list[list[Path]], + input_paths: list[Path], + params: dict, + ) -> list[str]: + return [f"{path.stem}.xcf" for path in input_paths] + + @classmethod + def parameters(cls) -> dict[str, Param]: + return { + "include_input": Param( + "bool", + default=True, + help="Include pipeline input/ directory as bottom layer", + ), + "skip_missing": Param( + "bool", + default=False, + help="Skip missing step outputs instead of failing", + ), + } + + def run(self, ctx: ModuleContext) -> None: + include_input = ctx.params["include_input"] + skip_missing = ctx.params["skip_missing"] + + layer_sources: list[tuple[str, Path]] = [] + if include_input: + layer_sources.append(("input", ctx.pipeline_output_root / "input")) + layer_sources.extend(ctx.prior_steps) + + if not layer_sources: + raise ValueError( + "xcf_stack has no layer sources " + "(include_input=False and no prior_steps)" + ) + + ctx.output_dir.mkdir(parents=True, exist_ok=True) + total = len(ctx.input_paths) + + for index, input_path in enumerate(ctx.input_paths, start=1): + self.log_image(ctx, index, total, input_path) + stem = input_path.stem + layers: list[tuple[str, Path]] = [] + + for step_id, directory in layer_sources: + image_path = find_image_by_stem(directory, stem) + if image_path is None: + if not skip_missing: + raise ValueError( + f"xcf_stack: no image with stem '{stem}' in step " + f"'{step_id}' ({directory})" + ) + continue + layers.append((step_id, image_path)) + + if not layers: + checked = ", ".join(step_id for step_id, _ in layer_sources) + raise ValueError( + f"xcf_stack: no layers found for stem '{stem}' " + f"(checked: {checked})" + ) + + dst = ctx.output_dir / f"{stem}.xcf" + stack_images_to_xcf(layers, dst, timeout=self.default_timeout) diff --git a/imagepipeline/utils/files.py b/imagepipeline/utils/files.py index 4750f3d..a866ac3 100644 --- a/imagepipeline/utils/files.py +++ b/imagepipeline/utils/files.py @@ -22,6 +22,15 @@ def stem_key(path: Path) -> str: return path.stem.lower() +def find_image_by_stem(directory: Path, stem: str) -> Path | None: + """Return first image in directory whose stem matches stem (case-insensitive).""" + target = stem.lower() + for path in directory.iterdir(): + if is_image(path) and stem_key(path) == target: + return path + return None + + def match_by_stem(sources: list[list[Path]]) -> list[list[Path]]: """Match image paths across multiple source lists by filename stem.""" if not sources: diff --git a/imagepipeline/utils/gimp.py b/imagepipeline/utils/gimp.py new file mode 100644 index 0000000..b8a469a --- /dev/null +++ b/imagepipeline/utils/gimp.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +from pathlib import Path + +from imagepipeline.utils.subprocess import require_command, run_command + + +def require_gimp() -> str: + """Return the GIMP executable name, raising DependencyError if missing.""" + return require_command("gimp") + + +def _scheme_string(value: str) -> str: + """Escape a Python string for use inside a Scheme double-quoted literal.""" + escaped = ( + value.replace("\\", "\\\\") + .replace('"', '\\"') + .replace("\n", "\\n") + .replace("\r", "\\r") + .replace("\t", "\\t") + ) + return f'"{escaped}"' + + +def _build_stack_script(layers: list[tuple[str, Path]], outfile: Path) -> str: + first_name, first_path = layers[0] + first_path_str = _scheme_string(str(first_path)) + outfile_str = _scheme_string(str(outfile)) + + lines = [ + "(let* (", + f" (loaded (gimp-file-load RUN-NONINTERACTIVE {first_path_str} {first_path_str}))", + " (image (car loaded))", + " (bottom-layer (cadr loaded))", + ")", + f" (gimp-layer-set-name bottom-layer {_scheme_string(first_name)})", + ] + + for layer_name, layer_path in layers[1:]: + path_str = _scheme_string(str(layer_path)) + lines.extend( + [ + " (let ((layer (car (gimp-file-load-layer RUN-NONINTERACTIVE image " + f"{path_str}))))", + " (gimp-image-insert-layer image layer 0 0)", + f" (gimp-layer-set-name layer {_scheme_string(layer_name)})", + " )", + ] + ) + + lines.extend( + [ + f" (gimp-xcf-save RUN-NONINTERACTIVE image bottom-layer {outfile_str} {outfile_str})", + " (gimp-image-delete image)", + ")", + ] + ) + return "\n".join(lines) + + +def stack_images_to_xcf( + layers: list[tuple[str, Path]], + outfile: Path, + *, + timeout: float | None = None, +) -> None: + """Stack images bottom-to-top into a single GIMP XCF file. + + ``layers`` is a list of ``(layer_name, image_path)`` tuples in bottom-to-top + order. Invokes GIMP headless via Script-Fu. + """ + if not layers: + raise ValueError("stack_images_to_xcf requires at least one layer") + + gimp = require_gimp() + resolved_layers: list[tuple[str, Path]] = [] + for layer_name, image_path in layers: + resolved = image_path.resolve() + if not resolved.is_file(): + raise FileNotFoundError( + f"Layer image not found for '{layer_name}': {resolved}" + ) + resolved_layers.append((layer_name, resolved)) + + outfile = outfile.resolve() + outfile.parent.mkdir(parents=True, exist_ok=True) + + script = _build_stack_script(resolved_layers, outfile) + cmd = [ + gimp, + "-idf", + "--batch-interpreter", + "plug-in-script-fu-eval", + "-b", + script, + "-b", + "(gimp-quit 0)", + ] + + try: + run_command(cmd, timeout=timeout) + except RuntimeError as exc: + layer_summary = ", ".join(name for name, _ in resolved_layers) + raise RuntimeError( + f"GIMP failed to stack layers [{layer_summary}] into {outfile}: {exc}" + ) from exc + + if not outfile.is_file(): + raise RuntimeError( + f"GIMP completed but XCF output was not created: {outfile}" + ) diff --git a/tests/test_xcf_stack.py b/tests/test_xcf_stack.py new file mode 100644 index 0000000..d5bf6d9 --- /dev/null +++ b/tests/test_xcf_stack.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +import shutil +from pathlib import Path +from unittest.mock import patch + +import pytest + +from imagepipeline.core.context import ModuleContext +from imagepipeline.modules.registry import get_module, list_modules +from imagepipeline.modules.xcf_stack import XcfStackModule +from imagepipeline.utils.files import find_image_by_stem +from tests.conftest import make_png + +has_gimp = bool(shutil.which("gimp")) +has_magick = bool(shutil.which("magick") or shutil.which("convert")) + + +def _gimp_headless_works() -> bool: + gimp = shutil.which("gimp") + if not gimp: + return False + try: + from imagepipeline.utils.subprocess import run_command + + run_command([gimp, "-idf", "-b", "(gimp-quit 0)"], timeout=10) + return True + except RuntimeError: + return False + + +has_working_gimp = _gimp_headless_works() + + +class TestFindImageByStem: + def test_finds_matching_stem_case_insensitively(self, tmp_path: Path) -> None: + make_png(tmp_path / "Photo.PNG") + found = find_image_by_stem(tmp_path, "photo") + assert found == tmp_path / "Photo.PNG" + + def test_returns_none_when_missing(self, tmp_path: Path) -> None: + make_png(tmp_path / "other.png") + assert find_image_by_stem(tmp_path, "photo") is None + + def test_handles_extension_mismatch(self, tmp_path: Path) -> None: + make_png(tmp_path / "photo.png") + # Input may be photo.jpg while step output is photo.png (stem-only match). + found = find_image_by_stem(tmp_path, Path("photo.jpg").stem) + assert found == tmp_path / "photo.png" + + +class TestModuleRegistration: + def test_xcf_stack_registered(self) -> None: + assert "xcf_stack" in list_modules() + + def test_get_module_returns_xcf_stack_class(self) -> None: + assert get_module("xcf_stack") is XcfStackModule + + +class TestExpectedOutputFilenames: + def test_returns_xcf_for_jpg_input(self) -> None: + names = XcfStackModule.expected_output_filenames( + matched_groups=[], + input_paths=[Path("photo.jpg")], + params={}, + ) + assert names == ["photo.xcf"] + + +def _make_stack_fixture(tmp_path: Path) -> dict[str, Path]: + root = tmp_path + input_dir = root / "input" + step_a = root / "step_a" + step_b = root / "step_b" + output_dir = root / "xcf_out" + for directory in (input_dir, step_a, step_b, output_dir): + directory.mkdir(parents=True, exist_ok=True) + + make_png(input_dir / "photo.png") + make_png(step_a / "photo.png", rgb=(10, 20, 30)) + make_png(step_b / "photo.png", rgb=(30, 20, 10)) + + return { + "root": root, + "input_dir": input_dir, + "step_a": step_a, + "step_b": step_b, + "output_dir": output_dir, + } + + +class TestXcfStackRun: + @patch("imagepipeline.modules.xcf_stack.stack_images_to_xcf") + def test_collects_layers_from_input_and_prior_steps( + self, mock_stack: object, tmp_path: Path + ) -> None: + paths = _make_stack_fixture(tmp_path) + input_path = paths["root"] / "refs" / "photo.jpg" + input_path.parent.mkdir() + input_path.touch() + + ctx = ModuleContext( + input_paths=[input_path], + output_dir=paths["output_dir"], + params=XcfStackModule.validate_module_params({}), + pipeline_output_root=paths["root"], + step_id="xcf_stack_01", + prior_steps=[ + ("step_a", paths["step_a"]), + ("step_b", paths["step_b"]), + ], + logger=None, + ) + XcfStackModule().run(ctx) + + mock_stack.assert_called_once() + layers, outfile = mock_stack.call_args[0] + assert outfile == paths["output_dir"] / "photo.xcf" + assert layers == [ + ("input", paths["input_dir"] / "photo.png"), + ("step_a", paths["step_a"] / "photo.png"), + ("step_b", paths["step_b"] / "photo.png"), + ] + + @patch("imagepipeline.modules.xcf_stack.stack_images_to_xcf") + def test_skip_missing_true_skips_missing_step( + self, mock_stack: object, tmp_path: Path + ) -> None: + paths = _make_stack_fixture(tmp_path) + (paths["step_b"] / "photo.png").unlink() + input_path = paths["root"] / "photo.jpg" + + ctx = ModuleContext( + input_paths=[input_path], + output_dir=paths["output_dir"], + params=XcfStackModule.validate_module_params({"skip_missing": True}), + pipeline_output_root=paths["root"], + step_id="xcf_stack_01", + prior_steps=[ + ("step_a", paths["step_a"]), + ("step_b", paths["step_b"]), + ], + logger=None, + ) + XcfStackModule().run(ctx) + + layers, _outfile = mock_stack.call_args[0] + step_ids = [step_id for step_id, _ in layers] + assert step_ids == ["input", "step_a"] + + @patch("imagepipeline.modules.xcf_stack.stack_images_to_xcf") + def test_skip_missing_false_raises_on_missing_step( + self, mock_stack: object, tmp_path: Path + ) -> None: + paths = _make_stack_fixture(tmp_path) + (paths["step_b"] / "photo.png").unlink() + input_path = paths["root"] / "photo.jpg" + + ctx = ModuleContext( + input_paths=[input_path], + output_dir=paths["output_dir"], + params=XcfStackModule.validate_module_params({"skip_missing": False}), + pipeline_output_root=paths["root"], + step_id="xcf_stack_01", + prior_steps=[ + ("step_a", paths["step_a"]), + ("step_b", paths["step_b"]), + ], + logger=None, + ) + with pytest.raises(ValueError, match="no image with stem 'photo' in step 'step_b'"): + XcfStackModule().run(ctx) + + mock_stack.assert_not_called() + + +@pytest.mark.skipif(not has_working_gimp, reason="GIMP headless not available") +@pytest.mark.skipif(not has_magick, reason="ImageMagick not installed") +class TestStackImagesToXcfIntegration: + def test_stacks_pngs_into_xcf(self, tmp_path: Path) -> None: + from imagepipeline.utils.gimp import stack_images_to_xcf + + bottom = tmp_path / "bottom.png" + top = tmp_path / "top.png" + magick = shutil.which("magick") or shutil.which("convert") + assert magick is not None + from imagepipeline.utils.subprocess import run_command + + run_command([magick, "-size", "8x8", "xc:#ff0000", str(bottom)]) + run_command([magick, "-size", "8x8", "xc:#0000ff", str(top)]) + + outfile = tmp_path / "stack.xcf" + stack_images_to_xcf( + [("bottom", bottom), ("top", top)], + outfile, + timeout=120.0, + ) + + assert outfile.is_file() + assert outfile.stat().st_size > 0