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

@@ -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)
 
 

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()