#!/usr/bin/env python3 """Run one identical MiniMax-H3 Ref2VA video-reference request per replica.""" from __future__ import annotations import argparse import json import statistics import time from pathlib import Path from typing import Any import requests def request(args: argparse.Namespace) -> dict[str, Any]: body = { "model": args.model, "prompt": args.prompt, "num_outputs_per_prompt": 1, "num_inference_steps": args.num_inference_steps, "flow_shift": args.flow_shift, "audio_flow_shift": args.audio_flow_shift, "seed": args.seed, "task": "ref2va", "conditions": [ { "type": "video", "uri": str(args.reference_video), "role": "reference", "start_time_seconds": 0.0, } ], "target": { "short_edge": args.short_edge, "aspect_ratio": args.aspect_ratio, "duration_seconds": args.duration_seconds, }, } started_epoch = time.time() started = time.monotonic() result: dict[str, Any] = { "request_id": f"ref2va-video-rep{args.replica_index}", "task": "ref2va", "reference_kind": "video", "reference_video": str(args.reference_video), "replica_index": args.replica_index, "port": args.port, "prompt": args.prompt, "seed": args.seed, "num_inference_steps": args.num_inference_steps, "short_edge": args.short_edge, "aspect_ratio": args.aspect_ratio, "duration_seconds": args.duration_seconds, "started_at_epoch": started_epoch, "success": False, "error": None, } try: response = requests.post( f"http://{args.host}:{args.port}/v1/videos", json=body, timeout=args.submit_timeout, ) if response.status_code != 200: raise RuntimeError(f"submit HTTP {response.status_code}: {response.text[:2000]}") status = response.json() video_id = status.get("id") if not video_id: raise RuntimeError(f"submit response has no id: {status}") result["video_id"] = video_id deadline = time.monotonic() + args.request_timeout while status.get("status") not in {"completed", "failed"}: if time.monotonic() >= deadline: raise TimeoutError(f"video job {video_id} exceeded {args.request_timeout}s") time.sleep(args.poll_interval) poll = requests.get( f"http://{args.host}:{args.port}/v1/videos/{video_id}", timeout=args.poll_timeout, ) if poll.status_code != 200: raise RuntimeError(f"poll HTTP {poll.status_code}: {poll.text[:2000]}") status = poll.json() if status.get("status") != "completed": raise RuntimeError(f"job failed: {status.get('error') or status}") result["success"] = True result["inference_time_s"] = status.get("inference_time_s") result["peak_memory_mb"] = status.get("peak_memory_mb") result["file_path"] = status.get("file_path") result["status"] = status except Exception as exc: result["error"] = f"{type(exc).__name__}: {exc}" result["latency_s"] = time.monotonic() - started result["finished_at_epoch"] = time.time() args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(result, ensure_ascii=False, indent=2), flush=True) return result def summarize(args: argparse.Namespace) -> int: rows = [json.loads(path.read_text(encoding="utf-8")) for path in sorted(args.input_dir.glob("client_*/result.json"))] ok = [row for row in rows if row.get("success")] starts = [float(row["started_at_epoch"]) for row in rows] finishes = [float(row["finished_at_epoch"]) for row in rows] wall_s = max(finishes) - min(starts) if starts and finishes else 0.0 latencies = [float(row["latency_s"]) for row in ok] summary = { "method": "base", "task": "ref2va", "reference_kind": "video", "tp": 2, "replicas": 4, "expected_requests": 4, "requests_recorded": len(rows), "completed": len(ok), "failed": len(rows) - len(ok), "machine_wall_s": wall_s, "machine_qps": len(ok) / wall_s if wall_s else 0.0, "latency_mean_s": statistics.fmean(latencies) if latencies else 0.0, "rows": rows, } args.output.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps({key: value for key, value in summary.items() if key != "rows"}, ensure_ascii=False, indent=2)) return int(len(rows) != 4 or len(ok) != 4) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() sub = parser.add_subparsers(dest="command", required=True) p = sub.add_parser("run") p.add_argument("--host", default="127.0.0.1") p.add_argument("--port", type=int, required=True) p.add_argument("--model", default="/data/hf_models/MiniMax-H3") p.add_argument("--replica-index", type=int, required=True) p.add_argument("--reference-video", type=Path, required=True) p.add_argument("--prompt", required=True) p.add_argument("--num-inference-steps", type=int, default=20) p.add_argument("--short-edge", type=int, default=768) p.add_argument("--duration-seconds", type=float, default=5.0) p.add_argument("--aspect-ratio", default="16:9") p.add_argument("--flow-shift", type=float, default=12.0) p.add_argument("--audio-flow-shift", type=float, default=3.0) p.add_argument("--seed", type=int, default=1101) p.add_argument("--submit-timeout", type=float, default=120.0) p.add_argument("--poll-timeout", type=float, default=30.0) p.add_argument("--poll-interval", type=float, default=1.0) p.add_argument("--request-timeout", type=float, default=3600.0) p.add_argument("--output", type=Path, required=True) p = sub.add_parser("summarize") p.add_argument("--input-dir", type=Path, required=True) p.add_argument("--output", type=Path, required=True) return parser.parse_args() def main() -> int: args = parse_args() if args.command == "summarize": return summarize(args) if not args.reference_video.is_file(): raise SystemExit(f"reference video not found: {args.reference_video}") return int(not request(args)["success"]) if __name__ == "__main__": raise SystemExit(main())