Add FastAPI HTTP service for background removal

Wrap the BGfilter matting pipeline in a long-running FastAPI service
(POST /remove-background, GET /healthz) exposing only screen_color and
seg_model, matching the birefnet-service request contract.

- bgfilter/pipeline.py: add in-memory run_rgb entry returning a
  MattingResult; allow injecting a shared runner/segmenter. CLI file
  path (run_image) behaviour is unchanged.
- bgfilter/service.py: PipelineManager caches one shared ViTMatte runner
  and per-backend segmenters, so screen_color/seg_model never reload a
  model.
- app.py: thin FastAPI layer -- param validation, image decode/encode,
  error codes (400/413/500), timing headers, global asyncio lock.
- requirements.txt: fastapi, uvicorn[standard], python-multipart.
- README + design doc: HTTP service usage and HF-mirror/proxy/Xet
  download notes.

Verified end-to-end (base ViTMatte + BiRefNet + anime-seg): 11/11
acceptance checks pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

bgfilter/pipeline.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import json
-from dataclasses import asdict
+from dataclasses import asdict, dataclass
 from pathlib import Path
 
 import numpy as np
@@ -22,11 +22,38 @@ from .trimap import (
 from .vitmatte_infer import ViTMatteRunner
 
 
+@dataclass
+class MattingResult:
+    """In-memory result of one matting run.
+
+    ``rgb`` is the despilled foreground (uint8 HxWx3) and ``alpha`` is float32 HxW in
+    0..1; together they compose the output RGBA. The remaining arrays are the
+    intermediate maps used for debug previews.
+    """
+
+    rgb: np.ndarray
+    alpha: np.ndarray
+    metadata: dict
+    bg_confidence: np.ndarray
+    trimap: np.ndarray
+    seg_mask: np.ndarray | None
+    color_mask: np.ndarray
+    input_rgb: np.ndarray
+
+
 class MattingPipeline:
-    def __init__(self, settings: PipelineSettings):
+    def __init__(
+        self,
+        settings: PipelineSettings,
+        *,
+        runner: ViTMatteRunner | None = None,
+        segmenter=None,
+    ):
         self.settings = settings
-        self._runner: ViTMatteRunner | None = None
-        self._segmenter = None
+        # Pre-loaded models can be injected so a long-lived service shares them
+        # across requests instead of reloading per call; None = lazy-load on first use.
+        self._runner: ViTMatteRunner | None = runner
+        self._segmenter = segmenter
 
     def _segment(self, rgb: np.ndarray) -> np.ndarray:
         if self._segmenter is None:
@@ -53,6 +80,10 @@ class MattingPipeline:
                 raise
             return trimap_to_alpha_seed(trimap, bg_confidence), "chroma_fallback"
 
+    def run_rgb(self, rgb: np.ndarray) -> MattingResult:
+        """Run the full pipeline on an in-memory RGB array (no disk I/O)."""
+        return _process_rgb(rgb, self)
+
     def run_image(
         self,
         input_path: str | Path,
@@ -71,14 +102,8 @@ def run_image(
     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:
+def _process_rgb(rgb: np.ndarray, pipeline: MattingPipeline) -> MattingResult:
     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
@@ -111,39 +136,56 @@ def _run_image(
     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)
+    corrected_rgb, color_mask = despill(foreground.rgb, alpha, model, settings.despill)
 
-    result = {
-        "input": str(input_path),
-        "output": str(output_path),
+    metadata = {
         "alpha_source": alpha_source,
         "background_model": model.to_dict(),
         "trimap": trimap_stats,
         "settings": asdict(settings),
     }
+    return MattingResult(
+        rgb=corrected_rgb,
+        alpha=alpha,
+        metadata=metadata,
+        bg_confidence=bg_confidence,
+        trimap=trimap,
+        seg_mask=seg_mask,
+        color_mask=color_mask,
+        input_rgb=rgb,
+    )
+
+
+def _run_image(
+    input_path: str | Path,
+    output_path: str | Path,
+    debug_dir: str | Path | None,
+    pipeline: MattingPipeline,
+) -> dict:
+    res = _process_rgb(load_rgb(input_path), pipeline)
+    save_rgba(output_path, res.rgb, res.alpha)
+
+    result = {"input": str(input_path), "output": str(output_path), **res.metadata}
 
     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)
-        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)
+        save_gray(debug / "bg_confidence.png", res.bg_confidence)
+        save_gray(debug / "trimap.png", res.trimap)
+        if res.seg_mask is not None:
+            save_gray(debug / "seg_mask.png", res.seg_mask)
+        save_gray(debug / "alpha.png", res.alpha)
+        save_rgb(debug / "foreground_rgb.png", res.rgb)
+        save_gray(debug / "color_mask.png", res.color_mask)
+        save_previews(debug, res.rgb, res.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),
+            "input": res.input_rgb,
+            "bg confidence": np.repeat((res.bg_confidence[..., None] * 255).astype(np.uint8), 3, axis=2),
+            "trimap": np.repeat(res.trimap[..., None], 3, axis=2),
+            "alpha": np.repeat((res.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)
+            tiles[f"preview {name}"] = composite(res.rgb, res.alpha, color)
         make_qa_grid(debug, tiles)
         write_text(debug / "metadata.json", json.dumps(result, indent=2, ensure_ascii=False))
 

bgfilter/service.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import logging
+import threading
+from dataclasses import replace
+
+import numpy as np
+
+from .config import _SEG_MODELS, override_settings
+from .pipeline import MattingPipeline, MattingResult
+from .segmentation import make_segmenter
+from .settings import PipelineSettings
+from .vitmatte_infer import ViTMatteRunner
+
+logger = logging.getLogger("bgfilter.service")
+
+# HTTP-facing seg_model values, in preference order (first = default).
+SEG_MODELS: tuple[str, ...] = tuple(_SEG_MODELS)
+
+
+class PipelineManager:
+    """Long-lived holder of the matting models.
+
+    ViTMatte is the matting model shared by both segmentation backends, so it is
+    loaded once and reused. Each segmentation backend is lazily loaded once on
+    first use and cached by name. Per-request parameters (``screen_color``,
+    ``seg_model``) never trigger a model reload -- ``screen_color`` only affects
+    this request's chroma/despill, and ``seg_model`` just selects an already-cached
+    (or first-time-loaded) segmenter.
+    """
+
+    def __init__(self, base_settings: PipelineSettings, preload: bool = True):
+        self._base = base_settings
+        self._runner: ViTMatteRunner | None = None
+        self._segmenters: dict[str, object] = {}
+        self._load_lock = threading.Lock()
+        if preload:
+            if base_settings.model.matting_method == "vitmatte":
+                self._get_runner()
+            if base_settings.segmentation.enabled:
+                self._get_segmenter(base_settings.segmentation.backend)
+
+    @property
+    def default_seg_model(self) -> str:
+        return self._base.segmentation.backend
+
+    @property
+    def device(self) -> str:
+        return self._base.model.device
+
+    def _get_runner(self) -> ViTMatteRunner:
+        if self._runner is None:
+            with self._load_lock:
+                if self._runner is None:
+                    logger.info(
+                        "loading ViTMatte: model=%s device=%s",
+                        self._base.model.model_name,
+                        self._base.model.device,
+                    )
+                    self._runner = ViTMatteRunner(self._base.model)
+        return self._runner
+
+    def _get_segmenter(self, backend: str):
+        seg = self._segmenters.get(backend)
+        if seg is None:
+            with self._load_lock:
+                seg = self._segmenters.get(backend)
+                if seg is None:
+                    seg_settings = replace(
+                        self._base.segmentation,
+                        backend=backend,
+                        model_name=_SEG_MODELS[backend],
+                    )
+                    logger.info(
+                        "loading segmenter: backend=%s model=%s device=%s",
+                        backend,
+                        seg_settings.model_name,
+                        seg_settings.device,
+                    )
+                    seg = make_segmenter(seg_settings)
+                    self._segmenters[backend] = seg
+        return seg
+
+    def process(
+        self, rgb: np.ndarray, *, screen_color: str | None, seg_model: str
+    ) -> MattingResult:
+        if seg_model not in _SEG_MODELS:
+            raise ValueError(f"invalid seg_model '{seg_model}'")
+        settings = override_settings(
+            self._base, screen_color=screen_color, seg_backend=seg_model
+        )
+        runner = (
+            self._get_runner()
+            if settings.model.matting_method == "vitmatte"
+            else None
+        )
+        segmenter = (
+            self._get_segmenter(seg_model) if settings.segmentation.enabled else None
+        )
+        pipeline = MattingPipeline(settings, runner=runner, segmenter=segmenter)
+        return pipeline.run_rgb(rgb)