Create app.py

app.py

@@ -0,0 +1,443 @@
+"""
+Gradio app: object-to-object distance estimation using SAM3 + Depth Anything 3.
+
+Deploy on Hugging Face Spaces:
+  1. Create a new Space -> SDK: Gradio -> hardware: GPU recommended (SAM3 + DA3
+     both run on CPU but are slow; a T4 or better is a big speedup).
+  2. Upload this file as app.py, plus requirements.txt (below) and README.md.
+  3. facebook/sam3 is gated: accept the license at
+     https://huggingface.co/facebook/sam3, then add a `HF_TOKEN` secret to
+     your Space (Settings -> Repository secrets) with a token that has access.
+"""
+
+import os
+import numpy as np
+import torch
+import cv2
+from PIL import Image, ExifTags
+from scipy import ndimage
+import gradio as gr
+
+from transformers import Sam3Processor, Sam3Model
+from depth_anything_3.api import DepthAnything3
+
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+HF_TOKEN = os.environ.get("HF_TOKEN")  # set as a Space secret if sam3 is gated for you
+ARUCO_DICT = cv2.aruco.getPredefinedDictionary(cv2.aruco.DICT_4X4_50)
+
+# --------------------------------------------------------------------------
+# Lazy, cached model loading — Spaces reload this module per-worker, so we
+# only want to pay the load cost once, not on every button click.
+# --------------------------------------------------------------------------
+
+_segmenter = None
+_depther = None
+
+
+def get_segmenter():
+    global _segmenter
+    if _segmenter is None:
+        model = Sam3Model.from_pretrained("facebook/sam3", token=HF_TOKEN).to(DEVICE)
+        processor = Sam3Processor.from_pretrained("facebook/sam3", token=HF_TOKEN)
+        _segmenter = (model, processor)
+    return _segmenter
+
+
+def get_depther(model_id: str):
+    global _depther
+    if _depther is None or _depther[0] != model_id:
+        model = DepthAnything3.from_pretrained(model_id).to(DEVICE)
+        _depther = (model_id, model)
+    return _depther[1]
+
+
+# --------------------------------------------------------------------------
+# Pipeline stages (same logic as the standalone script)
+# --------------------------------------------------------------------------
+
+def segment(image: Image.Image, text_prompt: str, score_threshold: float = 0.5) -> np.ndarray:
+    model, processor = get_segmenter()
+    inputs = processor(images=image, text=text_prompt, return_tensors="pt").to(DEVICE)
+    with torch.no_grad():
+        outputs = model(**inputs)
+
+    masks = processor.post_process_masks(outputs.pred_masks.cpu(), [image.size[::-1]])[0]
+    scores = outputs.pred_scores.cpu() if hasattr(outputs, "pred_scores") else None
+
+    if masks.shape[0] == 0:
+        raise gr.Error(f"No object found matching '{text_prompt}'. Try a more specific or different phrase.")
+
+    if scores is not None and scores.numel() > 0:
+        best_idx = int(torch.argmax(scores))
+        if float(scores[best_idx]) < score_threshold:
+            raise gr.Error(
+                f"Best match for '{text_prompt}' only scored {float(scores[best_idx]):.2f} "
+                f"(threshold {score_threshold}). Try rephrasing."
+            )
+    else:
+        best_idx = 0
+
+    return masks[best_idx].squeeze().numpy().astype(bool)
+
+
+def erode_mask(mask: np.ndarray, pixels: int = 3) -> np.ndarray:
+    if pixels <= 0:
+        return mask
+    eroded = ndimage.binary_erosion(mask, iterations=pixels)
+    return eroded if eroded.sum() > 20 else mask
+
+
+def robust_object_depth(depth, conf, mask, conf_percentile=40.0, mad_k=3.0):
+    d, c = depth[mask], conf[mask]
+    if d.size == 0:
+        raise gr.Error("Mask is empty after erosion — object may be too small in this image.")
+
+    conf_cut = np.percentile(c, conf_percentile)
+    d_kept = d[c >= conf_cut]
+    if d_kept.size < 5:
+        d_kept = d
+
+    med = np.median(d_kept)
+    mad = np.median(np.abs(d_kept - med)) + 1e-8
+    inliers = d_kept[np.abs(d_kept - med) <= mad_k * 1.4826 * mad]
+    if inliers.size == 0:
+        inliers = d_kept
+
+    return float(np.median(inliers)), float(np.std(inliers)), int(inliers.size)
+
+
+def backproject_to_3d(mask, robust_depth, intrinsics):
+    ys, xs = np.nonzero(mask)
+    cy, cx = np.median(ys), np.median(xs)
+    fx, fy = intrinsics[0, 0], intrinsics[1, 1]
+    px, py = intrinsics[0, 2], intrinsics[1, 2]
+    z = robust_depth
+    x = (cx - px) * z / fx
+    y = (cy - py) * z / fy
+    return np.array([x, y, z], dtype=np.float64)
+
+
+OBJECT_COLORS = [
+    (242, 183, 5),    # amber
+    (13, 148, 136),   # teal
+    (76, 29, 149),    # violet
+    (224, 102, 90),   # coral
+    (59, 130, 246),   # blue
+    (236, 72, 153),   # pink
+]
+
+
+def mask_overlay(image: Image.Image, masks: list) -> Image.Image:
+    """Tints each object's mask a distinct color (cycling through
+    OBJECT_COLORS if there are more objects than colors) for a quick
+    visual sanity-check of what got segmented."""
+    arr = np.array(image).astype(np.float32)
+    overlay = arr.copy()
+    for i, mask in enumerate(masks):
+        color = np.array(OBJECT_COLORS[i % len(OBJECT_COLORS)])
+        overlay[mask] = overlay[mask] * 0.4 + color * 0.6
+    return Image.fromarray(overlay.astype(np.uint8))
+
+
+# --------------------------------------------------------------------------
+# Addition 1: image-quality gate.
+# A blurry/motion-blurred view doesn't just give bad depth for itself — in
+# multi-view mode it can quietly drag down DA3's joint pose/depth solve for
+# every other view too. Flag it before it reaches the models.
+# --------------------------------------------------------------------------
+
+def check_blur(image: Image.Image, threshold: float = 100.0):
+    """Returns (sharpness_score, is_blurry). Variance of the Laplacian —
+    lower means blurrier. Threshold is scene-dependent; 100 is a reasonable
+    default for well-lit photos but tune it if you get false positives."""
+    gray = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2GRAY)
+    score = float(cv2.Laplacian(gray, cv2.CV_64F).var())
+    return score, score < threshold
+
+
+# --------------------------------------------------------------------------
+# Addition 2: EXIF-derived intrinsics as an alternative to DA3's estimated
+# ones. If you know the real camera, this removes a whole source of error
+# that no amount of downstream robust-statistics fixes.
+# --------------------------------------------------------------------------
+
+def intrinsics_from_exif(image: Image.Image):
+    """Approximates fx, fy in pixels from the 35mm-equivalent focal length
+    tag, assuming a 36mm-wide full-frame-equivalent sensor. Returns a 3x3
+    intrinsics matrix, or None if the tag isn't present. This is an
+    approximation, not a substitute for real calibration, but it's
+    typically closer than a single-image network estimate."""
+    try:
+        exif = image.getexif()
+        tag_map = {ExifTags.TAGS.get(k, k): v for k, v in exif.items()}
+        focal_35mm = tag_map.get("FocalLengthIn35mmFilm")
+        if not focal_35mm:
+            return None
+        w, h = image.size
+        fx = (float(focal_35mm) / 36.0) * w
+        fy = fx  # assume square pixels
+        cx, cy = w / 2.0, h / 2.0
+        return np.array([[fx, 0, cx], [0, fy, cy], [0, 0, 1]], dtype=np.float64)
+    except Exception:
+        return None
+
+
+# --------------------------------------------------------------------------
+# Addition 3: automatic scale calibration via an ArUco marker of known
+# physical size, instead of requiring the user to type in a measured
+# reference length by hand.
+#
+# Print a DICT_4X4_50 marker at a known side length (e.g. 5cm) and place it
+# flat in the scene. If found, this replaces the manual scale-calibration
+# inputs entirely.
+# --------------------------------------------------------------------------
+
+def detect_aruco_scale(image: Image.Image, depth: np.ndarray, intrinsics: np.ndarray,
+                        marker_real_size_m: float):
+    if not marker_real_size_m or marker_real_size_m <= 0:
+        return None, None
+
+    gray = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2GRAY)
+    detector = cv2.aruco.ArucoDetector(ARUCO_DICT, cv2.aruco.DetectorParameters())
+    corners, ids, _ = detector.detectMarkers(gray)
+    if ids is None or len(corners) == 0:
+        return None, None
+
+    quad = corners[0][0]  # 4x2 pixel corners of the first detected marker
+    h, w = depth.shape
+    points_3d = []
+    for (px, py) in quad:
+        xi, yi = int(np.clip(px, 0, w - 1)), int(np.clip(py, 0, h - 1))
+        z = float(depth[yi, xi])
+        fx, fy = intrinsics[0, 0], intrinsics[1, 1]
+        cx, cy = intrinsics[0, 2], intrinsics[1, 2]
+        x = (px - cx) * z / fx
+        y = (py - cy) * z / fy
+        points_3d.append(np.array([x, y, z]))
+
+    side_lengths = [np.linalg.norm(points_3d[i] - points_3d[(i + 1) % 4]) for i in range(4)]
+    measured_size = float(np.median(side_lengths))
+    if measured_size <= 0:
+        return None, None
+
+    scale_correction = marker_real_size_m / measured_size
+    return scale_correction, measured_size
+
+
+# --------------------------------------------------------------------------
+# Main entry point called by the Gradio UI
+# --------------------------------------------------------------------------
+
+def parse_object_list(objects_text: str) -> list:
+    names = [n.strip() for n in objects_text.split(",")]
+    names = [n for n in names if n]
+    return names
+
+
+def run_pipeline(files, objects_text, depth_model_id, erosion_px,
+                  use_exif_intrinsics, aruco_marker_size,
+                  true_reference_length, measured_reference_length):
+    if not files:
+        raise gr.Error("Please upload at least one image (the primary view).")
+
+    object_names = parse_object_list(objects_text or "")
+    if len(object_names) < 2:
+        raise gr.Error("Please list at least two objects, comma-separated, "
+                        "e.g. 'the red chair, the wooden table, the floor lamp'.")
+    if len(object_names) > len(OBJECT_COLORS):
+        gr.Warning(f"{len(object_names)} objects requested; overlay colors will repeat "
+                   f"after the first {len(OBJECT_COLORS)}.")
+
+    image_paths = [f.name if hasattr(f, "name") else f for f in files]
+    if len(image_paths) > 5:
+        gr.Warning(f"{len(image_paths)} images provided; accuracy gains from extra views "
+                   f"typically plateau well before this many.")
+
+    images = [Image.open(p).convert("RGB") for p in image_paths]
+
+    # Addition 1: quality gate — warn (don't silently fail) on blurry views,
+    # since a bad extra view can drag down the joint depth solve for all views.
+    for i, img in enumerate(images):
+        score, is_blurry = check_blur(img)
+        if is_blurry:
+            label = "primary image" if i == 0 else f"extra view {i}"
+            gr.Warning(f"{label} looks blurry (sharpness score {score:.0f}). "
+                       f"This can reduce accuracy — consider retaking it.")
+
+    primary_image = images[0]
+
+    # Segment each object independently. A bad prompt for one object
+    # shouldn't discard valid results for the others, so failures are
+    # collected and reported rather than raised immediately.
+    masks, valid_names, seg_warnings = [], [], []
+    for name in object_names:
+        try:
+            mask = erode_mask(segment(primary_image, name), erosion_px)
+            masks.append(mask)
+            valid_names.append(name)
+        except gr.Error as e:
+            seg_warnings.append(f"'{name}': {e}")
+
+    for w in seg_warnings:
+        gr.Warning(f"Skipped {w}")
+
+    if len(valid_names) < 2:
+        raise gr.Error("Fewer than two objects could be segmented — see warnings above for details.")
+
+    depther = get_depther(depth_model_id)
+    prediction = depther.inference(images)
+    depth, conf = prediction.depth[0], prediction.conf[0]
+
+    # Addition 2: prefer EXIF-derived intrinsics over DA3's estimated ones
+    # when available and requested — a known camera beats a network guess.
+    intrinsics = prediction.intrinsics[0]
+    intrinsics_source = "DA3 (estimated)"
+    if use_exif_intrinsics:
+        exif_intrinsics = intrinsics_from_exif(primary_image)
+        if exif_intrinsics is not None:
+            intrinsics = exif_intrinsics
+            intrinsics_source = "EXIF (35mm-equivalent focal length)"
+        else:
+            gr.Warning("No usable focal-length EXIF tag found on the primary image — "
+                       "falling back to DA3's estimated intrinsics.")
+
+    # Addition 3: automatic scale calibration via ArUco marker, falling back
+    # to manual true/measured length entry if no marker is found.
+    scale_correction = 1.0
+    scale_source = "none (raw metric depth)"
+    aruco_scale, aruco_measured = detect_aruco_scale(primary_image, depth, intrinsics, aruco_marker_size)
+    if aruco_scale is not None:
+        scale_correction = aruco_scale
+        scale_source = f"ArUco marker (measured {aruco_measured:.4f} m, expected {aruco_marker_size:.4f} m)"
+    elif true_reference_length and measured_reference_length and measured_reference_length > 0:
+        scale_correction = float(true_reference_length) / float(measured_reference_length)
+        scale_source = "manual reference length"
+    elif aruco_marker_size:
+        gr.Warning("ArUco marker size was set but no marker was detected in the primary image — "
+                   "check it's a DICT_4X4_50 marker, flat, and clearly visible.")
+
+    # Per-object depth, uncertainty, and 3D point — independent of object count.
+    points, stds, depths, pixel_counts = [], [], [], []
+    for mask in masks:
+        d, std, n = robust_object_depth(depth, conf, mask)
+        p = backproject_to_3d(mask, d, intrinsics) * scale_correction
+        points.append(p)
+        stds.append(std * scale_correction)
+        depths.append(d * scale_correction)
+        pixel_counts.append(n)
+
+    n_obj = len(valid_names)
+    dist_matrix = np.zeros((n_obj, n_obj))
+    unc_matrix = np.zeros((n_obj, n_obj))
+    for i in range(n_obj):
+        for j in range(n_obj):
+            if i == j:
+                continue
+            dist_matrix[i, j] = np.linalg.norm(points[i] - points[j])
+            unc_matrix[i, j] = np.sqrt(stds[i]**2 + stds[j]**2)
+
+    overlay_img = mask_overlay(primary_image, masks)
+
+    # Per-object table
+    per_object_rows = "\n".join(
+        f"| {name} | {depths[i]:.3f} m | {pixel_counts[i]} |"
+        for i, name in enumerate(valid_names)
+    )
+
+    # Pairwise distance matrix table (upper triangle to avoid repeating each pair twice)
+    header = "| |" + "".join(f" {n} |" for n in valid_names)
+    sep = "|---|" + "---|" * n_obj
+    rows = []
+    for i in range(n_obj):
+        cells = []
+        for j in range(n_obj):
+            if j <= i:
+                cells.append(" — |")
+            else:
+                cells.append(f" {dist_matrix[i, j]:.3f} ± {unc_matrix[i, j]:.3f} m |")
+        rows.append(f"| **{valid_names[i]}** |" + "".join(cells))
+    matrix_table = "\n".join([header, sep] + rows)
+
+    summary = (
+        f"### Per-object depth\n"
+        f"| Object | Depth | Pixels used |\n"
+        f"|---|---|---|\n"
+        f"{per_object_rows}\n\n"
+        f"### Pairwise distances\n"
+        f"{matrix_table}\n\n"
+        f"Views used: {len(images)} "
+        f"({'multi-view' if len(images) > 1 else 'single-view — add more views for better accuracy'})\n\n"
+        f"Camera intrinsics: {intrinsics_source}\n\n"
+        f"Scale calibration: {scale_source}"
+        + (f" (×{scale_correction:.4f})" if scale_correction != 1.0 else "")
+    )
+
+    return overlay_img, summary
+
+
+# --------------------------------------------------------------------------
+# UI
+# --------------------------------------------------------------------------
+
+with gr.Blocks(title="Object Distance Estimator — SAM3 + Depth Anything 3") as demo:
+    gr.Markdown(
+        "# Object Distance Estimator\n"
+        "Segment two or more objects with text prompts (SAM3), estimate metric depth "
+        "(Depth Anything 3), and compute the real-world pairwise distances between them.\n\n"
+        "**Tip:** upload the primary photo plus 1–4 extra photos of the *same scene* "
+        "from different angles for meaningfully better accuracy."
+    )
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            files = gr.File(
+                label="Images (first = primary view, rest = optional extra views)",
+                file_count="multiple",
+                file_types=["image"],
+            )
+            objects_text = gr.Textbox(
+                label="Objects (comma-separated, 2 or more)",
+                placeholder="e.g. the red chair, the wooden table, the floor lamp",
+            )
+
+            with gr.Accordion("Advanced settings", open=False):
+                depth_model_id = gr.Dropdown(
+                    label="Depth model (must be a metric checkpoint for real distances)",
+                    choices=[
+                        "depth-anything/da3metric-large",
+                        "depth-anything/da3nested-giant-large",
+                    ],
+                    value="depth-anything/da3metric-large",
+                )
+                erosion_px = gr.Slider(label="Mask erosion (pixels)", minimum=0, maximum=10, value=3, step=1)
+
+                gr.Markdown("**Camera intrinsics**")
+                use_exif_intrinsics = gr.Checkbox(
+                    label="Prefer EXIF focal length over DA3's estimated intrinsics (if available)",
+                    value=True,
+                )
+
+                gr.Markdown(
+                    "**Scale calibration** — pick one: place a printed ArUco `DICT_4X4_50` "
+                    "marker of known size in the scene (automatic), or enter a known length manually."
+                )
+                aruco_marker_size = gr.Number(label="ArUco marker side length (m)", value=None)
+                true_reference_length = gr.Number(label="Manual: true length (m)", value=None)
+                measured_reference_length = gr.Number(label="Manual: measured length from this pipeline (m)", value=None)
+
+            run_btn = gr.Button("Estimate distances", variant="primary")
+
+        with gr.Column(scale=1):
+            overlay_out = gr.Image(label="Mask overlay (each object gets a distinct color)")
+            result_out = gr.Markdown()
+
+    run_btn.click(
+        fn=run_pipeline,
+        inputs=[files, objects_text, depth_model_id, erosion_px,
+                use_exif_intrinsics, aruco_marker_size,
+                true_reference_length, measured_reference_length],
+        outputs=[overlay_out, result_out],
+    )
+
+if __name__ == "__main__":
+    demo.launch()