130 lines
4.4 KiB
Python
130 lines
4.4 KiB
Python
"""
|
|
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)
|
|
)
|