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 <cursoragent@cursor.com>
This commit is contained in:
Frank Schwenk
@@ -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
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
@@ -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:
|
||||
|
||||
@@ -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}"
|
||||
)
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user