fix: xcf_stack GIMP 3 headless script execution
Use gimp-console when available, GIMP 3 Script-Fu API, and temp .scm batch files so layer stacking works without a display. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -16,7 +16,7 @@ class XcfStackModule(SubprocessModule):
|
|||||||
description = (
|
description = (
|
||||||
"Stack all prior pipeline step outputs as GIMP layers into one XCF per image"
|
"Stack all prior pipeline step outputs as GIMP layers into one XCF per image"
|
||||||
)
|
)
|
||||||
command_candidates = ("gimp",)
|
command_candidates = ("gimp-console", "gimp")
|
||||||
default_timeout = 600.0
|
default_timeout = 600.0
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
|
|||||||
+74
-34
@@ -1,13 +1,18 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
import subprocess
|
||||||
|
import tempfile
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from imagepipeline.utils.subprocess import require_command, run_command
|
from imagepipeline.utils.subprocess import require_command
|
||||||
|
|
||||||
|
_GIMP_CANDIDATES = ("gimp-console", "gimp")
|
||||||
|
|
||||||
|
|
||||||
def require_gimp() -> str:
|
def require_gimp() -> str:
|
||||||
"""Return the GIMP executable name, raising DependencyError if missing."""
|
"""Return a GIMP executable for headless batch use."""
|
||||||
return require_command("gimp")
|
return require_command(*_GIMP_CANDIDATES)
|
||||||
|
|
||||||
|
|
||||||
def _scheme_string(value: str) -> str:
|
def _scheme_string(value: str) -> str:
|
||||||
@@ -23,34 +28,35 @@ def _scheme_string(value: str) -> str:
|
|||||||
|
|
||||||
|
|
||||||
def _build_stack_script(layers: list[tuple[str, Path]], outfile: Path) -> str:
|
def _build_stack_script(layers: list[tuple[str, Path]], outfile: Path) -> str:
|
||||||
|
"""Build a GIMP 3 Script-Fu snippet that stacks images into one XCF."""
|
||||||
first_name, first_path = layers[0]
|
first_name, first_path = layers[0]
|
||||||
first_path_str = _scheme_string(str(first_path))
|
first_path_str = _scheme_string(str(first_path))
|
||||||
outfile_str = _scheme_string(str(outfile))
|
outfile_str = _scheme_string(str(outfile))
|
||||||
|
|
||||||
lines = [
|
lines = [
|
||||||
|
"(script-fu-use-v3)",
|
||||||
"(let* (",
|
"(let* (",
|
||||||
f" (loaded (gimp-file-load RUN-NONINTERACTIVE {first_path_str} {first_path_str}))",
|
f" (image (gimp-file-load RUN-NONINTERACTIVE {first_path_str}))",
|
||||||
" (image (car loaded))",
|
" (bottom (vector-ref (gimp-image-get-selected-drawables image) 0))",
|
||||||
" (bottom-layer (cadr loaded))",
|
|
||||||
")",
|
")",
|
||||||
f" (gimp-layer-set-name bottom-layer {_scheme_string(first_name)})",
|
f" (gimp-item-set-name bottom {_scheme_string(first_name)})",
|
||||||
]
|
]
|
||||||
|
|
||||||
for layer_name, layer_path in layers[1:]:
|
for layer_name, layer_path in layers[1:]:
|
||||||
path_str = _scheme_string(str(layer_path))
|
path_str = _scheme_string(str(layer_path))
|
||||||
lines.extend(
|
lines.extend(
|
||||||
[
|
[
|
||||||
" (let ((layer (car (gimp-file-load-layer RUN-NONINTERACTIVE image "
|
" (let ((layer",
|
||||||
f"{path_str}))))",
|
f" (gimp-file-load-layer RUN-NONINTERACTIVE image {path_str})))",
|
||||||
" (gimp-image-insert-layer image layer 0 0)",
|
" (gimp-image-insert-layer image layer image 0)",
|
||||||
f" (gimp-layer-set-name layer {_scheme_string(layer_name)})",
|
f" (gimp-item-set-name layer {_scheme_string(layer_name)})",
|
||||||
" )",
|
" )",
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
|
||||||
lines.extend(
|
lines.extend(
|
||||||
[
|
[
|
||||||
f" (gimp-xcf-save RUN-NONINTERACTIVE image bottom-layer {outfile_str} {outfile_str})",
|
f" (gimp-file-save RUN-NONINTERACTIVE image {outfile_str})",
|
||||||
" (gimp-image-delete image)",
|
" (gimp-image-delete image)",
|
||||||
")",
|
")",
|
||||||
]
|
]
|
||||||
@@ -58,6 +64,27 @@ def _build_stack_script(layers: list[tuple[str, Path]], outfile: Path) -> str:
|
|||||||
return "\n".join(lines)
|
return "\n".join(lines)
|
||||||
|
|
||||||
|
|
||||||
|
def _run_gimp_batch(
|
||||||
|
cmd: list[str],
|
||||||
|
*,
|
||||||
|
timeout: float | None,
|
||||||
|
env: dict[str, str],
|
||||||
|
) -> subprocess.CompletedProcess[str]:
|
||||||
|
try:
|
||||||
|
return subprocess.run(
|
||||||
|
cmd,
|
||||||
|
check=False,
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
timeout=timeout,
|
||||||
|
env=env,
|
||||||
|
)
|
||||||
|
except subprocess.TimeoutExpired as exc:
|
||||||
|
raise RuntimeError(
|
||||||
|
f"Command timed out after {timeout}s: {' '.join(cmd)}"
|
||||||
|
) from exc
|
||||||
|
|
||||||
|
|
||||||
def stack_images_to_xcf(
|
def stack_images_to_xcf(
|
||||||
layers: list[tuple[str, Path]],
|
layers: list[tuple[str, Path]],
|
||||||
outfile: Path,
|
outfile: Path,
|
||||||
@@ -67,7 +94,7 @@ def stack_images_to_xcf(
|
|||||||
"""Stack images bottom-to-top into a single GIMP XCF file.
|
"""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
|
``layers`` is a list of ``(layer_name, image_path)`` tuples in bottom-to-top
|
||||||
order. Invokes GIMP headless via Script-Fu.
|
order. Invokes GIMP headless via Script-Fu (GIMP 3 compatible).
|
||||||
"""
|
"""
|
||||||
if not layers:
|
if not layers:
|
||||||
raise ValueError("stack_images_to_xcf requires at least one layer")
|
raise ValueError("stack_images_to_xcf requires at least one layer")
|
||||||
@@ -86,26 +113,39 @@ def stack_images_to_xcf(
|
|||||||
outfile.parent.mkdir(parents=True, exist_ok=True)
|
outfile.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
script = _build_stack_script(resolved_layers, outfile)
|
script = _build_stack_script(resolved_layers, outfile)
|
||||||
cmd = [
|
env = {**os.environ, "GIMP_NO_DISPLAY": "1"}
|
||||||
gimp,
|
|
||||||
"-idf",
|
with tempfile.NamedTemporaryFile(
|
||||||
"--batch-interpreter",
|
mode="w",
|
||||||
"plug-in-script-fu-eval",
|
suffix=".scm",
|
||||||
"-b",
|
delete=False,
|
||||||
script,
|
encoding="utf-8",
|
||||||
"-b",
|
) as handle:
|
||||||
"(gimp-quit 0)",
|
handle.write(script)
|
||||||
]
|
script_path = Path(handle.name)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
run_command(cmd, timeout=timeout)
|
cmd = [
|
||||||
except RuntimeError as exc:
|
gimp,
|
||||||
layer_summary = ", ".join(name for name, _ in resolved_layers)
|
"-d",
|
||||||
raise RuntimeError(
|
"-i",
|
||||||
f"GIMP failed to stack layers [{layer_summary}] into {outfile}: {exc}"
|
"--quit",
|
||||||
) from exc
|
"--batch-interpreter=plug-in-script-fu-eval",
|
||||||
|
"--batch",
|
||||||
if not outfile.is_file():
|
f"(load {_scheme_string(str(script_path))})",
|
||||||
raise RuntimeError(
|
]
|
||||||
f"GIMP completed but XCF output was not created: {outfile}"
|
try:
|
||||||
)
|
result = _run_gimp_batch(cmd, timeout=timeout, env=env)
|
||||||
|
except RuntimeError:
|
||||||
|
raise
|
||||||
|
if not outfile.is_file():
|
||||||
|
stderr = (result.stderr or "").strip()
|
||||||
|
stdout = (result.stdout or "").strip()
|
||||||
|
detail = stderr or stdout or f"exit code {result.returncode}"
|
||||||
|
layer_summary = ", ".join(name for name, _ in resolved_layers)
|
||||||
|
raise RuntimeError(
|
||||||
|
f"GIMP failed to stack layers [{layer_summary}] into {outfile}: "
|
||||||
|
f"{detail}"
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
script_path.unlink(missing_ok=True)
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ def run_command(
|
|||||||
*,
|
*,
|
||||||
timeout: float | None = None,
|
timeout: float | None = None,
|
||||||
cwd: Path | None = None,
|
cwd: Path | None = None,
|
||||||
|
env: dict[str, str] | None = None,
|
||||||
) -> subprocess.CompletedProcess[str]:
|
) -> subprocess.CompletedProcess[str]:
|
||||||
try:
|
try:
|
||||||
return subprocess.run(
|
return subprocess.run(
|
||||||
@@ -25,6 +26,7 @@ def run_command(
|
|||||||
text=True,
|
text=True,
|
||||||
timeout=timeout,
|
timeout=timeout,
|
||||||
cwd=str(cwd) if cwd else None,
|
cwd=str(cwd) if cwd else None,
|
||||||
|
env=env,
|
||||||
)
|
)
|
||||||
except subprocess.CalledProcessError as exc:
|
except subprocess.CalledProcessError as exc:
|
||||||
stderr = (exc.stderr or "").strip()
|
stderr = (exc.stderr or "").strip()
|
||||||
|
|||||||
+5
-18
@@ -12,26 +12,10 @@ from imagepipeline.modules.xcf_stack import XcfStackModule
|
|||||||
from imagepipeline.utils.files import find_image_by_stem
|
from imagepipeline.utils.files import find_image_by_stem
|
||||||
from tests.conftest import make_png
|
from tests.conftest import make_png
|
||||||
|
|
||||||
has_gimp = bool(shutil.which("gimp"))
|
has_gimp = bool(shutil.which("gimp-console") or shutil.which("gimp"))
|
||||||
has_magick = bool(shutil.which("magick") or shutil.which("convert"))
|
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:
|
class TestFindImageByStem:
|
||||||
def test_finds_matching_stem_case_insensitively(self, tmp_path: Path) -> None:
|
def test_finds_matching_stem_case_insensitively(self, tmp_path: Path) -> None:
|
||||||
make_png(tmp_path / "Photo.PNG")
|
make_png(tmp_path / "Photo.PNG")
|
||||||
@@ -174,7 +158,7 @@ class TestXcfStackRun:
|
|||||||
mock_stack.assert_not_called()
|
mock_stack.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.skipif(not has_working_gimp, reason="GIMP headless not available")
|
@pytest.mark.skipif(not has_gimp, reason="GIMP not installed")
|
||||||
@pytest.mark.skipif(not has_magick, reason="ImageMagick not installed")
|
@pytest.mark.skipif(not has_magick, reason="ImageMagick not installed")
|
||||||
class TestStackImagesToXcfIntegration:
|
class TestStackImagesToXcfIntegration:
|
||||||
def test_stacks_pngs_into_xcf(self, tmp_path: Path) -> None:
|
def test_stacks_pngs_into_xcf(self, tmp_path: Path) -> None:
|
||||||
@@ -198,3 +182,6 @@ class TestStackImagesToXcfIntegration:
|
|||||||
|
|
||||||
assert outfile.is_file()
|
assert outfile.is_file()
|
||||||
assert outfile.stat().st_size > 0
|
assert outfile.stat().st_size > 0
|
||||||
|
names = outfile.read_bytes()
|
||||||
|
assert b"bottom" in names
|
||||||
|
assert b"top" in names
|
||||||
|
|||||||
Reference in New Issue
Block a user