Frank Schwenk
cbca473f06
Defer xcf_stack to the final pipeline step via runs_last, expose per-image timeout (default 120s), accept XCF output when gimp-console hangs on quit, and discover .xcf outputs for resume. Co-authored-by: Cursor <cursoragent@cursor.com>
101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
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"
|
|
runs_last = True
|
|
description = (
|
|
"Stack all prior pipeline step outputs as GIMP layers into one XCF per image"
|
|
)
|
|
command_candidates = ("gimp-console", "gimp")
|
|
|
|
@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",
|
|
),
|
|
"timeout": Param(
|
|
"float",
|
|
default=120.0,
|
|
help="Seconds to wait for gimp-console per image (default 120)",
|
|
),
|
|
}
|
|
|
|
def run(self, ctx: ModuleContext) -> None:
|
|
include_input = ctx.params["include_input"]
|
|
skip_missing = ctx.params["skip_missing"]
|
|
timeout = ctx.params["timeout"]
|
|
|
|
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=timeout)
|
|
|
|
def list_output_images(self, ctx: ModuleContext) -> list[Path]:
|
|
return sorted(
|
|
p for p in ctx.output_dir.iterdir() if p.is_file() and p.suffix.lower() == ".xcf"
|
|
)
|