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

Translate Tandem profile information to Nightscout format

Browse Source
This commit is contained in:
James Woglom
2023-01-16 14:25:31 -05:00
parent df4e9e4ed0
commit d69a760f5b
8 changed files with 352 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tconnectsync/api/webui.py
+29 -5
View File
@@ -1,4 +1,4 @@
from typing import List
from typing import List, Tuple
import requests
import urllib
import datetime
@@ -8,7 +8,7 @@ import logging
from bs4 import BeautifulSoup
from tconnectsync.domain.device_settings import Device, Profile, ProfileSegment
from tconnectsync.domain.device_settings import Device, Profile, ProfileSegment, DeviceSettings
from tconnectsync.util import removesuffix, removeprefix
from tconnectsync.util.constants import MMOLL_TO_MGDL
@@ -108,9 +108,9 @@ class WebUIScraper:
"""
Returns a parsed representation of a pump's settings.
Note that pump_guid is NOT the serial number of the pump, and
should be obtained from my_devices()[str(serial_number)]['guid']
should be obtained from my_devices()[str(serial_number)].guid
"""
def device_settings_from_guid(self, pump_guid: str) -> List[Profile]:
def device_settings_from_guid(self, pump_guid: str) -> Tuple[List[Profile], DeviceSettings]:
profiles = []
settings = {}
r = self.get('myaccount/DeviceSettings.aspx?guid=%s' % pump_guid)
@@ -126,7 +126,15 @@ class WebUIScraper:
else:
settings.update(self._parse_settings_tbl(tbl))
return profiles, settings
low_bg_threshold, high_bg_threshold = self._extract_bg_thresholds(settings)
dev_settings = DeviceSettings(
low_bg_threshold=low_bg_threshold,
high_bg_threshold=high_bg_threshold,
raw_settings=settings
)
return profiles, dev_settings
def _parse_profile_tbl(self, tbl) -> Profile:
profile = {}
@@ -246,6 +254,22 @@ class WebUIScraper:
return settings
def _extract_bg_thresholds(self, settings):
# Nightscout needs default values
low_bg_threshold = 70
high_bg_threshold = 180
if 'CGM Alerts' in settings:
if 'Low Alert' in settings['CGM Alerts']:
low = settings['CGM Alerts']['Low Alert']
if low['value']:
low_bg_threshold = int(low['text'].split(' mg/dL')[0])
if 'High Alert' in settings['CGM Alerts']:
high = settings['CGM Alerts']['High Alert']
if high['value']:
high_bg_threshold = int(high['text'].split(' mg/dL')[0])
return low_bg_threshold, high_bg_threshold
"""
Wraps a call to my_devices to identify the device GUID from the
given pump serial, and then returns device_settings_from_guid.
tconnectsync/domain/device_settings.py
+7
View File
@@ -25,3 +25,10 @@ class Profile:
calculated_total_daily_basal: float # in units
insulin_duration_min: int
carbs_enabled: bool
# Settings stored globally in the pump that are stored per-profile in Nightscout
@dataclass
class DeviceSettings:
low_bg_threshold: int
high_bg_threshold: int
raw_settings: dict
tconnectsync/nightscout.py
+25
View File
@@ -145,3 +145,28 @@ class NightscoutApi:
if status.status_code != 200:
raise Exception('HTTP error status code (%d) from Nightscout: %s' % (status.status_code, status.text))
return status.json()
"""
Returns information on configured Nightscout profiles, over the optional time range.
Will only return the most recent profile for the given range.
"""
def profiles(self, time_start=None, time_end=None):
def internal(t_to_space):
dateFilter = time_range('created_at', time_start, time_end, t_to_space=t_to_space)
latest = requests.get(urljoin(self.url, 'api/v1/profile.json?' + dateFilter + '&ts=' + str(time.time())), headers={
'api-secret': hashlib.sha1(self.secret.encode()).hexdigest()
}, verify=self.verify)
if latest.status_code != 200:
raise ApiException(latest.status_code, "Nightscout profiles response: %s" % latest.text)
j = latest.json()
if j and len(j) > 0:
return j[0]
return None
ret = internal(False)
if ret is None and (time_start or time_end):
ret = internal(True)
if ret is not None:
logger.warning("profiles with time_start=%s time_end=%s only returned data when timestamps contained a space" % (time_start, time_end))
return ret
tconnectsync/parser/nightscout.py
+73
View File
@@ -1,5 +1,8 @@
import arrow
from ..domain.device_settings import Profile, DeviceSettings
from ..secret import TIMEZONE_NAME, NIGHTSCOUT_PROFILE_CARBS_HR_VALUE, NIGHTSCOUT_PROFILE_DELAY_VALUE
ENTERED_BY = "Pump (tconnectsync)"
BASAL_EVENTTYPE = "Temp Basal"
@@ -107,5 +110,75 @@ class NightscoutEntry:
"enteredBy": ENTERED_BY
}
# Tandem-scraped profile to Nightscout profile store entry
@staticmethod
def profile_store(profile: Profile, device_settings: DeviceSettings) -> dict:
return {
# insulin duration in hours
"dia": (profile.insulin_duration_min / 60),
"carbratio": [
{
"time": tandem_to_ns_time(segment.time),
"timeAsSeconds": tandem_to_ns_time_seconds(segment.time),
"value": segment.carb_ratio
} for segment in profile.segments
],
"carbs_hr": NIGHTSCOUT_PROFILE_CARBS_HR_VALUE,
"delay": NIGHTSCOUT_PROFILE_DELAY_VALUE,
"sens": [ # Correction factor
{
"time": tandem_to_ns_time(segment.time),
"timeAsSeconds": tandem_to_ns_time_seconds(segment.time),
"value": segment.correction_factor
} for segment in profile.segments
],
"basal": [
{
"time": tandem_to_ns_time(segment.time),
"timeAsSeconds": tandem_to_ns_time_seconds(segment.time),
"value": segment.basal_rate
} for segment in profile.segments
],
"target_low": [
{
"time": "00:00",
"timeAsSeconds": 0,
"value": device_settings.low_bg_threshold
}
],
"target_high": [
{
"time": "00:00",
"timeAsSeconds": 0,
"value": device_settings.high_bg_threshold
}
],
"timezone": TIMEZONE_NAME, # tconnectsync settings timezone
"startDate": "1970-01-01T00:00:00.000Z",
"units": "mg/dl"
}
def tandem_to_ns_time(tandem_time: str) -> str:
numbers, ampm = tandem_time.split(' ')
hr, min = numbers.split(':')
if ampm.lower().strip() == 'am':
return "%02d:%02d" % (int(hr) % 12, int(min))
elif ampm.lower().strip() == 'pm':
return "%02d:%02d" % (12 + (int(hr) % 12), int(min))
raise InvalidTimeException(tandem_time)
def tandem_to_ns_time_seconds(tandem_time: str) -> int:
numbers, ampm = tandem_time.split(' ')
hr, min = numbers.split(':')
if ampm.lower().strip() == 'am':
return 60 * (60 * (int(hr) % 12) + int(min))
elif ampm.lower().strip() == 'pm':
return 60 * (60 * (12 + (int(hr) % 12)) + int(min))
raise InvalidTimeException(tandem_time)
class InvalidBolusTypeException(RuntimeError):
pass
class InvalidTimeException(RuntimeError):
pass
tconnectsync/secret.py
+4
View File
@@ -59,6 +59,10 @@ AUTOUPDATE_FAILURE_MINUTES = get_number('AUTOUPDATE_FAILURE_MINUTES', '15') # 15
AUTOUPDATE_RESTART_ON_FAILURE = get_bool('AUTOUPDATE_RESTART_ON_FAILURE', 'true')
AUTOUPDATE_MAX_LOOP_INVOCATIONS = get_number('AUTOUPDATE_MAX_LOOP_INVOCATIONS', '-1')
# Default Nightscout profile segment fields which aren't stored by Tandem
NIGHTSCOUT_PROFILE_CARBS_HR_VALUE = get('NIGHTSCOUT_PROFILE_CARBS_HR_VALUE', '20')
NIGHTSCOUT_PROFILE_DELAY_VALUE = get('NIGHTSCOUT_PROFILE_DELAY_VALUE', '20')
ENABLE_TESTING_MODES = get_bool('ENABLE_TESTING_MODES', 'false')
SKIP_NS_LAST_UPLOADED_CHECK = get_bool('SKIP_NS_LAST_UPLOADED_CHECK', 'false')
REQUESTS_PROXY = get('REQUESTS_PROXY', '')
tconnectsync/sync/profile.py
+45
View File
@@ -0,0 +1,45 @@
from typing import List, Tuple
import logging
from ..api import TConnectApi
from ..domain.device_settings import Profile
from ..secret import PUMP_SERIAL_NUMBER
logger = logging.getLogger(__name__)
def get_pump_profiles(tconnect: TConnectApi) -> List[Profile]:
all_devices = tconnect.webui.my_devices()
if str(PUMP_SERIAL_NUMBER) not in all_devices:
logger.warn("Could not find entry for provided pump serial number in t:connect device list: %s, received: %s", PUMP_SERIAL_NUMBER, all_devices)
return []
device = all_devices[str(PUMP_SERIAL_NUMBER)]
logger.info("Getting profile settings for %s", device)
device_profiles, device_settings = tconnect.webui.device_settings_from_guid(device.guid)
logger.debug("device_profiles: %s", device_profiles)
logger.debug("device_settings: %s", device_settings)
logger.info("Found pump profiles: %s", ["%s%s" % (profile.title, " (active)" if profile.active else "") for profile in device_profiles])
return device_profiles
"""
Compare pump device and Nightscout profiles, and return a final dictionary of
Nightscout profile objects, with the pump profile settings overriding what is
currently in Nightscout.
ns_profile_obj is the output from NightscoutApi.profiles() and should be the most
recent profile object in mongo.
"""
def compare_profiles(device_profiles: List[Profile], ns_profile_obj: dict):
device = {profile.title: profile for profile in device_profiles}
ns = ns_profile_obj.get('store', {})
logger.info("compare_profiles profile names: device: %s ns: %s", device.keys(), ns.keys())
missing_profiles_in_ns = set(device.keys()) - set(ns.keys())
for profile_name in missing_profiles_in_ns:
logger.info("Missing profile in Nightscout: %s: %s", profile_name, device.get(profile_name))
pump_configured_profile = device[profile_name]
tests/api/test_webui.py
+3 -2
View File
@@ -953,7 +953,6 @@ class TestWebUIScraper(unittest.TestCase):
with requests_mock.Mocker() as m:
m.get('https://tconnect.tandemdiabetes.com/myaccount/DeviceSettings.aspx?guid=00000000-0000-0000-0000-000000000001',
request_headers=base_headers(),
text=self.PUMP_SETTINGS_HTML)
profiles, settings = webui.device_settings_from_guid('00000000-0000-0000-0000-000000000001')
@@ -994,7 +993,7 @@ class TestWebUIScraper(unittest.TestCase):
'carbs_enabled': True
})])
self.assertDictEqual(settings, {
self.assertDictEqual(settings.raw_settings, {
'Alerts': {
'Alert: Auto-Off': {'text': '18 hrs', 'value': True},
'Alert: Low Insulin': {'text': '35 u'}
@@ -1043,6 +1042,8 @@ class TestWebUIScraper(unittest.TestCase):
},
'upload_date': 'Jan 09, 2022'
})
self.assertEqual(settings.low_bg_threshold, 80)
self.assertEqual(settings.high_bg_threshold, 200)
tests/parser/test_nightscout.py
+165 -1
View File
@@ -1,9 +1,12 @@
#!/usr/bin/env python3
import unittest
from tconnectsync.parser.nightscout import NightscoutEntry, InvalidBolusTypeException
from tconnectsync.parser.nightscout import NightscoutEntry, InvalidBolusTypeException, tandem_to_ns_time, tandem_to_ns_time_seconds
from tconnectsync.domain.device_settings import Profile, ProfileSegment, DeviceSettings
from tconnectsync.secret import NIGHTSCOUT_PROFILE_CARBS_HR_VALUE, NIGHTSCOUT_PROFILE_DELAY_VALUE, TIMEZONE_NAME
class TestNightscoutEntry(unittest.TestCase):
maxDiff = None
def test_basal(self):
self.assertEqual(
NightscoutEntry.basal(
@@ -192,5 +195,166 @@ class TestNightscoutEntry(unittest.TestCase):
)
def test_profile_store(self):
self.assertEqual(
NightscoutEntry.profile_store(
profile=Profile(
title='A',
active=True,
segments=[
ProfileSegment(
display_time='Midnight',
time='12:00 AM',
basal_rate=0.8,
correction_factor=30.0,
carb_ratio=6.0,
target_bg_mgdl=110.0),
ProfileSegment(
display_time='6:00 AM',
time='6:00 AM',
basal_rate=1.25,
correction_factor=30.0,
carb_ratio=6.0,
target_bg_mgdl=110.0),
ProfileSegment(
display_time='11:00 AM',
time='11:00 AM',
basal_rate=1.0,
correction_factor=30.0,
carb_ratio=6.0,
target_bg_mgdl=110.0),
ProfileSegment(
display_time='Noon',
time='12:00 PM',
basal_rate=0.8,
correction_factor=30.0,
carb_ratio=6.0,
target_bg_mgdl=110.0)
],
calculated_total_daily_basal=21.65,
insulin_duration_min=300,
carbs_enabled=True
),
device_settings=DeviceSettings(
low_bg_threshold=80,
high_bg_threshold=200,
raw_settings={}
)
),
{
"dia": 5.0,
"carbratio": [
{
"time": "00:00",
"timeAsSeconds": 0,
"value": 6.0
},
{
"time": "06:00",
"timeAsSeconds": 6*60*60,
"value": 6.0
},
{
"time": "11:00",
"timeAsSeconds": 11*60*60,
"value": 6.0
},
{
"time": "12:00",
"timeAsSeconds": 12*60*60,
"value": 6.0
}
],
"carbs_hr": NIGHTSCOUT_PROFILE_CARBS_HR_VALUE,
"delay": NIGHTSCOUT_PROFILE_DELAY_VALUE,
"sens": [
{
"time": "00:00",
"timeAsSeconds": 0,
"value": 30.0
},
{
"time": "06:00",
"timeAsSeconds": 6*60*60,
"value": 30.0
},
{
"time": "11:00",
"timeAsSeconds": 11*60*60,
"value": 30.0
},
{
"time": "12:00",
"timeAsSeconds": 12*60*60,
"value": 30.0
}
],
"basal": [
{
"time": "00:00",
"timeAsSeconds": 0,
"value": 0.8
},
{
"time": "06:00",
"timeAsSeconds": 6*60*60,
"value": 1.25
},
{
"time": "11:00",
"timeAsSeconds": 11*60*60,
"value": 1.0
},
{
"time": "12:00",
"timeAsSeconds": 12*60*60,
"value": 0.8
}
],
"target_low": [
{
"time": "00:00",
"timeAsSeconds": 0,
"value": 80
}
],
"target_high": [
{
"time": "00:00",
"timeAsSeconds": 0,
"value": 200
}
],
"timezone": TIMEZONE_NAME,
"startDate": "1970-01-01T00:00:00.000Z",
"units": "mg/dl"
}
)
class TestTandemNightscoutTime(unittest.TestCase):
def test_tandem_to_ns_time(self):
self.assertEqual(tandem_to_ns_time('12:00 AM'), '00:00')
self.assertEqual(tandem_to_ns_time('12:30 AM'), '00:30')
self.assertEqual(tandem_to_ns_time('6:00 AM'), '06:00')
self.assertEqual(tandem_to_ns_time('6:30 AM'), '06:30')
self.assertEqual(tandem_to_ns_time('11:30 AM'), '11:30')
self.assertEqual(tandem_to_ns_time('12:00 PM'), '12:00')
self.assertEqual(tandem_to_ns_time('12:30 PM'), '12:30')
self.assertEqual(tandem_to_ns_time('06:30 PM'), '18:30')
self.assertEqual(tandem_to_ns_time('11:30 PM'), '23:30')
def test_tandem_to_ns_time_seconds(self):
self.assertEqual(tandem_to_ns_time_seconds('12:00 AM'), 0)
self.assertEqual(tandem_to_ns_time_seconds('12:30 AM'), 30*60)
self.assertEqual(tandem_to_ns_time_seconds('6:00 AM'), 6*60*60)
self.assertEqual(tandem_to_ns_time_seconds('6:30 AM'), 6*60*60 + 30*60)
self.assertEqual(tandem_to_ns_time_seconds('11:30 AM'), 11*60*60 + 30*60)
self.assertEqual(tandem_to_ns_time_seconds('12:00 PM'), 12*60*60)
self.assertEqual(tandem_to_ns_time_seconds('12:30 PM'), 12*60*60 + 30*60)
self.assertEqual(tandem_to_ns_time_seconds('06:30 PM'), 12*60*60 + 6*60*60 + 30*60)
self.assertEqual(tandem_to_ns_time_seconds('11:30 PM'), 12*60*60 + 11*60*60 + 30*60)
if __name__ == '__main__':
unittest.main()