""" Nightscout API client. Praat met een bestaande Nightscout-instantie (CGM-data, insuline/koolhydraten en pomp-/sensorwissels). Authenticatie loopt via environment variables: - NIGHTSCOUT_URL: basis-URL van je Nightscout-instantie, bijv. "https://jouw-nightscout.example.com" (verplicht). - NIGHTSCOUT_TOKEN: een Nightscout API-token (rol-gebaseerd), OF - NIGHTSCOUT_API_SECRET: het "API secret" van je instantie (wordt lokaal ge-hashed, nooit in plaintext verstuurd). Gebruik precies één van de twee auth-opties; als beide gezet zijn heeft het token voorrang. """ from __future__ import annotations import hashlib import os from datetime import datetime, timedelta, timezone import pandas as pd import requests MGDL_PER_MMOL = 18.0182 INSULIN_EVENT_TYPES = {"Bolus", "Meal Bolus", "Correction Bolus", "Combo Bolus", "SMB"} SITE_CHANGE_EVENT_TYPES = {"Site Change"} SENSOR_CHANGE_EVENT_TYPES = {"Sensor Change", "Sensor Start"} _REQUEST_TIMEOUT = 10 class NightscoutConfigError(RuntimeError): """NIGHTSCOUT_URL ontbreekt of is ongeldig.""" def mgdl_to_mmol(value: float) -> float: """Reken mg/dL (Nightscout's interne eenheid) om naar mmol/L.""" return round(value / MGDL_PER_MMOL, 1) def _base_url() -> str: url = os.environ.get("NIGHTSCOUT_URL", "").strip().rstrip("/") if not url: raise NightscoutConfigError( "NIGHTSCOUT_URL is niet ingesteld. Zet deze environment variable, " "bijv. https://jouw-nightscout.example.com" ) return url def _auth() -> tuple[dict, dict]: """Geeft (headers, params) terug voor authenticatie met Nightscout.""" token = os.environ.get("NIGHTSCOUT_TOKEN", "").strip() api_secret = os.environ.get("NIGHTSCOUT_API_SECRET", "").strip() if token: return {}, {"token": token} if api_secret: hashed = hashlib.sha1(api_secret.encode("utf-8")).hexdigest() return {"API-SECRET": hashed}, {} return {}, {} def _get(path: str, params: dict) -> list: headers, auth_params = _auth() response = requests.get( f"{_base_url()}{path}", headers=headers, params={**params, **auth_params}, timeout=_REQUEST_TIMEOUT, ) response.raise_for_status() return response.json() def fetch_entries(hours: int = 24) -> pd.DataFrame: """Haal glucosemetingen (SGV) op van de laatste `hours` uur, in mmol/L. Geeft een DataFrame terug met kolommen 'timestamp' en 'mmol', oplopend gesorteerd op tijd. """ since = datetime.now(timezone.utc) - timedelta(hours=hours) # Nightscout geeft resultaten terug in aflopende volgorde; vraag ruim # genoeg entries op (CGM meet doorgaans elke 5 min) en filter lokaal op # tijd. Dit is robuuster dan server-side find[]-query's, die niet op elke # Nightscout-versie identiek werken. count = max(300, hours * 15) data = _get("/api/v1/entries/sgv.json", {"count": count}) if not data: return pd.DataFrame(columns=["timestamp", "mmol"]) df = pd.DataFrame(data) df["timestamp"] = pd.to_datetime(df["date"], unit="ms", utc=True).dt.tz_localize(None) df["mmol"] = df["sgv"].astype(float).apply(mgdl_to_mmol) df = df[df["timestamp"] >= since.replace(tzinfo=None)] return df[["timestamp", "mmol"]].sort_values("timestamp").reset_index(drop=True) def fetch_treatments(hours: int = 24) -> pd.DataFrame: """Haal behandelingen op van de laatste `hours` uur: insuline, koolhydraten en pomp-/sensorwissels. Geeft een DataFrame terug met kolommen 'timestamp', 'eventType', 'insulin' en 'carbs' (de laatste twee NaN waar niet van toepassing). """ since = datetime.now(timezone.utc) - timedelta(hours=hours) count = max(200, hours * 10) data = _get("/api/v1/treatments.json", {"count": count}) if not data: return pd.DataFrame(columns=["timestamp", "eventType", "insulin", "carbs"]) df = pd.DataFrame(data) df["timestamp"] = pd.to_datetime(df["created_at"], utc=True).dt.tz_localize(None) for col in ("insulin", "carbs"): if col not in df.columns: df[col] = pd.NA df[col] = pd.to_numeric(df[col], errors="coerce") if "eventType" not in df.columns: df["eventType"] = "" df = df[df["timestamp"] >= since.replace(tzinfo=None)] return ( df[["timestamp", "eventType", "insulin", "carbs"]] .sort_values("timestamp") .reset_index(drop=True) )