code/hamer/server/app.py · changh95/hamer-p150 at main

hamer-p150 / code /hamer /server /app.py
p150 ETH-dispatch compliance (2026-10-05): default ETH dispatch, 1 CQ, 12x10 in Python API and server; numbers re-measured
843986e verified about 1 month ago
Raw History Blame Contribute Delete
26.9 kB
# SPDX-License-Identifier: Apache-2.0
"""ASGI serving app for HaMeR on Tenstorrent Blackhole.
Served by tt-model-manager as ``kind: tt-dit-server``::
    runtime:
      app: hamer.server.app:app
uvicorn runs this module.  Weights are resolved, the chip is opened, the
model is uploaded and warmed (JIT + trace capture) inside the ASGI
**lifespan**, so uvicorn's ``Application startup complete`` -- the line
``tt-model serve`` waits for -- means the device is claimed and warm.
I/O: full frame + one hand bbox -> MANO parameters (16 rotation matrices,
10 betas, weak-perspective cam); when ``MANO_RIGHT.pkl`` is available also
the posed mesh (778 vertices), 21 joints and their 2-D projection in the
original image.  One hand per request; call once per hand.
Environment (read in the lifespan, never at import):
    HF_MODEL              weights repo id (default changh95/hamer-weights)
    TT_WEIGHTS_REVISION   commit sha of that repo (default: its main branch)
    HAMER_WEIGHTS_DIR     directory holding hamer.ckpt + mano_mean_params.npz
                          (overrides the Hub lookup; offline/host use)
    HAMER_MANO_PATH       MANO_RIGHT.pkl (file or directory). Default:
                          $TT_DIT_CACHE_DIR/MANO_RIGHT.pkl, i.e. on the host
                          ~/.cache/tt-model/hamer-p150/weights/MANO_RIGHT.pkl
    TT_MESH_SHAPE         '1x1' (also '(1, 1)' / '1,1'); anything else fails
    TT_DEVICE_ID          chip to open (default 0; HAMER_DEVICE honoured too)
    HAMER_TRACE           '1' capture+replay a ttnn trace (default), '0' eager
    HAMER_WARMUP_ITERS    warm-up forwards (default 3, minimum 2). Forwards
                          1..N-1 use one fixed input (eager JIT, then trace
                          capture + replay); forward N uses a DIFFERENT input
                          so the copy-into-trace-buffer path every real request
                          takes runs before READY.
    HAMER_WARMUP_MIN_PCC  gate for the warm-up outputs vs the CPU reference
                          (default 0.99; the harness measures 0.9988)
"""
from __future__ import annotations

import base64
import io
import os
import sys
import threading
import time
from contextlib import asynccontextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

import numpy as np
import torch
from fastapi import FastAPI, HTTPException
from PIL import Image
from pydantic import BaseModel, Field

from hamer.preprocess import (
    BBOX_CROP_SIZE, IMAGE_HEIGHT, IMAGE_WIDTH, center_square_bbox, hamer_bbox_size, normalize_raw, preprocess_pil,
)
from hamer.rendering import FOCAL_LENGTH, MODEL_IMAGE_SIZE, cam_crop_to_full, project_mesh

MODEL_NAME = "hamer-p150"
DEFAULT_WEIGHTS_REPO = "changh95/hamer-weights"
CKPT_FILE = "hamer.ckpt"
MEAN_FILE = "mano_mean_params.npz"
MANO_FILE = "MANO_RIGHT.pkl"
PORT_SOURCE = "https://github.com/changh95/tt-hamer"
PORT_COMMIT = "cc1a512f2919df804b4b60864d20ec258b2283d7"
UPSTREAM = "https://github.com/geopavlakos/hamer"
LICENSE_NOTE = (
    "HaMeR code: MIT (geopavlakos/hamer). Weights: released by the HaMeR authors, mirrored at "
    "changh95/hamer-weights. MANO hand model (MANO_RIGHT.pkl): Max Planck non-commercial research "
    "licence, registration required at https://mano.is.tue.mpg.de/ -- NOT redistributed; without it "
    "this server returns MANO parameters only. Non-commercial use only."
)
MAX_IMAGE_PIXELS = 50_000_000  # decoded-frame cap (~8k x 6k); the crop, not the frame, reaches the chip
MIN_CROP_PIXELS = 2.0  # square crop side (rescale_factor * max(w, h)) below this is not a hand: 400

STATE: Dict[str, Any] = {}
_LOCK = threading.Lock()  # TtHamer is not thread-safe: one device call at a time
# Audit integration 2026-10-04: vertices + joints_21 from ONE MANO shape blend / kinematic walk
# (MANO.forward_with_joints, bit-identical).  HAMER_MANO_ONEPASS=0 restores the two-walk path.
MANO_ONEPASS = os.environ.get("HAMER_MANO_ONEPASS", "1") != "0"


def _log(msg: str) -> None:
    """Boot/serve log line on stdout (uvicorn's stream; tt-model's boot checklist reads it)."""
    print(f"[hamer.server] {msg}", flush=True)


# --------------------------------------------------------------------------- #
# Config
# --------------------------------------------------------------------------- #
def parse_mesh_shape(value: Optional[str]) -> Tuple[int, int]:
    """Accept ``'1x1'`` (what tt-model exports), ``'(1, 1)'`` and ``'1,1'``."""
    if value is None or not value.strip():
        return (1, 1)
    s = value.strip().lower().replace("(", "").replace(")", "").replace(" ", "")
    for sep in ("x", ","):
        if sep in s:
            parts = s.split(sep)
            if len(parts) == 2 and all(p.isdigit() for p in parts):
                return (int(parts[0]), int(parts[1]))
    raise RuntimeError(
        f"TT_MESH_SHAPE={value!r} is not understood; expected '1x1', '(1, 1)' or '1,1'."
    )


def _flag(name: str, default: str) -> bool:
    return os.environ.get(name, default).strip().lower() not in ("0", "false", "no", "off", "")


@dataclass
class Config:
    weights_repo: str
    revision: Optional[str]
    weights_dir: Optional[str]
    mano_path: Optional[str]
    mano_path_explicit: bool
    device_id: int
    mesh_shape: Tuple[int, int]
    use_trace: bool
    warmup_iters: int
    warmup_min_pcc: float


def load_config() -> Config:
    mano_env = os.environ.get("HAMER_MANO_PATH")
    if mano_env:
        mano_path, explicit = mano_env, True
    else:
        cache = os.environ.get("TT_DIT_CACHE_DIR")
        base = Path(cache) if cache else Path.home() / ".cache" / "tt-model" / MODEL_NAME / "weights"
        mano_path, explicit = str(base / MANO_FILE), False
    return Config(
        weights_repo=os.environ.get("HF_MODEL", DEFAULT_WEIGHTS_REPO),
        revision=os.environ.get("TT_WEIGHTS_REVISION") or None,
        weights_dir=os.environ.get("HAMER_WEIGHTS_DIR") or None,
        mano_path=mano_path,
        mano_path_explicit=explicit,
        device_id=int(os.environ.get("TT_DEVICE_ID", os.environ.get("HAMER_DEVICE", "0"))),
        mesh_shape=parse_mesh_shape(os.environ.get("TT_MESH_SHAPE")),
        use_trace=_flag("HAMER_TRACE", "1"),
        warmup_iters=max(2, int(os.environ.get("HAMER_WARMUP_ITERS", "3"))),
        warmup_min_pcc=float(os.environ.get("HAMER_WARMUP_MIN_PCC", "0.99")),
    )


# --------------------------------------------------------------------------- #
# Weights
# --------------------------------------------------------------------------- #
def resolve_weights(cfg: Config) -> Tuple[Path, Path, Dict[str, Any]]:
    """Locate hamer.ckpt + mano_mean_params.npz: a local dir, else the Hub
    cache (a hit when tt-model pre-downloaded the pinned snapshot)."""
    if cfg.weights_dir:
        d = Path(cfg.weights_dir).expanduser()
        ck, mean = d / CKPT_FILE, d / MEAN_FILE
        for f in (ck, mean):
            if not f.is_file():
                raise RuntimeError(f"HAMER_WEIGHTS_DIR={d} does not contain {f.name}")
        return ck, mean, {"source": "local_dir", "dir": str(d), "repo": None, "revision": None}
    from huggingface_hub import hf_hub_download  # noqa: WPS433

    kw = {"repo_id": cfg.weights_repo, "revision": cfg.revision}
    ck = Path(hf_hub_download(filename=CKPT_FILE, **kw))
    mean = Path(hf_hub_download(filename=MEAN_FILE, **kw))
    return ck, mean, {"source": "huggingface_hub", "dir": None, "repo": cfg.weights_repo, "revision": cfg.revision}


def resolve_mano_path(cfg: Config) -> Optional[Path]:
    p = Path(cfg.mano_path).expanduser()
    if p.is_dir():
        p = p / MANO_FILE
    if p.is_file():
        return p
    if cfg.mano_path_explicit:
        raise RuntimeError(f"HAMER_MANO_PATH={cfg.mano_path} does not exist")
    return None


def _pcc(a: torch.Tensor, b: torch.Tensor) -> float:
    a = a.reshape(-1).double() - a.double().mean()
    b = b.reshape(-1).double() - b.double().mean()
    return float((a @ b) / (a.norm() * b.norm()).clamp_min(1e-12))


# --------------------------------------------------------------------------- #
# Lifespan: weights -> device -> upload -> warm-up (JIT + trace) -> READY
# --------------------------------------------------------------------------- #
@asynccontextmanager
async def lifespan(_app: FastAPI):
    torch.set_grad_enabled(False)
    boot0 = time.perf_counter()
    cfg = load_config()
    if cfg.mesh_shape != (1, 1):
        raise RuntimeError(
            f"TT_MESH_SHAPE={cfg.mesh_shape[0]}x{cfg.mesh_shape[1]}: this port drives exactly one "
            "Blackhole chip (1x1 mesh)."
        )

    _log(f"Loading weights ({CKPT_FILE}, {MEAN_FILE}) from "
         f"{cfg.weights_dir or cfg.weights_repo + '@' + (cfg.revision or 'main')} ...")
    ck, mean, weights_info = resolve_weights(cfg)
    from hamer.checkpoint import load_hamer_checkpoint  # noqa: WPS433
    from hamer.reference import Hamer, sample_input  # noqa: WPS433

    t = time.perf_counter()
    ref = Hamer()
    ref.eval()
    report = load_hamer_checkpoint(ref, ck, mean)
    if report["loaded"] == 0 or report["missing_in_ref"] != 0 or report["unexpected_in_ref"] != 0:
        raise RuntimeError(f"Checkpoint {ck} does not map cleanly onto the reference: {report}")
    if not bool(ref.head.init_cam.abs().sum() > 0):
        raise RuntimeError(f"{mean} did not populate the MANO mean parameters")
    _log(f"Loading checkpoint done: {report['loaded']} tensors mapped in {time.perf_counter() - t:.1f}s")

    mano = None
    mano_path = resolve_mano_path(cfg)
    if mano_path is not None:
        from hamer.mano import MANO  # noqa: WPS433

        mano = MANO(mano_path)
        mano.eval()
        _log(f"MANO model loaded from {mano_path} (vertices/joints/keypoints_2d enabled)")
    else:
        _log(f"MANO_RIGHT.pkl not found at {cfg.mano_path}; serving MANO parameters only "
             "(set HAMER_MANO_PATH or copy the file into ~/.cache/tt-model/hamer-p150/weights/)")

    _log(f"Opening tt-nn device {cfg.device_id} (1x1 mesh, trace={'on' if cfg.use_trace else 'off'}) ...")
    from hamer.model import build_from_reference  # noqa: WPS433

    t = time.perf_counter()
    tt_model = build_from_reference(ref, device_id=cfg.device_id, strict=True, use_trace=cfg.use_trace)
    try:
        if tt_model.device is None or tt_model._vit_params is None or tt_model._head_params is None:
            raise RuntimeError("tt-nn device/parameters not available after build_from_reference")
        _log(f"Device dispatch={tt_model.dispatch} command_queues={tt_model.num_cqs} compute_grid={tt_model.grid}")
        _log(f"Device open + parameter upload took {time.perf_counter() - t:.1f}s "
             f"(~{tt_model.peak_dram_bytes / 2**30:.2f} GiB bf16 on DRAM)")

        # Warm-up = the whole request path, gated against the CPU reference so a
        # broken device path fails the boot instead of serving garbage:
        #   forward 1      x_a  eager (JIT of every kernel)
        #   forward 2..N-1 x_a  trace capture on the 2nd, then replay (cache hit,
        #                       no upload)
        #   forward N      x_b  a DIFFERENT input -> ttnn.copy_host_to_device_tensor
        #                       into the trace's token buffer + replay; this is the
        #                       path every real request with a new image takes.
        n = cfg.warmup_iters
        _log(f"Warming up ({n} forwards on fixed (1,3,{IMAGE_HEIGHT},{IMAGE_WIDTH}) inputs; "
             f"the last one is a fresh input through the copy-into-trace-buffer path) ...")
        raw_in = getattr(getattr(tt_model, "fused_opts", None), "input", "norm") == "u8"
        if raw_in:
            # TT_FUSED_INPUT=u8: the device graph takes uint8 crops (what predict() hands it); the CPU reference
            # gets the bit-identical normalized tensor
            g = torch.Generator().manual_seed(1)
            u_a = torch.randint(0, 256, (1, 3, IMAGE_HEIGHT, IMAGE_WIDTH), generator=g, dtype=torch.uint8)
            u_b = torch.randint(0, 256, (1, 3, IMAGE_HEIGHT, IMAGE_WIDTH), generator=g, dtype=torch.uint8)
            x_a, x_b = normalize_raw(u_a), normalize_raw(u_b)
            dev_in = {id(x_a): u_a, id(x_b): u_b}
        else:
            x_a = sample_input(batch=1, seed=0)
            x_b = sample_input(batch=1, seed=1)
            dev_in = {}
        inputs = [x_a] * (n - 1) + [x_b]
        outs: List[torch.Tensor] = []
        with torch.inference_mode():
            for i, x in enumerate(inputs):
                t = time.perf_counter()
                out = tt_model(dev_in.get(id(x), x))
                dt = (time.perf_counter() - t) * 1000.0
                if tuple(out.shape) != (1, 157) or not torch.isfinite(out).all():
                    raise RuntimeError(f"warm-up forward {i + 1} produced an invalid output {tuple(out.shape)}")
                outs.append(out.detach().clone())
                _log(f"warm-up forward {i + 1}/{n} ({'fresh' if x is x_b else 'fixed'} input): {dt:.0f} ms, "
                     f"trace_active={tt_model.trace_active}")
            if cfg.use_trace and not tt_model.trace_active:
                raise RuntimeError("ttnn trace was not captured during warm-up")
            t = time.perf_counter()
            ref_a = ref(x_a)
            ref_b = ref(x_b)
            _log(f"CPU reference forwards for the warm-up gate: {(time.perf_counter() - t) * 1000.0:.0f} ms")
        pcc_eager = _pcc(outs[0], ref_a)                    # eager device path vs CPU
        pcc_replay = _pcc(outs[-2], outs[0]) if n >= 3 else None  # replay vs eager, same input
        pcc_fresh = _pcc(outs[-1], ref_b)                   # copy path + replay vs CPU, new input
        warmup = {"eager_vs_ref_pcc": round(pcc_eager, 5),
                  "replay_vs_eager_pcc": None if pcc_replay is None else round(pcc_replay, 5),
                  "fresh_input_vs_ref_pcc": round(pcc_fresh, 5), "min_pcc": cfg.warmup_min_pcc}
        _log(f"warm-up PCC: eager-vs-ref {pcc_eager:.5f}, replay-vs-eager "
             f"{'n/a' if pcc_replay is None else f'{pcc_replay:.5f}'}, fresh-input-vs-ref {pcc_fresh:.5f} "
             f"(gate {cfg.warmup_min_pcc})")
        if pcc_eager < cfg.warmup_min_pcc:
            raise RuntimeError(f"eager tt-nn output disagrees with the CPU reference (PCC {pcc_eager:.4f})")
        if pcc_replay is not None and pcc_replay < cfg.warmup_min_pcc:
            raise RuntimeError(f"trace replay disagrees with the eager forward on the same input (PCC {pcc_replay:.4f})")
        if pcc_fresh < cfg.warmup_min_pcc:
            raise RuntimeError(
                f"a fresh input copied into the trace buffer disagrees with the CPU reference (PCC {pcc_fresh:.4f}); "
                "the copy_host_to_device_tensor -> execute_trace path is broken (HAMER_TRACE=0 runs eager)"
            )
        from hamer import native  # noqa: WPS433

        native.load()
        _log(f"native preprocess (crop + resize): {native.STATUS}")
        _log(f"Warmup complete in {time.perf_counter() - boot0:.1f}s total")
    except Exception:
        tt_model.close()
        raise

    STATE.update(
        model=tt_model,
        mano=mano,
        mano_path=str(mano_path) if mano_path else None,
        cfg=cfg,
        weights={**weights_info, "ckpt": str(ck), "mean_params": str(mean), "report": report},
        warmup=warmup,
        boot_seconds=round(time.perf_counter() - boot0, 1),
        ready=True,
    )
    try:
        yield
    finally:
        STATE["ready"] = False
        model = STATE.pop("model", None)
        STATE.pop("mano", None)
        if model is not None:
            with _LOCK:
                model.close()
        _log("tt-nn device closed")


app = FastAPI(title="HaMeR on Blackhole", lifespan=lifespan)


# --------------------------------------------------------------------------- #
# Schemas
# --------------------------------------------------------------------------- #
class PredictRequest(BaseModel):
    """One hand in one frame.  ``image`` is the FULL frame, base64 PNG/JPEG."""

    image: str = Field(..., description="base64-encoded PNG or JPEG of the full frame")
    bbox: Optional[List[float]] = Field(
        None, min_length=4, max_length=4,
        description="hand box [x1, y1, x2, y2] in original pixel coords; omitted -> centre square "
                    "(geometrically meaningless camera)",
    )
    is_right: bool = Field(True, description="False for a left hand (crop is mirrored; outputs mirrored back)")
    rescale_factor: float = Field(2.0, gt=0.5, le=5.0, description="square crop side = factor * max(bbox w, h)")
    return_vertices: bool = Field(True, description="include the 778x3 mesh when MANO is available")
    return_faces: bool = Field(False, description="include the static 1538x3 MANO face indices")


def _decode_image(b64: str) -> Image.Image:
    try:
        raw = base64.b64decode(b64, validate=True)
        im = Image.open(io.BytesIO(raw))
        im.load()
    except Exception as e:  # noqa: BLE001
        raise HTTPException(status_code=400, detail=f"bad image: {e}") from None
    if im.width * im.height > MAX_IMAGE_PIXELS:
        raise HTTPException(status_code=400, detail=f"image too large ({im.width}x{im.height})")
    return im if im.mode == "RGB" else im.convert("RGB")  # round 13: no copy of an RGB image (same pixels)


def _validate_bbox(
    bbox: Optional[List[float]], img: Image.Image, rescale_factor: float = 2.0,
) -> Tuple[float, float, float, float]:
    if bbox is None:
        return center_square_bbox(img)
    try:
        x1, y1, x2, y2 = (float(v) for v in bbox)
    except (TypeError, ValueError):
        raise HTTPException(status_code=400, detail="bbox must be four numbers") from None
    if not all(np.isfinite([x1, y1, x2, y2])) or x2 <= x1 or y2 <= y1:
        raise HTTPException(status_code=400, detail=f"bbox {bbox} is not a valid [x1, y1, x2, y2] with x2 > x1, y2 > y1")
    if x2 < 0 or y2 < 0 or x1 > img.width or y1 > img.height:
        raise HTTPException(status_code=400, detail=f"bbox {bbox} lies entirely outside the {img.width}x{img.height} image")
    crop_side = hamer_bbox_size((x1, y1, x2, y2), rescale_factor)
    if crop_side < MIN_CROP_PIXELS:
        # A sub-2-pixel crop is not a hand; without this it would reach PIL as a
        # 0x0 / 1x1 image and surface as a 500 instead of a client error.
        raise HTTPException(
            status_code=400,
            detail=f"bbox {bbox} is degenerate: crop side {crop_side:.3f} px "
                   f"(rescale_factor {rescale_factor} * max(w, h)) is below {MIN_CROP_PIXELS:g} px",
        )
    return (x1, y1, x2, y2)


def _round(arr: Any, nd: int = 6) -> Any:
    return np.round(np.asarray(arr, dtype=np.float64), nd).tolist()


# --------------------------------------------------------------------------- #
# Routes
# --------------------------------------------------------------------------- #
@app.get("/health")
def health() -> dict:
    ready = bool(STATE.get("ready"))
    return {
        "status": "ok" if ready else "starting",
        "model": MODEL_NAME,
        "device": "Tenstorrent Blackhole p150 (1x1 mesh)" if ready else None,
    }


@app.get("/info")
def info() -> dict:
    cfg: Optional[Config] = STATE.get("cfg")
    model = STATE.get("model")
    weights = STATE.get("weights") or {}
    return {
        "model": "HaMeR (Hand Mesh Recovery, Pavlakos et al. ICCV 2023) -- ViT-H/16 + MANO regression head",
        "name": MODEL_NAME,
        "task": "single-hand 3D hand pose/shape regression from a hand crop",
        "io": "full frame + hand bbox (+is_right) -> 16 rotation matrices, 10 betas, weak-perspective cam; "
              "with MANO_RIGHT.pkl also 778x3 vertices, 21x3 joints, 21x2 keypoints in original pixels",
        "hardware": "Tenstorrent Blackhole p150a, single chip, tt-nn (mesh 1x1, ttnn trace replay)",
        "mesh_shape": f"{cfg.mesh_shape[0]}x{cfg.mesh_shape[1]}" if cfg else None,
        "device_id": cfg.device_id if cfg else None,
        "dispatch": getattr(model, "dispatch", None),
        "num_command_queues": getattr(model, "num_cqs", None),
        "grid": getattr(model, "grid", None),
        "status": "ok" if STATE.get("ready") else "starting",
        "weights": {
            "repo": weights.get("repo") or (cfg.weights_repo if cfg else DEFAULT_WEIGHTS_REPO),
            "revision": weights.get("revision") or (cfg.revision if cfg else None),
            "source": weights.get("source"),
            "files": [CKPT_FILE, MEAN_FILE],
            "checkpoint_report": weights.get("report"),
        },
        "mano": {
            "available": STATE.get("mano") is not None,
            "path": STATE.get("mano_path"),
            "one_pass": MANO_ONEPASS,
            "how_to_enable": "place MANO_RIGHT.pkl at ~/.cache/tt-model/hamer-p150/weights/ on the host "
                             "(mounted as $TT_DIT_CACHE_DIR) or set HAMER_MANO_PATH",
        },
        "source": {"port": PORT_SOURCE, "port_commit": PORT_COMMIT, "upstream": UPSTREAM},
        "input": {
            "network_input": [1, 3, IMAGE_HEIGHT, IMAGE_WIDTH],
            "preprocess": f"square crop of side rescale_factor*max(w,h) around the bbox (grey padding) -> "
                          f"{BBOX_CROP_SIZE}x{BBOX_CROP_SIZE} bilinear -> mirror if left hand -> W slice "
                          f"[32:224] -> ImageNet normalisation",
            "hands_per_request": 1,
            "max_image_pixels": MAX_IMAGE_PIXELS,
            "default_rescale_factor": 2.0,
        },
        "output": {
            "rotmats": "[16,3,3] row-major; index 0 = global orientation, 1..15 = MANO hand pose",
            "betas": "[10] MANO shape coefficients",
            "cam": "[s, tx, ty] weak-perspective camera in crop space (as the network emits it)",
            "cam_t_full": "[tx, ty, tz] camera translation for the full frame (cam_crop_to_full, "
                          "focal_length = 5000/256*max(W,H)); x mirrored back for left hands",
            "vertices/joints": "metres, MANO mesh frame, x mirrored back for left hands (MANO only)",
            "keypoints_2d": "[21,2] pixels in the original frame (MANO only)",
        },
        "trace_active": bool(getattr(model, "trace_active", False)),
        "warmup": STATE.get("warmup"),
        "boot_seconds": STATE.get("boot_seconds"),
        "license": LICENSE_NOTE,
    }


@app.get("/v1/models")
def v1_models() -> dict:
    cfg: Optional[Config] = STATE.get("cfg")
    repo = cfg.weights_repo if cfg else os.environ.get("HF_MODEL", DEFAULT_WEIGHTS_REPO)
    return {"object": "list", "data": [{"id": repo, "object": "model", "owned_by": "changh95"}]}


@app.post("/predict")
def predict(req: PredictRequest) -> dict:
    if not STATE.get("ready"):
        raise HTTPException(status_code=503, detail="model is still starting")
    model = STATE["model"]
    mano = STATE.get("mano")

    t0 = time.perf_counter()
    img = _decode_image(req.image)
    bbox = _validate_bbox(req.bbox, img, req.rescale_factor)
    W, H = img.size
    raw_in = getattr(getattr(model, "fused_opts", None), "input", "norm") == "u8"  # TT_FUSED_INPUT=u8: uint8 crop
    x, (cx, cy, bbox_size) = preprocess_pil(img, bbox, rescale_factor=req.rescale_factor, is_right=req.is_right,
                                            raw=raw_in, contiguous=not raw_in)
    if tuple(x.shape) != (1, 3, IMAGE_HEIGHT, IMAGE_WIDTH):
        raise HTTPException(status_code=500, detail=f"preprocess produced {tuple(x.shape)}")
    t1 = time.perf_counter()

    try:
        xin = x if raw_in else x.contiguous()  # u8: the HWC view, centred by the model (round 13)
        with _LOCK, torch.inference_mode():
            out = model(xin)
    except Exception as e:  # noqa: BLE001
        raise HTTPException(status_code=500, detail=f"tt-nn forward failed: {e}") from None
    t2 = time.perf_counter()
    if tuple(out.shape) != (1, 157) or not torch.isfinite(out).all():
        raise HTTPException(status_code=500, detail="model produced a non-finite or malformed output")

    out = out[0].float().cpu()
    rotmats = out[:144].reshape(16, 3, 3)
    betas = out[144:154]
    cam = out[154:157]
    focal = FOCAL_LENGTH / MODEL_IMAGE_SIZE * max(W, H)
    cam_full_np = cam.numpy().astype(np.float64).copy()
    if not req.is_right:
        cam_full_np[1] *= -1.0  # mirror the camera x back for a left hand (upstream demo.py)
    cam_t = cam_crop_to_full(cam_full_np, np.array([cx, cy]), bbox_size, W, H, focal)

    resp: Dict[str, Any] = {
        "image_size": [W, H],
        "bbox": _round(bbox, 3),
        "bbox_source": "request" if req.bbox is not None else "center_square",
        "is_right": req.is_right,
        "rescale_factor": req.rescale_factor,
        "bbox_center": _round([cx, cy], 3),
        "bbox_size": round(float(bbox_size), 3),
        "focal_length": round(float(focal), 3),
        "rotmats": _round(rotmats.numpy()),
        "betas": _round(betas.numpy()),
        "cam": _round(cam.numpy()),
        "cam_t_full": _round(cam_t),
        "mano_available": mano is not None,
        "trace_active": bool(getattr(model, "trace_active", False)),
    }

    t3 = t2
    if mano is not None:
        with torch.inference_mode():
            if MANO_ONEPASS:
                verts, joints = (t[0] for t in mano.forward_with_joints(rotmats[None], betas[None]))
            else:
                verts = mano(rotmats[None], betas[None])[0]          # (778, 3)
                joints = mano.joints_21(rotmats[None], betas[None])[0]  # (21, 3)
        kp2d, _ = project_mesh(joints, cam, (cx, cy), bbox_size, W, H, is_right=req.is_right)
        verts_np = verts.numpy().astype(np.float64).copy()
        joints_np = joints.numpy().astype(np.float64).copy()
        if not req.is_right:
            verts_np[:, 0] *= -1.0
            joints_np[:, 0] *= -1.0
        resp["joints"] = _round(joints_np)
        resp["keypoints_2d"] = _round(kp2d, 3)
        if req.return_vertices:
            resp["vertices"] = _round(verts_np)
        if req.return_faces:
            resp["faces"] = np.asarray(mano.faces, dtype=np.int64).tolist()
        t3 = time.perf_counter()

    resp["timing_ms"] = {
        "preprocess": round((t1 - t0) * 1000.0, 2),
        "device": round((t2 - t1) * 1000.0, 2),
        "mano": round((t3 - t2) * 1000.0, 2),
        "total": round((time.perf_counter() - t0) * 1000.0, 2),
    }
    return resp


if __name__ == "__main__":  # host convenience: python -m hamer.server.app
    import uvicorn  # noqa: WPS433

    # the served HTTP stack (tt-model.yaml serve.args): uvloop + httptools, no per-request access log
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", "20000")), lifespan="on",
                loop="uvloop", http="httptools", access_log=False)
    sys.exit(0)