""" Parsing van het Nightscout-pompprofiel (/api/v1/profile): basaalschema, insulinegevoeligheid (ISF), koolhydraatratio, DIA en streefwaarden. Framework-vrije module (geen streamlit-import), zelfde opzet als reporting.py: puur functies op dicts, zodat dit los van de UI te unit-testen is. Zie nightscout_client.fetch_profile() voor de ruwe fetch; deze module normaliseert alleen (uren -> minuten sinds middernacht, mg/dl -> mmol/L waar nodig). """ from __future__ import annotations from nightscout_client import mgdl_to_mmol SEGMENT_FIELDS = ("basal", "sens", "carbratio", "target_low", "target_high") def _time_to_minutes(time_str: str) -> int: """Zet 'HH:MM' (Nightscout's segmenttijd-notatie) om naar minuten sinds middernacht.""" hours, minutes = time_str.split(":") return int(hours) * 60 + int(minutes) def _parse_segments(raw_segments: list[dict], *, to_mmol: bool = False) -> list[dict]: """Zet Nightscout-segmenten ([{'time': 'HH:MM', 'value': ...}, ...]) om naar [{'minutes': int, 'value': float}, ...], oplopend gesorteerd op tijd.""" segments = [] for seg in raw_segments: value = float(seg["value"]) if to_mmol: value = mgdl_to_mmol(value) segments.append({"minutes": _time_to_minutes(seg["time"]), "value": value}) return sorted(segments, key=lambda s: s["minutes"]) def parse_profile(raw: dict) -> dict: """Parse één Nightscout-profieldocument (zoals teruggegeven door nightscout_client.fetch_profile()) naar een genormaliseerde dict: { "name": str, "units": "mmol", # altijd genormaliseerd, ongeacht bronformaat "dia": float, # uur "basal": [{"minutes": int, "value": float}, ...], # E/uur "sens": [{"minutes": int, "value": float}, ...], # mmol/L per E (ISF) "carbratio": [{"minutes": int, "value": float}, ...], # g koolhydraten per E "target_low": [{"minutes": int, "value": float}, ...], # mmol/L "target_high": [{"minutes": int, "value": float}, ...], # mmol/L } Geeft {} terug als er geen (bruikbaar) profiel is. """ if not raw: return {} default_name = raw.get("defaultProfile") store = raw.get("store", {}) profile = store.get(default_name) if default_name else None if profile is None and store: # defaultProfile ontbreekt of matcht niet met een sleutel in store: # val terug op het enige/eerste beschikbare profiel. profile = next(iter(store.values())) if profile is None: return {} units = str(profile.get("units", "mg/dl")).strip().lower() to_mmol = units != "mmol" return { "name": default_name or "", "units": "mmol", "dia": float(profile.get("dia", 0) or 0), "basal": _parse_segments(profile.get("basal", [])), "sens": _parse_segments(profile.get("sens", []), to_mmol=to_mmol), "carbratio": _parse_segments(profile.get("carbratio", [])), "target_low": _parse_segments(profile.get("target_low", []), to_mmol=to_mmol), "target_high": _parse_segments(profile.get("target_high", []), to_mmol=to_mmol), } def value_at(segments: list[dict], minutes: int) -> float | None: """Geef de waarde terug die actief is op het gegeven tijdstip (minuten sinds middernacht) — het laatste segment met `minutes <= t`. Ligt `t` vóór het eerste segment van de dag, dan geldt het laatste segment (dat loopt door vanaf de vorige dag, want een profiel dekt altijd de volle 24 uur). Geeft None terug als `segments` leeg is. """ if not segments: return None active = None for seg in segments: if seg["minutes"] <= minutes: active = seg else: break if active is None: active = segments[-1] return active["value"] def _format_time(minutes: int) -> str: """Zet minuten sinds middernacht om naar 'HH:MM'.""" return f"{minutes // 60:02d}:{minutes % 60:02d}" def segments_table(segments: list[dict]) -> list[dict]: """Zet segmenten om naar rijen met een start- en eindtijd voor tabelweergave: [{'start': 'HH:MM', 'end': 'HH:MM', 'value': float}, ...]. Het laatste segment loopt door tot 24:00 (een profiel dekt altijd de volle dag).""" rows = [] for i, seg in enumerate(segments): end_minutes = segments[i + 1]["minutes"] if i + 1 < len(segments) else 24 * 60 rows.append( { "start": _format_time(seg["minutes"]), "end": "24:00" if end_minutes == 24 * 60 else _format_time(end_minutes), "value": seg["value"], } ) return rows def total_daily_basal(segments: list[dict]) -> float: """Totale basaaldosis per dag in E: som van elk segment se duur (uur) x snelheid (E/uur).""" if not segments: return 0.0 total = 0.0 for i, seg in enumerate(segments): end_minutes = segments[i + 1]["minutes"] if i + 1 < len(segments) else 24 * 60 total += seg["value"] * (end_minutes - seg["minutes"]) / 60 return total def segments_to_hourly_bars(segments: list[dict]) -> list[dict]: """Zet segmenten om naar een lijst geschikt voor een staafgrafiek per uur van de dag: [{'hour': float, 'value': float}, ...], met een extra punt op uur 24 (gelijk aan het eerste segment) zodat een staaf/stap-grafiek de volle dag dekt. Gebruikt door de basaalgrafiek op de Insulineprofiel-pagina. """ if not segments: return [] bars = [{"hour": seg["minutes"] / 60, "value": seg["value"]} for seg in segments] bars.append({"hour": 24.0, "value": segments[0]["value"]}) return bars