Frank Schwenk
0daf3e2315
Replace prior_steps and runs_last with inputs=[...] step refs so GIMP export waits only on listed layers. Add rezepttest pipeline and bokeh-oktagon recipe. Co-authored-by: Cursor <cursoragent@cursor.com>
77 lines
2.2 KiB
Python
77 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from pathlib import Path
|
|
from typing import Any, ClassVar
|
|
|
|
from imagepipeline.core.context import ModuleContext
|
|
from imagepipeline.core.params import Param, validate_params
|
|
from imagepipeline.utils.files import is_image
|
|
|
|
|
|
class BaseModule(ABC):
|
|
"""Base class for all pipeline modules."""
|
|
|
|
name: ClassVar[str]
|
|
description: ClassVar[str] = ""
|
|
supported_input_formats: ClassVar[tuple[str, ...]] = (
|
|
".jpg",
|
|
".jpeg",
|
|
".png",
|
|
".tif",
|
|
".tiff",
|
|
".webp",
|
|
)
|
|
|
|
@classmethod
|
|
def parameters(cls) -> dict[str, Param]:
|
|
return {}
|
|
|
|
@classmethod
|
|
def validate_module_params(cls, raw: dict[str, Any]) -> dict[str, Any]:
|
|
return validate_params(cls.parameters(), raw)
|
|
|
|
@classmethod
|
|
def check_dependencies(cls) -> None:
|
|
"""Raise DependencyError if required external tools are missing."""
|
|
|
|
@abstractmethod
|
|
def run(self, ctx: ModuleContext) -> None:
|
|
"""Process ctx.input_paths and write outputs into ctx.output_dir."""
|
|
|
|
@classmethod
|
|
def expected_output_filenames(
|
|
cls,
|
|
*,
|
|
matched_groups: list[list[Path]],
|
|
input_paths: list[Path],
|
|
params: dict[str, Any],
|
|
) -> list[str]:
|
|
return [path.name for path in input_paths]
|
|
|
|
def log_image(self, ctx: ModuleContext, index: int, total: int, path: Path) -> None:
|
|
ctx.log_image(self.name, index, total, path)
|
|
|
|
def list_output_images(self, ctx: ModuleContext) -> list[Path]:
|
|
return sorted(p for p in ctx.output_dir.iterdir() if is_image(p))
|
|
|
|
|
|
class SubprocessModule(BaseModule):
|
|
"""Base class for modules that shell out to CLI tools."""
|
|
|
|
command_candidates: ClassVar[tuple[str, ...]] = ()
|
|
default_timeout: ClassVar[float | None] = None
|
|
|
|
@classmethod
|
|
def check_dependencies(cls) -> None:
|
|
from imagepipeline.utils.subprocess import require_command
|
|
|
|
if cls.command_candidates:
|
|
require_command(*cls.command_candidates)
|
|
|
|
@classmethod
|
|
def resolve_command(cls) -> str:
|
|
from imagepipeline.utils.subprocess import require_command
|
|
|
|
return require_command(*cls.command_candidates)
|