Add measure_latency/src/measure_latency/cli.py

measure_latency/src/measure_latency/cli.py

@@ -0,0 +1,65 @@
+import argparse
+import json
+from pathlib import Path
+import sys
+from typing import Any
+
+from measure_latency import core
+
+
+OUTPUT_DIR = Path("output")
+OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def _validate_positive_int(value: str) -> int:
+    try:
+        ivalue = int(value)
+        if ivalue <= 0:
+            raise ValueError
+        return ivalue
+    except ValueError:
+        raise argparse.ArgumentTypeError(f"Invalid number of runs: '{value}' (must be positive integer)")
+
+
+def main(argv: list[str] | None = None) -> None:
+    parser = argparse.ArgumentParser(description="Measure and analyze upload latency across systems.")
+    parser.add_argument(
+        "--n-runs",
+        required=True,
+        type=_validate_positive_int,
+        help="Number of measurement runs to perform (positive integer).",
+    )
+
+    args = parser.parse_args(argv)
+
+    n_runs: int = args.n_runs
+
+    results = core.measure_latencies(n_runs)
+    analysis = core.analyze_results(results)
+
+    results_path = OUTPUT_DIR / "latency_results.json"
+    summary_path = OUTPUT_DIR / "analysis_summary.json"
+
+    # Prepare JSON-serializable output
+    def _default(obj: Any):
+        import datetime
+        if isinstance(obj, datetime.datetime):
+            return obj.isoformat()
+        return str(obj)
+
+    with results_path.open("w", encoding="utf-8") as f:
+        json.dump([r.__dict__ for r in results], f, indent=2, default=_default)
+
+    with summary_path.open("w", encoding="utf-8") as f:
+        json.dump(analysis, f, indent=2, default=_default)
+
+    print(f"Saved latency results to {results_path}")
+    print(f"Saved analysis summary to {summary_path}")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)