Add 3_cost_scaling_analysis/src/cost_scaling_analysis/core.py

3_cost_scaling_analysis/src/cost_scaling_analysis/core.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+import json
+import logging
+from dataclasses import dataclass, asdict
+from pathlib import Path
+from typing import Any
+import pandas as pd
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class CostMetrics:
+    """Repräsentiert Kostenmetriken für eine gegebene Workeranzahl."""
+    worker_count: int
+    cost_per_worker: float
+    total_cost: float
+
+    def to_json(self) -> str:
+        return json.dumps(asdict(self), indent=2)
+
+
+def _validate_input(worker_count: int) -> None:
+    if not isinstance(worker_count, int):
+        raise TypeError("worker_count muss ein Integer sein.")
+    if worker_count <= 0:
+        raise ValueError("worker_count muss größer als 0 sein.")
+
+
+def analyze_cost_scalability(worker_count: int) -> float:
+    """Analysiert, wie sich die Kosten pro Worker und Gesamtkosten bei variabler Worker-Anzahl verändern.
+
+    Args:
+        worker_count (int): Anzahl der eingesetzten Worker-Instanzen.
+
+    Returns:
+        float: Berechnete Gesamtkosten-Metrik basierend auf der Worker-Skalierung.
+    """
+    _validate_input(worker_count)
+    
+    logger.debug("Starte Kostenanalyse für %d Worker", worker_count)
+
+    # Simuliere Kostenfunktion mithilfe logarithmischer Skalierung mit pandas
+    data = pd.DataFrame({
+        'workers': range(1, worker_count + 1)
+    })
+    base_cost = 10.0
+    inefficiency_factor = (data['workers'] ** 0.1)
+    data['cost_per_worker'] = base_cost * inefficiency_factor
+    data['total_cost'] = data['cost_per_worker'] * data['workers']
+
+    result_row = data.iloc[-1]
+    metrics = CostMetrics(
+        worker_count=int(result_row['workers']),
+        cost_per_worker=float(result_row['cost_per_worker']),
+        total_cost=float(result_row['total_cost'])
+    )
+
+    output_path = Path('output/cost_metrics.json')
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    with output_path.open('w') as f:
+        f.write(metrics.to_json())
+
+    logger.info("Kostenanalyse abgeschlossen: %s", metrics)
+    return metrics.total_cost