File size: 8,444 Bytes
4d9b003
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
"""C08 client: build / send ``POST /predict`` requests and run smoke checks. Standard library only.

This module imports nothing from ttaw (no relative imports), so it works three ways: as ``<pkg>.ttaw.server.client``,
as a script, and loaded by path from a bundle's ``smoke_test.py`` running under the host's system ``python3``
(``container_smoke.sh``)::

    HERE = Path(__file__).resolve().parent                    # code/<pkg>/server
    sys.path.insert(0, str(HERE.parent / "ttaw" / "server"))
    import client as ttaw_client

CLI::

    python client.py --points sample.npz --out req.json                    # write the request
    python client.py --points sample.npz --url http://127.0.0.1:20000      # send it, print the response
    python client.py --image CAM_FRONT=f.jpg --calib-preset rig --url ...  # cameras
    python client.py --inputs scene.npz --url ...                          # planner
"""
from __future__ import annotations

import argparse
import base64
import json
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

__all__ = ["POINT_SUFFIXES", "b64file", "build_request", "post", "get_json", "wait_ready", "check_service",
           "check_bad_request", "smoke_line", "main"]

POINT_SUFFIXES = {".bin": "bin", ".npy": "npy", ".npz": "npz", ".pcd": "pcd"}


def b64file(path: Any) -> str:
    return base64.b64encode(Path(path).read_bytes()).decode("ascii")


def build_request(points: Optional[str] = None, fields: Optional[Sequence[str]] = None, images: Iterable[str] = (),
                  calib: Optional[str] = None, calib_preset: Optional[str] = None, inputs: Optional[str] = None,
                  params: Optional[Dict[str, Any]] = None, stream_id: Optional[str] = None, reset: bool = False,
                  output_format: str = "json") -> Dict[str, Any]:
    """A ``/predict`` body from files: ``points`` path (.bin needs ``fields``), ``images`` as ``NAME=PATH``,
    ``calib`` JSON file or ``calib_preset`` name, ``inputs`` .npz (planner), ``params``, stream id / reset."""
    req: Dict[str, Any] = {}
    if points:
        p = Path(points)
        fmt = POINT_SUFFIXES.get(p.suffix.lower())
        if fmt is None:
            raise ValueError(f"points: unknown suffix {p.suffix!r} (use .bin / .npy / .npz / .pcd)")
        req["points"] = {"format": fmt, "data": b64file(p)}
        if fields:
            req["points"]["fields"] = list(fields)
    images = list(images)
    if images:
        req["images"] = []
        for spec in images:
            name, _, path = spec.partition("=")
            if not path:
                raise ValueError(f"image expects NAME=PATH, got {spec!r}")
            req["images"].append({"camera": name, "data": b64file(path)})
    if calib:
        req["calibration"] = json.loads(Path(calib).read_text())
    elif calib_preset:
        req["calibration"] = {"preset": calib_preset}
    if inputs:
        req["inputs"] = {"format": "npz", "data": b64file(inputs)}
    if params:
        req["params"] = dict(params)
    if stream_id is not None or reset:
        req["stream"] = {"id": stream_id or "default", "reset": bool(reset)}
    if output_format != "json":
        req["output_format"] = output_format
    return req


def post(url: str, payload: Dict[str, Any], timeout: float = 120.0) -> Tuple[int, Dict[str, Any]]:
    """POST ``payload`` to ``<url>/predict`` -> ``(status, json body)`` (HTTP errors are returned, not raised)."""
    data = json.dumps(payload).encode("utf-8")
    r = urllib.request.Request(url.rstrip("/") + "/predict", data=data, headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(r, timeout=timeout) as resp:
            return resp.status, json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", "replace")
        try:
            return e.code, json.loads(body)
        except ValueError:
            return e.code, {"detail": body}


def get_json(url: str, timeout: float = 10.0) -> Dict[str, Any]:
    with urllib.request.urlopen(url, timeout=timeout) as r:
        return json.loads(r.read().decode("utf-8"))


def wait_ready(base: str, wait_s: float = 0.0, poll_s: float = 5.0) -> Dict[str, Any]:
    """Poll ``/health`` until ``ok`` / ``error`` or ``wait_s`` elapses; returns the last health body."""
    deadline = time.time() + wait_s
    last: Dict[str, Any] = {}
    while True:
        try:
            last = get_json(base.rstrip("/") + "/health")
            if last.get("status") in ("ok", "error"):
                return last
        except (urllib.error.URLError, OSError, ValueError) as e:
            last = {"status": f"unreachable ({e})"}
        if time.time() >= deadline:
            return last
        time.sleep(poll_s)


def check_service(base: str, *, expect_dispatch: Optional[str] = "eth",
                  expect_grid: Optional[str] = "12x10") -> Tuple[Dict[str, Any], List[str]]:
    """``/info`` and ``/v1/models`` answer; ``/info`` reports the pinned dispatch and grid (PLAN.md 0.3 item 6:
    the smoke must assert ETH dispatch and 12x10, not just print them). Returns ``(info, failures)``."""
    fails: List[str] = []
    try:
        info = get_json(base.rstrip("/") + "/info")
        models = get_json(base.rstrip("/") + "/v1/models")
    except (urllib.error.URLError, OSError, ValueError) as e:
        return {}, [f"/info or /v1/models: {e}"]
    if not models.get("data"):
        fails.append("/v1/models is empty")
    device = info.get("device") or {}
    if expect_dispatch and device.get("dispatch") != expect_dispatch:
        fails.append(f"dispatch is {device.get('dispatch')!r}, expected {expect_dispatch!r}")
    if expect_grid and device.get("grid") != expect_grid:
        fails.append(f"grid is {device.get('grid')!r}, expected {expect_grid!r}")
    return info, fails


def check_bad_request(base: str, payload: Optional[Dict[str, Any]] = None) -> Optional[str]:
    """A malformed request must answer 400 (not 500). Returns a failure message or None."""
    code, _ = post(base, payload or {"points": {"format": "npy", "data": "not base64!"}})
    return None if code == 400 else f"malformed request answered {code}, expected 400"


def smoke_line(model: str, summary: str, failures: Sequence[str]) -> str:
    """The one-line verdict container_smoke.sh greps: ``PASS <model>: ...`` / ``FAIL <model>: ... :: reasons``."""
    if failures:
        return f"FAIL {model}: {summary} :: " + "; ".join(list(failures)[:5])
    return f"PASS {model}: {summary}"


def _param(text: str) -> Tuple[str, Any]:
    k, _, v = text.partition("=")
    try:
        return k, json.loads(v)
    except ValueError:
        return k, v


def main(argv: Optional[Sequence[str]] = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--points")
    ap.add_argument("--fields", help="comma list, e.g. x,y,z,intensity (needed for .bin)")
    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("--inputs", help=".npz of named input tensors (planner)")
    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)
    req = build_request(a.points, a.fields.split(",") if a.fields else None, a.image, a.calib, a.calib_preset,
                        a.inputs, dict(_param(p) for p in a.param), a.stream_id, a.reset, a.output_format)
    if a.out:
        Path(a.out).write_text(json.dumps(req))
    if a.url:
        code, body = 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())