code/tt_moge/server/postprocess.py

# SPDX-License-Identifier: Apache-2.0
"""Host-side pre/post-processing for the MoGe-2 server.

Two jobs, both pure torch/numpy/PIL on the host:

* ``fit_to_canvas``: put ANY client image onto the single canonical canvas the
  decoder trace was captured for (default 1920x1080, the README's validated
  setting: ``num_tokens=1800`` -> 32x57 token grid). ``fit="pad"`` keeps the
  aspect ratio (letterbox, centred, neutral gray) so the geometry stays faithful;
  ``fit="stretch"`` squashes the image onto the canvas.
* ``canvas_to_camera`` + ``to_original``: the tail of the upstream
  ``MoGeModel.infer()`` (moge/model/v2.py) -- recover focal + z-shift from the
  affine point map, build normalized intrinsics, apply the metric scale -- then
  crop the letterbox region away, resize to the client's original resolution,
  re-project points from the resized depth (``force_projection``) and apply the
  validity mask.

Imports of the vendored ``moge`` / ``utils3d`` happen inside the functions:
importing ``tt_moge`` puts them on ``sys.path`` and their import chain pulls in
cv2/scipy, so nothing here runs at module import.
"""
from __future__ import annotations

import math
from typing import Dict, Optional, Tuple

import numpy as np
import torch
import torch.nn.functional as F
from PIL import Image

Region = Tuple[int, int, int, int]  # x0, y0, x1, y1 on the canvas (pixels)

PAD_GRAY = (128, 128, 128)


def token_grid(width: int, height: int, num_tokens: int) -> Tuple[int, int]:
    """(base_h, base_w) MoGe uses for an image of this size -- mirrors MoGeModel.forward."""
    aspect_ratio = width / height
    return round((num_tokens / aspect_ratio) ** 0.5), round((num_tokens * aspect_ratio) ** 0.5)


def fit_to_canvas(im: Image.Image, canvas: Tuple[int, int], fit: str,
                  contiguous: bool = True) -> Tuple[torch.Tensor, Region]:
    """RGB PIL image -> (float32 [1,3,Hc,Wc] in [0,1], region the image occupies).

    ``canvas`` is (width, height). ``contiguous=False`` returns the channels-last view (same values,
    no NCHW copy; the C++ input prep reads any strides).
    """
    cw, ch = canvas
    if fit == "stretch":
        pasted = im.resize((cw, ch), Image.BILINEAR)
        region: Region = (0, 0, cw, ch)
    elif fit == "pad":
        s = min(cw / im.width, ch / im.height)
        rw, rh = max(1, round(im.width * s)), max(1, round(im.height * s))
        x0, y0 = (cw - rw) // 2, (ch - rh) // 2
        pasted = Image.new("RGB", (cw, ch), PAD_GRAY)
        pasted.paste(im.resize((rw, rh), Image.BILINEAR), (x0, y0))
        region = (x0, y0, x0 + rw, y0 + rh)
    else:
        raise ValueError(f"unknown fit mode {fit!r}")
    # one pass: float32(u8) / 255 in float32 (== np.asarray(pasted, np.float32) / 255.0, bit for bit)
    arr = np.true_divide(np.asarray(pasted), np.float32(255.0), dtype=np.float32)
    x = torch.from_numpy(arr).permute(2, 0, 1)[None]
    return (x.contiguous() if contiguous else x), region


def region_mask(canvas: Tuple[int, int], region: Region) -> torch.Tensor:
    """bool [Hc, Wc]: True inside the pasted image, False on the letterbox bars."""
    cw, ch = canvas
    m = torch.zeros(ch, cw, dtype=torch.bool)
    x0, y0, x1, y1 = region
    m[y0:y1, x0:x1] = True
    return m


@torch.inference_mode()
def canvas_to_camera(
    out: Dict[str, torch.Tensor],
    valid: Optional[torch.Tensor],
    fov_x: Optional[float] = None,
) -> Dict[str, torch.Tensor]:
    """``MoGeModel.infer()``'s recovery step on one canvas-sized TtMoGe output.

    ``out`` is the TtMoGe / MoGeModel.forward dict (batch 1): ``points`` [1,H,W,3]
    affine, ``normal`` [1,H,W,3], ``mask`` [1,H,W] sigmoid, ``metric_scale`` [1].
    ``valid`` (bool [H,W]) restricts the focal/shift solve to real (non-letterbox)
    pixels. Returns canvas-resolution tensors, metric-scaled, UNMASKED:
    ``depth`` [H,W], ``points`` [H,W,3] (affine map shifted into camera space --
    the exact upstream output when force_projection=False), ``normal`` [H,W,3],
    ``mask`` bool [H,W], ``intrinsics`` [3,3] normalized (cx=cy=0.5), plus scalars
    ``focal``, ``shift``, ``metric_scale``.
    """
    from moge.utils.geometry_torch import recover_focal_shift
    import utils3d

    points = out["points"][0].float().clone()      # (H, W, 3)
    normal = out["normal"][0].float()
    mask_prob = out["mask"][0].float()
    metric_scale = out["metric_scale"].float().reshape(-1)[0]
    height, width = points.shape[:2]
    aspect_ratio = width / height

    mask_binary = mask_prob > 0.5
    if valid is not None:
        mask_binary &= valid

    p, m = points[None], mask_binary[None]
    if fov_x is None:
        focal, shift = recover_focal_shift(p, m)
    else:
        focal = torch.tensor(
            [aspect_ratio / (1 + aspect_ratio ** 2) ** 0.5 / math.tan(math.radians(fov_x) / 2)],
            dtype=p.dtype,
        )
        _, shift = recover_focal_shift(p, m, focal=focal)
    focal = focal.reshape(-1)[0]
    shift = shift.reshape(-1)[0]

    fx = focal / 2 * (1 + aspect_ratio ** 2) ** 0.5 / aspect_ratio
    fy = focal / 2 * (1 + aspect_ratio ** 2) ** 0.5
    half = torch.tensor(0.5, dtype=p.dtype)
    intrinsics = utils3d.pt.intrinsics_from_focal_center(fx[None], fy[None], half, half)[0]

    points[..., 2] += shift
    mask_binary &= points[..., 2] > 0          # negative depth should never happen; drop it if it does
    depth = points[..., 2].clone()

    points *= metric_scale
    depth *= metric_scale
    return {
        "depth": depth, "points": points, "normal": normal, "mask": mask_binary,
        "intrinsics": intrinsics, "focal": focal, "shift": shift, "metric_scale": metric_scale,
    }


def fov_x_to_canvas(fov_x_deg: float, canvas: Tuple[int, int], region: Region) -> float:
    """A client's horizontal FoV (of ITS image) -> the horizontal FoV of the padded canvas.

    The image occupies ``region`` (width rw) on a canvas of width cw with the same focal
    length in pixels, so tan(fov_c/2) = tan(fov/2) * cw / rw. Identity for ``fit="stretch"``.
    """
    cw = canvas[0]
    rw = region[2] - region[0]
    return math.degrees(2 * math.atan(math.tan(math.radians(fov_x_deg) / 2) * cw / rw))


def crop_intrinsics(intrinsics: torch.Tensor, canvas: Tuple[int, int], region: Region) -> torch.Tensor:
    """Normalized canvas intrinsics -> normalized intrinsics of the cropped region.

    Normalized intrinsics are invariant to a uniform resize, so the result also holds
    for the client's original image (the region has its aspect ratio).
    """
    cw, ch = canvas
    x0, y0, x1, y1 = region
    rw, rh = x1 - x0, y1 - y0
    k = intrinsics.clone()
    k[0, 0] = intrinsics[0, 0] * cw / rw
    k[1, 1] = intrinsics[1, 1] * ch / rh
    k[0, 2] = (intrinsics[0, 2] * cw - x0) / rw
    k[1, 2] = (intrinsics[1, 2] * ch - y0) / rh
    return k


def _resize_map(t: torch.Tensor, size: Tuple[int, int], mode: str) -> torch.Tensor:
    """[H,W] or [H,W,C] -> same rank at ``size`` (height, width)."""
    squeeze = t.dim() == 2
    x = t[None, None] if squeeze else t.permute(2, 0, 1)[None]
    kw = {} if mode == "nearest" else {"align_corners": False}
    y = F.interpolate(x, size, mode=mode, **kw)
    return y[0, 0] if squeeze else y[0].permute(1, 2, 0)


@torch.inference_mode()
def to_original(
    cam: Dict[str, torch.Tensor],
    canvas: Tuple[int, int],
    region: Region,
    size: Tuple[int, int],
    force_projection: bool = True,
    apply_mask: bool = True,
) -> Dict[str, torch.Tensor]:
    """Crop the letterbox away, resize to the original (height, width), finish infer().

    Returns ``depth`` [H,W] f32 (metres), ``points`` [H,W,3] f32 (camera space, metres),
    ``normal`` [H,W,3] f32 (unit), ``mask`` bool [H,W], ``intrinsics`` [3,3] normalized.
    With ``apply_mask`` invalid pixels get depth/points = +inf and normal = 0, exactly
    like the upstream ``infer()``.
    """
    import utils3d

    x0, y0, x1, y1 = region
    sl = (slice(y0, y1), slice(x0, x1))
    depth = _resize_map(cam["depth"][sl], size, "bilinear")
    normal = _resize_map(cam["normal"][sl], size, "bilinear")
    normal = F.normalize(normal, dim=-1)
    mask = _resize_map(cam["mask"][sl].float(), size, "nearest") > 0.5
    intrinsics = crop_intrinsics(cam["intrinsics"], canvas, region)

    if force_projection:
        points = utils3d.pt.depth_map_to_point_map(depth, intrinsics)
    else:
        points = _resize_map(cam["points"][sl], size, "bilinear")

    mask &= torch.isfinite(depth) & (depth > 0)
    if apply_mask:
        inf = torch.tensor(float("inf"), dtype=depth.dtype)
        depth = torch.where(mask, depth, inf)
        points = torch.where(mask[..., None], points, inf)
        normal = torch.where(mask[..., None], normal, torch.zeros_like(normal))

    return {"depth": depth, "points": points, "normal": normal, "mask": mask, "intrinsics": intrinsics}


def fov_deg(intrinsics: torch.Tensor) -> Tuple[float, float]:
    """Horizontal / vertical FoV in degrees from normalized intrinsics."""
    fx, fy = float(intrinsics[0, 0]), float(intrinsics[1, 1])
    return math.degrees(2 * math.atan(0.5 / fx)), math.degrees(2 * math.atan(0.5 / fy))