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

# Nightscout's /api/v1/entries en /api/v1/treatments negeren een groot
# `count`-verzoek: het antwoord wordt hoe dan ook stilzwijgend afgeknipt rond
# een paar honderd tot ~1150 documenten, ongeacht hoe ver de opgevraagde
# periode teruggaat (waargenomen: count=100000 op entries gaf slechts de
# meest recente ~1150 metingen terug — nog geen 4 dagen — terwijl er wel
# degelijk weken oudere data in Nightscout beschikbaar bleek te zijn). Eén
# verzoek is dus niet genoeg voor meerdaagse rapportages; _paginated_get()
# loopt terug in de tijd met `find[date][$lte]`-cursors tot de gevraagde
# `since`-grens bereikt is of de server minder dan een volle pagina teruggeeft
# (= geen oudere data meer beschikbaar, wat verschilt van "afgeknipt").
_PAGE_SIZE = 1000
_MAX_PAGES = 150  # 150 000 documenten; ruim boven wat een 90-dagenrapport nodig heeft


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 _paginated_get(path: str, since: datetime, until: datetime | None = None) -> list:
    """Haal alle documenten op tussen `since` en `until` (of tot nu), pagina voor pagina.

    Zie de toelichting bij _PAGE_SIZE hierboven voor waarom dit niet één
    verzoek met een groot `count` kan zijn (BUG: dat leverde bij een
    meerdaags rapport stilzwijgend alleen de laatste ~3-4 dagen op, ongeacht
    de gekozen periode). Elke pagina wordt opgevraagd met `find[date][$lte]`
    net onder de oudste timestamp van de vorige pagina, zodat elk document
    precies één keer terugkomt. Stopt zodra de oudste timestamp in een pagina
    op of voorbij `since` ligt, of zodra de server minder dan een volle
    pagina teruggeeft (dan is er geen oudere data meer, punt).
    """
    since_ms = int(since.timestamp() * 1000)
    cursor_ms = int(until.timestamp() * 1000) if until is not None else None

    results: list = []
    for _ in range(_MAX_PAGES):
        params = {"count": _PAGE_SIZE}
        if cursor_ms is not None:
            params["find[date][$lte]"] = cursor_ms
        page = _get(path, params)
        if not page:
            break
        results.extend(page)

        dates = [item["date"] for item in page if "date" in item]
        if not dates:
            break
        oldest_ms = min(dates)
        if oldest_ms <= since_ms or len(page) < _PAGE_SIZE:
            break
        cursor_ms = oldest_ms - 1
    return results


def _entries_since(since: datetime, until: datetime | None = None) -> pd.DataFrame:
    """Haal entries op sinds `since` (en optioneel tot `until`), lokaal gefilterd."""
    data = _paginated_get("/api/v1/entries/sgv.json", since, until)

    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, until: datetime | None = None) -> pd.DataFrame:
    """Haal treatments op sinds `since` (en optioneel tot `until`), lokaal gefilterd."""
    data = _paginated_get("/api/v1/treatments.json", since, until)

    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)
    return _entries_since(since)


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


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