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

Merge pull request #8 from jwoglom/develop

Browse Source 
version 0.2.0
This commit is contained in:
James Woglom
2021-04-21 01:43:43 -04:00
committed by GitHub
parent 8dcbee4280 a3c7aa69a9
commit 71ceb94270
21 changed files with 710 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/python-package.yml
+2 -2
View File
@@ -5,9 +5,9 @@ name: Python package
on:
push:
branches: [ master ]
branches: [ master, develop ]
pull_request:
branches: [ master ]
branches: [ master, develop ]
jobs:
build:
Pipfile
+2 -1
View File
@@ -13,4 +13,5 @@ lxml = "*"
python-dotenv = "*"
[scripts]
tconnectsync = "python3 main.py"
tconnectsync = "python3 main.py"
test = "python3 -m unittest discover -vv"
main.py
+8 -3
View File
@@ -9,11 +9,14 @@ from tconnectsync.api import TConnectApi
from tconnectsync.process import process_time_range
from tconnectsync.autoupdate import process_auto_update
from tconnectsync.check import check_login
from tconnectsync.nightscout import NightscoutApi
try:
from tconnectsync.secret import (
TCONNECT_EMAIL,
TCONNECT_PASSWORD
TCONNECT_PASSWORD,
NS_URL,
NS_SECRET
)
except Exception:
print('Unable to read secret.py')
@@ -49,15 +52,17 @@ def main():
tconnect = TConnectApi(TCONNECT_EMAIL, TCONNECT_PASSWORD)
nightscout = NightscoutApi(NS_URL, NS_SECRET)
if args.check_login:
return check_login(tconnect, time_start, time_end)
if args.auto_update:
print("Starting auto-update between", time_start, "and", time_end, "(PRETEND)" if args.pretend else "")
process_auto_update(tconnect, time_start, time_end, args.pretend)
process_auto_update(tconnect, nightscout, time_start, time_end, args.pretend)
else:
print("Processing data between", time_start, "and", time_end, "(PRETEND)" if args.pretend else "")
added = process_time_range(tconnect, time_start, time_end, args.pretend)
added = process_time_range(tconnect, nightscout, time_start, time_end, args.pretend)
print("Added", added, "items")
if __name__ == '__main__':
tconnectsync/api/__init__.py
+1 -1
View File
@@ -18,7 +18,7 @@ class TConnectApi:
@property
def controliq(self):
if self._ciq:
if self._ciq and not self._ciq.needs_relogin():
return self._ciq
self._ciq = ControlIQApi(self.email, self.password)
tconnectsync/api/controliq.py
+6
View File
@@ -1,6 +1,8 @@
import requests
import urllib
import datetime
import arrow
from bs4 import BeautifulSoup
from .common import parse_date, base_headers, ApiException, ApiLoginException
@@ -46,6 +48,10 @@ class ControlIQApi:
self.accessTokenExpiresAt = req.cookies['accessTokenExpiresAt']
return True
def needs_relogin(self):
diff = (arrow.get(self.accessTokenExpiresAt) - arrow.get())
return (diff.seconds <= 5 * 60)
def api_headers(self):
if not self.accessToken:
raise Exception('No access token provided')
tconnectsync/autoupdate.py
+2 -2
View File
@@ -12,7 +12,7 @@ from .secret import (
Performs the auto-update functionality. Runs indefinitely in a loop
until stopped (ctrl+c).
"""
def process_auto_update(tconnect, time_start, time_end, pretend):
def process_auto_update(tconnect, nightscout, time_start, time_end, pretend):
# Read from android api, find exact interval to cut down on API calls
# Refresh API token. If failure, die, have wrapper script re-run.
@@ -28,7 +28,7 @@ def process_auto_update(tconnect, time_start, time_end, pretend):
if pretend:
print('Would update now')
else:
added = process_time_range(tconnect, time_start, time_end, pretend)
added = process_time_range(tconnect, nightscout, time_start, time_end, pretend)
print('Added', added, 'items')
if last_event_index:
tconnectsync/check.py
+10 -5
View File
@@ -1,7 +1,4 @@
from .nightscout import api_status
from .secret import PUMP_SERIAL_NUMBER
from .nightscout import NightscoutApi
"""
Attempts to authenticate with each t:connect API,
@@ -32,16 +29,24 @@ def check_login(tconnect, time_start, time_end):
summary = tconnect.android.user_profile()
print("Android user profile: %s" % summary)
from .secret import PUMP_SERIAL_NUMBER
event = tconnect.android.last_event_uploaded(PUMP_SERIAL_NUMBER)
print("\nAndroid last uploaded event: %s" % event)
except ImportError:
print("Error: Unable to load config file.")
except Exception as e:
print("Error occurred querying Android API: %s" % e)
errors += 1
print("\nLogging in to Nightscout...")
try:
status = api_status()
from .secret import NS_URL, NS_SECRET
status = NightscoutApi(NS_URL, NS_SECRET).api_status()
print("\nNightscout status: %s" % status)
except ImportError:
print("Error: Unable to load config file.")
except Exception as e:
print("Error occurred querying Nightscout API: %s" % e)
errors += 1
tconnectsync/nightscout.py
+67 -52
View File
@@ -6,64 +6,79 @@ import urllib.parse
from urllib.parse import urljoin
from .api.common import ApiException
from .parser.nightscout import ENTERED_BY
try:
from .secret import NS_URL, NS_SECRET, TIMEZONE_NAME
except Exception:
print('Unable to import Nightscout secrets from secret.py')
sys.exit(1)
# try:
# from .secret import NS_URL, NS_SECRET
# except Exception:
# print('Unable to import Nightscout secrets from secret.py')
# sys.exit(1)
class NightscoutApi:
def __init__(self, url, secret):
self.url = url
self.secret = secret
def upload_nightscout(ns_format, entity='treatments'):
upload = requests.post(urljoin(NS_URL, 'api/v1/' + entity + '?api_secret=' + NS_SECRET), json=ns_format, headers={
'Accept': 'application/json',
'Content-Type': 'application/json',
'api-secret': hashlib.sha1(NS_SECRET.encode()).hexdigest()
})
print("Nightscout upload status:", upload.status_code, upload.text)
def upload_entry(self, ns_format, entity='treatments'):
r = requests.post(urljoin(self.url, 'api/v1/' + entity + '?api_secret=' + self.secret), json=ns_format, headers={
'Accept': 'application/json',
'Content-Type': 'application/json',
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
})
if r.status_code != 200:
raise ApiException(r.status_code, "Nightscout upload response: %s" % r.text)
def delete_nightscout(entity):
upload = requests.delete(urljoin(NS_URL, 'api/v1/' + entity + '?api_secret=' + NS_SECRET), json={}, headers={
'Accept': 'application/json',
'Content-Type': 'application/json',
'api-secret': hashlib.sha1(NS_SECRET.encode()).hexdigest()
})
print("Nightscout delete status:", upload.status_code, upload.text)
def delete_entry(self, entity):
r = requests.delete(urljoin(self.url, 'api/v1/' + entity + '?api_secret=' + self.secret), json={}, headers={
'Accept': 'application/json',
'Content-Type': 'application/json',
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
})
if r.status_code != 200:
raise ApiException(r.status_code, "Nightscout delete response: %s" % r.text)
def put_nightscout(ns_format, entity):
upload = requests.put(urljoin(NS_URL, 'api/v1/' + entity + '?api_secret=' + NS_SECRET), json=ns_format, headers={
'Accept': 'application/json',
'Content-Type': 'application/json',
'api-secret': hashlib.sha1(NS_SECRET.encode()).hexdigest()
})
print("Nightscout put status:", upload.status_code, upload.text)
def put_entry(self, ns_format, entity):
r = requests.put(urljoin(self.url, 'api/v1/' + entity + '?api_secret=' + self.secret), json=ns_format, headers={
'Accept': 'application/json',
'Content-Type': 'application/json',
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
})
if r.status_code != 200:
raise ApiException(r.status_code, "Nightscout put response: %s" % r.text)
def last_uploaded_nightscout_entry(eventType):
latest = requests.get(urljoin(NS_URL, 'api/v1/treatments?count=1&find[enteredBy]=' + urllib.parse.quote(ENTERED_BY) + '&find[eventType]=' + urllib.parse.quote(eventType) + '&ts=' + str(time.time())), headers={
'api-secret': hashlib.sha1(NS_SECRET.encode()).hexdigest()
})
j = latest.json()
if j and len(j) > 0:
return j[0]
return None
def last_uploaded_entry(self, eventType):
latest = requests.get(urljoin(self.url, 'api/v1/treatments?count=1&find[enteredBy]=' + urllib.parse.quote(ENTERED_BY) + '&find[eventType]=' + urllib.parse.quote(eventType) + '&ts=' + str(time.time())), headers={
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
})
if latest.status_code != 200:
raise ApiException(latest.status_code, "Nightscout treatments response: %s" % latest.text)
def last_uploaded_nightscout_activity(activityType):
latest = requests.get(urljoin(NS_URL, 'api/v1/activity?find[enteredBy]=' + urllib.parse.quote(ENTERED_BY) + '&find[activityType]=' + urllib.parse.quote(activityType) + '&ts=' + str(time.time())), headers={
'api-secret': hashlib.sha1(NS_SECRET.encode()).hexdigest()
})
j = latest.json()
if j and len(j) > 0:
return j[0]
return None
j = latest.json()
if j and len(j) > 0:
return j[0]
return None
"""
Returns general status information about the Nightscout server.
"""
def api_status():
status = requests.get(urljoin(NS_URL, 'api/v1/status.json'), headers={
'api-secret': hashlib.sha1(NS_SECRET.encode()).hexdigest()
})
if status.status_code != 200:
raise Exception('HTTP error status code (%d) from Nightscout: %s' % (status.status_code, status.text))
return status.json()
def last_uploaded_activity(self, activityType):
latest = requests.get(urljoin(self.url, 'api/v1/activity?find[enteredBy]=' + urllib.parse.quote(ENTERED_BY) + '&find[activityType]=' + urllib.parse.quote(activityType) + '&ts=' + str(time.time())), headers={
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
})
if latest.status_code != 200:
raise ApiException(latest.status_code, "Nightscout activity response: %s" % latest.text)
j = latest.json()
if j and len(j) > 0:
return j[0]
return None
"""
Returns general status information about the Nightscout server.
"""
def api_status(self):
status = requests.get(urljoin(self.url, 'api/v1/status.json'), headers={
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
})
if status.status_code != 200:
raise Exception('HTTP error status code (%d) from Nightscout: %s' % (status.status_code, status.text))
return status.json()
tconnectsync/parser/nightscout.py
+2 -2
View File
@@ -25,8 +25,8 @@ class NightscoutEntry:
return {
"eventType": BOLUS_EVENTTYPE,
"created_at": created_at,
"carbs": carbs,
"insulin": bolus,
"carbs": int(carbs),
"insulin": float(bolus),
"notes": notes,
"enteredBy": ENTERED_BY,
}
tconnectsync/process.py
+4 -5
View File
@@ -20,7 +20,7 @@ Given a TConnectApi object and start/end range, performs a single
cycle of synchronizing data within the time range.
If pretend is true, then doesn't actually write data to Nightscout.
"""
def process_time_range(tconnect, time_start, time_end, pretend):
def process_time_range(tconnect, nightscout, time_start, time_end, pretend):
print("Downloading t:connect ControlIQ data")
try:
ciqTherapyTimelineData = tconnect.controliq.therapy_timeline(time_start, time_end)
@@ -51,13 +51,12 @@ def process_time_range(tconnect, time_start, time_end, pretend):
if csvBasalData:
add_csv_basal_events(basalEvents, csvBasalData)
added += ns_write_basal_events(basalEvents, pretend=pretend)
added += ns_write_basal_events(nightscout, basalEvents, pretend=pretend)
bolusEvents = process_bolus_events(bolusData)
added += ns_write_bolus_events(bolusEvents, pretend=pretend)
added += ns_write_bolus_events(nightscout, bolusEvents, pretend=pretend)
iobEvents = process_iob_events(iobData)
added += ns_write_iob_events(iobEvents, pretend=pretend)
added += ns_write_iob_events(nightscout, iobEvents, pretend=pretend)
return added
tconnectsync/sync/basal.py
+9 -9
View File
@@ -4,11 +4,6 @@ from ..parser.nightscout import (
BASAL_EVENTTYPE,
NightscoutEntry
)
from ..nightscout import (
last_uploaded_nightscout_entry,
put_nightscout,
upload_nightscout
)
from ..parser.tconnect import TConnectEntry
@@ -64,8 +59,8 @@ def add_csv_basal_events(basalEvents, data):
"""
Given processed basal data, adds basal events to Nightscout.
"""
def ns_write_basal_events(basalEvents, pretend=False):
last_upload = last_uploaded_nightscout_entry(BASAL_EVENTTYPE)
def ns_write_basal_events(nightscout, basalEvents, pretend=False):
last_upload = nightscout.last_uploaded_entry(BASAL_EVENTTYPE)
last_upload_time = None
if last_upload:
last_upload_time = arrow.get(last_upload["created_at"])
@@ -84,6 +79,11 @@ def ns_write_basal_events(basalEvents, pretend=False):
# has newer info, then delete and recreate it.
recent_needs_update = (round(last_upload["duration"]) < round(event["duration_mins"]))
# If the timestamps are identical, and the duration is identical,
# then don't upload a duplicate entry of what we already have.
if not recent_needs_update:
continue
reason = event["delivery_type"]
if "suspendReason" in reason:
reason += " (" + reason["suspendReason"] + ")"
@@ -102,8 +102,8 @@ def ns_write_basal_events(basalEvents, pretend=False):
print("Replacing last uploaded entry:", last_upload)
if not pretend:
entry['_id'] = last_upload['_id']
put_nightscout(entry, entity='treatments')
nightscout.put_entry(entry, entity='treatments')
elif not pretend:
upload_nightscout(entry)
nightscout.upload_entry(entry)
return add_count
tconnectsync/sync/bolus.py
+3 -8
View File
@@ -4,11 +4,6 @@ from ..parser.nightscout import (
BOLUS_EVENTTYPE,
NightscoutEntry
)
from ..nightscout import (
last_uploaded_nightscout_entry,
put_nightscout,
upload_nightscout
)
from ..parser.tconnect import TConnectEntry
"""
@@ -35,8 +30,8 @@ def process_bolus_events(bolusdata):
"""
Given processed bolus data, adds bolus events to Nightscout.
"""
def ns_write_bolus_events(bolusEvents, pretend=False):
last_upload = last_uploaded_nightscout_entry(BOLUS_EVENTTYPE)
def ns_write_bolus_events(nightscout, bolusEvents, pretend=False):
last_upload = nightscout.last_uploaded_entry(BOLUS_EVENTTYPE)
last_upload_time = None
if last_upload:
last_upload_time = arrow.get(last_upload["created_at"])
@@ -60,6 +55,6 @@ def ns_write_bolus_events(bolusEvents, pretend=False):
print(" Processing bolus:", event, "entry:", entry)
if not pretend:
upload_nightscout(entry)
nightscout.upload_entry(entry)
return add_count
tconnectsync/sync/iob.py
+4 -9
View File
@@ -4,11 +4,6 @@ from ..parser.nightscout import (
IOB_ACTIVITYTYPE,
NightscoutEntry
)
from ..nightscout import (
last_uploaded_nightscout_activity,
delete_nightscout,
upload_nightscout
)
from ..parser.tconnect import TConnectEntry
"""
@@ -26,8 +21,8 @@ def process_iob_events(iobdata):
"""
Given processed IOB data, creates a single Nightscout activity definition to store IOB.
"""
def ns_write_iob_events(iobEvents, pretend=False):
last_upload = last_uploaded_nightscout_activity(IOB_ACTIVITYTYPE)
def ns_write_iob_events(nightscout, iobEvents, pretend=False):
last_upload = nightscout.last_uploaded_activity(IOB_ACTIVITYTYPE)
last_upload_time = None
if last_upload:
last_upload_time = arrow.get(last_upload["created_at"])
@@ -49,12 +44,12 @@ def ns_write_iob_events(iobEvents, pretend=False):
print(" Processing iob:", event, "entry:", entry)
if not pretend:
upload_nightscout(entry, entity='activity')
nightscout.upload_entry(entry, entity='activity')
# Delete the previous activity
if last_upload and '_id' in last_upload:
print(" Deleting old iob entry:", last_upload)
if not pretend:
delete_nightscout('activity/{}'.format(last_upload['_id']))
nightscout.delete_entry('activity/{}'.format(last_upload['_id']))
return 1
tests/api/__init__.py
View File
tests/api/fake.py
+46
View File
@@ -0,0 +1,46 @@
import tconnectsync.api
class ControlIQApi(tconnectsync.api.controliq.ControlIQApi):
def __init__(self):
self.BASE_URL = 'invalid://'
self.LOGIN_URL = 'invalid://'
def login(self, email, password):
raise NotImplementedError
def needs_relogin(self):
return False
def get(self, endpoint, query):
raise NotImplementedError
class WS2Api(tconnectsync.api.ws2.WS2Api):
def __init__(self):
self.BASE_URL = 'invalid://'
def get(self, endpoint, query):
raise NotImplementedError
def get_jsonp(self, endpoint):
raise NotImplementedError
class AndroidApi(tconnectsync.api.android.AndroidApi):
def __init__(self):
self.BASE_URL = 'invalid://'
def login(self, email, password):
raise NotImplementedError
def needs_relogin(self):
return False
def get(self, endpoint, query={}, **kwargs):
raise NotImplementedError
class TConnectApi(tconnectsync.api.TConnectApi):
def __init__(self):
pass
_ciq = ControlIQApi()
_ws2 = WS2Api()
_android = AndroidApi()
tests/nightscout_fake.py
+31
View File
@@ -0,0 +1,31 @@
import collections
import tconnectsync.nightscout
class NightscoutApi(tconnectsync.nightscout.NightscoutApi):
def __init__(self):
self.url = 'invalid://'
self.secret = 'invalid'
self.uploaded_entries = collections.defaultdict(list)
self.deleted_entries = collections.defaultdict(list)
self.put_entries = collections.defaultdict(list)
def upload_entry(self, ns_format, entity='treatments'):
self.uploaded_entries[entity].append(ns_format)
def delete_entry(self, ns_format, entity):
self.deleted_entries[entity].append(ns_format)
def put_entry(self, ns_format, entity):
self.put_entries[entity].append(ns_format)
def last_uploaded_entry(self, eventType):
raise NotImplementedError
def last_uploaded_activity(self, activityType):
raise NotImplementedError
def api_status(self):
raise NotImplementedError
tests/parser/test_nightscout.py
+89
View File
@@ -0,0 +1,89 @@
#!/usr/bin/env python3
import unittest
from tconnectsync.parser.nightscout import NightscoutEntry
class TestNightscoutEntry(unittest.TestCase):
def test_basal(self):
self.assertEqual(
NightscoutEntry.basal(
value=1.05,
duration_mins=30,
created_at="2021-03-16 00:25:21-04:00"),
{
"eventType": "Temp Basal",
"reason": "",
"duration": 30,
"absolute": 1.05,
"created_at": "2021-03-16 00:25:21-04:00",
"carbs": None,
"insulin": None,
"enteredBy": "Pump (tconnectsync)"
}
)
self.assertEqual(
NightscoutEntry.basal(
value=0.95,
duration_mins=5,
created_at="2021-03-16 12:25:21-04:00",
reason="Correction"),
{
"eventType": "Temp Basal",
"reason": "Correction",
"duration": 5,
"absolute": 0.95,
"created_at": "2021-03-16 12:25:21-04:00",
"carbs": None,
"insulin": None,
"enteredBy": "Pump (tconnectsync)"
}
)
def test_bolus(self):
self.assertEqual(
NightscoutEntry.bolus(
bolus=7.5,
carbs=45,
created_at="2021-03-16 00:25:21-04:00"),
{
"eventType": "Combo Bolus",
"created_at": "2021-03-16 00:25:21-04:00",
"carbs": 45,
"insulin": 7.5,
"notes": "",
"enteredBy": "Pump (tconnectsync)"
}
)
self.assertEqual(
NightscoutEntry.bolus(
bolus=0.5,
carbs=5,
created_at="2021-03-16 12:25:21-04:00"),
{
"eventType": "Combo Bolus",
"created_at": "2021-03-16 12:25:21-04:00",
"carbs": 5,
"insulin": 0.5,
"notes": "",
"enteredBy": "Pump (tconnectsync)"
}
)
def test_iob(self):
self.assertEqual(
NightscoutEntry.iob(
iob=2.05,
created_at="2021-03-16 00:25:21-04:00"),
{
"activityType": "tconnect_iob",
"iob": 2.05,
"created_at": "2021-03-16 00:25:21-04:00",
"enteredBy": "Pump (tconnectsync)"
}
)
if __name__ == '__main__':
unittest.main()
tests/parser/test_tconnect.py
+184 -3
View File
@@ -3,9 +3,9 @@
import unittest
from tconnectsync.parser.tconnect import TConnectEntry
class TestTConnectEntry(unittest.TestCase):
class TestTConnectEntryBasal(unittest.TestCase):
def test_parse_ciq_basal_entry(self):
self.assertEquals(
self.assertEqual(
TConnectEntry.parse_ciq_basal_entry({
"y": 0.8,
"duration": 1221,
@@ -19,7 +19,7 @@ class TestTConnectEntry(unittest.TestCase):
}
)
self.assertEquals(
self.assertEqual(
TConnectEntry.parse_ciq_basal_entry({
"y": 0.797,
"duration": 300,
@@ -33,6 +33,187 @@ class TestTConnectEntry(unittest.TestCase):
}
)
class TestTConnectEntryBolus(unittest.TestCase):
entryStdCorrection = {
"Type": "Bolus",
"Description": "Standard/Correction",
"BG": "141",
"IOB": "",
"BolusRequestID": "7001.000",
"BolusCompletionID": "7001.000",
"CompletionDateTime": "2021-04-01T12:58:26",
"InsulinDelivered": "13.53",
"FoodDelivered": "12.50",
"CorrectionDelivered": "1.03",
"CompletionStatusID": "3",
"CompletionStatusDesc": "Completed",
"BolusIsComplete": "1",
"BolexCompletionID": "",
"BolexSize": "",
"BolexStartDateTime": "",
"BolexCompletionDateTime": "",
"BolexInsulinDelivered": "",
"BolexIOB": "",
"BolexCompletionStatusID": "",
"BolexCompletionStatusDesc": "",
"ExtendedBolusIsComplete": "",
"EventDateTime": "2021-04-01T12:53:36",
"RequestDateTime": "2021-04-01T12:53:36",
"BolusType": "Carb",
"BolusRequestOptions": "Standard/Correction",
"StandardPercent": "100.00",
"Duration": "0",
"CarbSize": "75",
"UserOverride": "0",
"TargetBG": "110",
"CorrectionFactor": "30.00",
"FoodBolusSize": "12.50",
"CorrectionBolusSize": "1.03",
"ActualTotalBolusRequested": "13.53",
"IsQuickBolus": "0",
"EventHistoryReportEventDesc": "0",
"EventHistoryReportDetails": "Correction & Food Bolus",
"NoteID": "CF 1:30 - Carb Ratio 1:6 - Target BG 110",
"IndexID": "0",
"Note": "1181649"
}
def test_parse_bolus_entry_std_correction(self):
self.assertEqual(
TConnectEntry.parse_bolus_entry(self.entryStdCorrection),
{
"description": "Standard/Correction",
"complete": "1",
"completion": "Completed",
"request_time": "2021-04-01 12:53:36-04:00",
"completion_time": "2021-04-01 12:58:26-04:00",
"insulin": "13.53",
"carbs": "75",
"user_override": "0",
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
})
entryStd = {
"Type": "Bolus",
"Description": "Standard",
"BG": "159",
"IOB": "2.13",
"BolusRequestID": "7007.000",
"BolusCompletionID": "7007.000",
"CompletionDateTime": "2021-04-01T23:23:17",
"InsulinDelivered": "1.25",
"FoodDelivered": "0.00",
"CorrectionDelivered": "0.00",
"CompletionStatusID": "3",
"CompletionStatusDesc": "Completed",
"BolusIsComplete": "1",
"BolexCompletionID": "",
"BolexSize": "",
"BolexStartDateTime": "",
"BolexCompletionDateTime": "",
"BolexInsulinDelivered": "",
"BolexIOB": "",
"BolexCompletionStatusID": "",
"BolexCompletionStatusDesc": "",
"ExtendedBolusIsComplete": "",
"EventDateTime": "2021-04-01T23:21:58",
"RequestDateTime": "2021-04-01T23:21:58",
"BolusType": "Carb",
"BolusRequestOptions": "Standard",
"StandardPercent": "100.00",
"Duration": "0",
"CarbSize": "0",
"UserOverride": "1",
"TargetBG": "110",
"CorrectionFactor": "30.00",
"FoodBolusSize": "0.00",
"CorrectionBolusSize": "0.00",
"ActualTotalBolusRequested": "1.25",
"IsQuickBolus": "0",
"EventHistoryReportEventDesc": "0",
"EventHistoryReportDetails": "Food Bolus",
"NoteID": "CF 1:30 - Carb Ratio 1:6 - Target BG 110 | Override: Pump calculated Bolus = 0.0 units",
"IndexID": "0",
"Note": "1182867"
}
def test_parse_bolus_entry_std(self):
self.assertEqual(
TConnectEntry.parse_bolus_entry(self.entryStd),
{
"description": "Standard",
"complete": "1",
"completion": "Completed",
"request_time": "2021-04-01 23:21:58-04:00",
"completion_time": "2021-04-01 23:23:17-04:00",
"insulin": "1.25",
"carbs": "0",
"user_override": "1",
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
})
entryStdAutomatic = {
"Type": "Bolus",
"Description": "Automatic Bolus/Correction",
"BG": "",
"IOB": "3.24",
"BolusRequestID": "7010.000",
"BolusCompletionID": "7010.000",
"CompletionDateTime": "2021-04-02T01:00:47",
"InsulinDelivered": "1.70",
"FoodDelivered": "0.00",
"CorrectionDelivered": "1.70",
"CompletionStatusID": "3",
"CompletionStatusDesc": "Completed",
"BolusIsComplete": "1",
"BolexCompletionID": "",
"BolexSize": "",
"BolexStartDateTime": "",
"BolexCompletionDateTime": "",
"BolexInsulinDelivered": "",
"BolexIOB": "",
"BolexCompletionStatusID": "",
"BolexCompletionStatusDesc": "",
"ExtendedBolusIsComplete": "",
"EventDateTime": "2021-04-02T00:59:13",
"RequestDateTime": "2021-04-02T00:59:13",
"BolusType": "Automatic Correction",
"BolusRequestOptions": "Automatic Bolus/Correction",
"StandardPercent": "100.00",
"Duration": "0",
"CarbSize": "0",
"UserOverride": "0",
"TargetBG": "160",
"CorrectionFactor": "30.00",
"FoodBolusSize": "0.00",
"CorrectionBolusSize": "1.70",
"ActualTotalBolusRequested": "1.70",
"IsQuickBolus": "0",
"EventHistoryReportEventDesc": "0",
"EventHistoryReportDetails": "Correction Bolus",
"NoteID": "CF 1:30 - Carb Ratio 1:0 - Target BG 160",
"IndexID": "0",
"Note": "1183132"
}
def test_parse_bolus_entry_std_automatic(self):
self.assertEqual(
TConnectEntry.parse_bolus_entry(self.entryStdAutomatic),
{
"description": "Automatic Bolus/Correction",
"complete": "1",
"completion": "Completed",
"request_time": "2021-04-02 00:59:13-04:00",
"completion_time": "2021-04-02 01:00:47-04:00",
"insulin": "1.70",
"carbs": "0",
"user_override": "0",
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
})
if __name__ == '__main__':
tests/sync/test_basal.py
+8 -2
View File
@@ -18,8 +18,9 @@ class TestBasalSync(unittest.TestCase):
"pumpFeatures": []
}
def test_process_ciq_basal_events(self):
data = self.base.copy()
@staticmethod
def get_example_ciq_basal_events():
data = TestBasalSync.base.copy()
data["basal"]["tempDeliveryEvents"] = [
{
"y": 0.8,
@@ -55,6 +56,11 @@ class TestBasalSync(unittest.TestCase):
},
]
return data
def test_process_ciq_basal_events(self):
data = TestBasalSync.get_example_ciq_basal_events()
basalEvents = process_ciq_basal_events(data)
self.assertEqual(len(basalEvents), 4)
tests/sync/test_bolus.py
+34
View File
@@ -0,0 +1,34 @@
#!/usr/bin/env python3
import unittest
from tconnectsync.sync.bolus import process_bolus_events
from tconnectsync.parser.tconnect import TConnectEntry
from ..parser.test_tconnect import TestTConnectEntryBolus
class TestBolusSync(unittest.TestCase):
@staticmethod
def get_example_csv_bolus_events():
return [
TestTConnectEntryBolus.entryStdCorrection,
TestTConnectEntryBolus.entryStd,
TestTConnectEntryBolus.entryStdAutomatic
]
def test_process_bolus_events(self):
bolusData = TestBolusSync.get_example_csv_bolus_events()
bolusEvents = process_bolus_events(bolusData)
self.assertEqual(len(bolusEvents), 3)
self.assertListEqual(bolusEvents, [
TConnectEntry.parse_bolus_entry(bolusData[0]),
TConnectEntry.parse_bolus_entry(bolusData[1]),
TConnectEntry.parse_bolus_entry(bolusData[2])
])
if __name__ == '__main__':
unittest.main()
tests/test_process.py
+198
View File
@@ -0,0 +1,198 @@
#!/usr/bin/env python3
import unittest
import datetime
import pprint
from tconnectsync.process import process_time_range
from tconnectsync.parser.nightscout import NightscoutEntry
from .api.fake import TConnectApi
from .nightscout_fake import NightscoutApi
from .sync.test_basal import TestBasalSync
from .sync.test_bolus import TestBolusSync
class TestProcessTimeRange(unittest.TestCase):
maxDiff = None
def stub_therapy_timeline(self, time_start, time_end):
pass
def stub_therapy_timeline_csv(self, time_start, time_end):
return {
"readingData": [],
"iobData": [],
"basalData": [],
"bolusData": []
}
def stub_last_uploaded_entry(self, event_type):
return None
def stub_last_uploaded_activity(self, activity_type):
return None
"""No data in Nightscout. Uploads all basal data from tconnect."""
def test_new_ciq_basal_data(self):
tconnect = TConnectApi()
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
def fake_therapy_timeline(time_start, time_end):
self.assertEqual(time_start, start)
self.assertEqual(time_end, end)
return TestBasalSync.get_example_ciq_basal_events()
tconnect.controliq.therapy_timeline = fake_therapy_timeline
tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv
nightscout = NightscoutApi()
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False)
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 4)
self.assertDictEqual(dict(nightscout.uploaded_entries), {
"treatments": [
NightscoutEntry.basal(0.8, 20.35, "2021-03-16 00:00:00-04:00", reason="tempDelivery"),
NightscoutEntry.basal(0.799, 5.0, "2021-03-16 00:20:21-04:00", reason="profileDelivery"),
NightscoutEntry.basal(0.797, 5.0, "2021-03-16 00:25:21-04:00", reason="algorithmDelivery"),
NightscoutEntry.basal(0, 2693/60, "2021-03-16 00:30:21-04:00", reason="algorithmDelivery")
]})
self.assertDictEqual(nightscout.put_entries, {})
self.assertDictEqual(nightscout.deleted_entries, {})
"""Two basal entries in Nightscout. Two new basal entries in tconnect."""
def test_partial_ciq_basal_data(self):
tconnect = TConnectApi()
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
def fake_therapy_timeline(time_start, time_end):
self.assertEqual(time_start, start)
self.assertEqual(time_end, end)
return TestBasalSync.get_example_ciq_basal_events()
tconnect.controliq.therapy_timeline = fake_therapy_timeline
tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv
nightscout = NightscoutApi()
def fake_last_uploaded_entry(event_type):
if event_type == "Temp Basal":
return {
"created_at": "2021-03-16 00:20:21-04:00",
"duration": 5
}
nightscout.last_uploaded_entry = fake_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False)
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 2)
self.assertDictEqual(dict(nightscout.uploaded_entries), {
"treatments": [
NightscoutEntry.basal(0.797, 5.0, "2021-03-16 00:25:21-04:00", reason="algorithmDelivery"),
NightscoutEntry.basal(0, 2693/60, "2021-03-16 00:30:21-04:00", reason="algorithmDelivery")
]})
self.assertDictEqual(nightscout.put_entries, {})
self.assertDictEqual(nightscout.deleted_entries, {})
"""
Two basal entries in Nightscout, the latter which needs to be updated
with a longer duration. Two entirely new entries in tconnect."""
def test_with_updated_duration_ciq_basal_data(self):
tconnect = TConnectApi()
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
def fake_therapy_timeline(time_start, time_end):
self.assertEqual(time_start, start)
self.assertEqual(time_end, end)
return TestBasalSync.get_example_ciq_basal_events()
tconnect.controliq.therapy_timeline = fake_therapy_timeline
tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv
nightscout = NightscoutApi()
def fake_last_uploaded_entry(event_type):
if event_type == "Temp Basal":
return {
"created_at": "2021-03-16 00:20:21-04:00",
"duration": 3,
"_id": "nightscout_id"
}
nightscout.last_uploaded_entry = fake_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False)
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 2)
self.assertDictEqual(nightscout.uploaded_entries, {
"treatments": [
NightscoutEntry.basal(0.797, 5.0, "2021-03-16 00:25:21-04:00", reason="algorithmDelivery"),
NightscoutEntry.basal(0, 2693/60, "2021-03-16 00:30:21-04:00", reason="algorithmDelivery")
]})
self.assertEqual(len(nightscout.put_entries["treatments"]), 1)
self.assertDictEqual(dict(nightscout.put_entries), {
"treatments": [
{
"_id": "nightscout_id",
**NightscoutEntry.basal(0.799, 5.0, "2021-03-16 00:20:21-04:00", reason="profileDelivery")
}
]
})
self.assertDictEqual(nightscout.deleted_entries, {})
"""No data in Nightscout. Uploads all bolus data from tconnect."""
def test_new_ciq_bolus_data(self):
tconnect = TConnectApi()
start = datetime.datetime(2021, 4, 20, 12, 0)
end = datetime.datetime(2021, 4, 21, 12, 0)
tconnect.controliq.therapy_timeline = self.stub_therapy_timeline
def fake_therapy_timeline_csv(time_start, time_end):
return {
**self.stub_therapy_timeline_csv(time_start, time_end),
"bolusData": TestBolusSync.get_example_csv_bolus_events(),
}
tconnect.ws2.therapy_timeline_csv = fake_therapy_timeline_csv
nightscout = NightscoutApi()
nightscout.last_uploaded_entry = self.stub_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
process_time_range(tconnect, nightscout, start, end, pretend=False)
pprint.pprint(nightscout.uploaded_entries)
self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 3)
self.assertDictEqual(dict(nightscout.uploaded_entries), {
"treatments": [
NightscoutEntry.bolus(13.53, 75, "2021-04-01 12:58:26-04:00", notes="Standard/Correction"),
NightscoutEntry.bolus(1.25, 0, "2021-04-01 23:23:17-04:00", notes="Standard (Override)"),
NightscoutEntry.bolus(1.7, 0, "2021-04-02 01:00:47-04:00", notes="Automatic Bolus/Correction"),
]})
self.assertDictEqual(nightscout.put_entries, {})
self.assertDictEqual(nightscout.deleted_entries, {})
if __name__ == '__main__':
unittest.main()