bckelley/tconnectsync
1
0
Fork 0
mirror of https://github.com/bckelley/tconnectsync.git synced 2026-08-24 03:34:12 -05:00
Code Issues Packages Projects Releases Wiki Activity

move Bolus internal state from dictionary to domain object

Browse Source
This commit is contained in:
James Woglom
2022-08-09 22:54:51 -04:00
parent 1b400d3b22
commit a18c2f3d17
8 changed files with 115 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tconnectsync/domain/__init__.py
View File
tconnectsync/domain/bolus.py
+18
View File
@@ -0,0 +1,18 @@
from dataclasses import dataclass
@dataclass
class Bolus:
description: str
complete: str # "1" / "0"
completion: str
request_time: str # _datetime_parse timestamp
completion_time: str # _datetime_parse timestamp
insulin: str
requested_insulin: str
carbs: str
bg: str # potentially ""
user_override: str
extended_bolus: str # "1" / "0"
bolex_completion_time: str
bolex_start_time: str
tconnectsync/parser/ciq_therapy_events.py
+16
View File
@@ -0,0 +1,16 @@
from tconnectsync.domain.therapy_event import BolusTherapyEvent, CGMTherapyEvent
from tconnectsync.parser.tconnect import TConnectEntry
def split_therapy_events(ciqTherapyEvents):
bolusEvents = []
cgmEvents = []
for e in ciqTherapyEvents['event']:
event = TConnectEntry.parse_therapy_event(e)
if type(event) == BolusTherapyEvent:
bolusEvents.append(event)
elif type(event) == CGMTherapyEvent:
cgmEvents.append(event)
return bolusEvents, cgmEvents
tconnectsync/parser/tconnect.py
+20 -4
View File
@@ -1,6 +1,9 @@
from os import stat
import sys
import arrow
from tconnectsync.domain.bolus import Bolus
from tconnectsync.domain.therapy_event import BolusTherapyEvent, CGMTherapyEvent
try:
from ..secret import TIMEZONE_NAME
@@ -105,7 +108,7 @@ class TConnectEntry:
complete = is_complete(data["ExtendedBolusIsComplete"]) or is_complete(data["BolusIsComplete"])
extended_bolus = ("extended" in data["Description"].lower())
return {
return Bolus(**{
"description": data["Description"],
"complete": "1" if complete else "",
"completion": data["CompletionStatusDesc"] if not extended_bolus else data["BolexCompletionStatusDesc"],
@@ -120,7 +123,7 @@ class TConnectEntry:
# Note: completion time can be empty if the extended bolus is in progress
"bolex_completion_time": TConnectEntry._datetime_parse(data["BolexCompletionDateTime"]).format() if data["BolexCompletionDateTime"] and complete and extended_bolus else None,
"bolex_start_time": TConnectEntry._datetime_parse(data["BolexStartDateTime"]).format() if data["BolexStartDateTime"] and complete and extended_bolus else None,
}
})
@staticmethod
def parse_reading_entry(data):
@@ -189,7 +192,16 @@ class TConnectEntry:
"time": time.format(),
"event_type": TConnectEntry.BASALSUSPENSION_EVENTS[data["SuspendReason"]]
}
# Parses an entry from controliq.therapy_events() and returns a TherapyEvent
@staticmethod
def parse_therapy_event(data):
if data["type"] == "Bolus":
return BolusTherapyEvent.parse(data)
elif data["type"] == "CGM":
return CGMTherapyEvent.parse(data)
raise UnknownTherapyEventException(data)
class UnknownCIQActivityEventException(Exception):
def __init__(self, data):
@@ -197,4 +209,8 @@ class UnknownCIQActivityEventException(Exception):
class UnknownBasalSuspensionEventException(Exception):
def __init__(self, data):
super().__init__("Unknown basal suspension event type: %s" % data)
super().__init__("Unknown basal suspension event type: %s" % data)
class UnknownTherapyEventException(Exception):
def __init__(self, data):
super().__init__("Unknown therapy event type: " % data)
tconnectsync/process.py
+30 -6
View File
@@ -3,6 +3,8 @@ import datetime
import arrow
import time
from tconnectsync.parser.ciq_therapy_events import split_therapy_events
from .util import timeago
from .api.common import ApiException
from .sync.basal import (
@@ -56,7 +58,24 @@ def process_time_range(tconnect, nightscout, time_start, time_end, pretend, feat
csvBasalData = None
csvBolusData = None
if CGM in features or BOLUS in features or BOLUS_BG in features or IOB in features:
ciqBolusData = None
ciqReadingData = None
if BOLUS in features:
logger.info("Downloading t:connect therapy_events")
ciqTherapyEventsData = tconnect.controliq.therapy_events(time_start, time_end)
ciqBolusData, ciqReadingData = split_therapy_events(ciqTherapyEventsData)
if ciqReadingData and len(ciqReadingData) > 0:
lastReading = ciqReadingData[-1].eventDateTime
lastReading = TConnectEntry._datetime_parse(lastReading)
logger.debug(ciqReadingData[-1])
logger.info("Last CGM reading from t:connect CIQ: %s (%s)" % (lastReading, timeago(lastReading)))
else:
logger.warning("No last CGM reading is able to be determined from CIQ")
if (BOLUS in features and not ciqBolusData) or \
(CGM in features and not ciqReadingData) or \
BOLUS_BG in features or IOB in features:
logger.info("Downloading t:connect CSV data")
csvdata = tconnect.ws2.therapy_timeline_csv(time_start, time_end)
@@ -69,9 +88,10 @@ def process_time_range(tconnect, nightscout, time_start, time_end, pretend, feat
lastReading = csvReadingData[-1]['EventDateTime'] if 'EventDateTime' in csvReadingData[-1] else 0
lastReading = TConnectEntry._datetime_parse(lastReading)
logger.debug(csvReadingData[-1])
logger.info("Last CGM reading from t:connect: %s (%s)" % (lastReading, timeago(lastReading)))
logger.info("Last CGM reading from t:connect CSV: %s (%s)" % (lastReading, timeago(lastReading)))
else:
logger.warning("No last CGM reading is able to be determined")
logger.warning("No last CGM reading is able to be determined from CSV")
added = 0
@@ -108,10 +128,14 @@ def process_time_range(tconnect, nightscout, time_start, time_end, pretend, feat
added += ns_write_pump_events(nightscout, pumpEvents, pretend=pretend, time_start=time_start, time_end=time_end)
if csvBolusData:
if BOLUS in features:
if BOLUS in features:
bolusEvents = []
if ciqBolusData:
bolusEvents = process_bolus_therapy_events(ciqBolusData)
if csvBolusData and not bolusEvents:
bolusEvents = process_bolus_events(csvBolusData)
added += ns_write_bolus_events(nightscout, bolusEvents, pretend=pretend, include_bg=(BOLUS_BG in features), time_start=time_start, time_end=time_end)
added += ns_write_bolus_events(nightscout, bolusEvents, pretend=pretend, include_bg=(BOLUS_BG in features), time_start=time_start, time_end=time_end)
if csvIobData:
if IOB in features:
tconnectsync/sync/bolus.py
+17 -17
View File
@@ -20,21 +20,21 @@ def process_bolus_events(bolusdata, cgmEvents=None):
for b in bolusdata:
parsed = TConnectEntry.parse_bolus_entry(b)
if parsed["completion"] != "Completed":
if parsed["insulin"] and float(parsed["insulin"]) > 0:
if parsed.completion != "Completed":
if parsed.insulin and float(parsed.insulin) > 0:
# Count non-completed bolus if any insulin was delivered (vs. the amount of insulin requested)
parsed["description"] += " (%s: requested %s units)" % (parsed["completion"], parsed["requested_insulin"])
parsed.description += " (%s: requested %s units)" % (parsed.completion, parsed.requested_insulin)
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)
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["request_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
@@ -72,27 +72,27 @@ def ns_write_bolus_events(nightscout, bolusEvents, pretend=False, include_bg=Fal
add_count = 0
for event in bolusEvents:
created_at = event["completion_time"] if not event["extended_bolus"] else event["bolex_start_time"]
created_at = event.completion_time if not event.extended_bolus else event.bolex_start_time
if last_upload_time and arrow.get(created_at) <= last_upload_time:
if pretend:
logger.info("Skipping basal event before last upload time: %s (time range: %s - %s)" % (event, time_start, time_end))
continue
if include_bg and event["bg"]:
if include_bg and event.bg:
entry = NightscoutEntry.bolus(
bolus=event["insulin"],
carbs=event["carbs"],
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"]
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"],
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 "")
notes="{}{}{}".format(event.description, " (Override)" if event.user_override == "1" else "", " (Extended)" if event.extended_bolus == "1" else "")
)
add_count += 1
tests/parser/test_tconnect.py
+11 -10
View File
@@ -1,6 +1,7 @@
#!/usr/bin/env python3
import unittest
from tconnectsync.domain.bolus import Bolus
from tconnectsync.parser.tconnect import TConnectEntry, UnknownBasalSuspensionEventException, UnknownCIQActivityEventException
class TestTConnectEntryBasal(unittest.TestCase):
@@ -177,7 +178,7 @@ class TestTConnectEntryBolus(unittest.TestCase):
def test_parse_bolus_entry_std_correction(self):
self.assertEqual(
TConnectEntry.parse_bolus_entry(self.entryStdCorrection),
{
Bolus(**{
"description": "Standard/Correction",
"complete": "1",
"completion": "Completed",
@@ -191,7 +192,7 @@ class TestTConnectEntryBolus(unittest.TestCase):
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
})
}))
entryStd = {
"Type": "Bolus",
@@ -239,7 +240,7 @@ class TestTConnectEntryBolus(unittest.TestCase):
def test_parse_bolus_entry_std(self):
self.assertEqual(
TConnectEntry.parse_bolus_entry(self.entryStd),
{
Bolus(**{
"description": "Standard",
"complete": "1",
"completion": "Completed",
@@ -253,7 +254,7 @@ class TestTConnectEntryBolus(unittest.TestCase):
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
})
}))
entryStdAutomatic = {
"Type": "Bolus",
@@ -301,7 +302,7 @@ class TestTConnectEntryBolus(unittest.TestCase):
def test_parse_bolus_entry_std_automatic(self):
self.assertEqual(
TConnectEntry.parse_bolus_entry(self.entryStdAutomatic),
{
Bolus(**{
"description": "Automatic Bolus/Correction",
"complete": "1",
"completion": "Completed",
@@ -315,7 +316,7 @@ class TestTConnectEntryBolus(unittest.TestCase):
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
})
}))
entryStdIncompleteZero = {
"Type": "Bolus",
@@ -363,7 +364,7 @@ class TestTConnectEntryBolus(unittest.TestCase):
def test_parse_bolus_entry_std_incomplete_zero(self):
self.assertEqual(
TConnectEntry.parse_bolus_entry(self.entryStdIncompleteZero),
{
Bolus(**{
"description": "Standard",
"complete": "",
"completion": "User Aborted",
@@ -377,7 +378,7 @@ class TestTConnectEntryBolus(unittest.TestCase):
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
})
}))
entryStdIncompletePartial = {
"Type": "Bolus",
@@ -425,7 +426,7 @@ class TestTConnectEntryBolus(unittest.TestCase):
def test_parse_bolus_entry_std_incomplete_partial(self):
self.assertEqual(
TConnectEntry.parse_bolus_entry(self.entryStdIncompletePartial),
{
Bolus(**{
"description": "Standard/Correction",
"complete": "",
"completion": "Terminated by Alarm",
@@ -439,7 +440,7 @@ class TestTConnectEntryBolus(unittest.TestCase):
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
})
}))
class TestTConnectEntryReading(unittest.TestCase):
entry1 = {
tests/sync/test_bolus.py
+3 -3
View File
@@ -52,7 +52,7 @@ class TestBolusSync(unittest.TestCase):
self.assertEqual(len(bolusEvents), len(bolusData))
def set_bg_type(entry, type):
entry["bg_type"] = 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
@@ -111,7 +111,7 @@ class TestBolusSync(unittest.TestCase):
self.assertEqual(len(bolusEvents), len(bolusData))
def set_bg_type(entry, type):
entry["bg_type"] = type
entry.bg_type = type
return entry
expected = [
@@ -145,7 +145,7 @@ class TestBolusSync(unittest.TestCase):
]
for e in partialEntries:
e["description"] += " (%s: requested %s units)" % (e["completion"], e["requested_insulin"])
e.description += " (%s: requested %s units)" % (e.completion, e.requested_insulin)
self.assertListEqual(bolusEvents, [
TConnectEntry.parse_bolus_entry(d) for d in stdData