code/tt_meteor/server/client.py

#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
"""Build (and optionally send) a ``POST /predict`` request for meteor-p150. Standard library only.

    # the shipped sample: eight camera files + its calibration preset + ego speed + stream, from its manifest
    python3 code/tt_meteor/server/client.py --sample code/tt_meteor/samples/synthetic_8cam.json --out req.json
    curl -s localhost:20000/predict -H 'Content-Type: application/json' -d @req.json

    # your own frame: the eight cameras (an absent narrow camera as an all-zero image), a calibration JSON
    # (SERVING.md 3.1) or a preset listed by /info, and the ego speed in m/s; send it and print the response
    python3 code/tt_meteor/server/client.py --image CAM_FRONT_WIDE=fw.jpg ... --image CAM_BACK_NARROW=bn.jpg \
        --calib rig.json --ego-speed 8.3 --stream-id car1 --url http://127.0.0.1:20000

Everything but ``--sample`` and ``--ego-speed`` is the vendored ``ttaw/server/client.py`` (C08): ``--image NAME=PATH``
(repeat), ``--calib`` / ``--calib-preset``, ``--param k=v`` (repeat), ``--stream-id``, ``--reset``,
``--output-format json|npz``, ``--out``, ``--url``. Runs from the model repository with any Python 3.9+, without numpy
and without installing the package. In Python, use ``tt_meteor.ttaw.server.client`` (``build_request``, ``post``,
``wait_ready``, ...) and ``tt_meteor.server.smoke_test.build_sample_request``.
"""
import argparse
import json
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
sys.path.insert(0, str(HERE.parent / "ttaw" / "server"))      # first: its client.py, not this file
import client as ttaw_client  # noqa: E402  (the vendored stdlib-only client)


def _param(spec: str):
    k, _, v = spec.partition("=")
    try:
        return k, json.loads(v)
    except ValueError:
        return k, v


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("--sample", help="a sample manifest (samples/<name>.json): images, calibration, ego speed, stream")
    ap.add_argument("--image", action="append", default=[], help="NAME=PATH, repeat per camera")
    ap.add_argument("--calib", help="calibration JSON file")
    ap.add_argument("--calib-preset", help="a preset name listed by /info")
    ap.add_argument("--ego-speed", type=float, help="ego speed in m/s (required unless --sample)")
    ap.add_argument("--param", action="append", default=[], help="k=v (JSON value), repeat")
    ap.add_argument("--stream-id")
    ap.add_argument("--reset", action="store_true")
    ap.add_argument("--output-format", default="json", choices=["json", "npz"])
    ap.add_argument("--out", help="write the request JSON here")
    ap.add_argument("--url", help="POST it to this server and print the response")
    a = ap.parse_args(argv)
    if a.sample:
        from smoke_test import build_sample_request  # noqa: E402  (stdlib only)

        req = build_sample_request(Path(a.sample))
        if a.param:
            req["params"] = dict(_param(p) for p in a.param)
        if a.ego_speed is not None:
            req["ego_speed"] = a.ego_speed
        if a.output_format != "json":
            req["output_format"] = a.output_format
    else:
        if a.ego_speed is None:
            ap.error("--ego-speed is required (m/s), unless --sample gives it")
        req = ttaw_client.build_request(images=a.image, calib=a.calib, calib_preset=a.calib_preset,
                                        params=dict(_param(p) for p in a.param), stream_id=a.stream_id,
                                        reset=a.reset, output_format=a.output_format)
        req["ego_speed"] = a.ego_speed
    if a.out:
        Path(a.out).write_text(json.dumps(req))
    if a.url:
        code, body = ttaw_client.post(a.url, req)
        print(json.dumps(body, indent=1)[:20000])
        return 0 if code == 200 else 1
    if not a.out:
        json.dump(req, sys.stdout)
    return 0


if __name__ == "__main__":
    sys.exit(main())