Add artifact_1_bandwidth_analysis/src/artifact_1_bandwidth_analysis/core.py

artifact_1_bandwidth_analysis/src/artifact_1_bandwidth_analysis/core.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+import pandas as pd
+from statistics import median
+from typing import List, Dict, Any
+import logging
+
+
+logger = logging.getLogger(__name__)
+logger.addHandler(logging.NullHandler())
+
+
+def _validate_input_data(data: List[Dict[str, Any]], label: str) -> pd.DataFrame:
+    """Validates and converts input data to a pandas DataFrame."""
+    if not isinstance(data, list):
+        raise TypeError(f"{label} must be a list of dicts, got {type(data)}")
+    if not data:
+        raise ValueError(f"{label} is empty; cannot perform analysis.")
+
+    for idx, item in enumerate(data):
+        if not isinstance(item, dict):
+            raise TypeError(f"{label}[{idx}] must be a dict, got {type(item)}")
+        if 'timestamp' not in item:
+            raise KeyError(f"Missing 'timestamp' in {label}[{idx}]")
+        if not isinstance(item['timestamp'], (float, int)):
+            raise TypeError(
+                f"{label}[{idx}]['timestamp'] must be float or int, got {type(item['timestamp'])}"
+            )
+
+    df = pd.DataFrame(data)
+    if 'timestamp' not in df.columns:
+        raise KeyError(f"{label} must contain a 'timestamp' column.")
+    return df
+
+
+def analyze_bandwidth(
+    data_enforced: List[Dict[str, Any]],
+    data_randomized: List[Dict[str, Any]]
+) -> Dict[str, Dict[str, float]]:
+    """Vergleicht Bandbreite und Bandzentrum zweier Datensätze.
+
+    Parameters
+    ----------
+    data_enforced : List[Dict[str, Any]]
+        Messdaten im 'affinity enforced'-Modus.
+    data_randomized : List[Dict[str, Any]]
+        Messdaten im 'affinity off'-Modus.
+
+    Returns
+    -------
+    Dict[str, Dict[str, float]]
+        Ergebnisse pro Modus mit 'band_center' und 'band_width'.
+    """
+    logger.debug("Starting analyze_bandwidth computation.")
+
+    enforced_df = _validate_input_data(data_enforced, 'data_enforced')
+    random_df = _validate_input_data(data_randomized, 'data_randomized')
+
+    def _compute_metrics(df: pd.DataFrame, label: str) -> Dict[str, float]:
+        timestamps = df['timestamp'].dropna()
+        if timestamps.empty:
+            raise ValueError(f"No valid timestamps found in {label}.")
+
+        center = float(median(timestamps))
+        q75 = float(timestamps.quantile(0.75))
+        q25 = float(timestamps.quantile(0.25))
+        width = q75 - q25
+        logger.debug(f"Computed {label}: center={center}, width={width}")
+        return {'band_center': center, 'band_width': width}
+
+    result = {
+        'affinity_enforced': _compute_metrics(enforced_df, 'affinity_enforced'),
+        'affinity_off': _compute_metrics(random_df, 'affinity_off')
+    }
+
+    # CI assertion sanity checks
+    assert all('band_center' in v and 'band_width' in v for v in result.values()), \
+        "Result structure validation failed."
+
+    logger.info("Bandwidth analysis completed successfully.")
+    return result