Add policy_grid_evaluation/src/policy_grid_evaluation/core.py

policy_grid_evaluation/src/policy_grid_evaluation/core.py

@@ -0,0 +1,117 @@
+import argparse
+import json
+import itertools
+import logging
+import os
+from dataclasses import dataclass
+from typing import List
+import pandas as pd
+
+
+# Configure logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
+logger = logging.getLogger(__name__)
+
+
+class PolicyGridError(Exception):
+    """Custom exception for policy grid evaluation errors."""
+    pass
+
+
+@dataclass
+class GridResult:
+    """Repräsentiert ein einzelnes Policy-Kombinationsergebnis."""
+    policy: str
+    unknown_conversion: float
+    real_missing_cases: int
+    additional_wait_time: float
+
+    def __post_init__(self):
+        if not isinstance(self.policy, str):
+            raise TypeError('policy must be of type str')
+        if not isinstance(self.unknown_conversion, (int, float)):
+            raise TypeError('unknown_conversion must be a number')
+        if not isinstance(self.real_missing_cases, int):
+            raise TypeError('real_missing_cases must be an int')
+        if not isinstance(self.additional_wait_time, (int, float)):
+            raise TypeError('additional_wait_time must be a number')
+
+
+def evaluate_grid(grace_values: List[int], delay_values: List[int], policies: List[str]) -> List[GridResult]:
+    """Evaluierung aller Kombinationen von grace- und delay-Werten über verschiedene Policies."""
+    if not grace_values or not delay_values or not policies:
+        raise ValueError('Input lists must not be empty.')
+
+    results: List[GridResult] = []
+    for policy, grace, delay in itertools.product(policies, grace_values, delay_values):
+        # Beispielhafte Metrikberechnungen (Mocked Logic)
+        try:
+            unknown_conversion = round(100.0 / (grace + delay + 1), 4)
+            real_missing_cases = int((grace * delay) % 7)
+            additional_wait_time = float(grace * delay * 0.5)
+            result = GridResult(
+                policy=f"{policy}_g{grace}_d{delay}",
+                unknown_conversion=unknown_conversion,
+                real_missing_cases=real_missing_cases,
+                additional_wait_time=additional_wait_time
+            )
+            results.append(result)
+        except Exception as e:
+            logger.error(f'Error evaluating combination {policy}, {grace}, {delay}: {e}')
+            raise PolicyGridError from e
+
+    assert all(isinstance(r, GridResult) for r in results), 'All results must be GridResult instances.'
+    logger.info(f'Evaluated {len(results)} grid combinations.')
+    return results
+
+
+def save_results_to_csv(grid_results: List[GridResult], filename: str) -> None:
+    """Speichert die berechneten Resultate als CSV-Datei."""
+    if not grid_results:
+        raise ValueError('grid_results must not be empty.')
+    if not isinstance(filename, str) or not filename:
+        raise TypeError('filename must be a valid string path.')
+
+    data = [r.__dict__ for r in grid_results]
+    df = pd.DataFrame(data)
+    os.makedirs(os.path.dirname(filename), exist_ok=True)
+    df.to_csv(filename, index=False)
+    logger.info(f'Saved results to {filename}')
+
+
+def _load_input_parameters(json_path: str):
+    """Hilfsfunktion zum Laden der Eingangsparameter aus JSON."""
+    if not os.path.exists(json_path):
+        raise FileNotFoundError(f'Input file not found: {json_path}')
+
+    with open(json_path, 'r', encoding='utf-8') as f:
+        data = json.load(f)
+
+    try:
+        grace_values = data['grace_values']
+        delay_values = data['delay_values']
+        policies = data['policies']
+    except KeyError as e:
+        raise PolicyGridError(f'Missing key in input JSON: {e}')
+
+    if not all(isinstance(x, int) for x in grace_values + delay_values):
+        raise PolicyGridError('Grace and delay values must be integers.')
+    if not all(isinstance(p, str) for p in policies):
+        raise PolicyGridError('Policies must be strings.')
+
+    return grace_values, delay_values, policies
+
+
+def main():
+    parser = argparse.ArgumentParser(description='Evaluate policy parameter grid.')
+    parser.add_argument('--input', required=True, help='Pfad zu JSON-Datei mit Policy-Parametern.')
+    parser.add_argument('--output', required=False, default='output/grid_results.csv', help='Pfad zur Ausgabe-CSV-Datei.')
+    args = parser.parse_args()
+
+    grace_values, delay_values, policies = _load_input_parameters(args.input)
+    results = evaluate_grid(grace_values, delay_values, policies)
+    save_results_to_csv(results, args.output)
+
+
+if __name__ == '__main__':
+    main()