From f5ef34174f3dd46a11b6eb9f96298ec366c54f5a Mon Sep 17 00:00:00 2001 From: James Woglom Date: Sun, 24 Oct 2021 00:36:48 -0400 Subject: [PATCH] in-progress: support BOLUS_BG and CGM features. BOLUS_BG will add BG readings which are associated with boluses on the pump into the Nightscout treatment object. It will determine whether the BG reading was automatically filled via the Dexcom connection on the pump or was manually entered by seeing if the BG reading matches the current CGM reading as known to the pump at that time. Support for this is nearly complete. CGM will add Dexcom CGM readings from the pump to Nightscout as SGV (sensor glucose value) entries. This should only be used in a situation where xDrip is not used and the pump connection to the CGM will be the only source of CGM data to Nightscout. This requires additional testing before it should be considered ready. Both options are hidden behind the ENABLE_TESTING_MODES=true environment variable. Run tconnectsync with this environment variable and BOLUS_BG and CGM will be usable with the --features argument. --- tconnectsync/features.py | 15 ++++- tconnectsync/nightscout.py | 14 ++++- tconnectsync/parser/nightscout.py | 37 +++++++++++- tconnectsync/parser/tconnect.py | 9 +++ tconnectsync/process.py | 17 +++++- tconnectsync/secret.py | 6 +- tconnectsync/sync/bolus.py | 51 ++++++++++++++--- tconnectsync/sync/cgm.py | 69 +++++++++++++++++++++++ tests/parser/test_nightscout.py | 79 +++++++++++++++++++++++++- tests/parser/test_tconnect.py | 75 ++++++++++++++++++++++++ tests/sync/test_bolus.py | 94 ++++++++++++++++++++++++++++++- tests/sync/test_cgm.py | 51 +++++++++++++++++ 12 files changed, 493 insertions(+), 24 deletions(-) create mode 100644 tconnectsync/sync/cgm.py create mode 100644 tests/sync/test_cgm.py diff --git a/tconnectsync/features.py b/tconnectsync/features.py index 1e3ad82..63cbcaf 100644 --- a/tconnectsync/features.py +++ b/tconnectsync/features.py @@ -1,8 +1,11 @@ -"""Supported synchronization features.""" +from .secret import ENABLE_TESTING_MODES +"""Supported synchronization features.""" BASAL = "BASAL" BOLUS = "BOLUS" IOB = "IOB" +BOLUS_BG = "BOLUS_BG" +CGM = "CGM" DEFAULT_FEATURES = [ BASAL, @@ -14,4 +17,12 @@ BASAL, BOLUS, IOB -] \ No newline at end of file +] + + +# These modes are not yet ready for wide use. +if ENABLE_TESTING_MODES: + ALL_FEATURES += [ + BOLUS_BG, + CGM + ] \ No newline at end of file diff --git a/tconnectsync/nightscout.py b/tconnectsync/nightscout.py index ff071e5..16d7d53 100644 --- a/tconnectsync/nightscout.py +++ b/tconnectsync/nightscout.py @@ -53,7 +53,19 @@ def last_uploaded_entry(self, eventType): 'api-secret': hashlib.sha1(self.secret.encode()).hexdigest() }) if latest.status_code != 200: - raise ApiException(latest.status_code, "Nightscout treatments response: %s" % latest.text) + raise ApiException(latest.status_code, "Nightscout last_uploaded_entry response: %s" % latest.text) + + j = latest.json() + if j and len(j) > 0: + return j[0] + return None + + def last_uploaded_bg_entry(self): + latest = requests.get(urljoin(self.url, 'api/v1/entries.json?count=1&find[device]=' + urllib.parse.quote(ENTERED_BY) + '&ts=' + str(time.time())), headers={ + 'api-secret': hashlib.sha1(self.secret.encode()).hexdigest() + }) + if latest.status_code != 200: + raise ApiException(latest.status_code, "Nightscout last_uploaded_bg_entry response: %s" % latest.text) j = latest.json() if j and len(j) > 0: diff --git a/tconnectsync/parser/nightscout.py b/tconnectsync/parser/nightscout.py index 481dbc7..e7eefad 100644 --- a/tconnectsync/parser/nightscout.py +++ b/tconnectsync/parser/nightscout.py @@ -1,8 +1,11 @@ +import arrow + ENTERED_BY = "Pump (tconnectsync)" BASAL_EVENTTYPE = "Temp Basal" BOLUS_EVENTTYPE = "Combo Bolus" IOB_ACTIVITYTYPE = "tconnect_iob" + """ Conversion methods for parsing data into Nightscout objects. """ @@ -21,9 +24,14 @@ def basal(value, duration_mins, created_at, reason=""): "enteredBy": ENTERED_BY } + # Note that Nightscout is not consistent and uses "Sensor"/"Finger" + # for treatment objects, unlike "sgv"/"mbg" for entries + SENSOR = "Sensor" + FINGER = "Finger" + @staticmethod - def bolus(bolus, carbs, created_at, notes=""): - return { + def bolus(bolus, carbs, created_at, notes="", bg="", bg_type=""): + data = { "eventType": BOLUS_EVENTTYPE, "created_at": created_at, "carbs": int(carbs), @@ -31,6 +39,15 @@ def bolus(bolus, carbs, created_at, notes=""): "notes": notes, "enteredBy": ENTERED_BY, } + if bg: + if bg_type not in (NightscoutEntry.SENSOR, NightscoutEntry.FINGER): + raise InvalidBolusTypeException + + data.update({ + "glucose": str(bg), + "glucoseType": bg_type + }) + return data @staticmethod def iob(iob, created_at): @@ -39,4 +56,18 @@ def iob(iob, created_at): "iob": float(iob), "created_at": created_at, "enteredBy": ENTERED_BY - } \ No newline at end of file + } + + @staticmethod + def entry(sgv, created_at): + return { + "type": "sgv", + "sgv": int(sgv), + "date": int(1000 * arrow.get(created_at).timestamp()), + "dateString": arrow.get(created_at).strftime('%Y-%m-%dT%H:%M:%S%z'), + "device": ENTERED_BY, + # delta, direction are undefined + } + +class InvalidBolusTypeException(RuntimeError): + pass \ No newline at end of file diff --git a/tconnectsync/parser/tconnect.py b/tconnectsync/parser/tconnect.py index 7ab4cda..2fe59f5 100644 --- a/tconnectsync/parser/tconnect.py +++ b/tconnectsync/parser/tconnect.py @@ -100,8 +100,17 @@ def is_complete(s): "insulin": data["InsulinDelivered"], "requested_insulin": data["ActualTotalBolusRequested"], "carbs": data["CarbSize"], + "bg": data["BG"], # Note: can be empty string for automatic Control-IQ boluses "user_override": data["UserOverride"], "extended_bolus": "1" if extended_bolus else "", "bolex_completion_time": TConnectEntry._datetime_parse(data["BolexCompletionDateTime"]).format() if complete and extended_bolus else None, "bolex_start_time": TConnectEntry._datetime_parse(data["BolexStartDateTime"]).format() if complete and extended_bolus else None, + } + + @staticmethod + def parse_reading_entry(data): + return { + "time": TConnectEntry._datetime_parse(data["EventDateTime"]).format(), + "bg": data["Readings (CGM / BGM)"], + "type": data["Description"] } \ No newline at end of file diff --git a/tconnectsync/process.py b/tconnectsync/process.py index b526aff..fc4604b 100644 --- a/tconnectsync/process.py +++ b/tconnectsync/process.py @@ -18,8 +18,12 @@ process_iob_events, ns_write_iob_events ) +from .sync.cgm import ( + process_cgm_events, + ns_write_cgm_events +) from .parser.tconnect import TConnectEntry -from .features import BASAL, BOLUS, IOB, DEFAULT_FEATURES +from .features import BASAL, BOLUS, IOB, BOLUS_BG, CGM, DEFAULT_FEATURES logger = logging.getLogger(__name__) @@ -60,6 +64,15 @@ def process_time_range(tconnect, nightscout, time_start, time_end, pretend, feat added = 0 + cgmData = None + if CGM in features or BOLUS_BG in features: + logger.debug("Processing CGM events") + cgmData = process_cgm_events(readingData) + + if CGM in features: + logger.debug("Writing CGM events") + added += ns_write_cgm_events(nightscout, cgmData, pretend) + if BASAL in features: basalEvents = process_ciq_basal_events(ciqTherapyTimelineData) if csvBasalData: @@ -72,7 +85,7 @@ def process_time_range(tconnect, nightscout, time_start, time_end, pretend, feat if BOLUS in features: bolusEvents = process_bolus_events(bolusData) - added += ns_write_bolus_events(nightscout, bolusEvents, pretend=pretend) + added += ns_write_bolus_events(nightscout, bolusEvents, pretend=pretend, include_bg=(BOLUS_BG in features)) if IOB in features: iobEvents = process_iob_events(iobData) diff --git a/tconnectsync/secret.py b/tconnectsync/secret.py index 51e86e1..b8f5376 100644 --- a/tconnectsync/secret.py +++ b/tconnectsync/secret.py @@ -36,11 +36,7 @@ def get_bool(name, default): AUTOUPDATE_FAILURE_MINUTES = get_number('AUTOUPDATE_FAILURE_MINUTES', '180') # 3 hours AUTOUPDATE_RESTART_ON_FAILURE = get_bool('AUTOUPDATE_RESTART_ON_FAILURE', 'false') -_config = ['TCONNECT_EMAIL', 'TCONNECT_PASSWORD', 'PUMP_SERIAL_NUMBER', - 'NS_URL', 'NS_SECRET', 'TIMEZONE_NAME', - 'AUTOUPDATE_DEFAULT_SLEEP_SECONDS', 'AUTOUPDATE_MAX_SLEEP_SECONDS', - 'AUTOUPDATE_USE_FIXED_SLEEP', 'AUTOUPDATE_FAILURE_MINUTES', - 'AUTOUPDATE_RESTART_ON_FAILURE'] +ENABLE_TESTING_MODES = get_bool('ENABLE_TESTING_MODES', 'false') if __name__ == '__main__': for k in locals(): diff --git a/tconnectsync/sync/bolus.py b/tconnectsync/sync/bolus.py index 075f0d7..ff67f4f 100644 --- a/tconnectsync/sync/bolus.py +++ b/tconnectsync/sync/bolus.py @@ -1,6 +1,8 @@ import arrow import logging +from tconnectsync.sync.cgm import find_event_at + from ..parser.nightscout import ( BOLUS_EVENTTYPE, NightscoutEntry @@ -12,7 +14,7 @@ """ Given bolus data input from the therapy timeline CSV, converts it into a digestable format. """ -def process_bolus_events(bolusdata): +def process_bolus_events(bolusdata, cgmEvents=None): bolusEvents = [] for b in bolusdata: @@ -24,16 +26,37 @@ def process_bolus_events(bolusdata): else: logger.warning("Skipping non-completed bolus data (was a bolus in progress?): %s parsed: %s" % (b, parsed)) continue + + if parsed["bg"] and cgmEvents: + requested_at = parsed["request_time"] if not parsed["extended_bolus"] else parsed["bolex_start_time"] + parsed["bg_type"] = guess_bolus_bg_type(parsed["bg"], requested_at, cgmEvents) + bolusEvents.append(parsed) - bolusEvents.sort(key=lambda event: arrow.get(event["completion_time"] if not event["extended_bolus"] else event["bolex_start_time"])) + bolusEvents.sort(key=lambda event: arrow.get(event["request_time"] if not event["extended_bolus"] else event["bolex_start_time"])) return bolusEvents +""" +Determine whether the given BG specified in the bolus is identical to the +most recent CGM reading at that time. If it is, return SENSOR. +Otherwise, return FINGER. +""" +def guess_bolus_bg_type(bg, created_at, cgmEvents): + if not cgmEvents: + return NightscoutEntry.FINGER + + event = find_event_at(cgmEvents, created_at) + if event and str(event["bg"]) == str(bg): + return NightscoutEntry.SENSOR + + return NightscoutEntry.FINGER + + """ Given processed bolus data, adds bolus events to Nightscout. """ -def ns_write_bolus_events(nightscout, bolusEvents, pretend=False): +def ns_write_bolus_events(nightscout, bolusEvents, pretend=False, include_bg=False, reading_events=None): logger.debug("ns_write_bolus_events: querying for last uploaded entry") last_upload = nightscout.last_uploaded_entry(BOLUS_EVENTTYPE) last_upload_time = None @@ -49,12 +72,22 @@ def ns_write_bolus_events(nightscout, bolusEvents, pretend=False): logger.info("Skipping basal event before last upload time: %s" % event) continue - entry = NightscoutEntry.bolus( - bolus=event["insulin"], - carbs=event["carbs"], - created_at=created_at, - notes="{}{}{}".format(event["description"], " (Override)" if event["user_override"] == "1" else "", " (Extended)" if event["extended_bolus"] == "1" else "") - ) + if include_bg and event["bg"]: + entry = NightscoutEntry.bolus( + bolus=event["insulin"], + carbs=event["carbs"], + created_at=created_at, + notes="{}{}{}".format(event["description"], " (Override)" if event["user_override"] == "1" else "", " (Extended)" if event["extended_bolus"] == "1" else ""), + bg=event["bg"], + bg_type=event["bg_type"] + ) + else: + entry = NightscoutEntry.bolus( + bolus=event["insulin"], + carbs=event["carbs"], + created_at=created_at, + notes="{}{}{}".format(event["description"], " (Override)" if event["user_override"] == "1" else "", " (Extended)" if event["extended_bolus"] == "1" else "") + ) add_count += 1 diff --git a/tconnectsync/sync/cgm.py b/tconnectsync/sync/cgm.py new file mode 100644 index 0000000..081666c --- /dev/null +++ b/tconnectsync/sync/cgm.py @@ -0,0 +1,69 @@ +import json +import arrow +import logging + +from ..parser.tconnect import TConnectEntry +from ..parser.nightscout import NightscoutEntry + +logger = logging.getLogger(__name__) + +def process_cgm_events(readingData): + data = [] + for r in readingData: + data.append(TConnectEntry.parse_reading_entry(r)) + + return data + +""" +Given reading data and a time, finds the BG reading event which would have +been the current one at that time. e.g., it looks before the given time, +not after. +This is a heuristic for checking whether the BG component of a bolus was +manually entered or inferred based on the pump's CGM. +""" +def find_event_at(cgmEvents, find_time): + find_t = arrow.get(find_time) + events = list(map(lambda x: (arrow.get(x["time"]), x), cgmEvents)) + events.sort() + + closestReading = None + for t, r in events: + if t > find_t: + break + closestReading = r + + + return closestReading + + +""" +Given processed CGM data, adds reading entries to Nightscout. +""" +def ns_write_cgm_events(nightscout, cgmEvents, pretend=False): + logger.debug("ns_write_cgm_events: querying for last uploaded entry") + last_upload = nightscout.last_uploaded_bg_entry() + last_upload_time = None + if last_upload: + last_upload_time = arrow.get(last_upload["dateString"]) + logger.info("Last Nightscout CGM upload: %s" % last_upload_time) + + add_count = 0 + for event in cgmEvents: + created_at = event["time"] + if last_upload_time and arrow.get(created_at) <= last_upload_time: + if pretend: + logger.info("Skipping CGM event before last upload time: %s" % event) + continue + + entry = NightscoutEntry.entry( + sgv=event["bg"], + created_at=created_at + ) + + add_count += 1 + + logger.info(" Processing cgm reading: %s entry: %s" % (event, entry)) + if not pretend: + nightscout.upload_entry(entry, entity='entries') + + return add_count diff --git a/tests/parser/test_nightscout.py b/tests/parser/test_nightscout.py index 0f2da95..6a36d95 100644 --- a/tests/parser/test_nightscout.py +++ b/tests/parser/test_nightscout.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 import unittest -from tconnectsync.parser.nightscout import NightscoutEntry +from tconnectsync.parser.nightscout import NightscoutEntry, InvalidBolusTypeException class TestNightscoutEntry(unittest.TestCase): def test_basal(self): @@ -73,6 +73,69 @@ def test_bolus(self): } ) + def test_bolus_with_bg(self): + self.assertEqual( + NightscoutEntry.bolus( + bolus=7.5, + carbs=45, + created_at="2021-03-16 00:25:21-04:00", + bg="123", + bg_type=NightscoutEntry.SENSOR), + { + "eventType": "Combo Bolus", + "created_at": "2021-03-16 00:25:21-04:00", + "carbs": 45, + "insulin": 7.5, + "notes": "", + "enteredBy": "Pump (tconnectsync)", + "glucose": "123", + "glucoseType": "Sensor" + } + ) + + self.assertEqual( + NightscoutEntry.bolus( + bolus=0.5, + carbs=5, + created_at="2021-03-16 12:25:21-04:00", + bg="150", + bg_type=NightscoutEntry.FINGER), + { + "eventType": "Combo Bolus", + "created_at": "2021-03-16 12:25:21-04:00", + "carbs": 5, + "insulin": 0.5, + "notes": "", + "enteredBy": "Pump (tconnectsync)", + "glucose": "150", + "glucoseType": "Finger" + } + ) + + def test_bolus_with_bg_invalid_type(self): + self.assertRaises(InvalidBolusTypeException, + NightscoutEntry.bolus, + bolus=0.5, + carbs=5, + created_at="2021-03-16 12:25:21-04:00", + bg="150") + + self.assertRaises(InvalidBolusTypeException, + NightscoutEntry.bolus, + bolus=0.5, + carbs=5, + created_at="2021-03-16 12:25:21-04:00", + bg="150", + bg_type="") + + self.assertRaises(InvalidBolusTypeException, + NightscoutEntry.bolus, + bolus=0.5, + carbs=5, + created_at="2021-03-16 12:25:21-04:00", + bg="150", + bg_type="unknown") + def test_iob(self): self.assertEqual( NightscoutEntry.iob( @@ -86,6 +149,20 @@ def test_iob(self): } ) + def test_entry(self): + self.assertEqual( + NightscoutEntry.entry( + sgv=152, + created_at="2021-10-23 22:17:14-04:00"), + { + "type": "sgv", + "sgv": 152, + "date": 1635041834000, + "dateString": "2021-10-23T22:17:14-0400", + "device": "Pump (tconnectsync)", + } + ) + if __name__ == '__main__': unittest.main() \ No newline at end of file diff --git a/tests/parser/test_tconnect.py b/tests/parser/test_tconnect.py index 9f2d7bd..41db82d 100644 --- a/tests/parser/test_tconnect.py +++ b/tests/parser/test_tconnect.py @@ -166,6 +166,7 @@ def test_parse_bolus_entry_std_correction(self): "insulin": "13.53", "requested_insulin": "13.53", "carbs": "75", + "bg": "141", "user_override": "0", "extended_bolus": "", "bolex_completion_time": None, @@ -227,6 +228,7 @@ def test_parse_bolus_entry_std(self): "insulin": "1.25", "requested_insulin": "1.25", "carbs": "0", + "bg": "159", "user_override": "1", "extended_bolus": "", "bolex_completion_time": None, @@ -288,6 +290,7 @@ def test_parse_bolus_entry_std_automatic(self): "insulin": "1.70", "requested_insulin": "1.70", "carbs": "0", + "bg": "", "user_override": "0", "extended_bolus": "", "bolex_completion_time": None, @@ -349,6 +352,7 @@ def test_parse_bolus_entry_std_incomplete_zero(self): "insulin": "0.00", "requested_insulin": "0.50", "carbs": "0", + "bg": "144", "user_override": "1", "extended_bolus": "", "bolex_completion_time": None, @@ -410,11 +414,82 @@ def test_parse_bolus_entry_std_incomplete_partial(self): "insulin": "1.82", "requested_insulin": "2.63", "carbs": "0", + "bg": "189", "user_override": "0", "extended_bolus": "", "bolex_completion_time": None, "bolex_start_time": None }) +class TestTConnectEntryReading(unittest.TestCase): + entry1 = { + "DeviceType": "t:slim X2 Insulin Pump", + "SerialNumber": "90556643", + "Description": "EGV", + "EventDateTime": "2021-10-23T12:55:53", + "Readings (CGM / BGM)": "135" + } + def test_parse_reading_entry1(self): + self.assertEqual( + TConnectEntry.parse_reading_entry(self.entry1), + { + "time": "2021-10-23 12:55:53-04:00", + "bg": "135", + "type": "EGV" + } + ) + + entry2 = { + "DeviceType": "t:slim X2 Insulin Pump", + "SerialNumber": "90556643", + "Description": "EGV", + "EventDateTime": "2021-10-23T16:15:52", + "Readings (CGM / BGM)": "93" + } + def test_parse_reading_entry2(self): + self.assertEqual( + TConnectEntry.parse_reading_entry(self.entry2), + { + "time": "2021-10-23 16:15:52-04:00", + "bg": "93", + "type": "EGV" + } + ) + + entry3 = { + "DeviceType": "t:slim X2 Insulin Pump", + "SerialNumber": "90556643", + "Description": "EGV", + "EventDateTime": "2021-10-23T16:20:52", + "Readings (CGM / BGM)": "100" + } + def test_parse_reading_entry3(self): + self.assertEqual( + TConnectEntry.parse_reading_entry(self.entry3), + { + "time": "2021-10-23 16:20:52-04:00", + "bg": "100", + "type": "EGV" + } + ) + + entry4 = { + "DeviceType": "t:slim X2 Insulin Pump", + "SerialNumber": "90556643", + "Description": "EGV", + "EventDateTime": "2021-10-23T16:25:52", + "Readings (CGM / BGM)": "107" + } + def test_parse_reading_entry4(self): + self.assertEqual( + TConnectEntry.parse_reading_entry(self.entry4), + { + "time": "2021-10-23 16:25:52-04:00", + "bg": "107", + "type": "EGV" + } + ) + + if __name__ == '__main__': unittest.main() \ No newline at end of file diff --git a/tests/sync/test_bolus.py b/tests/sync/test_bolus.py index 2410ba9..8f26580 100644 --- a/tests/sync/test_bolus.py +++ b/tests/sync/test_bolus.py @@ -3,10 +3,12 @@ import unittest import random + from tconnectsync.sync.bolus import process_bolus_events from tconnectsync.parser.tconnect import TConnectEntry +from tconnectsync.parser.nightscout import NightscoutEntry -from ..parser.test_tconnect import TestTConnectEntryBolus +from ..parser.test_tconnect import TestTConnectEntryBolus, TestTConnectEntryCGM, TestTConnectEntryReading class TestBolusSync(unittest.TestCase): @@ -33,6 +35,96 @@ def test_process_bolus_events_standard(self): TConnectEntry.parse_bolus_entry(d) for d in bolusData ]) + def test_process_bolus_events_cgmevents_not_matching(self): + bolusData = [ + TestTConnectEntryBolus.entryStdCorrection, + TestTConnectEntryBolus.entryStd, + TestTConnectEntryBolus.entryStdAutomatic + ] + + cgmEvents = [ + TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry1), + TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry2), + TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry3) + ] + + bolusEvents = process_bolus_events(bolusData, cgmEvents=cgmEvents) + self.assertEqual(len(bolusEvents), len(bolusData)) + + def set_bg_type(entry, type): + entry["bg_type"] = type + return entry + + # Expect FINGER for bolus entries with a BG because there's no matching event with the same BG + expected = [ + set_bg_type(TConnectEntry.parse_bolus_entry(bolusData[0]), NightscoutEntry.FINGER), + set_bg_type(TConnectEntry.parse_bolus_entry(bolusData[1]), NightscoutEntry.FINGER), + # No BG specified for the automatic bolus + TConnectEntry.parse_bolus_entry(bolusData[2]) + ] + + self.assertListEqual(bolusEvents, expected) + + def test_process_bolus_events_cgmevents_matches(self): + bolusData = [ + TestTConnectEntryBolus.entryStdCorrection, + TestTConnectEntryBolus.entryStd, + TestTConnectEntryBolus.entryStdAutomatic + ] + + cgmEvents = [ + { + "time": "2021-04-01 12:45:30-04:00", + "bg": "100", + "type": "EGV" + }, + # Matches entryStdCorrection time but with wrong BG + { + "time": "2021-04-01 12:50:30-04:00", + "bg": "105", + "type": "EGV" + }, + { + "time": "2021-04-01 13:00:30-04:00", + "bg": "110", + "type": "EGV" + }, + { + "time": "2021-04-01 23:15:30-04:00", + "bg": "150", + "type": "EGV" + }, + # Matches entryStd time with correct BG + { + "time": "2021-04-01 23:20:30-04:00", + "bg": "159", + "type": "EGV" + }, + { + "time": "2021-04-01 23:25:30-04:00", + "bg": "160", + "type": "EGV" + }, + ] + + bolusEvents = process_bolus_events(bolusData, cgmEvents=cgmEvents) + self.assertEqual(len(bolusEvents), len(bolusData)) + + def set_bg_type(entry, type): + entry["bg_type"] = type + return entry + + expected = [ + # Time found but BG doesn't match + set_bg_type(TConnectEntry.parse_bolus_entry(bolusData[0]), NightscoutEntry.FINGER), + # Time found and BG matches + set_bg_type(TConnectEntry.parse_bolus_entry(bolusData[1]), NightscoutEntry.SENSOR), + # No BG specified for the automatic bolus + TConnectEntry.parse_bolus_entry(bolusData[2]) + ] + + self.assertListEqual(bolusEvents, expected) + def test_process_bolus_events_update_partial_description(self): stdData = [ TestTConnectEntryBolus.entryStdCorrection, diff --git a/tests/sync/test_cgm.py b/tests/sync/test_cgm.py new file mode 100644 index 0000000..62105ec --- /dev/null +++ b/tests/sync/test_cgm.py @@ -0,0 +1,51 @@ +#!/usr/bin/env python3 + +import unittest + +from tconnectsync.sync.cgm import find_event_at, process_cgm_events +from tconnectsync.parser.tconnect import TConnectEntry + +from ..parser.test_tconnect import TestTConnectEntryReading + +class TestProcessCGMEvents(unittest.TestCase): + def test_process_cgm_events(self): + rawReadings = [ + TestTConnectEntryReading.entry1, + TestTConnectEntryReading.entry2, + TestTConnectEntryReading.entry3, + TestTConnectEntryReading.entry4 + ] + self.assertListEqual( + process_cgm_events(rawReadings), + [TConnectEntry.parse_reading_entry(r) for r in rawReadings] + ) + +class TestFindEventAt(unittest.TestCase): + readingData = [ + TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry1), + TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry2), + TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry3), + TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry4) + ] + + def test_find_event_at_exact(self): + for r in self.readingData: + self.assertEqual(find_event_at(self.readingData, r["time"]), r) + + def test_find_event_at_before_not_found(self): + self.assertEqual(find_event_at(self.readingData, "2021-10-22 10:30:00-04:00"), None) + + def test_find_event_at_large_gap(self): + self.assertEqual(find_event_at(self.readingData, "2021-10-23 13:30:00-04:00"), self.readingData[0]) + + def test_find_event_at_between_close(self): + self.assertEqual(find_event_at(self.readingData, "2021-10-23 16:17:52-04:00"), self.readingData[1]) + self.assertEqual(find_event_at(self.readingData, "2021-10-23 16:21:52-04:00"), self.readingData[2]) + self.assertEqual(find_event_at(self.readingData, "2021-10-23 16:25:59-04:00"), self.readingData[3]) + + def test_find_event_at_most_recent(self): + self.assertEqual(find_event_at(self.readingData, "2021-10-23 18:00:00-04:00"), self.readingData[3]) + + +if __name__ == '__main__': + unittest.main() \ No newline at end of file