Add cpu_pinning_analysis/src/cpu_pinning_analysis/core.py

cpu_pinning_analysis/src/cpu_pinning_analysis/core.py

@@ -0,0 +1,75 @@
+import statistics
+from typing import Dict, Any, List
+
+
+def _extract_values(data: Dict[str, Any]) -> List[float]:
+    """Extract latency values list from provided JSON-like data structure."""
+    if not isinstance(data, dict):
+        raise ValueError("Data must be a dictionary containing 'latencies' or 'values' list.")
+
+    values = data.get("latencies") or data.get("values")
+    if not isinstance(values, list) or not all(isinstance(v, (int, float)) for v in values):
+        raise ValueError("Data dictionary must have a list of numeric latency values under 'latencies' or 'values'.")
+
+    return sorted(values)
+
+
+def _percentile(data: List[float], percent: float) -> float:
+    """Compute a given percentile for a sorted list of numeric values."""
+    if not data:
+        return 0.0
+    k = (len(data) - 1) * percent / 100.0
+    f, c = int(k), int(k) + 1
+    if c >= len(data):
+        return float(data[-1])
+    d0 = data[f] * (c - k)
+    d1 = data[c] * (k - f)
+    return float(d0 + d1)
+
+
+def _build_histogram(values: List[float], bins: int = 10) -> Dict[str, int]:
+    """Build a simple histogram with a given number of bins."""
+    if not values:
+        return {}
+
+    min_val, max_val = min(values), max(values)
+    if min_val == max_val:
+        return {f"{min_val:.2f}": len(values)}
+
+    step = (max_val - min_val) / bins
+    hist = {}
+    for v in values:
+        idx = int((v - min_val) / step)
+        if idx == bins:
+            idx -= 1
+        bin_label = f"{(min_val + idx * step):.2f}-{(min_val + (idx + 1) * step):.2f}"
+        hist[bin_label] = hist.get(bin_label, 0) + 1
+    return hist
+
+
+def analyze_latencies(pinned_data: Dict[str, Any], unpinned_data: Dict[str, Any]) -> Dict[str, Any]:
+    """Vergleicht Latenzmetriken, berechnet P95, P99 und Histogramme.
+
+    Args:
+        pinned_data: JSON-basiertes Dict mit Latency-Werten für pinned CPUs.
+        unpinned_data: JSON-basiertes Dict mit Latency-Werten für unpinned CPUs.
+
+    Returns:
+        Dict mit Schlüsseln 'pinned' und 'unpinned', jeweils mit P95, P99 und Histogramm.
+    """
+    pinned_values = _extract_values(pinned_data)
+    unpinned_values = _extract_values(unpinned_data)
+
+    pinned_metrics = {
+        "P95": _percentile(pinned_values, 95),
+        "P99": _percentile(pinned_values, 99),
+        "histogram": _build_histogram(pinned_values),
+    }
+
+    unpinned_metrics = {
+        "P95": _percentile(unpinned_values, 95),
+        "P99": _percentile(unpinned_values, 99),
+        "histogram": _build_histogram(unpinned_values),
+    }
+
+    return {"pinned": pinned_metrics, "unpinned": unpinned_metrics}