glucose-dashboard/nightscout_client.py

"""
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 date, datetime, time, 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 _entries_since(since: datetime, count: int, until: datetime | None = None) -> pd.DataFrame:
    """Haal entries op sinds `since` (en optioneel tot `until`), lokaal gefilterd.

    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.
    """
    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)]
    if until is not None:
        df = df[df["timestamp"] <= until.replace(tzinfo=None)]
    return df[["timestamp", "mmol"]].sort_values("timestamp").reset_index(drop=True)


def _treatments_since(since: datetime, count: int, until: datetime | None = None) -> pd.DataFrame:
    """Haal treatments op sinds `since` (en optioneel tot `until`), lokaal gefilterd."""
    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)]
    if until is not None:
        df = df[df["timestamp"] <= until.replace(tzinfo=None)]
    return (
        df[["timestamp", "eventType", "insulin", "carbs"]]
        .sort_values("timestamp")
        .reset_index(drop=True)
    )


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)
    count = max(300, hours * 15)
    return _entries_since(since, count)


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)
    return _treatments_since(since, count)


def fetch_entries_range(start_date: date, end_date: date) -> pd.DataFrame:
    """Haal glucosemetingen op tussen start_date en end_date (beide inclusief).

    Bedoeld voor meerdaagse rapportages (bijv. 90 dagen) waarbij het
    uren-schalende `fetch_entries` niet praktisch is. Zelfde kolommen als
    `fetch_entries`.
    """
    since = datetime.combine(start_date, time.min, tzinfo=timezone.utc)
    until = datetime.combine(end_date, time.max, tzinfo=timezone.utc)
    days = (end_date - start_date).days + 1
    count = min(days * 300, 100_000)  # ruime marge boven 5-min-interval
    return _entries_since(since, count, until)


def fetch_treatments_range(start_date: date, end_date: date) -> pd.DataFrame:
    """Haal behandelingen op tussen start_date en end_date (beide inclusief).

    Zelfde kolommen als `fetch_treatments`.
    """
    since = datetime.combine(start_date, time.min, tzinfo=timezone.utc)
    until = datetime.combine(end_date, time.max, tzinfo=timezone.utc)
    days = (end_date - start_date).days + 1
    count = min(days * 50, 50_000)
    return _treatments_since(since, count, until)