introduced categorization of buckets

Author: Sigrid Tofte Thiis

.venv/Lib/site-packages/_pytest/__pycache__/terminalprogress.cpython-310-pytest-9.0.2.pyc

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
 from typing import Any, Optional
 
 import pandas as pd
@@ -22,6 +23,86 @@ class AutotunePrepConfig:
     cgm_col: str = "CGM"
     bgi_col: str = "BGI"
 
+    # --- oref0-like categorization knobs (simplified) ---
+    # classify points as CSF for X minutes after any announced carb entry
+    meal_exclusion_minutes: int = 240
+    # classify as UAM if deviation > threshold (mg/dL per 5 min)
+    uam_deviation_threshold: float = 6.0
+    # avoid using near-zero BGI points for ISF (prevents ratio explosions)
+    min_bgi_abs_for_isf: float = 0.5
+
+
+def _parse_loop_ts(s: str) -> datetime:
+    # your fixtures use '...Z'; datetime.fromisoformat doesn't parse 'Z' in py<3.11 reliably
+    if s.endswith("Z"):
+        s = s[:-1] + "+00:00"
+    dt = datetime.fromisoformat(s)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)
+
+
+def _categorize_points_oref0_like(
+    points: list[dict[str, Any]],
+    *,
+    loop_algorithm_input: dict,
+    cfg: AutotunePrepConfig,
+) -> dict[str, list[dict[str, Any]]]:
+    """
+    Simplified oref0-like categorization:
+      - CSF: within meal_exclusion_minutes of announced carbs
+      - UAM: deviation > uam_deviation_threshold
+      - ISF: remaining points with |BGI| >= min_bgi_abs_for_isf
+      - basal: everything else
+    """
+    carb_entries = loop_algorithm_input.get("carbEntries", []) or []
+    carb_times = []
+    for c in carb_entries:
+        if "date" in c and float(c.get("grams", 0) or 0) > 0:
+            carb_times.append(_parse_loop_ts(c["date"]))
+    carb_times.sort()
+
+    def in_meal_window(t: datetime) -> bool:
+        if not carb_times:
+            return False
+        # find latest carb time <= t
+        # (linear scan is fine for small fixtures; can optimize later)
+        last = None
+        for ct in carb_times:
+            if ct <= t:
+                last = ct
+            else:
+                break
+        if last is None:
+            return False
+        return t <= last + timedelta(minutes=cfg.meal_exclusion_minutes)
+
+    csf: list[dict[str, Any]] = []
+    uam: list[dict[str, Any]] = []
+    isf: list[dict[str, Any]] = []
+    basal: list[dict[str, Any]] = []
+
+    for p in points:
+        t = _parse_loop_ts(p["date"]) if isinstance(p["date"], str) else p["date"]
+        bgi = float(p["BGI"])
+        dev = float(p["deviation"])
+
+        if in_meal_window(t):
+            csf.append(p)
+        elif dev > cfg.uam_deviation_threshold:
+            uam.append(p)
+        elif abs(bgi) >= cfg.min_bgi_abs_for_isf:
+            isf.append(p)
+        else:
+            basal.append(p)
+
+    return {
+        "ISFGlucoseData": isf,
+        "CSFGlucoseData": csf,
+        "UAMGlucoseData": uam,
+        "basalGlucoseData": basal,
+    }
+
 
 def prepare_for_autotune_isf(
     df: pd.DataFrame,
@@ -30,12 +111,16 @@ def prepare_for_autotune_isf(
     config: Optional[AutotunePrepConfig] = None,
 ) -> dict[str, Any]:
     """
-    Prepare the exact inputs autotune_isf.tune_isf_like_oref0 needs.
+    Prepare inputs for autotune_isf.tune_isf_like_oref0, with oref0-like categorization.
 
     Returns:
       {
         "df": df_with_BGI,
-        "isf_glucose_data": list[{"date","avgDelta","BGI","deviation"}...]
+        "isf_glucose_data": points to use for ISF tuning (ISFGlucoseData),
+        "ISFGlucoseData": ...,
+        "CSFGlucoseData": ...,
+        "UAMGlucoseData": ...,
+        "basalGlucoseData": ...,
       }
     """
     cfg = config or AutotunePrepConfig()
@@ -48,12 +133,19 @@ def prepare_for_autotune_isf(
         history_hours=cfg.history_hours,
     )
 
-    df2, isf_glucose_data = prepare_isf_glucose_data(
+    df2, all_points = prepare_isf_glucose_data(
         df,
         loop_algorithm_input=loop_algorithm_input,
         config=bgi_cfg,
         cgm_col=cfg.cgm_col,
         bgi_col=cfg.bgi_col,
     )
 
-    return {"df": df2, "isf_glucose_data": isf_glucose_data}
+    buckets = _categorize_points_oref0_like(all_points, loop_algorithm_input=loop_algorithm_input, cfg=cfg)
+
+    # Backward compatible: return ISF-only points as isf_glucose_data
+    return {
+        "df": df2,
+        "isf_glucose_data": buckets["ISFGlucoseData"],
+        **buckets,
+    }

loop_to_python_adaptive/__pycache__/autotune_prep.cpython-310.pyc

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+from loop_to_python_adaptive.autotune_prep import AutotunePrepConfig, prepare_for_autotune_isf
+from loop_to_python_adaptive.autotune_isf import tune_isf_like_oref0
+
+
+def find_repo_root(start: Path) -> Path:
+    p = start
+    while True:
+        if (p / "loop_to_python_adaptive").exists():
+            return p
+        if p.parent == p:
+            raise RuntimeError("Could not find repo root (no loop_to_python_adaptive directory found).")
+        p = p.parent
+
+
+def test_autotune_isf_runs_on_loop_algorithm_input_fixture():
+    repo_root = find_repo_root(Path(__file__).resolve())
+    file = repo_root / "tests" / "test_files" / "loop_algorithm_input.json"
+    assert file.exists(), f"Missing file: {file}"
+
+    loop_input = json.loads(file.read_text(encoding="utf-8"))
+
+    glucose = loop_input["glucoseHistory"]
+    assert len(glucose) > 20
+
+    idx = pd.to_datetime([g["date"] for g in glucose], utc=True)
+    df = pd.DataFrame({"CGM": [float(g["value"]) for g in glucose]}, index=idx).sort_index()
+
+    # Key change: fixture basal schedule starts ~6h before first glucose point
+    prep = prepare_for_autotune_isf(
+        df,
+        loop_algorithm_input=loop_input,
+        config=AutotunePrepConfig(history_hours=6),
+    )
+
+    res = tune_isf_like_oref0(
+        isf_current=50.0,
+        pump_isf=None,
+        isf_glucose_data=prep["isf_glucose_data"],
+    )
+    print("Autotune ISF result:", res)
+    print(len(prep["ISFGlucoseData"]), len(prep["CSFGlucoseData"]), len(prep["UAMGlucoseData"]), len(prep["basalGlucoseData"]))
+
+    assert len(prep["ISFGlucoseData"]) > 0
+    assert len(prep["CSFGlucoseData"]) >= 0
+    assert len(prep["UAMGlucoseData"]) >= 0
+    assert res["reason"] == "OK", res
+    assert res["n_points"] >= 10, res
+    assert np.isfinite(res["newISF"]), res
+    assert abs(res["newISF"] - 80.0) > 0.01, res