BGfilter/bgfilter/pipeline.py

from __future__ import annotations

import json
from dataclasses import asdict
from pathlib import Path

import numpy as np

from .alpha_post import (
    clean_alpha,
    cross_check_alpha,
    enforce_trimap,
    suppress_alpha_by_chroma,
)
from .chroma import (
    bg_hue_projection,
    compute_bg_confidence,
    convert_color_spaces,
    parse_hex_color,
)
from .despill import despill
from .foreground import estimate_foreground_rgb
from .io import load_rgb, save_gray, save_rgb, save_rgba, write_text
from .qa import BACKGROUND_COLORS, composite, make_qa_grid, save_previews
from .settings import PipelineSettings, SegmentationSettings
from .trimap import (
    fuse_trimap,
    fuse_trimap_directional,
    generate_trimap,
    trimap_to_alpha_seed,
)
from .vitmatte_infer import ViTMatteRunner


class MattingPipeline:
    def __init__(self, settings: PipelineSettings):
        self.settings = settings
        self._runner: ViTMatteRunner | None = None
        self._segmenter = None
        self._cross_checker = None

    def _segment(self, rgb: np.ndarray) -> np.ndarray:
        if self._segmenter is None:
            from .segmentation import make_segmenter

            self._segmenter = make_segmenter(self.settings.segmentation)
        return self._segmenter.mask(rgb)

    def _second_opinion(self, rgb: np.ndarray) -> np.ndarray:
        if self._cross_checker is None:
            from .segmentation import BiRefNetSegmenter

            check = self.settings.cross_check
            self._cross_checker = BiRefNetSegmenter(
                SegmentationSettings(
                    model_name=check.model_name,
                    device=self.settings.segmentation.device,
                    input_size=check.input_size,
                    precision=check.precision,
                )
            )
        return self._cross_checker.mask(rgb)

    def _predict_alpha(self, rgb: np.ndarray, trimap: np.ndarray, bg_confidence: np.ndarray) -> tuple[np.ndarray, str]:
        if self.settings.model.matting_method == "chroma":
            return trimap_to_alpha_seed(trimap, bg_confidence), "chroma"
        if self.settings.model.matting_method != "vitmatte":
            raise RuntimeError(
                f"Unsupported matting method '{self.settings.model.matting_method}'. "
                "Use 'vitmatte' or 'chroma'."
            )

        try:
            if self._runner is None:
                self._runner = ViTMatteRunner(self.settings.model)
            return self._runner.predict(rgb, trimap), "vitmatte"
        except RuntimeError:
            if not self.settings.model.fallback_to_chroma_alpha:
                raise
            return trimap_to_alpha_seed(trimap, bg_confidence), "chroma_fallback"

    def run_image(
        self,
        input_path: str | Path,
        output_path: str | Path,
        debug_dir: str | Path | None,
    ) -> dict:
        return _run_image(input_path, output_path, debug_dir, self)


def run_image(
    input_path: str | Path,
    output_path: str | Path,
    debug_dir: str | Path | None,
    settings: PipelineSettings,
) -> dict:
    return MattingPipeline(settings).run_image(input_path, output_path, debug_dir)


def _run_image(
    input_path: str | Path,
    output_path: str | Path,
    debug_dir: str | Path | None,
    pipeline: MattingPipeline,
) -> dict:
    settings = pipeline.settings
    rgb = load_rgb(input_path)
    screen_color = parse_hex_color(settings.screen_color) if settings.screen_color else None
    bg_confidence, model = compute_bg_confidence(
        rgb, settings=settings.chroma, screen_color=screen_color
    )
    second_alpha = None
    if settings.segmentation.enabled:
        if settings.cross_check.enabled and settings.cross_check.reuse_as_seg:
            # One HR-matting forward serves both the trimap topology and the
            # cross-check second opinion; the primary seg model is never loaded.
            second_alpha = pipeline._second_opinion(rgb)
            seg_mask = second_alpha
        else:
            seg_mask = pipeline._segment(rgb)
        mode = settings.trimap.mode
        if mode == "seg":
            trimap, trimap_stats = fuse_trimap(seg_mask, bg_confidence, settings.trimap)
        elif mode in ("directional", "directional-hard-bg"):
            lab = convert_color_spaces(rgb)[1]
            trimap, trimap_stats = fuse_trimap_directional(
                seg_mask, bg_confidence, lab, model.lab_center, settings.trimap
            )
        else:
            raise RuntimeError(
                f"Unknown trimap.mode '{mode}'. "
                "Use 'directional', 'seg', or 'directional-hard-bg'."
            )
    else:
        seg_mask = None
        trimap, trimap_stats = generate_trimap(bg_confidence, settings.trimap)
    raw_alpha, alpha_source = pipeline._predict_alpha(rgb, trimap, bg_confidence)

    alpha = enforce_trimap(raw_alpha, trimap)
    alpha = suppress_alpha_by_chroma(
        alpha, bg_confidence, trimap, settings.alpha_post,
        # The matte-confidence gate only makes sense for a real matting prediction;
        # a chroma-seeded alpha is itself colour evidence, so no gate there.
        raw_alpha=raw_alpha if alpha_source == "vitmatte" else None,
    )
    alpha = clean_alpha(alpha, trimap, settings.alpha_post)
    if settings.cross_check.enabled:
        lab = convert_color_spaces(rgb)[1]
        if second_alpha is None:
            second_alpha = pipeline._second_opinion(rgb)
        alpha = cross_check_alpha(
            alpha,
            second_alpha,
            bg_hue_projection(lab, model.lab_center),
            lab[..., 0],
            trimap,
            settings.cross_check,
        )
    foreground = estimate_foreground_rgb(
        rgb, alpha, bg_confidence, model, settings.foreground
    )
    corrected_rgb, color_mask = despill(
        foreground.rgb, alpha, model, settings.despill
    )
    save_rgba(output_path, corrected_rgb, alpha)

    result = {
        "input": str(input_path),
        "output": str(output_path),
        "alpha_source": alpha_source,
        "background_model": model.to_dict(),
        "trimap": trimap_stats,
        "settings": asdict(settings),
    }

    if debug_dir is not None:
        debug = Path(debug_dir)
        debug.mkdir(parents=True, exist_ok=True)
        save_gray(debug / "bg_confidence.png", bg_confidence)
        save_gray(debug / "trimap.png", trimap)
        if seg_mask is not None:
            save_gray(debug / "seg_mask.png", seg_mask)
        if second_alpha is not None:
            save_gray(debug / "cross_check_alpha.png", second_alpha)
        save_gray(debug / "alpha.png", alpha)
        save_rgb(debug / "foreground_rgb.png", corrected_rgb)
        save_gray(debug / "color_mask.png", color_mask)
        save_previews(debug, corrected_rgb, alpha)
        tiles = {
            "input": rgb,
            "bg confidence": np.repeat((bg_confidence[..., None] * 255).astype(np.uint8), 3, axis=2),
            "trimap": np.repeat(trimap[..., None], 3, axis=2),
            "alpha": np.repeat((alpha[..., None] * 255).astype(np.uint8), 3, axis=2),
        }
        for name, color in BACKGROUND_COLORS.items():
            tiles[f"preview {name}"] = composite(corrected_rgb, alpha, color)
        make_qa_grid(debug, tiles)
        write_text(debug / "metadata.json", json.dumps(result, indent=2, ensure_ascii=False))

    return result