Add basal/bolus/carb/target overlays to the Live and Dagoverzicht charts
willemjan/glucose-dashboard/pipeline/pr-feature/insulin-profile This commit looks good

Overlay the full insulin picture onto the interactive glucose charts:
basal rate as a semi-transparent step-area (profile schedule overridden
by active temp basal), bolus doses as size-scaled dots split into
food/correction/SMB categories (food and correction labeled with units,
SMB left unlabeled), carbs as bars, and the pump's target glucose as a
line — each with its own color, alpha, and legend entry. Shared trace-
building helpers live in ui_common.py so both pages (and the PDF report
in a follow-up PR) render identically.

Builds on the pump profile work in #29 (profile_data.py, load_profile()).
This commit is contained in:
2026-09-05 22:30:10 +02:00
parent 79249958c2
commit c54dcf616a
6 changed files with 729 additions and 42 deletions
+161
View File
@@ -15,6 +15,7 @@ import numpy as np
import pandas as pd
from nightscout_client import INSULIN_EVENT_TYPES
from profile_data import value_at
# Klinische drempels in mmol/L (Battelino et al. 2019, Diabetes Care 42(8):1593-1603).
VERY_LOW = 3.0
@@ -440,6 +441,166 @@ def bolus_stats(treatments: pd.DataFrame) -> dict:
}
BOLUS_CATEGORY_FOOD = "food"
BOLUS_CATEGORY_CORRECTION = "correction"
BOLUS_CATEGORY_SMB = "smb"
BOLUS_CATEGORY_COLUMNS = ["timestamp", "eventType", "insulin", "carbs", "category"]
def classify_bolus(treatments: pd.DataFrame) -> pd.DataFrame:
"""Splits bolus-insuline op in 'food'/'correction'/'smb', voor grafiekoverlays
(gitea insulineprofiel-werk, vervolg op bolus_stats()'s SMB-detectie).
Herkenningsregels, in volgorde:
- SMB: `eventType == "SMB"`, of `eventType == "Correction Bolus"` met
`type == "SMB"` (zie bolus_stats()) — automatische AndroidAPS-microbolus.
- food: `eventType == "Meal Bolus"`, of een andere bolus-eventType
(`Bolus`/`Combo Bolus`) met koolhydraten op dezelfde treatment-rij (een
sterke aanwijzing dat het om een maaltijdbolus gaat).
- correction: alle overige bolus-rijen (typisch `Correction Bolus` zonder
SMB-markering, of een generieke `Bolus`/`Combo Bolus` zonder koolhydraten).
Geeft een DataFrame terug met kolommen 'timestamp', 'eventType', 'insulin',
'carbs', 'category' — alleen de rijen die daadwerkelijk bolusinsuline zijn.
"""
if treatments.empty:
return pd.DataFrame(columns=BOLUS_CATEGORY_COLUMNS)
bolus = treatments[treatments["eventType"].isin(INSULIN_EVENT_TYPES) & treatments["insulin"].notna()].copy()
if bolus.empty:
return pd.DataFrame(columns=BOLUS_CATEGORY_COLUMNS)
is_smb = (bolus["eventType"] == SMB_TYPE) | (
(bolus["eventType"] == "Correction Bolus") & (bolus.get("type", "") == SMB_TYPE)
)
has_carbs = pd.to_numeric(bolus["carbs"], errors="coerce").fillna(0) > 0
is_food = ~is_smb & ((bolus["eventType"] == "Meal Bolus") | has_carbs)
bolus["category"] = BOLUS_CATEGORY_CORRECTION
bolus.loc[is_food, "category"] = BOLUS_CATEGORY_FOOD
bolus.loc[is_smb, "category"] = BOLUS_CATEGORY_SMB
return bolus[BOLUS_CATEGORY_COLUMNS].sort_values("timestamp").reset_index(drop=True)
def glucose_min_max(glucose: pd.DataFrame) -> tuple[float, float]:
"""Minimum en maximum glucosewaarde (mmol/L) in de gegeven data."""
if glucose.empty:
return (0.0, 0.0)
return (round(glucose["mmol"].min(), 1), round(glucose["mmol"].max(), 1))
def _temp_basal_intervals(treatments: pd.DataFrame, day_start, day_end) -> list[tuple]:
"""Temp Basal-intervallen (start, eind, rate) die (deels) binnen [day_start, day_end) vallen,
geclipt aan de daggrenzen."""
if treatments is None or treatments.empty:
return []
temp_basal = treatments[
(treatments["eventType"] == BASAL_EVENT_TYPE)
& treatments["absolute"].notna()
& treatments["duration"].notna()
]
intervals = []
for _, row in temp_basal.iterrows():
start = row["timestamp"]
end = start + pd.Timedelta(float(row["duration"]), unit="min")
if end <= day_start or start >= day_end:
continue
intervals.append((max(start, day_start), min(end, day_end), row["absolute"]))
return intervals
def _basal_breakpoints(basal_segments: list[dict], temp_intervals: list[tuple], day_start, day_end) -> list:
"""Alle tijdstippen waarop de effectieve basaalsnelheid kan wijzigen: daggrens,
profielsegment-grenzen, en temp-basal start/eind — oplopend gesorteerd."""
breakpoints = {day_start}
for seg in basal_segments:
ts = day_start + pd.Timedelta(float(seg["minutes"]), unit="min")
if day_start <= ts < day_end:
breakpoints.add(ts)
for start, end, _ in temp_intervals:
breakpoints.add(start)
breakpoints.add(end)
return sorted(t for t in breakpoints if day_start <= t < day_end)
def _rate_at(ts, basal_segments: list[dict], temp_intervals: list[tuple], day_start) -> float:
"""Effectieve basaalsnelheid op tijdstip `ts`: een actieve Temp Basal wint van het profieltempo."""
for start, end, rate in temp_intervals:
if start <= ts < end:
return rate
minutes_of_day = int((ts - day_start).total_seconds() / 60) % MINUTES_PER_DAY
value = value_at(basal_segments, minutes_of_day)
return value if value is not None else 0.0
def effective_basal_series(profile: dict | None, treatments: pd.DataFrame, day_start, day_end) -> pd.DataFrame:
"""Basaalsnelheid (E/uur) als stapfunctie over [day_start, day_end).
Basislijn is het pompprofiel se basaalschema (`profile["basal"]`); waar een
Temp Basal-treatment actief is (rate/duur uit de treatment) overschrijft
die het profieltempo voor de duur van die tijdelijke aanpassing — dit is de
daadwerkelijk afgeleverde basaal, in tegenstelling tot basal_stats() (die
alleen temp-basal-periodes optelt en periodes zonder actieve temp basal
mist omdat het profiel daar niet bij betrokken wordt).
Zonder profiel (`profile` is None/leeg) is de basislijn 0 — dan toont dit
alleen de temp-basal-periodes, net als basal_stats().
Geeft een DataFrame terug met kolommen 'timestamp' en 'rate': elke rij
geldt tot de volgende rij (stapfunctie), geschikt voor een
step-area-plot (line_shape='hv', fill='tozeroy').
"""
day_start = pd.Timestamp(day_start)
day_end = pd.Timestamp(day_end)
basal_segments = (profile or {}).get("basal", [])
temp_intervals = _temp_basal_intervals(treatments, day_start, day_end)
breakpoints = _basal_breakpoints(basal_segments, temp_intervals, day_start, day_end)
rows = [
{"timestamp": ts, "rate": _rate_at(ts, basal_segments, temp_intervals, day_start)} for ts in breakpoints
]
return pd.DataFrame(rows, columns=["timestamp", "rate"])
def _target_midpoint_at(minutes_of_day: int, target_low: list[dict], target_high: list[dict]) -> float | None:
low = value_at(target_low, minutes_of_day) if target_low else None
high = value_at(target_high, minutes_of_day) if target_high else None
if low is not None and high is not None:
return (low + high) / 2
return low if low is not None else high
def target_glucose_series(profile: dict | None, day_start, day_end) -> pd.DataFrame:
"""Streefglucose (mmol/L) uit het profiel als stapfunctie over [day_start, day_end) —
het midden van target_low/target_high per tijdsegment (de pomp kent vaak één
streefwaarde per segment; met losse low/high-segmenten is dat gewoon het
gemiddelde). Geeft een lege DataFrame terug als het profiel geen streefwaarden
bevat.
Zelfde stapfunctie-vorm als effective_basal_series(): elke rij geldt tot de
volgende rij.
"""
day_start = pd.Timestamp(day_start)
day_end = pd.Timestamp(day_end)
target_low = (profile or {}).get("target_low", [])
target_high = (profile or {}).get("target_high", [])
if not target_low and not target_high:
return pd.DataFrame(columns=["timestamp", "target"])
breakpoints = _basal_breakpoints(target_low + target_high, [], day_start, day_end)
rows = [
{
"timestamp": ts,
"target": _target_midpoint_at(
int((ts - day_start).total_seconds() / 60) % MINUTES_PER_DAY, target_low, target_high
),
}
for ts in breakpoints
]
return pd.DataFrame(rows, columns=["timestamp", "target"])
def basal_stats(treatments: pd.DataFrame, period_days: float) -> dict:
"""Totale basaalinsuline uit Temp Basal-treatments (rate x duur), zonder pompprofiel.