From b362bbbf3da5cf62854fb33a54b61bdfaa21177f Mon Sep 17 00:00:00 2001 From: James Woglom Date: Wed, 1 Jul 2026 02:27:20 +0000 Subject: [PATCH] Rewire pump_events() onto the pump-logs endpoint with date windowing Fetch pre-parsed JSON events from get_pump_logs instead of decoding the retired reportsfacade binary stream: - page the requested range into inclusive windows of at most 28 days (the endpoint caps each request at ~4 weeks), covering short ranges and single days correctly - dedupe events that span windows by (sequenceGroup, sequenceNumber) - count but skip clockChanges (LID_TIME_CHANGED/LID_DATE_CHANGED are not consumed by any processor) - parse via Events_from_json Remove the dead reportsfacade pump_events_raw() and the now-unused pump_event_metadata()/PumpEventMetadata/LastUpload types. --- tconnectsync/api/tandemsource.py | 128 ++++++++++++---------------- tests/api/test_tandemsource.py | 140 +++++++++++++++++++++++++++++++ 2 files changed, 192 insertions(+), 76 deletions(-) diff --git a/tconnectsync/api/tandemsource.py b/tconnectsync/api/tandemsource.py index 1ea050f..01f7c33 100644 --- a/tconnectsync/api/tandemsource.py +++ b/tconnectsync/api/tandemsource.py @@ -9,7 +9,7 @@ import os import jwt import pickle -from typing import Any, Dict, Iterator, List, Optional +from typing import Any, Dict, Iterator, List, Optional, Tuple try: from typing import TypedDict except ImportError: # Python 3.7 @@ -25,42 +25,11 @@ from jwt.algorithms import RSAAlgorithm from ..util import timeago, cap_length from .common import parse_ymd_date, base_headers, base_session, ApiException, ApiLoginException from ..secret import CACHE_CREDENTIALS, CACHE_CREDENTIALS_PATH -from ..eventparser.generic import Events, decode_raw_events, EVENT_LEN +from ..eventparser.generic import Events_from_json logger = logging.getLogger(__name__) -class LastUpload(TypedDict, total=False): - """The 'lastUpload' object within a PumpEventMetadata entry. - - 'settings' is the raw pump settings blob consumed by - tconnectsync.domain.tandemsource.pump_settings.PumpSettings.from_dict(). - """ - settings: dict - - -class PumpEventMetadata(TypedDict): - """One entry returned by TandemSourceApi.pump_event_metadata(). - - Field names mirror the JSON returned by the - api/reports/reportsfacade/{pumperId}/pumpeventmetadata endpoint. - The *DateWithEvents fields are ISO-8601 datetime strings (parsed via - arrow.get()); tconnectDeviceId and serialNumber are numeric-looking - strings. - """ - tconnectDeviceId: str - serialNumber: str - modelNumber: str - minDateWithEvents: str - maxDateWithEvents: str - lastUpload: LastUpload - patientName: str - patientDateOfBirth: str - patientCareGiver: str - softwareVersion: str - partNumber: str - - class JwtClaims(TypedDict, total=False): """Decoded OIDC id_token claims stored on TandemSourceApi.jwtData. @@ -150,7 +119,7 @@ class BffPumper(TypedDict, total=False): class PumpMetadata(TypedDict, total=False): """Normalized per-pump metadata the sync code consumes, adapted from a BffPump (see TandemSourceApi.pump_metadata()). This is the stable - replacement for the old reportsfacade PumpEventMetadata shape. + replacement for the old reportsfacade pump-event-metadata shape. deviceId is the UUID assignmentId used as the pump-logs path segment (it replaces the old numeric tconnectDeviceId). date fields are ISO-8601 @@ -599,21 +568,12 @@ class TandemSourceApi: def pumper_info(self) -> Any: return self.get('api/pumpers/pumpers/%s' % (self.pumperId), {}) - """ - Returns metadata for pump events. Returns a list of dict's per-pump on the account. - [ - {'tconnectDeviceId', 'serialNumber', 'modelNumber', 'minDateWithEvents', 'maxDateWithEvents', 'lastUpload', 'patientName', 'patientDateOfBirth', 'patientCareGiver', 'softwareVersion', 'partNumber'}, - ] - """ - def pump_event_metadata(self) -> List[PumpEventMetadata]: - return self.get('api/reports/reportsfacade/%s/pumpeventmetadata' % (self.pumperId), {}) - def get_pumper(self) -> BffPumper: """Returns the pumper's profile plus the list of pumps on the account - (BffPumper.pumps) from the new BFF endpoint. Replaces - pump_event_metadata(): pumps[].assignmentId is the UUID device id used - by the pump-logs endpoint, and pumps[].settings.details carries the - pump settings blob.""" + (BffPumper.pumps) from the new BFF endpoint. Replaces the old + reportsfacade pump-event-metadata endpoint: pumps[].assignmentId is the + UUID device id used by the pump-logs endpoint, and + pumps[].settings.details carries the pump settings blob.""" return self.get('api/reports/bff/pumper/%s' % (self.pumperId), {}) @staticmethod @@ -636,7 +596,7 @@ class TandemSourceApi: def pump_metadata(self) -> List[PumpMetadata]: """Normalized device list adapted from get_pumper(). This is the - BFF-backed replacement for pump_event_metadata().""" + BFF-backed replacement for the old reportsfacade pump-event-metadata.""" pumper = self.get_pumper() return [self._bff_pump_to_metadata(p) for p in pumper.get('pumps', [])] @@ -662,39 +622,55 @@ class TandemSourceApi: }) return self.get('api/reports/bff/pump-logs/%s?%s' % (device_id, query), {}) - """ - Returns raw unparsed string for pump events. - tconnect_device_id is the device id from pump_metadata() (deviceId). - """ - def pump_events_raw(self, tconnect_device_id: str, min_date: Optional[str] = None, max_date: Optional[str] = None, event_ids_filter: Optional[List[int]] = DEFAULT_EVENT_IDS) -> str: - minDate = parse_ymd_date(min_date) - maxDate = parse_ymd_date(max_date) - logger.debug(f'pump_events_raw({tconnect_device_id}, {minDate}, {maxDate})') + # The pump-logs endpoint caps each request at roughly four weeks, so a + # longer range is paged in windows no larger than this. + PUMP_LOGS_WINDOW_DAYS = 28 - eventIdsFilter = '%2C'.join(map(str, event_ids_filter)) if event_ids_filter else None - return self.get('api/reports/reportsfacade/pumpevents/%s/%s?minDate=%s&maxDate=%s%s' % ( - self.pumperId, - tconnect_device_id, - minDate, - maxDate, - '&eventIds=%s' % eventIdsFilter if eventIdsFilter else '' - ), {}) + @classmethod + def _pump_log_windows(cls, min_date: Optional[str], max_date: Optional[str]) -> List[Tuple[str, str]]: + """Split the (min_date, max_date) range into inclusive date windows no + larger than PUMP_LOGS_WINDOW_DAYS. A None bound defaults to today (via + parse_ymd_date), so an unset range yields a single one-day window.""" + start = arrow.get(parse_ymd_date(min_date)) + end = arrow.get(parse_ymd_date(max_date)) + if end < start: + start, end = end, start + + windows = [] + cur = start + while cur <= end: + win_end = min(cur.shift(days=cls.PUMP_LOGS_WINDOW_DAYS - 1), end) + windows.append((cur.format('YYYY-MM-DD'), win_end.format('YYYY-MM-DD'))) + cur = win_end.shift(days=1) + return windows """ - Fetch and decode pump events using eventparser. - Default of fetch_all_events=False will filter to the same eventids used in the Tandem Source backend. - If fetch_all_events=True, then all event types from the history log will be returned. + Fetch and parse pump events from the pump-logs endpoint. + Default of fetch_all_event_types=False will filter to the same event ids used in the Tandem Source backend. + If fetch_all_event_types=True, then all event types from the history log will be returned. + tconnect_device_id is the UUID assignmentId from pump_metadata() (deviceId). """ def pump_events(self, tconnect_device_id: str, min_date: Optional[str] = None, max_date: Optional[str] = None, fetch_all_event_types: bool = False) -> Iterator: - pump_events_raw = self.pump_events_raw( - tconnect_device_id, - min_date, - max_date, - event_ids_filter=None if fetch_all_event_types else self.DEFAULT_EVENT_IDS - ) + event_ids_filter = None if fetch_all_event_types else self.DEFAULT_EVENT_IDS - pump_events_decoded = decode_raw_events(pump_events_raw) - logger.info(f"Read {len(pump_events_decoded)} bytes (est. {len(pump_events_decoded)/EVENT_LEN} events)") - return Events(pump_events_decoded) + # Page across date windows, deduplicating events that appear in more + # than one window by their (sequenceGroup, sequenceNumber) identity. + seen = set() + events = [] + clock_change_count = 0 + for window_start, window_end in self._pump_log_windows(min_date, max_date): + resp = self.get_pump_logs(tconnect_device_id, window_start, window_end, event_ids_filter) + clock_change_count += len(resp.get('clockChanges') or []) + for event in resp.get('events') or []: + key = (event.get('sequenceGroup'), event.get('sequenceNumber')) + if key in seen: + continue + seen.add(key) + events.append(event) + + # clockChanges (LID_TIME_CHANGED/LID_DATE_CHANGED) are not consumed by any + # processor, so they are counted for visibility but not parsed. + logger.info(f"Read {len(events)} events ({clock_change_count} clock changes skipped)") + return Events_from_json(events) diff --git a/tests/api/test_tandemsource.py b/tests/api/test_tandemsource.py index c93b4b2..4d2b5d0 100644 --- a/tests/api/test_tandemsource.py +++ b/tests/api/test_tandemsource.py @@ -1,11 +1,13 @@ #!/usr/bin/env python3 +import arrow import datetime import unittest import urllib.parse from unittest.mock import patch from tconnectsync.api.tandemsource import TandemSourceApi +from tconnectsync.eventparser import events as eventtypes # Representative GET api/reports/bff/pumper/{pumperId} response, mirroring the @@ -249,5 +251,143 @@ class TestGetPumpLogs(unittest.TestCase): self.assertEqual(qs["endDate"], ["%sT23:59:59Z" % today]) +def _ev(group, num, event_code=16, pump_date_time="2024-01-10T08:15:30", **props): + """Trimmed real-shape pump-log event; eventCode 16 parses to LidBgReadingTaken.""" + return { + "deviceAssignmentId": "1b493210-9336-4901-a329-a352775738c5", + "eventCode": event_code, + "sequenceGroup": group, + "sequenceNumber": num, + "pumpDateTime": pump_date_time, + "eventProperties": props or {"iob": 1.25, "bg": 112}, + "estimatedDateTime": pump_date_time + "Z", + } + + +class TestPumpLogWindows(unittest.TestCase): + """#10: the range is paged into inclusive windows no larger than 28 days.""" + maxDiff = None + + def test_single_day(self): + self.assertEqual(TandemSourceApi._pump_log_windows("2024-01-01", "2024-01-01"), + [("2024-01-01", "2024-01-01")]) + + def test_short_range_is_one_window(self): + # A span shorter than the window must still yield a covering window. + self.assertEqual(TandemSourceApi._pump_log_windows("2024-01-01", "2024-01-15"), + [("2024-01-01", "2024-01-15")]) + + def test_exactly_28_days_is_one_window(self): + self.assertEqual(TandemSourceApi._pump_log_windows("2024-01-01", "2024-01-28"), + [("2024-01-01", "2024-01-28")]) + + def test_29_days_splits(self): + self.assertEqual(TandemSourceApi._pump_log_windows("2024-01-01", "2024-01-29"), + [("2024-01-01", "2024-01-28"), ("2024-01-29", "2024-01-29")]) + + def test_long_range_windows_are_contiguous_and_bounded(self): + windows = TandemSourceApi._pump_log_windows("2024-01-01", "2024-03-01") + self.assertEqual(windows, [ + ("2024-01-01", "2024-01-28"), + ("2024-01-29", "2024-02-25"), + ("2024-02-26", "2024-03-01"), + ]) + # each window <= 28 days, and windows are contiguous (no gaps/overlaps) + for start, end in windows: + self.assertLessEqual((arrow.get(end) - arrow.get(start)).days, 27) + for (_, prev_end), (next_start, _) in zip(windows, windows[1:]): + self.assertEqual(arrow.get(next_start), arrow.get(prev_end).shift(days=1)) + + def test_reversed_dates_are_swapped(self): + self.assertEqual(TandemSourceApi._pump_log_windows("2024-03-01", "2024-01-01"), + TandemSourceApi._pump_log_windows("2024-01-01", "2024-03-01")) + + def test_none_dates_default_to_single_today_window(self): + windows = TandemSourceApi._pump_log_windows(None, None) + self.assertEqual(len(windows), 1) + self.assertEqual(windows[0][0], windows[0][1]) + + +class TestPumpEvents(unittest.TestCase): + """#16: pump_events pages get_pump_logs by window, dedupes, skips + clockChanges, and yields parsed event objects.""" + maxDiff = None + + def _api(self): + api = TandemSourceApi.__new__(TandemSourceApi) + api.pumperId = "PUMPER123" + return api + + def test_single_window_one_call_with_default_event_ids(self): + api = self._api() + resp = {"events": [_ev(0, 1)], "clockChanges": []} + with patch.object(TandemSourceApi, "get_pump_logs", return_value=resp) as m: + out = list(api.pump_events("dev-uuid", "2024-01-01", "2024-01-10")) + m.assert_called_once_with("dev-uuid", "2024-01-01", "2024-01-10", + TandemSourceApi.DEFAULT_EVENT_IDS) + self.assertEqual([type(e).__name__ for e in out], ["LidBgReadingTaken"]) + + def test_fetch_all_event_types_passes_none_filter(self): + api = self._api() + resp = {"events": [], "clockChanges": []} + with patch.object(TandemSourceApi, "get_pump_logs", return_value=resp) as m: + list(api.pump_events("dev", "2024-01-01", "2024-01-10", fetch_all_event_types=True)) + self.assertIsNone(m.call_args.args[3]) + + def test_multi_window_paging_boundaries(self): + api = self._api() + responses = [ + {"events": [_ev(0, 1)], "clockChanges": []}, + {"events": [_ev(0, 2)], "clockChanges": []}, + {"events": [_ev(0, 3)], "clockChanges": []}, + ] + with patch.object(TandemSourceApi, "get_pump_logs", side_effect=responses) as m: + out = list(api.pump_events("dev", "2024-01-01", "2024-03-01")) + windows = [(c.args[1], c.args[2]) for c in m.call_args_list] + self.assertEqual(windows, [ + ("2024-01-01", "2024-01-28"), + ("2024-01-29", "2024-02-25"), + ("2024-02-26", "2024-03-01"), + ]) + self.assertEqual([e.seqNum for e in out], [1, 2, 3]) + + def test_dedupes_across_windows_by_group_and_number(self): + api = self._api() + # Same (sequenceGroup, sequenceNumber) appears in two windows -> kept once. + responses = [ + {"events": [_ev(0, 100), _ev(0, 101)], "clockChanges": []}, + {"events": [_ev(0, 100), _ev(0, 102)], "clockChanges": []}, + ] + with patch.object(TandemSourceApi, "get_pump_logs", side_effect=responses): + out = list(api.pump_events("dev", "2024-01-01", "2024-02-15")) + self.assertEqual([e.seqNum for e in out], [100, 101, 102]) + + def test_same_number_different_group_not_deduped(self): + api = self._api() + responses = [ + {"events": [_ev(0, 100)], "clockChanges": []}, + {"events": [_ev(1, 100)], "clockChanges": []}, + ] + with patch.object(TandemSourceApi, "get_pump_logs", side_effect=responses): + out = list(api.pump_events("dev", "2024-01-01", "2024-02-15")) + self.assertEqual(len(out), 2) + + def test_clock_changes_are_skipped(self): + api = self._api() + resp = { + "events": [_ev(0, 1)], + "clockChanges": [_ev(0, 5, event_code=13), _ev(0, 6, event_code=14)], + } + with patch.object(TandemSourceApi, "get_pump_logs", return_value=resp): + out = list(api.pump_events("dev", "2024-01-01", "2024-01-10")) + self.assertEqual([e.eventId for e in out], [16]) + + def test_missing_events_key_is_tolerated(self): + api = self._api() + with patch.object(TandemSourceApi, "get_pump_logs", return_value={}): + out = list(api.pump_events("dev", "2024-01-01", "2024-01-10")) + self.assertEqual(out, []) + + if __name__ == "__main__": unittest.main()