froxxxy/imagepipeline
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

A long time ago, in a galaxy far far away...

Browse Source
This commit is contained in:
Frank Schwenk
2026-06-21 10:26:10 +02:00
parent e7cdb8dd6f
commit 980c7f3b8b
15 changed files with 645 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
imagepipeline/core/log.py
+6
View File
@@ -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("")
imagepipeline/core/pipeline.py
+9
View File
@@ -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
imagepipeline/core/resume.py
+110
View File
@@ -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"))
imagepipeline/core/runner.py
+75
View File
@@ -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,
imagepipeline/modules/__init__.py
+2
View File
@@ -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
imagepipeline/modules/color_to_alpha.py
+63
View File
@@ -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)],
)
imagepipeline/modules/gmic.py
+2 -1
View File
@@ -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)
imagepipeline/modules/gmic_grayscale.py
+2 -1
View File
@@ -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)
imagepipeline/modules/imagemagick_resize.py
+45
View File
@@ -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)])
imagepipeline/utils/gmic.py
+11
View File
@@ -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)
pipelines/pipeline_2000px.py
+27
View File
@@ -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()
pipelines/pipeline_baxxter.py
+11
View File
@@ -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")
pipelines/pipeline_baxxter_2.py
+75
View File
@@ -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()
tests/test_resume.py
+118
View File
@@ -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"])
tests/test_workflow_modules.py
+89
View File
@@ -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: