""" Rapportage-logica voor de klinische rapportagepagina's. Zelfde opzet als data_processing.py: pure functies, geen Streamlit-imports, dus los te unit-testen. Zie docs/requirements-reporting.md voor de klinische definities waarop dit gebaseerd is (Battelino et al. 2019, Bergenstal et al. 2018). """ from __future__ import annotations import pandas as pd # Klinische drempels in mmol/L (Battelino et al. 2019, Diabetes Care 42(8):1593-1603). VERY_LOW = 3.0 LOW = 3.9 HIGH = 10.0 VERY_HIGH = 13.9 MINUTES_PER_DAY = 24 * 60 def gmi_percent(mean_mmol: float) -> float: """Geschat HbA1c (Glucose Management Indicator) uit gemiddelde glucose. GMI% = 3.31 + 0.4311 * gemiddelde(mmol/L) — Bergenstal et al. 2018, Diabetes Care 41(11):2275-2280. Dit is GEEN vervanging voor een labwaarde (kan tot ~0.5 procentpunt afwijken door o.a. rode-bloedcel-factoren) en moet als zodanig gelabeld worden in de UI. """ return round(3.31 + 0.4311 * mean_mmol, 1) def percent_cv(glucose: pd.DataFrame) -> float: """Glucosevariabiliteit als %CV (standaarddeviatie / gemiddelde * 100).""" if glucose.empty: return 0.0 mean = glucose["mmol"].mean() if mean == 0: return 0.0 return round(glucose["mmol"].std() / mean * 100, 1) def tir_breakdown_5band(glucose: pd.DataFrame) -> dict[str, float]: """Percentage metingen in elk van de 5 klinische glucosebanden. Banden: very_low (<3.0), low (3.0-3.9), in_range (3.9-10.0), high (10.0-13.9), very_high (>13.9). Sluit exact aan op de 5-bands AGP- indeling uit de consensus. """ if glucose.empty: return {"very_low": 0.0, "low": 0.0, "in_range": 0.0, "high": 0.0, "very_high": 0.0} def _band(mmol: float) -> str: if mmol < VERY_LOW: return "very_low" if mmol < LOW: return "low" if mmol <= HIGH: return "in_range" if mmol <= VERY_HIGH: return "high" return "very_high" counts = glucose["mmol"].apply(_band).value_counts(normalize=True) * 100 return { band: round(counts.get(band, 0.0), 1) for band in ("very_low", "low", "in_range", "high", "very_high") } def minutes_per_day(percentage: float) -> float: """Reken een percentage van de tijd om naar minuten per dag (ruwe vuistregel).""" return round(percentage / 100 * MINUTES_PER_DAY, 0) def sensor_active_percent(glucose: pd.DataFrame, start, end, expected_interval_minutes: int = 5) -> float: """Percentage van de periode waarin daadwerkelijk metingen binnenkwamen. Vuistregel op basis van het verwachte meetinterval (CGM's meten doorgaans elke 5 minuten) — geen exacte sensor-uptime-registratie, maar voldoende om een periode met te veel hiaten te signaleren (zie FR9 / NFR "minimaal geldig rapport"). """ total_minutes = (pd.Timestamp(end) - pd.Timestamp(start)).total_seconds() / 60 if total_minutes <= 0: return 0.0 expected_readings = total_minutes / expected_interval_minutes if expected_readings <= 0: return 0.0 return round(min(len(glucose) / expected_readings, 1.0) * 100, 1) def _state_blocks(df: pd.DataFrame, out: pd.Series) -> list[dict]: """Groepeer aaneengesloten metingen met dezelfde in/out-status. Eindtijd van elk blok = starttijd van het volgende blok (de status geldt tot de eerstvolgende meting anders aangeeft); het laatste blok behoudt zijn eigen laatste meting als eind. """ block_id = (out != out.shift()).cumsum() blocks = [ { "is_out": bool(out.loc[group.index[0]]), "start": group["timestamp"].iloc[0], "end": group["timestamp"].iloc[-1], } for _, group in df.groupby(block_id) ] for i in range(len(blocks) - 1): blocks[i]["end"] = blocks[i + 1]["start"] return blocks def _maybe_add_episode(events: list[dict], start, end, min_duration_minutes: int) -> None: duration = (end - start).total_seconds() / 60 if duration >= min_duration_minutes: events.append({"start": start, "end": end, "duration_minutes": round(duration, 1)}) def _scan_episodes(blocks: list[dict], min_duration_minutes: int) -> list[dict]: """Statelijke doorloop van de blokken volgens de persistentieregel. Een lopende episode wordt alleen verlengd voorbij een kort in-bereik-blok als er daarna weer een out-blok komt (het korte herstel wordt dan met terugwerkende kracht "geabsorbeerd"). Een kort slot-blok waarna de data simpelweg ophoudt telt dus NIET automatisch mee als voortzetting — anders zou een niet-bevestigd herstel de duur van het event kunstmatig oprekken. """ events: list[dict] = [] episode_start = None episode_end = None for block in blocks: if block["is_out"]: episode_start = episode_start or block["start"] episode_end = block["end"] continue if episode_start is None: continue recovery_minutes = (block["end"] - block["start"]).total_seconds() / 60 if recovery_minutes >= min_duration_minutes: _maybe_add_episode(events, episode_start, episode_end, min_duration_minutes) episode_start = None episode_end = None # anders: kortstondig herstel — episode blijft open, wordt pas # verlengd als er weer een out-blok volgt (hierboven). if episode_start is not None: _maybe_add_episode(events, episode_start, episode_end, min_duration_minutes) return events def detect_events( glucose: pd.DataFrame, threshold: float, direction: str, min_duration_minutes: int = 15, ) -> pd.DataFrame: """Detecteer discrete hypo-/hyperglykemie-events volgens de persistentieregel. Een event begint zodra glucose ononderbroken voorbij de drempel is geweest voor >= min_duration_minutes, en eindigt pas zodra glucose ononderbroken hersteld is voor >= min_duration_minutes — een korte terugval binnen die hersteltijd beëindigt het event dus niet (Battelino et al. 2019). `direction` is "below" (hypo) of "above" (hyper). Geeft een DataFrame terug met kolommen 'start', 'end', 'duration_minutes'. """ if glucose.empty: return pd.DataFrame(columns=["start", "end", "duration_minutes"]) df = glucose.sort_values("timestamp").reset_index(drop=True) if direction == "below": out = df["mmol"] < threshold elif direction == "above": out = df["mmol"] > threshold else: raise ValueError("direction moet 'below' of 'above' zijn") blocks = _state_blocks(df, out) events = _scan_episodes(blocks, min_duration_minutes) return pd.DataFrame(events, columns=["start", "end", "duration_minutes"]) def summarize_events(events: pd.DataFrame, period_days: float) -> dict[str, float]: """Samenvattende statistieken over een set gedetecteerde events.""" if events.empty or period_days <= 0: return { "count": 0, "avg_duration_minutes": 0.0, "longest_duration_minutes": 0.0, "total_minutes": 0.0, "events_per_week": 0.0, } total_minutes = events["duration_minutes"].sum() return { "count": len(events), "avg_duration_minutes": round(events["duration_minutes"].mean(), 1), "longest_duration_minutes": round(events["duration_minutes"].max(), 1), "total_minutes": round(total_minutes, 1), "events_per_week": round(len(events) / period_days * 7, 1), } AGP_PERCENTILE_COLUMNS = ["minute_of_day", "p5", "p25", "p50", "p75", "p95"] def agp_percentile_bands(glucose: pd.DataFrame, bin_minutes: int = 30) -> pd.DataFrame: """Aggregeer glucosewaarden per tijdstip-van-de-dag naar percentielbanden. Groepeert alle metingen in de periode naar hun tijd-van-de-dag (in bins van `bin_minutes`), en berekent per bin de 5e/25e/50e/75e/95e percentiel — de standaard AGP-percentielgrafiek over een samengesteld 24-uursbeeld (Danne et al. 2017, Diabetes Care 40(12):1631-1640). """ if glucose.empty: return pd.DataFrame(columns=AGP_PERCENTILE_COLUMNS) df = glucose.copy() minute_of_day = df["timestamp"].dt.hour * 60 + df["timestamp"].dt.minute df["bin"] = (minute_of_day // bin_minutes) * bin_minutes grouped = df.groupby("bin")["mmol"].quantile([0.05, 0.25, 0.5, 0.75, 0.95]).unstack() grouped.columns = ["p5", "p25", "p50", "p75", "p95"] grouped = grouped.reset_index().rename(columns={"bin": "minute_of_day"}) return grouped[AGP_PERCENTILE_COLUMNS].sort_values("minute_of_day").reset_index(drop=True)