66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Send a test request to inferenceserver /predict/")
|
|
parser.add_argument("--url", default="http://127.0.0.1:8003/predict/", help="Predict endpoint URL")
|
|
parser.add_argument("--image", default="image.jpg", help="Path to image file")
|
|
parser.add_argument("--timeout", type=float, default=10.0, help="HTTP timeout in seconds")
|
|
parser.add_argument("--conf", type=float, default=None, help="Optional confidence override")
|
|
parser.add_argument("--scale", type=float, default=None, help="Optional scale override")
|
|
parser.add_argument("--infer-scale", type=float, default=None, help="Optional infer_scale override")
|
|
parser.add_argument("--task", choices=["auto", "detect", "segment"], default=None, help="Optional task override")
|
|
args = parser.parse_args()
|
|
|
|
image_path = Path(args.image)
|
|
if not image_path.exists():
|
|
print(f"ERROR: image not found: {image_path.resolve()}")
|
|
return 2
|
|
|
|
params: dict[str, float] = {}
|
|
if args.conf is not None:
|
|
params["conf"] = float(args.conf)
|
|
if args.scale is not None:
|
|
params["scale"] = float(args.scale)
|
|
if args.infer_scale is not None:
|
|
params["infer_scale"] = float(args.infer_scale)
|
|
if args.task is not None:
|
|
params["task"] = args.task
|
|
|
|
print(f"POST {args.url}")
|
|
print(f"image={image_path.resolve()} timeout={args.timeout}s params={params}")
|
|
|
|
try:
|
|
with image_path.open("rb") as f:
|
|
response = requests.post(
|
|
args.url,
|
|
files={"file": (image_path.name, f, "application/octet-stream")},
|
|
params=params,
|
|
timeout=args.timeout,
|
|
)
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"REQUEST FAILED: {e}")
|
|
print("Hint: keep SSH tunnel open in a separate terminal:")
|
|
print(" ssh -N -L 8002:x10sa-spark-01:8002 gotthard_g@x10sa-gw")
|
|
return 1
|
|
|
|
print(f"status={response.status_code}")
|
|
print("response headers:", dict(response.headers))
|
|
|
|
try:
|
|
print("json:", response.json())
|
|
except ValueError:
|
|
print("non-json body:")
|
|
print(response.text[:2000])
|
|
|
|
return 0 if response.ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|