From 980c7f3b8bc0890cd90bc02e519e2a7f89be5dab Mon Sep 17 00:00:00 2001 From: Frank Schwenk Date: Sun, 21 Jun 2026 10:26:10 +0200 Subject: [PATCH] A long time ago, in a galaxy far far away... --- imagepipeline/core/log.py | 6 + imagepipeline/core/pipeline.py | 9 ++ imagepipeline/core/resume.py | 110 ++++++++++++++++++ imagepipeline/core/runner.py | 75 +++++++++++++ imagepipeline/modules/__init__.py | 2 + imagepipeline/modules/color_to_alpha.py | 63 +++++++++++ imagepipeline/modules/gmic.py | 3 +- imagepipeline/modules/gmic_grayscale.py | 3 +- imagepipeline/modules/imagemagick_resize.py | 45 ++++++++ imagepipeline/utils/gmic.py | 11 ++ pipelines/pipeline_2000px.py | 27 +++++ pipelines/pipeline_baxxter.py | 11 ++ pipelines/pipeline_baxxter_2.py | 75 +++++++++++++ tests/test_resume.py | 118 ++++++++++++++++++++ tests/test_workflow_modules.py | 89 +++++++++++++++ 15 files changed, 645 insertions(+), 2 deletions(-) create mode 100644 imagepipeline/core/resume.py create mode 100644 imagepipeline/modules/color_to_alpha.py create mode 100644 imagepipeline/modules/imagemagick_resize.py create mode 100644 imagepipeline/utils/gmic.py create mode 100644 pipelines/pipeline_2000px.py create mode 100644 pipelines/pipeline_baxxter_2.py create mode 100644 tests/test_resume.py diff --git a/imagepipeline/core/log.py b/imagepipeline/core/log.py index a9f383b..c950e48 100644 --- a/imagepipeline/core/log.py +++ b/imagepipeline/core/log.py @@ -78,5 +78,11 @@ class PipelineLogger: def step_done(self, step_id: str, output_dir: str, count: int) -> None: self.info(f" Done: {count} image(s) -> {output_dir}/") + def step_skipped(self, step_id: str, count: int) -> None: + self.info(f" Skipped step {step_id} ({count} existing output(s))") + + def step_reused(self, step_id: str, source_dir: str, count: int) -> None: + self.info(f" Reused external output for {step_id}: {source_dir} ({count} file(s))") + def blank(self) -> None: self.info("") diff --git a/imagepipeline/core/pipeline.py b/imagepipeline/core/pipeline.py index f645fde..dfc3c5a 100644 --- a/imagepipeline/core/pipeline.py +++ b/imagepipeline/core/pipeline.py @@ -24,12 +24,18 @@ class Pipeline: output_base: Path | str | None = None, symlink_input: bool = True, verbose: bool = True, + existing_outputs: dict[str, Path | str] | None = None, + continue_from: Path | str | None = None, + skip_completed: bool = True, ) -> None: self.input_dir = Path(input_dir) self.name = name self.output_base = Path(output_base) if output_base else None self.symlink_input = symlink_input self.verbose = verbose + self.existing_outputs = existing_outputs + self.continue_from = Path(continue_from) if continue_from else None + self.skip_completed = skip_completed self._steps: list[StepDefinition] = [] self._module_counters: dict[str, int] = defaultdict(int) self._output_root: Path | None = None @@ -75,6 +81,9 @@ class Pipeline: steps=self._steps, symlink_input=self.symlink_input, verbose=self.verbose, + existing_outputs=self.existing_outputs, + continue_from=self.continue_from, + skip_completed=self.skip_completed, ) self._output_root = runner.run() return self._output_root diff --git a/imagepipeline/core/resume.py b/imagepipeline/core/resume.py new file mode 100644 index 0000000..36f3187 --- /dev/null +++ b/imagepipeline/core/resume.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import json +import os +import shutil +from pathlib import Path + +from imagepipeline.core.exceptions import StepError +from imagepipeline.core.step import StepDefinition +from imagepipeline.utils.files import is_image, list_images, stem_key + + +def expected_output_filenames( + step: StepDefinition, + *, + matched_groups: list[list[Path]], + input_paths: list[Path], + params: dict, +) -> list[str]: + if step.module_name == "composite": + output_ext = params.get("output_ext", ".png") + return [f"{group[-1].stem}{output_ext}" for group in matched_groups] + return [path.name for path in input_paths] + + +def expected_output_paths( + output_dir: Path, + step: StepDefinition, + *, + matched_groups: list[list[Path]], + input_paths: list[Path], + params: dict, +) -> list[Path]: + return [ + output_dir / name + for name in expected_output_filenames( + step, + matched_groups=matched_groups, + input_paths=input_paths, + params=params, + ) + ] + + +def step_outputs_complete(expected_paths: list[Path]) -> bool: + return bool(expected_paths) and all( + path.is_file() and is_image(path) for path in expected_paths + ) + + +def source_stems_for_step( + step: StepDefinition, + *, + matched_groups: list[list[Path]], + input_paths: list[Path], +) -> list[str]: + if step.module_name == "composite": + return [stem_key(group[-1]) for group in matched_groups] + return [stem_key(path) for path in input_paths] + + +def materialize_external_outputs( + external_dir: Path, + output_dir: Path, + step: StepDefinition, + *, + matched_groups: list[list[Path]], + input_paths: list[Path], + params: dict, + symlink: bool = True, +) -> list[Path]: + external_dir = external_dir.resolve() + if not external_dir.is_dir(): + raise StepError(f"External output directory not found: {external_dir}") + + external_by_stem = {stem_key(path): path for path in list_images(external_dir)} + output_names = expected_output_filenames( + step, + matched_groups=matched_groups, + input_paths=input_paths, + params=params, + ) + stems = source_stems_for_step( + step, + matched_groups=matched_groups, + input_paths=input_paths, + ) + output_dir.mkdir(parents=True, exist_ok=True) + + output_paths: list[Path] = [] + for output_name, stem in zip(output_names, stems, strict=True): + source = external_by_stem.get(stem) + if source is None: + raise StepError( + f"External output for step '{step.step_id}' is missing stem {stem!r} " + f"in {external_dir}" + ) + destination = output_dir / output_name + if destination.exists() or destination.is_symlink(): + destination.unlink() + if symlink: + os.symlink(source, destination) + else: + shutil.copy2(source, destination) + output_paths.append(destination) + return output_paths + + +def read_manifest(path: Path) -> dict: + return json.loads(path.read_text(encoding="utf-8")) diff --git a/imagepipeline/core/runner.py b/imagepipeline/core/runner.py index 2ae6b82..d7d5fd7 100644 --- a/imagepipeline/core/runner.py +++ b/imagepipeline/core/runner.py @@ -13,6 +13,11 @@ from imagepipeline.core.manifest import ( utc_now_iso, write_manifest, ) +from imagepipeline.core.resume import ( + expected_output_paths, + materialize_external_outputs, + step_outputs_complete, +) from imagepipeline.core.step import ( INPUT_SOURCE, StepDefinition, @@ -31,6 +36,9 @@ class PipelineRunner: steps: list[StepDefinition], symlink_input: bool = True, verbose: bool = True, + existing_outputs: dict[str, Path] | None = None, + continue_from: Path | None = None, + skip_completed: bool = True, ) -> None: from imagepipeline.core.log import PipelineLogger @@ -40,11 +48,26 @@ class PipelineRunner: self.steps = steps self.symlink_input = symlink_input self.logger = PipelineLogger(verbose=verbose) + self.existing_outputs = { + key: Path(value).resolve() + for key, value in (existing_outputs or {}).items() + } + self.continue_from = ( + Path(continue_from).resolve() if continue_from is not None else None + ) + self.skip_completed = skip_completed self.output_root = self._build_output_root() self._input_link_dir = self.output_root / "input" self._results: dict[str, StepResult] = {} def _build_output_root(self) -> Path: + if self.continue_from is not None: + if not self.continue_from.is_dir(): + raise ValidationError( + f"continue_from directory not found: {self.continue_from}" + ) + return self.continue_from + timestamp = datetime.now().strftime("%y%m%d%H%M%S") folder_name = f"{self.name}_{timestamp}" output_root = self.output_base / folder_name @@ -56,6 +79,11 @@ class PipelineRunner: self.logger.info(f"Pipeline: {self.name}") self.logger.info(f"Input: {self.input_dir}") self.logger.info(f"Output: {self.output_root}") + if self.continue_from is not None: + self.logger.info("Mode: continue existing run") + if self.existing_outputs: + mapped = ", ".join(sorted(self.existing_outputs)) + self.logger.info(f"External outputs: {mapped}") self.logger.blank() self.logger.info(f"Found {len(images)} photo(s)") self.logger.blank() @@ -84,6 +112,7 @@ class PipelineRunner: output_files=[str(p) for p in result.output_paths], ) ) + write_manifest(self.output_root / "pipeline_manifest.json", manifest) manifest.finished_at = utc_now_iso() write_manifest(self.output_root / "pipeline_manifest.json", manifest) @@ -128,6 +157,13 @@ class PipelineRunner: output_dir = self.output_root / step.output_dir_name output_dir.mkdir(parents=True, exist_ok=True) + expected_paths = expected_output_paths( + output_dir, + step, + matched_groups=matched_groups, + input_paths=input_paths, + params=validated, + ) self.logger.step_start( step_index, @@ -138,6 +174,45 @@ class PipelineRunner: params=validated, ) + if step.step_id in self.existing_outputs: + output_paths = materialize_external_outputs( + self.existing_outputs[step.step_id], + output_dir, + step, + matched_groups=matched_groups, + input_paths=input_paths, + params=validated, + symlink=self.symlink_input, + ) + self.logger.step_reused( + step.output_dir_name, + str(self.existing_outputs[step.step_id]), + len(output_paths), + ) + self.logger.blank() + return StepResult( + step_id=step.step_id, + output_dir_name=step.output_dir_name, + module_name=step.module_name, + output_dir=output_dir, + input_paths=input_paths, + output_paths=output_paths, + params=validated, + ) + + if self.skip_completed and step_outputs_complete(expected_paths): + self.logger.step_skipped(step.output_dir_name, len(expected_paths)) + self.logger.blank() + return StepResult( + step_id=step.step_id, + output_dir_name=step.output_dir_name, + module_name=step.module_name, + output_dir=output_dir, + input_paths=input_paths, + output_paths=expected_paths, + params=validated, + ) + ctx = ModuleContext( input_paths=input_paths, output_dir=output_dir, diff --git a/imagepipeline/modules/__init__.py b/imagepipeline/modules/__init__.py index ad32c77..5b10917 100644 --- a/imagepipeline/modules/__init__.py +++ b/imagepipeline/modules/__init__.py @@ -3,6 +3,7 @@ import imagepipeline.modules.ai_exposure # noqa: F401 import imagepipeline.modules.ai_tone_map # noqa: F401 import imagepipeline.modules.comfy_flux_edit # noqa: F401 +import imagepipeline.modules.color_to_alpha # noqa: F401 import imagepipeline.modules.composite # noqa: F401 import imagepipeline.modules.crop_square # noqa: F401 import imagepipeline.modules.darktable_style # noqa: F401 @@ -10,6 +11,7 @@ import imagepipeline.modules.gmic # noqa: F401 import imagepipeline.modules.gmic_grayscale # noqa: F401 import imagepipeline.modules.imagemagick_fill # noqa: F401 import imagepipeline.modules.imagemagick_grayscale # noqa: F401 +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 diff --git a/imagepipeline/modules/color_to_alpha.py b/imagepipeline/modules/color_to_alpha.py new file mode 100644 index 0000000..d037d2e --- /dev/null +++ b/imagepipeline/modules/color_to_alpha.py @@ -0,0 +1,63 @@ +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.imagemagick_fill import normalize_color +from imagepipeline.modules.registry import register +from imagepipeline.utils.subprocess import run_command + + +def build_color_to_alpha_args(*, color: str, fuzz: float) -> list[str]: + """ImageMagick arguments to make ``color`` fully transparent.""" + c = normalize_color(color) + args = ["-alpha", "on"] + if fuzz > 0: + args.extend(["-fuzz", f"{fuzz}%"]) + args.extend(["-transparent", c]) + return args + + +@register +class ColorToAlphaModule(SubprocessModule): + name = "color_to_alpha" + description = ( + "Make a solid color transparent (GIMP-style color to alpha). " + "Outputs PNG with alpha." + ) + command_candidates = ("magick", "convert") + + @classmethod + def parameters(cls) -> dict[str, Param]: + return { + "color": Param( + "string", + required=True, + help="Color to make transparent (hex, e.g. #ffffff or ffffff)", + ), + "fuzz": Param( + "float", + default=0.0, + help=( + "Match tolerance in percent (ImageMagick -fuzz); " + "0 = exact color only" + ), + ), + } + + def run(self, ctx: ModuleContext) -> None: + command = self.resolve_command() + color = ctx.params["color"] + fuzz = ctx.params["fuzz"] + transparent_args = build_color_to_alpha_args(color=color, fuzz=fuzz) + ctx.output_dir.mkdir(parents=True, exist_ok=True) + total = len(ctx.input_paths) + + for index, src in enumerate(ctx.input_paths, start=1): + self.log_image(ctx, index, total, src) + dst = ctx.output_dir / f"{src.stem}.png" + run_command( + [command, str(src), *transparent_args, str(dst)], + ) diff --git a/imagepipeline/modules/gmic.py b/imagepipeline/modules/gmic.py index 444c91b..e32b973 100644 --- a/imagepipeline/modules/gmic.py +++ b/imagepipeline/modules/gmic.py @@ -4,6 +4,7 @@ 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.gmic import split_gmic_command from imagepipeline.utils.subprocess import run_command @@ -32,5 +33,5 @@ class GmicModule(SubprocessModule): for index, src in enumerate(ctx.input_paths, start=1): self.log_image(ctx, index, total, src) dst = ctx.output_dir / src.name - cmd = ["gmic", str(src), gmic_command, "-output", str(dst)] + cmd = ["gmic", str(src), *split_gmic_command(gmic_command), "-output", str(dst)] run_command(cmd, timeout=self.default_timeout) diff --git a/imagepipeline/modules/gmic_grayscale.py b/imagepipeline/modules/gmic_grayscale.py index 7fdf210..3561afe 100644 --- a/imagepipeline/modules/gmic_grayscale.py +++ b/imagepipeline/modules/gmic_grayscale.py @@ -4,6 +4,7 @@ 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.gmic import split_gmic_command from imagepipeline.utils.subprocess import run_command @@ -32,5 +33,5 @@ class GmicGrayscale(SubprocessModule): for index, src in enumerate(ctx.input_paths, start=1): self.log_image(ctx, index, total, src) dst = ctx.output_dir / src.name - cmd = ["gmic", str(src), gmic_command, "-output", str(dst)] + cmd = ["gmic", str(src), *split_gmic_command(gmic_command), "-output", str(dst)] run_command(cmd, timeout=self.default_timeout) diff --git a/imagepipeline/modules/imagemagick_resize.py b/imagepipeline/modules/imagemagick_resize.py new file mode 100644 index 0000000..cc41071 --- /dev/null +++ b/imagepipeline/modules/imagemagick_resize.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +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.subprocess import run_command + + +def build_resize_arguments(*, max_edge: int) -> list[str]: + if max_edge <= 0: + raise ValueError("max_edge must be positive") + return ["-auto-orient", "-resize", f"{max_edge}x{max_edge}>"] + + +@register +class ImageMagickResizeModule(SubprocessModule): + name = "imagemagick_resize" + description = ( + "Resize images so the longer side is at most max_edge pixels " + "(aspect ratio preserved; never upscales)" + ) + command_candidates = ("magick", "convert") + + @classmethod + def parameters(cls) -> dict[str, Param]: + return { + "max_edge": Param( + "int", + default=2000, + help="Maximum length of the longer side in pixels", + ), + } + + def run(self, ctx: ModuleContext) -> None: + command = self.resolve_command() + max_edge = ctx.params["max_edge"] + resize_args = build_resize_arguments(max_edge=max_edge) + ctx.output_dir.mkdir(parents=True, exist_ok=True) + total = len(ctx.input_paths) + + for index, src in enumerate(ctx.input_paths, start=1): + self.log_image(ctx, index, total, src) + dst = ctx.output_dir / src.name + run_command([command, str(src), *resize_args, str(dst)]) diff --git a/imagepipeline/utils/gmic.py b/imagepipeline/utils/gmic.py new file mode 100644 index 0000000..7dd6ed0 --- /dev/null +++ b/imagepipeline/utils/gmic.py @@ -0,0 +1,11 @@ +from __future__ import annotations + +import shlex + + +def split_gmic_command(command: str) -> list[str]: + """Split a G'MIC command string into argv tokens for subprocess.""" + command = command.strip() + if not command: + raise ValueError("G'MIC command must not be empty") + return shlex.split(command) diff --git a/pipelines/pipeline_2000px.py b/pipelines/pipeline_2000px.py new file mode 100644 index 0000000..3dc31be --- /dev/null +++ b/pipelines/pipeline_2000px.py @@ -0,0 +1,27 @@ +#!/usr/bin/env python3 +"""Resize exported images so the longer side is at most 2000 pixels.""" + +from pathlib import Path + +from imagepipeline import Pipeline + +INPUT = Path("/home/frank/pics/20260525_Shooting Baxxter Boys/darktable_exported") +OUTPUT_BASE = Path.home() / "pipeline_output" + +MAX_EDGE = 2000 + + +def main() -> None: + with Pipeline( + name="2000px", + input_dir=INPUT, + output_base=OUTPUT_BASE, + ) as p: + p.step("imagemagick_resize", inputs="input", max_edge=MAX_EDGE) + output_root = p.run() + + print(f"Pipeline finished. Output: {output_root}") + + +if __name__ == "__main__": + main() diff --git a/pipelines/pipeline_baxxter.py b/pipelines/pipeline_baxxter.py index 867b05e..958be29 100644 --- a/pipelines/pipeline_baxxter.py +++ b/pipelines/pipeline_baxxter.py @@ -8,6 +8,15 @@ from imagepipeline import Pipeline INPUT = Path("/home/frank/pics/20260525_Shooting Baxxter Boys/darktable_exported") OUTPUT_BASE = Path.home() / "pipeline_output" +# Reuse outputs from a previous run or external folder (key = step id, e.g. rembg_01). +EXISTING_OUTPUTS: dict[str, Path] = { + # "rembg_01": Path("/home/frank/pipeline_output/baxxter_260530102700/rembg_01"), +} + +# Resume an aborted run: point to its output root folder (or None for a fresh run). +CONTINUE_FROM: Path | None = Path("/home/frank/pipeline_output/baxxter_260530102700") +# CONTINUE_FROM = None + GRADIENT_COLOR1 = "#d7fd00ff" GRADIENT_COLOR2 = "#fc0adeff" @@ -36,6 +45,8 @@ def main() -> None: name="baxxter", input_dir=INPUT, output_base=OUTPUT_BASE, + existing_outputs=EXISTING_OUTPUTS or None, + continue_from=CONTINUE_FROM, ) as p: rembg_out = p.step("rembg", inputs="input") diff --git a/pipelines/pipeline_baxxter_2.py b/pipelines/pipeline_baxxter_2.py new file mode 100644 index 0000000..9b479d0 --- /dev/null +++ b/pipelines/pipeline_baxxter_2.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +"""Baxxter pipeline 2: composites using outputs from baxxter run 1.""" + +from pathlib import Path + +from imagepipeline import Pipeline + +INPUT = Path("/home/frank/pics/20260525_Shooting Baxxter Boys/darktable_exported") +OUTPUT_BASE = Path.home() / "pipeline_output" + +# Previous run (pipeline_baxxter.py). +PREV = Path("/home/frank/pipeline_output/baxxter_260530102700") + +# step_id -> folder from PREV. Third gmic step is gmic_03 here but reuses PREV/gmic_06. +EXISTING_OUTPUTS: dict[str, Path] = { + "rembg_01": PREV / "rembg_01", + "gmic_01": PREV / "gmic_01", + "gmic_02": PREV / "gmic_02", + "gmic_03": PREV / "gmic_06", +} + +YELLOW = "#d7fd00" + +# Commands only for step definition; reused steps are not executed. +GMIC_STEREO = "-gcd_stereo_img 0,0,2.028,1,1.714,3.06,4,1,0" +GMIC_DROP_SHADOW = "-fx_drop_shadow3d 0,0,0,10,1,1,2,0.5,252,10,222,200,0" +GMIC_JPR_SMOOTH = "-jpr_gradient_smooth 0,1.5" + + +def main() -> None: + with Pipeline( + name="baxxter_2", + input_dir=INPUT, + output_base=OUTPUT_BASE, + existing_outputs=EXISTING_OUTPUTS, + ) as p: + rembg = p.step("rembg", inputs="input") + + gmic_stereo = p.step("gmic", inputs=rembg, command=GMIC_STEREO) + gmic_shadow = p.step("gmic", inputs=rembg, command=GMIC_DROP_SHADOW) + gmic_smooth = p.step("gmic", inputs=rembg, command=GMIC_JPR_SMOOTH) + + gmic_stereo_alpha = p.step( + "color_to_alpha", inputs=gmic_stereo, color="#000000" + ) + yellow_bg = p.step("imagemagick_fill", inputs="input", color1=YELLOW) + + gmic_smooth_alpha = p.step( + "color_to_alpha", inputs=gmic_smooth, color="#7f7f7f" + ) + gmic_smooth_sized = p.step( + "imagemagick_scale_crop", + inputs=gmic_smooth_alpha, + scale=1.05, + ) + + # combine: original, gmic_01 (black to alpha), rembg + stereo_mid = p.step("composite", inputs=["input", gmic_stereo_alpha]) + p.step("composite", inputs=[stereo_mid, rembg]) + + # combine: yellow background, gmic_02, rembg + shadow_mid = p.step("composite", inputs=[yellow_bg, gmic_shadow]) + p.step("composite", inputs=[shadow_mid, rembg]) + + # combine: original, gmic_06 (#7f7f7f to alpha, scaled), rembg + smooth_mid = p.step("composite", inputs=["input", gmic_smooth_sized]) + p.step("composite", inputs=[smooth_mid, rembg]) + + output_root = p.run() + + print(f"Pipeline finished. Output: {output_root}") + + +if __name__ == "__main__": + main() diff --git a/tests/test_resume.py b/tests/test_resume.py new file mode 100644 index 0000000..11077bb --- /dev/null +++ b/tests/test_resume.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +import shutil +from pathlib import Path + +import pytest + +from imagepipeline.core.pipeline import Pipeline +from imagepipeline.core.resume import materialize_external_outputs, step_outputs_complete +from imagepipeline.core.step import StepDefinition +from imagepipeline.modules.imagemagick_grayscale import ImageMagickGrayscale +from imagepipeline.modules.registry import get_module +from imagepipeline.utils.gmic import split_gmic_command +from tests.conftest import make_png + + +class TestGmicCommandSplit: + def test_splits_command_and_arguments(self) -> None: + assert split_gmic_command("-gcd_stereo_img 0,0,2.028,1,1.714,3.06,4,1,0") == [ + "-gcd_stereo_img", + "0,0,2.028,1,1.714,3.06,4,1,0", + ] + + def test_preserves_quoted_empty_argument(self) -> None: + parts = split_gmic_command('-fx_custom_gradient 0,0,0,"",1,0') + assert parts == ["-fx_custom_gradient", "0,0,0,,1,0"] + + +class TestPipelineResume: + @pytest.mark.skipif(not shutil.which("magick"), reason="ImageMagick not installed") + def test_continue_skips_completed_steps(self, input_dir: Path, output_base: Path, capsys) -> None: + with Pipeline( + name="resume_test", + input_dir=input_dir, + output_base=output_base, + verbose=True, + ) as p: + first = p.step("imagemagick_grayscale", inputs="input") + p.step("imagemagick_grayscale", inputs=first) + root = p.run() + + capsys.readouterr() + + with Pipeline( + name="resume_test", + input_dir=input_dir, + output_base=output_base, + verbose=True, + continue_from=root, + ) as p: + step_a = p.step("imagemagick_grayscale", inputs="input") + p.step("imagemagick_grayscale", inputs=step_a) + resumed_root = p.run() + + assert resumed_root == root + output = capsys.readouterr().out + assert "Skipped step imagemagick_grayscale_01" in output + assert "Skipped step imagemagick_grayscale_02" in output + + @pytest.mark.skipif(not shutil.which("magick"), reason="ImageMagick not installed") + def test_existing_outputs_reuse_external_folder( + self, input_dir: Path, output_base: Path, tmp_path: Path, capsys + ) -> None: + external = tmp_path / "external_rembg" + external.mkdir() + for src in input_dir.iterdir(): + if src.is_file(): + make_png(external / src.name, width=4, height=4, rgb=(10, 20, 30)) + + with Pipeline( + name="external_test", + input_dir=input_dir, + output_base=output_base, + verbose=True, + existing_outputs={"imagemagick_grayscale_01": external}, + ) as p: + reused = p.step("imagemagick_grayscale", inputs="input") + p.step("imagemagick_grayscale", inputs=reused) + root = p.run() + + output = capsys.readouterr().out + assert "Reused external output for imagemagick_grayscale_01" in output + assert (root / "imagemagick_grayscale_01" / "photo_a.png").exists() + + +class TestMaterializeExternal: + def test_links_files_by_stem(self, tmp_path: Path) -> None: + external = tmp_path / "external" + external.mkdir() + make_png(external / "photo_a.png") + output_dir = tmp_path / "out" + step = StepDefinition( + step_id="imagemagick_grayscale_01", + module_name="imagemagick_grayscale", + module=ImageMagickGrayscale, + input_refs=["input"], + params={}, + output_dir_name="imagemagick_grayscale_01", + ) + input_paths = [tmp_path / "photo_a.png"] + paths = materialize_external_outputs( + external, + output_dir, + step, + matched_groups=[[path] for path in input_paths], + input_paths=input_paths, + params={}, + ) + assert len(paths) == 1 + assert paths[0].name == "photo_a.png" + assert paths[0].is_symlink() + + def test_step_outputs_complete(self, tmp_path: Path) -> None: + output_dir = tmp_path / "done" + output_dir.mkdir() + make_png(output_dir / "photo.png") + assert step_outputs_complete([output_dir / "photo.png"]) + assert not step_outputs_complete([output_dir / "missing.png"]) diff --git a/tests/test_workflow_modules.py b/tests/test_workflow_modules.py index 9d0eada..7126a10 100644 --- a/tests/test_workflow_modules.py +++ b/tests/test_workflow_modules.py @@ -6,6 +6,10 @@ from pathlib import Path import pytest from imagepipeline.core.params import validate_params +from imagepipeline.modules.color_to_alpha import ( + ColorToAlphaModule, + build_color_to_alpha_args, +) from imagepipeline.modules.composite import CompositeModule from imagepipeline.modules.crop_square import CropSquareModule from imagepipeline.modules.darktable_style import DarktableStyleModule @@ -14,6 +18,10 @@ from imagepipeline.modules.imagemagick_fill import ( build_fill_arguments, ) from imagepipeline.modules.imagemagick_grayscale import ImageMagickGrayscale +from imagepipeline.modules.imagemagick_resize import ( + ImageMagickResizeModule, + build_resize_arguments, +) from imagepipeline.modules.gmic_grayscale import GmicGrayscale from imagepipeline.modules.registry import get_module, list_modules from imagepipeline.modules.rembg import RembgModule @@ -31,6 +39,8 @@ class TestModuleRegistration: "darktable_style", "imagemagick_grayscale", "imagemagick_fill", + "color_to_alpha", + "imagemagick_resize", "crop_square", ): assert name in names @@ -42,6 +52,85 @@ class TestModuleRegistration: assert get_module("darktable_style") is DarktableStyleModule assert get_module("crop_square") is CropSquareModule assert get_module("imagemagick_fill") is ImageMagickFillModule + assert get_module("color_to_alpha") is ColorToAlphaModule + + +class TestImageMagickResize: + def test_build_resize_arguments(self) -> None: + assert build_resize_arguments(max_edge=2000) == [ + "-auto-orient", + "-resize", + "2000x2000>", + ] + + def test_rejects_non_positive_max_edge(self) -> None: + with pytest.raises(ValueError, match="positive"): + build_resize_arguments(max_edge=0) + + def test_default_max_edge(self) -> None: + params = ImageMagickResizeModule.validate_module_params({}) + assert params["max_edge"] == 2000 + + +class TestColorToAlpha: + def test_build_args_exact(self) -> None: + assert build_color_to_alpha_args(color="#00ff00", fuzz=0.0) == [ + "-alpha", + "on", + "-transparent", + "#00ff00", + ] + + def test_build_args_with_fuzz(self) -> None: + assert build_color_to_alpha_args(color="ffffff", fuzz=2.5) == [ + "-alpha", + "on", + "-fuzz", + "2.5%", + "-transparent", + "#ffffff", + ] + + def test_requires_color(self) -> None: + with pytest.raises(ValueError, match="required"): + ColorToAlphaModule.validate_module_params({}) + + @pytest.mark.skipif(not has_magick, reason="ImageMagick not installed") + def test_makes_matching_color_transparent(self, tmp_path: Path) -> None: + from imagepipeline.core.context import ModuleContext + from imagepipeline.utils.subprocess import run_command + + src = tmp_path / "green.png" + output_dir = tmp_path / "out" + output_dir.mkdir() + magick = shutil.which("magick") or shutil.which("convert") + run_command([magick, "-size", "8x8", "xc:#00ff00", str(src)]) + + ctx = ModuleContext( + input_paths=[src], + matched_groups=[], + output_dir=output_dir, + params=ColorToAlphaModule.validate_module_params({"color": "#00ff00"}), + pipeline_output_root=tmp_path, + step_id="color_to_alpha_01", + logger=None, + ) + ColorToAlphaModule().run(ctx) + + dst = output_dir / "green.png" + assert dst.is_file() + result = run_command( + [ + magick, + str(dst), + "-alpha", + "extract", + "-format", + "%[fx:mean]", + "info:", + ] + ) + assert float(result.stdout.strip()) == 0.0 class TestImageMagickFill: