willemjan
29d20ca441
_entries_since/_treatments_since requested a single page via `count`, but Nightscout's /api/v1/entries and /api/v1/treatments ignore a large `count` and cap the response around ~1150/~700 documents regardless of how far back the requested period goes. Every report page relying on fetch_entries_range/fetch_treatments_range (7/14/30/90-day selectors) was silently only getting the most recent ~3-4 days, no matter the selected period. Fixed by paginating with `find[date][$lte]` cursors in a new _paginated_get(), walking back until the requested `since` boundary is reached or the server returns a short/empty page (= genuinely no more data, confirmed by manually paginating against the real instance: entries go back 50+ days, treatments only ~4 days on this install). Also added reporting.completeness_warning(), shared by the interactive pages (via ui_common.render_completeness_warning) and the PDF export, that distinguishes "a structural gap at the start of the period" (data wasn't fetched/available) from "scattered gaps below the clinical AGP coverage threshold" (Danne et al. 2017) — the previous single generic "X% sensor coverage" message read as a CGM/sensor problem even when the real cause was this fetch bug.
407 lines
14 KiB
Python
407 lines
14 KiB
Python
from datetime import date
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from reporting import (
|
|
agp_percentile_bands,
|
|
bolus_stats,
|
|
completeness_warning,
|
|
day_of_week_pattern,
|
|
detect_events,
|
|
gmi_percent,
|
|
minutes_per_day,
|
|
overnight_daytime_split,
|
|
percent_cv,
|
|
sensor_active_percent,
|
|
site_interval_stats,
|
|
summarize_events,
|
|
tir_breakdown_5band,
|
|
)
|
|
|
|
|
|
def _series(pairs):
|
|
"""Bouw een glucose-DataFrame uit (tijd-string, mmol)-paren."""
|
|
timestamps, values = zip(*pairs, strict=True)
|
|
return pd.DataFrame({"timestamp": pd.to_datetime(list(timestamps)), "mmol": list(values)})
|
|
|
|
|
|
def test_gmi_percent_matches_bergenstal_formula():
|
|
# 3.31 + 0.4311 * 7.0 = 6.3277 -> 6.3
|
|
assert gmi_percent(7.0) == pytest.approx(6.3, abs=0.05)
|
|
|
|
|
|
def test_percent_cv_handles_empty_input():
|
|
empty = pd.DataFrame(columns=["timestamp", "mmol"])
|
|
assert percent_cv(empty) == 0.0
|
|
|
|
|
|
def test_percent_cv_computes_stdev_over_mean():
|
|
glucose = _series(
|
|
[
|
|
("2026-01-01 00:00", 6.0),
|
|
("2026-01-01 00:05", 8.0),
|
|
("2026-01-01 00:10", 6.0),
|
|
("2026-01-01 00:15", 8.0),
|
|
]
|
|
)
|
|
expected = glucose["mmol"].std() / glucose["mmol"].mean() * 100
|
|
assert percent_cv(glucose) == pytest.approx(round(expected, 1))
|
|
|
|
|
|
def test_tir_breakdown_5band_covers_all_bands_and_sums_to_100():
|
|
glucose = _series(
|
|
[
|
|
("2026-01-01 00:00", 2.5), # very_low
|
|
("2026-01-01 00:05", 3.5), # low
|
|
("2026-01-01 00:10", 6.0), # in_range
|
|
("2026-01-01 00:15", 11.0), # high
|
|
("2026-01-01 00:20", 14.0), # very_high
|
|
]
|
|
)
|
|
result = tir_breakdown_5band(glucose)
|
|
assert result == {
|
|
"very_low": 20.0,
|
|
"low": 20.0,
|
|
"in_range": 20.0,
|
|
"high": 20.0,
|
|
"very_high": 20.0,
|
|
}
|
|
assert sum(result.values()) == pytest.approx(100.0)
|
|
|
|
|
|
def test_tir_breakdown_5band_handles_empty_input():
|
|
empty = pd.DataFrame(columns=["timestamp", "mmol"])
|
|
result = tir_breakdown_5band(empty)
|
|
assert result == {"very_low": 0.0, "low": 0.0, "in_range": 0.0, "high": 0.0, "very_high": 0.0}
|
|
|
|
|
|
def test_minutes_per_day_converts_percentage():
|
|
assert minutes_per_day(50.0) == 720.0
|
|
assert minutes_per_day(100.0) == 1440.0
|
|
|
|
|
|
def test_sensor_active_percent_full_coverage():
|
|
# 12 metingen om de 5 minuten = precies 1 uur dekking
|
|
timestamps = pd.date_range("2026-01-01 00:00", periods=12, freq="5min")
|
|
glucose = pd.DataFrame({"timestamp": timestamps, "mmol": [6.0] * 12})
|
|
result = sensor_active_percent(glucose, "2026-01-01 00:00", "2026-01-01 01:00")
|
|
assert result == pytest.approx(100.0, abs=1.0)
|
|
|
|
|
|
def test_sensor_active_percent_handles_gaps():
|
|
timestamps = pd.date_range("2026-01-01 00:00", periods=6, freq="5min") # maar 30 min dekking
|
|
glucose = pd.DataFrame({"timestamp": timestamps, "mmol": [6.0] * 6})
|
|
result = sensor_active_percent(glucose, "2026-01-01 00:00", "2026-01-01 01:00")
|
|
assert result == pytest.approx(50.0, abs=1.0)
|
|
|
|
|
|
def test_completeness_warning_none_for_empty_glucose():
|
|
empty = pd.DataFrame(columns=["timestamp", "mmol"])
|
|
assert completeness_warning(empty, "2026-01-01", "2026-01-14") == "Geen glucosedata gevonden in deze periode."
|
|
|
|
|
|
def test_completeness_warning_none_when_period_is_fully_covered():
|
|
# 14 volle dagen aan 5-minuten-metingen, precies de gekozen periode
|
|
timestamps = pd.date_range("2026-01-01 00:00", "2026-01-14 23:55", freq="5min")
|
|
glucose = pd.DataFrame({"timestamp": timestamps, "mmol": [6.0] * len(timestamps)})
|
|
assert completeness_warning(glucose, date(2026, 1, 1), date(2026, 1, 14)) is None
|
|
|
|
|
|
def test_completeness_warning_flags_missing_days_at_start_of_period():
|
|
# gevraagd: 1 t/m 14 januari, maar data begint pas op 11 januari (de
|
|
# exacte bug: alleen de laatste ~4 dagen van een langere periode zijn
|
|
# opgehaald)
|
|
timestamps = pd.date_range("2026-01-11 00:00", "2026-01-14 23:55", freq="5min")
|
|
glucose = pd.DataFrame({"timestamp": timestamps, "mmol": [6.0] * len(timestamps)})
|
|
message = completeness_warning(glucose, date(2026, 1, 1), date(2026, 1, 14))
|
|
assert message is not None
|
|
assert "11-01-2026" in message
|
|
assert "10 dag" in message # 1 t/m 10 januari ontbreken volledig = 10 dagen
|
|
assert "sensorhiaten" in message
|
|
|
|
|
|
def test_completeness_warning_flags_low_coverage_without_start_gap():
|
|
# data start op tijd, maar met veel verspreide hiaten -> lage dekking
|
|
# zonder dat er een aaneengesloten gat aan het begin is
|
|
timestamps = pd.date_range("2026-01-01 00:00", periods=20, freq="5min")
|
|
glucose = pd.DataFrame({"timestamp": timestamps, "mmol": [6.0] * 20})
|
|
message = completeness_warning(glucose, date(2026, 1, 1), date(2026, 1, 14))
|
|
assert message is not None
|
|
assert "sensordekking" in message
|
|
assert "sensorhiaten" not in message # dit is de generieke, niet de "ontbreekt volledig"-boodschap
|
|
|
|
|
|
def test_detect_events_ignores_dip_shorter_than_minimum_duration():
|
|
# dip van 10 minuten (< 15 min drempel) mag niet meetellen
|
|
glucose = _series(
|
|
[
|
|
("2026-01-01 00:00", 6.0),
|
|
("2026-01-01 00:05", 3.5),
|
|
("2026-01-01 00:10", 3.4),
|
|
("2026-01-01 00:15", 6.0),
|
|
("2026-01-01 00:20", 6.1),
|
|
("2026-01-01 00:25", 6.2),
|
|
]
|
|
)
|
|
events = detect_events(glucose, threshold=3.9, direction="below")
|
|
assert events.empty
|
|
|
|
|
|
def test_detect_events_counts_dip_of_exactly_20_minutes():
|
|
glucose = _series(
|
|
[
|
|
("2026-01-01 00:00", 6.0),
|
|
("2026-01-01 00:05", 3.5),
|
|
("2026-01-01 00:10", 3.4),
|
|
("2026-01-01 00:15", 3.3),
|
|
("2026-01-01 00:20", 3.2),
|
|
("2026-01-01 00:25", 6.0),
|
|
("2026-01-01 00:30", 6.1),
|
|
("2026-01-01 00:35", 6.2),
|
|
("2026-01-01 00:40", 6.3),
|
|
]
|
|
)
|
|
events = detect_events(glucose, threshold=3.9, direction="below")
|
|
assert len(events) == 1
|
|
assert events.iloc[0]["duration_minutes"] == pytest.approx(20.0)
|
|
|
|
|
|
def test_detect_events_merges_across_brief_recovery():
|
|
# twee dips gescheiden door een kort (5 min) herstel moeten één
|
|
# doorlopend event vormen volgens de persistentieregel
|
|
glucose = _series(
|
|
[
|
|
("2026-01-01 00:00", 6.0),
|
|
("2026-01-01 00:05", 3.5),
|
|
("2026-01-01 00:10", 3.4),
|
|
("2026-01-01 00:15", 6.0), # kort herstel, 5 min
|
|
("2026-01-01 00:20", 3.3),
|
|
("2026-01-01 00:25", 3.2),
|
|
("2026-01-01 00:30", 3.1),
|
|
("2026-01-01 00:35", 6.0),
|
|
("2026-01-01 00:40", 6.1),
|
|
("2026-01-01 00:45", 6.2),
|
|
("2026-01-01 00:50", 6.3), # bevestigd herstel, 15 min
|
|
]
|
|
)
|
|
events = detect_events(glucose, threshold=3.9, direction="below")
|
|
assert len(events) == 1
|
|
assert events.iloc[0]["start"] == pd.Timestamp("2026-01-01 00:05")
|
|
assert events.iloc[0]["end"] == pd.Timestamp("2026-01-01 00:35")
|
|
assert events.iloc[0]["duration_minutes"] == pytest.approx(30.0)
|
|
|
|
|
|
def test_detect_events_does_not_inflate_duration_on_trailing_brief_blip():
|
|
# regressie: een out-periode van 10 min gevolgd door een niet-bevestigd
|
|
# (kort, en aan het eind van de data) herstel mag niet meetellen
|
|
glucose = _series(
|
|
[
|
|
("2026-01-01 00:00", 6.0),
|
|
("2026-01-01 00:05", 3.5),
|
|
("2026-01-01 00:10", 3.4),
|
|
("2026-01-01 00:15", 6.0),
|
|
("2026-01-01 00:20", 6.1),
|
|
("2026-01-01 00:25", 6.2),
|
|
]
|
|
)
|
|
events = detect_events(glucose, threshold=3.9, direction="below")
|
|
assert events.empty
|
|
|
|
|
|
def test_detect_events_supports_hyperglycemia_direction():
|
|
glucose = _series(
|
|
[
|
|
("2026-01-01 00:00", 6.0),
|
|
("2026-01-01 00:05", 12.0),
|
|
("2026-01-01 00:10", 12.5),
|
|
("2026-01-01 00:15", 13.0),
|
|
("2026-01-01 00:20", 12.8),
|
|
("2026-01-01 00:25", 6.0),
|
|
("2026-01-01 00:30", 6.1),
|
|
("2026-01-01 00:35", 6.2),
|
|
("2026-01-01 00:40", 6.3),
|
|
]
|
|
)
|
|
events = detect_events(glucose, threshold=10.0, direction="above")
|
|
assert len(events) == 1
|
|
assert events.iloc[0]["duration_minutes"] == pytest.approx(20.0)
|
|
|
|
|
|
def test_detect_events_rejects_invalid_direction():
|
|
glucose = _series([("2026-01-01 00:00", 6.0)])
|
|
with pytest.raises(ValueError, match="direction"):
|
|
detect_events(glucose, threshold=3.9, direction="sideways")
|
|
|
|
|
|
def test_detect_events_handles_empty_input():
|
|
empty = pd.DataFrame(columns=["timestamp", "mmol"])
|
|
events = detect_events(empty, threshold=3.9, direction="below")
|
|
assert events.empty
|
|
assert list(events.columns) == ["start", "end", "duration_minutes"]
|
|
|
|
|
|
def test_summarize_events_computes_expected_stats():
|
|
events = pd.DataFrame(
|
|
{
|
|
"start": pd.to_datetime(["2026-01-01 00:00", "2026-01-03 00:00"]),
|
|
"end": pd.to_datetime(["2026-01-01 00:20", "2026-01-03 01:00"]),
|
|
"duration_minutes": [20.0, 60.0],
|
|
}
|
|
)
|
|
result = summarize_events(events, period_days=7)
|
|
assert result["count"] == 2
|
|
assert result["avg_duration_minutes"] == pytest.approx(40.0)
|
|
assert result["longest_duration_minutes"] == pytest.approx(60.0)
|
|
assert result["total_minutes"] == pytest.approx(80.0)
|
|
assert result["events_per_week"] == pytest.approx(2.0)
|
|
|
|
|
|
def test_agp_percentile_bands_groups_by_time_of_day_across_days():
|
|
# zelfde tijdstip (08:00) op twee verschillende dagen, met bekende waarden
|
|
glucose = _series(
|
|
[
|
|
("2026-01-01 08:00", 6.0),
|
|
("2026-01-02 08:00", 8.0),
|
|
("2026-01-01 20:00", 10.0),
|
|
("2026-01-02 20:00", 10.0),
|
|
]
|
|
)
|
|
result = agp_percentile_bands(glucose, bin_minutes=30)
|
|
assert list(result.columns) == ["minute_of_day", "p5", "p25", "p50", "p75", "p95"]
|
|
|
|
morning_bin = 8 * 60
|
|
row = result[result["minute_of_day"] == morning_bin].iloc[0]
|
|
assert row["p50"] == pytest.approx(7.0) # mediaan van 6.0 en 8.0
|
|
|
|
evening_bin = 20 * 60
|
|
evening_row = result[result["minute_of_day"] == evening_bin].iloc[0]
|
|
assert evening_row["p50"] == pytest.approx(10.0)
|
|
|
|
|
|
def test_agp_percentile_bands_handles_empty_input():
|
|
empty = pd.DataFrame(columns=["timestamp", "mmol"])
|
|
result = agp_percentile_bands(empty)
|
|
assert result.empty
|
|
assert list(result.columns) == ["minute_of_day", "p5", "p25", "p50", "p75", "p95"]
|
|
|
|
|
|
def test_overnight_daytime_split_separates_by_hour():
|
|
glucose = _series(
|
|
[
|
|
("2026-01-01 02:00", 3.0), # nacht, laag
|
|
("2026-01-01 03:00", 3.0), # nacht, laag
|
|
("2026-01-01 12:00", 6.0), # dag, in bereik
|
|
("2026-01-01 13:00", 6.0), # dag, in bereik
|
|
]
|
|
)
|
|
result = overnight_daytime_split(glucose, overnight_start=0, overnight_end=6)
|
|
assert result["overnight"]["low"] == pytest.approx(100.0)
|
|
assert result["daytime"]["in_range"] == pytest.approx(100.0)
|
|
|
|
|
|
def test_overnight_daytime_split_handles_empty_input():
|
|
empty = pd.DataFrame(columns=["timestamp", "mmol"])
|
|
result = overnight_daytime_split(empty)
|
|
assert result["overnight"] == result["daytime"]
|
|
|
|
|
|
def test_day_of_week_pattern_computes_per_weekday_stats():
|
|
# 2026-01-05 is een maandag
|
|
glucose = _series(
|
|
[
|
|
("2026-01-05 08:00", 6.0),
|
|
("2026-01-05 09:00", 8.0),
|
|
("2026-01-12 08:00", 7.0), # ook een maandag
|
|
]
|
|
)
|
|
result = day_of_week_pattern(glucose)
|
|
monday = result[result["weekday"] == "Maandag"].iloc[0]
|
|
assert monday["count"] == 3
|
|
assert monday["mean_mmol"] == pytest.approx(7.0)
|
|
|
|
tuesday = result[result["weekday"] == "Dinsdag"].iloc[0]
|
|
assert tuesday["count"] == 0
|
|
assert pd.isna(tuesday["mean_mmol"])
|
|
|
|
|
|
def test_day_of_week_pattern_handles_empty_input():
|
|
empty = pd.DataFrame(columns=["timestamp", "mmol"])
|
|
result = day_of_week_pattern(empty)
|
|
assert result.empty
|
|
assert list(result.columns) == ["weekday", "mean_mmol", "in_range_pct", "count"]
|
|
|
|
|
|
def test_bolus_stats_splits_by_event_type():
|
|
treatments = pd.DataFrame(
|
|
{
|
|
"timestamp": pd.to_datetime(["2026-01-01 08:00", "2026-01-01 12:00", "2026-01-01 20:00"]),
|
|
"eventType": ["Meal Bolus", "Correction Bolus", "Site Change"],
|
|
"insulin": [4.0, 1.5, None],
|
|
"carbs": [45.0, None, None],
|
|
}
|
|
)
|
|
result = bolus_stats(treatments)
|
|
assert result["total_units"] == pytest.approx(5.5)
|
|
assert result["count"] == 2
|
|
assert result["by_type"] == {"Meal Bolus": 4.0, "Correction Bolus": 1.5}
|
|
|
|
|
|
def test_bolus_stats_handles_empty_input():
|
|
empty = pd.DataFrame(columns=["timestamp", "eventType", "insulin", "carbs"])
|
|
result = bolus_stats(empty)
|
|
assert result == {"total_units": 0.0, "count": 0, "by_type": {}}
|
|
|
|
|
|
def test_site_interval_stats_splits_by_site_change():
|
|
glucose = _series(
|
|
[
|
|
("2026-01-01 00:00", 6.0),
|
|
("2026-01-05 00:00", 6.0),
|
|
("2026-01-10 00:00", 8.0),
|
|
]
|
|
)
|
|
treatments = pd.DataFrame(
|
|
{
|
|
"timestamp": pd.to_datetime(["2026-01-02 00:00", "2026-01-06 00:00"]),
|
|
"eventType": ["Meal Bolus", "Meal Bolus"],
|
|
"insulin": [3.0, 4.0],
|
|
"carbs": [30.0, 40.0],
|
|
}
|
|
)
|
|
site_changes = pd.DataFrame(
|
|
{
|
|
"timestamp": pd.to_datetime(["2026-01-01 00:00", "2026-01-06 00:00"]),
|
|
"eventType": ["Site Change", "Site Change"],
|
|
"insulin": [None, None],
|
|
"carbs": [None, None],
|
|
}
|
|
)
|
|
|
|
result = site_interval_stats(glucose, treatments, site_changes)
|
|
assert len(result) == 2
|
|
assert result.iloc[0]["duration_days"] == pytest.approx(5.0)
|
|
assert result.iloc[0]["total_bolus_units"] == pytest.approx(3.0)
|
|
# de bolus op 6 jan valt binnen het tweede interval ([6 jan, 10 jan))
|
|
assert result.iloc[1]["total_bolus_units"] == pytest.approx(4.0)
|
|
|
|
|
|
def test_site_interval_stats_handles_no_site_changes():
|
|
glucose = _series([("2026-01-01 00:00", 6.0)])
|
|
treatments = pd.DataFrame(columns=["timestamp", "eventType", "insulin", "carbs"])
|
|
empty_site_changes = pd.DataFrame(columns=["timestamp", "eventType", "insulin", "carbs"])
|
|
result = site_interval_stats(glucose, treatments, empty_site_changes)
|
|
assert result.empty
|
|
|
|
|
|
def test_summarize_events_handles_empty_input():
|
|
empty = pd.DataFrame(columns=["start", "end", "duration_minutes"])
|
|
result = summarize_events(empty, period_days=14)
|
|
assert result == {
|
|
"count": 0,
|
|
"avg_duration_minutes": 0.0,
|
|
"longest_duration_minutes": 0.0,
|
|
"total_minutes": 0.0,
|
|
"events_per_week": 0.0,
|
|
}
|