Add 3_cost_scaling_analysis/src/cost_scaling_analysis/cli.py

3_cost_scaling_analysis/src/cost_scaling_analysis/cli.py

@@ -0,0 +1,76 @@
+import argparse
+import json
+import os
+from pathlib import Path
+from typing import Any, Dict
+
+from cost_scaling_analysis.core import analyze_cost_scalability
+
+def _validate_input_path(path: str) -> Path:
+    file_path = Path(path)
+    if not file_path.exists() or not file_path.is_file():
+        raise FileNotFoundError(f"Eingabedatei nicht gefunden: {path}")
+    if file_path.suffix.lower() != '.json':
+        raise ValueError("Eingabedatei muss im JSON-Format vorliegen.")
+    return file_path
+
+
+def _validate_output_path(path: str) -> Path:
+    out_path = Path(path)
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    if out_path.suffix.lower() != '.json':
+        raise ValueError("Ausgabedatei muss im JSON-Format gespeichert werden.")
+    return out_path
+
+
+def _load_worker_config(file_path: Path) -> Dict[str, Any]:
+    with file_path.open('r', encoding='utf-8') as f:
+        try:
+            data = json.load(f)
+        except json.JSONDecodeError as e:
+            raise ValueError(f"Ungültiges JSON in Eingabedatei: {e}") from e
+    if 'worker_count' not in data or not isinstance(data['worker_count'], int):
+        raise ValueError("Feld 'worker_count' fehlt oder ist kein Integer.")
+    return data
+
+
+def _save_metrics(output_path: Path, metrics: Dict[str, Any]) -> None:
+    with output_path.open('w', encoding='utf-8') as f:
+        json.dump(metrics, f, indent=2, ensure_ascii=False)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Analysiert die Kosten-Skalierung basierend auf einer Worker-Konfigurationsdatei."
+    )
+    parser.add_argument(
+        '--input', required=True, help='Pfad zur JSON-Eingabedatei mit Worker-Konfigurationsdaten.'
+    )
+    parser.add_argument(
+        '--output', required=True, help='Pfad zur Ausgabedatei für berechnete Kostenmetriken.'
+    )
+
+    args = parser.parse_args()
+
+    input_path = _validate_input_path(args.input)
+    output_path = _validate_output_path(args.output)
+
+    config = _load_worker_config(input_path)
+    worker_count = config['worker_count']
+
+    total_cost = analyze_cost_scalability(worker_count)
+
+    # Beispielhafte Kostenstruktur erstellen
+    metrics = {
+        'worker_count': worker_count,
+        'cost_per_worker': round(total_cost / worker_count, 3) if worker_count > 0 else 0.0,
+        'total_cost': round(total_cost, 3),
+    }
+
+    _save_metrics(output_path, metrics)
+
+    print(f"Kostenmetriken erfolgreich berechnet und gespeichert unter: {output_path}")
+
+
+if __name__ == '__main__':
+    main()