Support YAML pipeline configuration

bgfilter/config.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+from dataclasses import fields, replace
+from pathlib import Path
+from typing import Any, TypeVar
+
+from .settings import (
+    AlphaPostSettings,
+    ChromaSettings,
+    DespillSettings,
+    ModelSettings,
+    PipelineSettings,
+    TrimapSettings,
+)
+
+T = TypeVar("T")
+
+
+def _load_yaml(path: str | Path) -> dict[str, Any]:
+    try:
+        import yaml
+    except ModuleNotFoundError as exc:
+        raise RuntimeError("Missing dependency 'PyYAML'. Install it with: pip install PyYAML") from exc
+
+    with Path(path).open("r", encoding="utf-8") as handle:
+        data = yaml.safe_load(handle) or {}
+    if not isinstance(data, dict):
+        raise ValueError(f"Config file must contain a YAML mapping: {path}")
+    return data
+
+
+def _update_dataclass(instance: T, values: dict[str, Any] | None) -> T:
+    if not values:
+        return instance
+    allowed = {field.name for field in fields(instance)}
+    unknown = sorted(set(values) - allowed)
+    if unknown:
+        cls_name = type(instance).__name__
+        raise ValueError(f"Unknown {cls_name} field(s): {', '.join(unknown)}")
+    return replace(instance, **values)
+
+
+def settings_from_dict(data: dict[str, Any]) -> PipelineSettings:
+    allowed_sections = {"chroma", "trimap", "alpha_post", "despill", "model"}
+    unknown_sections = sorted(set(data) - allowed_sections)
+    if unknown_sections:
+        raise ValueError(f"Unknown config section(s): {', '.join(unknown_sections)}")
+
+    return PipelineSettings(
+        chroma=_update_dataclass(ChromaSettings(), data.get("chroma")),
+        trimap=_update_dataclass(TrimapSettings(), data.get("trimap")),
+        alpha_post=_update_dataclass(AlphaPostSettings(), data.get("alpha_post")),
+        despill=_update_dataclass(DespillSettings(), data.get("despill")),
+        model=_update_dataclass(ModelSettings(), data.get("model")),
+    )
+
+
+def load_settings(path: str | Path | None = None) -> PipelineSettings:
+    if path is None:
+        return PipelineSettings()
+    return settings_from_dict(_load_yaml(path))
+
+
+def override_settings(settings: PipelineSettings, **overrides: Any) -> PipelineSettings:
+    chroma = settings.chroma
+    trimap = settings.trimap
+    alpha_post = settings.alpha_post
+    despill = settings.despill
+    model = settings.model
+
+    model_updates = {
+        key: overrides[key]
+        for key in ["model_name", "device", "matting_method", "fallback_to_chroma_alpha"]
+        if overrides.get(key) is not None
+    }
+    trimap_updates = {
+        key: overrides[key]
+        for key in [
+            "sure_bg_threshold",
+            "sure_fg_threshold",
+            "unknown_radius_ratio",
+            "fg_safe_radius_ratio",
+        ]
+        if overrides.get(key) is not None
+    }
+    despill_updates: dict[str, Any] = {}
+    if overrides.get("despill") is not None:
+        despill_updates["enabled"] = overrides["despill"]
+
+    return PipelineSettings(
+        chroma=chroma,
+        trimap=_update_dataclass(trimap, trimap_updates),
+        alpha_post=alpha_post,
+        despill=_update_dataclass(despill, despill_updates),
+        model=_update_dataclass(model, model_updates),
+    )