Add runnable entry point for autotune ISF module

Author: Sigrid Tofte Thiis

loop_to_python_adaptive/autotune/autotune.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Optional
+
+import numpy as np
+
+
+@dataclass(frozen=True)
+class AutotuneISFConfig:
+    min_points: int = 10
+    adjustment_fraction: float = 0.2  # oref0 uses 20%
+    autosens_max: float = 1.2         # oref0 default
+    autosens_min: float = 0.7         # oref0 default
+    min_bgi_abs: float = 1e-6         # avoid divide by tiny numbers
+    ratio_clip_low: float = 0.3       # hard clip to avoid insane outliers
+    ratio_clip_high: float = 3.0
+
+
+def tune_isf_like_oref0(
+    *,
+    isf_current: float,
+    isf_glucose_data: list[dict[str, Any]],
+    pump_isf: Optional[float] = None,
+    config: AutotuneISFConfig | None = None,
+) -> dict[str, Any]:
+    """
+    oref0-like ISF tuning:
+      ratio = 1 + deviation / BGI
+      fullNewISF = ISF * median(ratio)
+      newISF = (1-adjustment_fraction)*ISF + adjustment_fraction*adjustedISF
+    """
+    cfg = config or AutotuneISFConfig()
+
+    ratios: list[float] = []
+    for p in isf_glucose_data:
+        dev = float(p["deviation"])
+        bgi = float(p["BGI"])
+        if abs(bgi) < cfg.min_bgi_abs:
+            continue
+        r = 1.0 + dev / bgi
+        if not np.isfinite(r):
+            continue
+        r = float(np.clip(r, cfg.ratio_clip_low, cfg.ratio_clip_high))
+        ratios.append(r)
+
+    if len(ratios) < cfg.min_points:
+        return {
+            "newISF": float(isf_current),
+            "p50_ratio": None,
+            "fullNewISF": None,
+            "adjustedISF": None,
+            "n_points": len(ratios),
+            "reason": f"Only {len(ratios)} usable ISF points (<{cfg.min_points}); leaving ISF unchanged.",
+        }
+
+    p50_ratio = float(np.median(ratios))
+    full_new_isf = float(isf_current * p50_ratio)
+
+    adjusted_isf = full_new_isf
+    if pump_isf is not None and pump_isf > 0:
+        # Match oref0 bounds:
+        # low autosens ratio = high ISF => maxISF = pumpISF / autosens_min
+        # high autosens ratio = low ISF => minISF = pumpISF / autosens_max
+        max_isf = pump_isf / cfg.autosens_min
+        min_isf = pump_isf / cfg.autosens_max
+        adjusted_isf = float(np.clip(adjusted_isf, min_isf, max_isf))
+
+    # Slow update like oref0
+    new_isf = (1.0 - cfg.adjustment_fraction) * float(isf_current) + cfg.adjustment_fraction * float(adjusted_isf)
+    print(f"tune_isf_like_oref0: current ISF={isf_current}, p50_ratio={p50_ratio}, full_new_isf={full_new_isf}, ")
+    return {
+        "newISF": float(new_isf),
+        "p50_ratio": p50_ratio,
+        "fullNewISF": full_new_isf,
+        "adjustedISF": float(adjusted_isf),
+        "n_points": len(ratios),
+        "reason": "OK",
+    }
+
+
+if __name__ == "__main__":
+    sample_points = [
+        {"deviation": 3.0, "BGI": 10.0},
+        {"deviation": -2.0, "BGI": 9.5},
+        {"deviation": 1.2, "BGI": 8.0},
+        {"deviation": -0.5, "BGI": 7.5},
+        {"deviation": 2.2, "BGI": 10.2},
+        {"deviation": -1.0, "BGI": 9.0},
+        {"deviation": 0.8, "BGI": 8.8},
+        {"deviation": 1.5, "BGI": 9.7},
+        {"deviation": -0.9, "BGI": 8.6},
+        {"deviation": 2.0, "BGI": 10.1},
+    ]
+
+    result = tune_isf_like_oref0(
+        isf_current=50.0,
+        isf_glucose_data=sample_points,
+        pump_isf=50.0,
+    )
+    print(result)

loop_to_python_adaptive/autotunePrep/categorize.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Optional
+
+import numpy as np
+
+
+@dataclass(frozen=True)
+class AutotuneISFConfig:
+    min_points: int = 10
+    adjustment_fraction: float = 0.2  # oref0 uses 20%
+    autosens_max: float = 1.2         # oref0 default
+    autosens_min: float = 0.7         # oref0 default
+    min_bgi_abs: float = 1e-6         # avoid divide by tiny numbers
+    ratio_clip_low: float = 0.3       # hard clip to avoid insane outliers
+    ratio_clip_high: float = 3.0
+
+
+def tune_isf_like_oref0(
+    *,
+    isf_current: float,
+    isf_glucose_data: list[dict[str, Any]],
+    pump_isf: Optional[float] = None,
+    config: AutotuneISFConfig | None = None,
+) -> dict[str, Any]:
+    """
+    oref0-like ISF tuning:
+      ratio = 1 + deviation / BGI
+      fullNewISF = ISF * median(ratio)
+      newISF = (1-adjustment_fraction)*ISF + adjustment_fraction*adjustedISF
+    """
+    cfg = config or AutotuneISFConfig()
+
+    ratios: list[float] = []
+    for p in isf_glucose_data:
+        dev = float(p["deviation"])
+        bgi = float(p["BGI"])
+        if abs(bgi) < cfg.min_bgi_abs:
+            continue
+        r = 1.0 + dev / bgi
+        if not np.isfinite(r):
+            continue
+        r = float(np.clip(r, cfg.ratio_clip_low, cfg.ratio_clip_high))
+        ratios.append(r)
+
+    if len(ratios) < cfg.min_points:
+        return {
+            "newISF": float(isf_current),
+            "p50_ratio": None,
+            "fullNewISF": None,
+            "adjustedISF": None,
+            "n_points": len(ratios),
+            "reason": f"Only {len(ratios)} usable ISF points (<{cfg.min_points}); leaving ISF unchanged.",
+        }
+
+    p50_ratio = float(np.median(ratios))
+    full_new_isf = float(isf_current * p50_ratio)
+
+    adjusted_isf = full_new_isf
+    if pump_isf is not None and pump_isf > 0:
+        # Match oref0 bounds:
+        # low autosens ratio = high ISF => maxISF = pumpISF / autosens_min
+        # high autosens ratio = low ISF => minISF = pumpISF / autosens_max
+        max_isf = pump_isf / cfg.autosens_min
+        min_isf = pump_isf / cfg.autosens_max
+        adjusted_isf = float(np.clip(adjusted_isf, min_isf, max_isf))
+
+    # Slow update like oref0
+    new_isf = (1.0 - cfg.adjustment_fraction) * float(isf_current) + cfg.adjustment_fraction * float(adjusted_isf)
+
+    return {
+        "newISF": float(new_isf),
+        "p50_ratio": p50_ratio,
+        "fullNewISF": full_new_isf,
+        "adjustedISF": float(adjusted_isf),
+        "n_points": len(ratios),
+        "reason": "OK",
+    }

loop_to_python_adaptive/autotunePrep/prep.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional
+
+import pandas as pd
+
+from . import api
+
+
+@dataclass(frozen=True)
+class ICEFeatureConfig:
+    insulin_type: str = "novolog"
+    batch_size: int = 300
+    overlap: int = 72  # 72 * 5 min = 6 hours; matches upstream default
+
+
+def add_ice_to_history_df(
+    df: pd.DataFrame,
+    *,
+    basal_scheduled_u_per_hr: float,
+    isf_mgdl_per_u: float,
+    cr_g_per_u: float,
+    config: Optional[ICEFeatureConfig] = None,
+) -> pd.DataFrame:
+    """
+    Compute insulin counteraction effect (ICE) time series using the Loop wrapper
+    and add it to a copy of df as df['ice'].
+
+    Expected df format (5-min index recommended):
+      - DatetimeIndex
+      - columns required by upstream:
+          'CGM' (mg/dL)
+          'basal' (U/hr)   (delivered basal rate per step)
+          'bolus' (U)      (bolus delivered at that timestamp; 0 otherwise)
+      - optional:
+          'carbs' (g) (not needed for ICE computation itself)
+
+    Returns:
+      A copy of df with 'ice' column (float). Some leading rows may be NaN due
+      to overlap/warmup requirements.
+    """
+    cfg = config or ICEFeatureConfig()
+
+    if not isinstance(df.index, pd.DatetimeIndex):
+        raise ValueError("df must have a DatetimeIndex (timestamps).")
+    required = {"CGM", "basal", "bolus"}
+    missing = required - set(df.columns)
+    if missing:
+        raise ValueError(f"df missing required columns: {sorted(missing)}")
+
+    # Upstream helper mutates the df in-place, so operate on a copy.
+    out = df.copy()
+
+    # This is provided by loop_to_python_api.api (re-exported via our api.py)
+    # It writes an 'ice' column aligned to timestamps.
+    out = api.add_insulin_counteraction_effect_to_df(
+        out,
+        basal_scheduled_u_per_hr,
+        isf_mgdl_per_u,
+        cr_g_per_u,
+        insulin_type=cfg.insulin_type,
+        batch_size=cfg.batch_size,
+        overlap=cfg.overlap,
+    )
+    return out
+
+
+def ensure_ice_column(
+    df: pd.DataFrame,
+    *,
+    basal_scheduled_u_per_hr: float,
+    isf_mgdl_per_u: float,
+    cr_g_per_u: float,
+    config: Optional[ICEFeatureConfig] = None,
+) -> pd.DataFrame:
+    """
+    Convenience: if df already has non-null 'ice' values, return df unchanged,
+    otherwise compute ICE and return df with 'ice'.
+    """
+    if "ice" in df.columns and df["ice"].notna().any():
+        return df
+    return add_ice_to_history_df(
+        df,
+        basal_scheduled_u_per_hr=basal_scheduled_u_per_hr,
+        isf_mgdl_per_u=isf_mgdl_per_u,
+        cr_g_per_u=cr_g_per_u,
+        config=config,
+    )