commit 15010cd5913fad9d028daa3266a7bc8bd9894fc5 Author: Mika Date: Mon Apr 6 16:12:07 2026 +0000 Add artifact.metrics_analysis/src/artifact_metrics_analysis/core.py diff --git a/artifact.metrics_analysis/src/artifact_metrics_analysis/core.py b/artifact.metrics_analysis/src/artifact_metrics_analysis/core.py new file mode 100644 index 0000000..bff2a3e --- /dev/null +++ b/artifact.metrics_analysis/src/artifact_metrics_analysis/core.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import logging +from typing import Dict, Any +import pandas as pd +from statistics import mean + +logger = logging.getLogger(__name__) +logging.basicConfig(level=logging.INFO) + + +class MetricResults: + """Repräsentation des Ergebnisses der Metrikanalyse inklusive Delta-Berechnung.""" + + def __init__(self, retry_tail_p99: float, band_width: float, delta_band_width: float) -> None: + assert isinstance(retry_tail_p99, (int, float)), "retry_tail_p99 muss numerisch sein" + assert isinstance(band_width, (int, float)), "band_width muss numerisch sein" + assert isinstance(delta_band_width, (int, float)), "delta_band_width muss numerisch sein" + + self.retry_tail_p99 = float(retry_tail_p99) + self.band_width = float(band_width) + self.delta_band_width = float(delta_band_width) + + def to_dict(self) -> Dict[str, float]: + """Serialisiert die Ergebnisse als dictionary für JSON-Ausgabe.""" + return { + "retry_tail_p99": self.retry_tail_p99, + "band_width": self.band_width, + "delta_band_width": self.delta_band_width, + } + + +class MetricDataError(Exception): + """Custom-Exception für ungültige Metrikdaten.""" + pass + + +def analyze_metrics(card_data_40: Dict[str, Any], card_data_42: Dict[str, Any]) -> MetricResults: + """Vergleicht Metriken von zwei Evidence Cards anhand der Eingabedaten und erstellt eine Ergebnisanalyse.""" + if not isinstance(card_data_40, dict) or not isinstance(card_data_42, dict): + raise MetricDataError("Beide Eingaben müssen Dictionaries sein.") + + required_fields = {"retry_tail_p99", "band_width"} + for cid, data in (("card_40", card_data_40), ("card_42", card_data_42)): + missing = required_fields - data.keys() + if missing: + raise MetricDataError(f"Fehlende Felder in {cid}: {missing}") + for field in required_fields: + if not isinstance(data[field], (int, float)): + raise MetricDataError(f"Ungültiger Typ für {field} in {cid}: {type(data[field])}") + + df = pd.DataFrame([ + {"card": "40", **card_data_40}, + {"card": "42", **card_data_42} + ]) + + retry_tail_p99_avg = mean(df["retry_tail_p99"]) + bw_40 = df.loc[df["card"] == "40", "band_width"].iloc[0] + bw_42 = df.loc[df["card"] == "42", "band_width"].iloc[0] + delta_band_width = bw_42 - bw_40 + + logger.info("Analyse durchgeführt: ΔBandbreite=%.4f", delta_band_width) + + result = MetricResults( + retry_tail_p99=retry_tail_p99_avg, + band_width=mean([bw_40, bw_42]), + delta_band_width=delta_band_width, + ) + + assert isinstance(result.retry_tail_p99, float) + assert isinstance(result.band_width, float) + assert isinstance(result.delta_band_width, float) + + return result \ No newline at end of file