Add artifact.1/src/artifact_1/cli.py

artifact.1/src/artifact_1/cli.py

@@ -0,0 +1,88 @@
+import argparse
+import json
+import logging
+import sys
+from pathlib import Path
+from typing import Any, List, Dict
+
+import pandas as pd
+
+from artifact_1 import core
+
+
+class CLIError(Exception):
+    """Custom exception for CLI related errors."""
+
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="[%(asctime)s] %(levelname)s in %(module)s: %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+def _validate_input_data(data: Any) -> List[Dict[str, Any]]:
+    if not isinstance(data, list):
+        raise CLIError("Input JSON must be a list of metric objects.")
+    required_fields = {"band_width", "retry_tail_p99", "band_center"}
+    for idx, item in enumerate(data):
+        if not isinstance(item, dict):
+            raise CLIError(f"Item {idx} is not a JSON object.")
+        missing = required_fields - item.keys()
+        if missing:
+            raise CLIError(f"Item {idx} missing fields: {', '.join(missing)}")
+        for field in required_fields:
+            if not isinstance(item[field], (int, float)):
+                raise CLIError(f"Field '{field}' in item {idx} must be numeric.")
+    return data
+
+
+def main(argv: List[str] | None = None) -> None:
+    parser = argparse.ArgumentParser(
+        description="Analysiert Benchmark-Metriken zur Bewertung der Parallelitätseffekte."
+    )
+    parser.add_argument(
+        "--input",
+        required=True,
+        help="Pfad zur JSON-Datei mit Benchmark-Metriken.",
+    )
+    parser.add_argument(
+        "--output",
+        required=True,
+        help="Zielpfad für die gespeicherte Analyseausgabe.",
+    )
+
+    args = parser.parse_args(argv)
+
+    input_path = Path(args.input)
+    output_path = Path(args.output)
+
+    try:
+        if not input_path.exists():
+            raise CLIError(f"Input file not found: {input_path}")
+
+        with input_path.open("r", encoding="utf-8") as f:
+            raw_data = json.load(f)
+
+        data = _validate_input_data(raw_data)
+
+        logger.info("Starting metric analysis for %d records.", len(data))
+        result = core.analyse_metrics(data)
+
+        with output_path.open("w", encoding="utf-8") as f:
+            json.dump(result, f, indent=2, ensure_ascii=False)
+
+        logger.info("Analysis successfully written to %s", output_path)
+    except CLIError as e:
+        logger.error("CLI error: %s", e)
+        sys.exit(2)
+    except json.JSONDecodeError as e:
+        logger.error("Invalid JSON input: %s", e)
+        sys.exit(3)
+    except Exception as e:
+        logger.exception("Unexpected error: %s", e)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()