code/tt_meteor/tests/test_server_host.py

# SPDX-License-Identifier: Apache-2.0
"""Host tests of the HTTP contract (no device, no weights): the real app of this bundle with a stub model, eight
cameras + calibration (inline or the shipped preset) + ``ego_speed``.

    TT_VISIBLE_DEVICES=none python -m pytest -q code/tt_meteor/tests/test_server_host.py
"""
from __future__ import annotations

import base64
import io

import numpy as np
import pytest

fastapi = pytest.importorskip("fastapi")
from fastapi.testclient import TestClient  # noqa: E402

from tt_meteor.api import METEOR  # noqa: E402
from tt_meteor.server import app as server  # noqa: E402
from tt_meteor.tests.stubs import StubModel  # noqa: E402
from tt_meteor.ttaw.server.app import decode_request  # noqa: E402

CAMS = list(METEOR.CAMERA_ORDER)


@pytest.fixture
def client(monkeypatch):
    monkeypatch.setenv("TT_MESH_SHAPE", "1x1")
    monkeypatch.setenv("TT_MODEL_WEIGHTS_REVISION", "0" * 40)
    monkeypatch.setattr(server.app.state.ttaw, "model_factory", StubModel)
    with TestClient(server.app) as c:
        yield c


def _png_b64(seed: int, hw=(432, 768), zero=False) -> str:
    from PIL import Image

    rng = np.random.default_rng(seed)
    a = np.zeros(hw + (3,), np.uint8) if zero else rng.integers(0, 256, size=hw + (3,), dtype=np.uint8)
    buf = io.BytesIO()
    Image.fromarray(a).save(buf, format="PNG")
    return base64.b64encode(buf.getvalue()).decode()


def _images(cams=CAMS, absent=()):
    return [{"camera": c, "data": _png_b64(i, zero=c in absent)} for i, c in enumerate(cams)]


def _inline_calib(cams=CAMS):
    K = [[400.0, 0.0, 384.0], [0.0, 400.0, 216.0], [0.0, 0.0, 1.0]]
    return {"cameras": {c: {"intrinsics": K, "T_ref_from_camera": {"translation": [1.0, 0.1 * i, 1.5],
                                                                    "rotation_wxyz": [0.5, -0.5, 0.5, -0.5]}}
                        for i, c in enumerate(cams)}}


def test_health_info_models(client):
    assert client.get("/health").json()["status"] == "ok"
    assert client.get("/v1/health").json()["status"] == "ok"
    info = client.get("/info").json()
    assert info["model"] == METEOR.MODEL_NAME and info["weights"]["revision"] == "0" * 40
    assert info["device"]["dispatch"] == "eth" and info["device"]["grid"] == "12x10"
    assert info["autoware"]["package"] == "none (METEOR runtime: tier4/METEOR deploy/cpp)"
    assert info["output"]["labels"] == list(METEOR.LABELS)
    assert info["input"]["camera_order"] == CAMS
    assert info["input"]["calibration_presets"] == ["synthetic_8cam"]          # PandaSet preset staged, not shipped
    assert info["variant"] == METEOR.DEFAULT_VARIANT
    assert client.get("/v1/models").json()["data"][0]["id"] == "AutowareFoundation/meteor"


@pytest.mark.parametrize("calibration", ["inline", "preset"])
def test_predict_ok_and_equals_api(client, calibration):
    cal = _inline_calib() if calibration == "inline" else {"preset": "synthetic_8cam"}
    body = {"images": _images(absent=("CAM_BACK_NARROW",)), "calibration": cal, "ego_speed": 10.9,
            "params": {"vehicle_threshold": 0.5}, "stream": {"id": "rig-1", "reset": False}}
    r = client.post("/predict", json=body)
    assert r.status_code == 200, r.text
    out = r.json()
    assert out["num_detections"] == 2 and out["frame_id"] == "base_link" and out["model"] == METEOR.MODEL_NAME
    assert set(out["detections"][0]) >= {"label", "label_id", "score", "center", "size", "yaw", "stationary"}
    assert out["trajectory"][0] == pytest.approx([10.9 * 0.5, 0.0]) and out["columns"] == ["x", "y"]
    assert out["lane"]["format"] == "png" and set(out["detections_2d"]) == set(CAMS)
    assert {"decode", "model_call", "total", "device"} <= set(out["timing_ms"])
    state = server.app.state.ttaw
    req = server.MeteorRequest(**{k: v for k, v in body.items() if k != "params"})
    api = state.model(**decode_request(req, state.spec, 1 << 30)).to_dict()["detections"]
    assert state.predict(req)["detections"] == api


@pytest.mark.parametrize("payload,code", [
    ({"images": [{"camera": "CAM_FRONT_WIDE", "data": "not base64!"}], "ego_speed": 1.0}, 400),   # undecodable
    ({"images": _images(CAMS[:5]), "calibration": _inline_calib(), "ego_speed": 1.0}, 400),        # cameras missing
    ({"images": _images(), "calibration": _inline_calib(CAMS[:5]), "ego_speed": 1.0}, 400),        # calibration missing
    ({"images": _images(), "calibration": {"preset": "../etc"}, "ego_speed": 1.0}, 400),           # bad preset
    ({"images": _images(), "calibration": {"preset": "synthetic_8cam"}}, 400),                         # no ego_speed
    ({"images": _images(), "calibration": {"preset": "synthetic_8cam"}, "ego_speed": 1.0,
      "params": {"vehicle_threshold": 2.0}}, 400),
    ({"images": _images(), "calibration": {"preset": "synthetic_8cam"}, "ego_speed": 1.0,
      "params": {"no_such_knob": 1}}, 400),
    ({"images": _images(), "calibration": {"preset": "synthetic_8cam"}, "ego_speed": "fast"}, 422),   # not a number
    ({}, 400),                                                                                       # no images
    ({"imagez": []}, 422),                                                                           # schema violation
])
def test_predict_errors(client, payload, code):
    r = client.post("/predict", json=payload)
    assert r.status_code == code, r.text


def test_503_while_starting(client, monkeypatch):
    monkeypatch.setattr(server.app.state.ttaw, "ready", False)
    assert client.post("/predict", json={}).status_code == 503
    assert client.get("/health").json()["status"] == "starting"


def test_mesh_shape_parser():
    p = server.parse_mesh_shape
    assert p("1x1") == p("(1, 1)") == p("1,1") == p(None) == (1, 1)
    with pytest.raises(RuntimeError):
        p("two by two")