diff --git a/3_cost_scaling_analysis/src/cost_scaling_analysis/cli.py b/3_cost_scaling_analysis/src/cost_scaling_analysis/cli.py new file mode 100644 index 0000000..a57c125 --- /dev/null +++ b/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() \ No newline at end of file