Remove dead code for legacy pre-Tandem Source APIs

Since the 2.0 migration to Tandem Source, the live sync path
(api.tandemsource + sync/tandemsource/*) no longer references the
legacy t:connect APIs. This removes that now-unreachable code.

Removed modules:
- api/controliq.py, api/ws2.py, api/android.py, api/webui.py
  (the legacy controliq / tconnectws2 / android / webui clients)
- process.py (old process_time_range; already broken since it
  imported sync submodules that no longer exist)
- parser/ciq_therapy_events.py, parser/tconnect.py (TConnectEntry)
- domain/therapy_event.py, domain/bolus.py, domain/device_settings.py,
  domain/utility.py

Trimmed dead wiring from live modules:
- api/__init__.py: dropped the controliq/ws2/android/webui properties,
  keeping only the tandemsource accessor
- check.py: removed the unused TConnectEntry import
- parser/nightscout.py: removed the unused legacy profile_store() plus
  the now-orphaned tandem_to_ns_time / tandem_to_ns_time_seconds helpers
  and InvalidTimeException (the live path uses tandemsource_profile_store)

Tests: removed suites covering the deleted modules; pared tests/api/fake.py
down to the TConnectApi fake still used by the tandemsource tests. README
"Tandem APIs" sections updated to reflect the single Tandem Source API.

Full test suite passes (48 passed, 1 skipped).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQNn3mBG1kXTAdQb9c2jfW
This commit is contained in:
Claude
2026-06-30 19:12:44 -04:00
committed by James Woglom
parent e5195b2613
commit ea7cc8f4ec
24 changed files with 9 additions and 5115 deletions
+6 -8
View File
@@ -419,13 +419,11 @@ If main.py doesn't exist in `C:\Users\<USERNAME>\AppData\Local\Programs\Python\<
## Tandem APIs
This application utilizes three separate Tandem APIs for obtaining t:connect data, referenced here by the identifying part of their URLs:
As of version 2.0, tconnectsync retrieves all of its data from a single Tandem API, [**tandemsource**](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/api/tandemsource.py), which powers [Tandem Source](https://source.tandemdiabetes.com/). After logging in, tconnectsync fetches the list of pumps on the account along with a stream of raw pump event data, which is decoded locally (see [`tconnectsync/eventparser`](https://github.com/jwoglom/tconnectsync/tree/master/tconnectsync/eventparser)) to extract basal, bolus, CGM, and other pump events.
* [**controliq**](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/api/controliq.py) - Contains Control:IQ related data, namely a timeline of all Basal events uploaded by the pump, separated by type (temp basals, algorithmically-updated basals, or profile-updated basals). Additionally includes CGM and Bolus data.
* [**android**](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/api/android.py) - Used internally by the t:connect Android app, these API endpoints were discovered by reverse-engineering the Android app. Most of the API endpoints are used for uploading pump data, and tconnectsync uses one endpoint which returns the most recent event ID uploaded by the pump, so we know when more data has been uploaded.
* [**tconnectws2**](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/api/ws2.py) - More legacy than the others, this seems to power the bulk of the main t:connect website. It is used as a last resort due to severe performance issues with this API (see https://github.com/jwoglom/tconnectsync/issues/43). We can use it to retrieve a CSV export of non-ControlIQ basal data, as well as bolus and IOB data. It is only used for bolus data as a fallback, and for pump-reported IOB data if requested. Full tracking of pump events also uses a limited version of this API.
> Earlier versions of tconnectsync (1.x) instead used three separate legacy t:connect APIs (`controliq`, `android`, and `tconnectws2`). Those APIs — and the code supporting them — were removed once t:connect was shut down in favor of Tandem Source.
I have only tested tconnectsync with a Tandem pump set in the US Eastern timezone. Tandem's (to us, undocumented) APIs are [a bit loose with timezones](https://github.com/jwoglom/tconnectsync/blob/master/tconnectsync/parser.py#L15), so please let me know if you notice any timezone-related bugs.
I have only tested tconnectsync with a Tandem pump set in the US Eastern timezone. Tandem's (to us, undocumented) APIs are a bit loose with timezones, so please let me know if you notice any timezone-related bugs.
## Backfilling t:connect Data
To backfill existing t:connect data in to Nightscout, you can use the `--start-date` and `--end-date` options. For example, the following will upload all t:connect data between January 1st and March 1st, 2020 to Nightscout:
@@ -438,14 +436,14 @@ In order to bulk-import a lot of data, you may need to use shorter intervals, an
One oddity when backfilling data is that the Control:IQ specific API endpoints return errors if they are queried before you updated your pump to utilize Control:IQ. This is [partially worked around in tconnectsync's code](https://github.com/jwoglom/tconnectsync/blob/d841c3811aeff3671d941a7d3ff4b80cce6a219e/main.py#L238), but you might need to update the logic if you did not switch to a Control:IQ enabled pump immediately after launch.
## t:connect API Testing
## Tandem Source API Testing
To test t:connect API endpoints in a Python shell, you can do something like the following:
To test Tandem Source API endpoints in a Python shell, you can do something like the following:
```python
import tconnectsync
tconnectsync.util.cli.enable_logging()
api = tconnectsync.util.cli.get_api()
# Make API calls, e.g.
therapy_timeline = api.controliq.therapy_timeline('2022-08-01', '2022-08-10')
pumps = api.tandemsource.pump_event_metadata()
```
+1 -55
View File
@@ -1,14 +1,10 @@
import logging
from .android import AndroidApi
from .controliq import ControlIQApi
from .ws2 import WS2Api
from .webui import WebUIScraper
from .tandemsource import TandemSourceApi
logger = logging.getLogger(__name__)
"""A wrapper for the three different t:connect API types."""
"""A wrapper for the Tandem Source API."""
class TConnectApi:
email = None
password = None
@@ -17,10 +13,6 @@ class TConnectApi:
self.email = email
self.password = password
self.region = region
self._ciq = None
self._ws2 = None
self._android = None
self._webui = None
self._tandemsource = None
@property
@@ -32,49 +24,3 @@ class TConnectApi:
self._tandemsource = TandemSourceApi(self.email, self.password, self.region)
return self._tandemsource
@property
def controliq(self):
if self._ciq and not self._ciq.needs_relogin():
return self._ciq
logger.debug("Instantiating new ControlIQApi")
self._ciq = ControlIQApi(self.email, self.password)
return self._ciq
@property
def ws2(self):
if self._ws2:
return self._ws2
logger.debug("Instantiating new WS2Api")
# Trigger login or re-login via controliq api if necessary
# so userGuid can be accessed from it
self.controliq
self._ws2 = WS2Api(self._ciq.userGuid)
return self._ws2
@property
def android(self):
if self._android and not self._android.needs_relogin():
return self._android
logger.debug("Instantiating new AndroidApi")
self._android = AndroidApi(self.email, self.password)
return self._android
@property
def webui(self):
if self._webui and not self._webui.needs_relogin():
return self._webui
logger.debug("Instantiating new WebUIScraper")
self._webui = WebUIScraper(self.controliq)
return self._webui
-175
View File
@@ -1,175 +0,0 @@
import requests
import json
import urllib
import datetime
import csv
import base64
import arrow
import time
import logging
from bs4 import BeautifulSoup
from ..util import timeago
from .common import ApiException, ApiLoginException, parse_date, base_session
logger = logging.getLogger(__name__)
"""
The AndroidApi class contains methods which are queried in the t:connect
Android application. These methods are a part of the tdc API which require
Android specific credentials.
"""
class AndroidApi:
BASE_URL = 'https://tdcservices.tandemdiabetes.com/'
OAUTH_TOKEN_PATH = 'cloud/oauth2/token'
OAUTH_SCOPES = 'cloud.account cloud.upload cloud.accepttcpp cloud.email cloud.password'
# These credentials are found in source code
ANDROID_API_USERNAME = base64.b64decode('QzIzMzFDRDYtRDQ1MC00OTVFLTlDMTktNjcyMTUyMzBDODVD').decode()
ANDROID_API_PASSWORD = base64.b64decode('dHo0MzNLVzVRREM5VjdmIXo2QF4ybyZZNlNHR1lo').decode()
# These credentials are used by tconnect web
TCONNECT_WEB_USERNAME = base64.b64decode('M0U2MzU3QkEtRjYyNS00REQyLUI2NUYtNEI1RTgxNDRBQTZG').decode()
TCONNECT_WEB_PASSWORD = base64.b64decode('cUMyaXFIc2w3OFFoR0RYdCpMenFwb1pxZTl3eHN6').decode()
ANDROID_USER_AGENT = 'Dalvik/2.1.0 (Linux; U; Android 12; Pixel 4a Build/SP2A.220305.012)'
# These tokens are separate from the "standard" tdcservices API
accessToken = None
accessTokenExpiresAt = None
refreshToken = None
refreshTokenExpiresAt = None
userId = None
patientObjectId = None
def __init__(self, email, password):
self.session = base_session()
self.login(email, password)
self._email = email
self._password = password
def login(self, email, password):
r = self.session.post(
self.BASE_URL + self.OAUTH_TOKEN_PATH,
{
'username': email,
'password': password,
'grant_type': 'password',
'scope': self.OAUTH_SCOPES
},
headers={
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'User-Agent': self.ANDROID_USER_AGENT
},
auth=requests.auth.HTTPBasicAuth(self.ANDROID_API_USERNAME, self.ANDROID_API_PASSWORD)
)
if r.status_code != 200:
raise ApiLoginException(r.status_code, 'Received HTTP %s during login: %s' % (r.status_code, r.text))
j = r.json()
# tconnect web returns a null user
# if "user" not in j or not j["user"]:
# raise ApiException(r.status_code, 'No user details present in AndroidApi oauth response: %s' % r.text)
self.accessToken = j["accessToken"]
self.accessTokenExpiresAt = j["accessTokenExpiresAt"]
# NOTE: the refresh token is currently unused, instead a new access
# token is obtained from scratch by re-logging in when it expires.
if "refreshToken" in j and "refreshTokenExpiresAt" in j:
self.refreshToken = j["refreshToken"]
self.refreshTokenExpiresAt = j["refreshTokenExpiresAt"]
self.userId = j["user"]["id"]
logger.info("Logged in to AndroidApi successfully (expiration: %s, %s)" % (self.accessTokenExpiresAt, timeago(self.accessTokenExpiresAt)))
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')
return {'Authorization': 'Bearer %s' % self.accessToken}
def _get(self, endpoint, query={}, **kwargs):
r = self.session.get(self.BASE_URL + endpoint, data=query, headers={
'User-Agent': self.ANDROID_USER_AGENT,
'Content-Type': 'application/json',
**self.api_headers()
}, **kwargs)
if r.status_code != 200:
raise ApiException(r.status_code, "Android API HTTP %s response: %s" % (str(r.status_code), r.text))
return r.json()
def get(self, endpoint, query={}, tries=0, **kwargs):
try:
return self._get(endpoint, query, **kwargs)
except ApiException as e:
if tries > 0:
raise ApiException(e.status_code, "Android API HTTP %s on retry #%d: %s" % (e.status_code, tries, e))
# Trigger automatic re-login, and try again once
if e.status_code == 401:
self.accessTokenExpiresAt = time.time()
self.login(self._email, self._password)
return self.get(endpoint, query, tries=tries+1, **kwargs)
if e.status_code == 500:
return self.get(endpoint, query, tries=tries+1, **kwargs)
raise e
def post(self, endpoint, query={}, **kwargs):
r = self.session.post(self.BASE_URL + endpoint, query, headers=self.api_headers(), **kwargs)
if r.status_code != 200:
raise ApiException(r.status_code, "Internal API HTTP %s response: %s" % (str(r.status_code), r.text))
return r.json()
"""
Returns the most recent event ID that was uploaded for the given pump.
{'maxPumpEventIndex': <integer>, 'processingStatus': 1}
"""
def last_event_uploaded(self, pump_serial_number):
return self.get('cloud/upload/getlasteventuploaded?sn=%d' % pump_serial_number)
"""
Returns user login information about a tconnect account.
{'firstName': <string>, 'lastName': <string>, 'birthDate': 'YYYY-MM-DDT00:00:00.000Z',
'emailAddress': <string>, 'secretQuestion': <string>, 'secretAnswer': <string>,
'secretQuestionId': <integer>}
"""
def patient_info(self):
return self.get('cloud/account/patient_info')
# TODO: these methods are used in the web app, not the Android app,
# but support the same auth tokens and are on this domain. They should
# be moved to a new Api class.
# 3/17/2022: the API appears to be more stringently checking scopes,
# and some of these endpoints no longer work with the API token scoped
# to the Android app.
"""
Returns BG and pump threshold values.
{'targetBGHigh': <integer>, 'targetBGLow': <integer>, 'hypoThreshold': <integer>,
'hyperThreshold': <integer>, 'siteChangeThreshold': <integer>,
'cartridgeChangeThreshold': <integer>, 'tubingChangeThreshold': <integer>}
"""
def therapy_thresholds(self):
return self.get('cloud/usersettings/api/therapythresholds?userId=%s' % self.userId)
"""
Returns therapy-related user information about a tconnect account.
{'userID': <string>, 'targetBgHigh': <integer>, 'targetBgLow': <integer>,
'hypoThreshold': <integer>, 'hyperThreshold': <integer>,
'dateOfBirth': 'YYYY-MM-DDT00:00:00', 'age': <integer>,
'patientFullName': <string>, 'caregiverDateOfBirth': <string>,
'hasCGM': <bool>, 'hasBASALIQ': <bool>, 'hasControlIQ': <bool>}
"""
def user_profile(self):
return self.get('cloud/usersettings/api/UserProfile?userId=%s' % self.userId)
-206
View File
@@ -1,206 +0,0 @@
import urllib
import arrow
import time
import logging
from bs4 import BeautifulSoup
from ..util import timeago, cap_length
from .common import parse_date, base_headers, base_session, ApiException, ApiLoginException
logger = logging.getLogger(__name__)
class ControlIQApi:
BASE_URL = 'https://tdcservices.tandemdiabetes.com/'
LOGIN_URL = 'https://tconnect.tandemdiabetes.com/login.aspx?ReturnUrl=%2f'
LAST_CONFIRMED_SOFTWARE_VERSION = 't:connect 7.17.2.3'
userGuid = None
accessToken = None
accessTokenExpiresAt = None
tconnect_software_ver = None
def __init__(self, email, password):
self.login(email, password)
self._email = email
self._password = password
def login(self, email, password):
logger.info("Logging in to ControlIQApi...")
with base_session() as s:
initial = s.get(self.LOGIN_URL, headers=base_headers())
soup = BeautifulSoup(initial.content, features='lxml')
data = self._build_login_data(email, password, soup)
req = s.post(self.LOGIN_URL, data=data, headers={'Referer': self.LOGIN_URL, **base_headers()}, allow_redirects=False)
# HTTP 200 is reported when credentials are incorrect
if req.status_code == 200:
login_error = self._find_login_error(req.text)
if not login_error:
login_error = 'Check your login credentials.'
raise ApiLoginException(None, 'Error logging in to t:connect: %s' % login_error)
if req.status_code != 302:
raise ApiLoginException(req.status_code, 'Error logging in to t:connect')
fwd = s.post(urllib.parse.urljoin(self.LOGIN_URL, req.headers['Location']), cookies=req.cookies, headers=base_headers())
if fwd.status_code != 200:
logger.warn("Received non-HTTP 200: %s" % req.text)
raise ApiException(fwd.status_code, 'Error retrieving t:connect login cookies.')
self.userGuid = req.cookies['UserGUID']
if 'accessToken' in req.cookies and 'accessTokenExpiresAt' in req.cookies:
self.accessToken = req.cookies['accessToken']
self.accessTokenExpiresAt = req.cookies['accessTokenExpiresAt']
logger.info("Logged in to ControlIQApi successfully via accessToken cookie (expiration: %s, %s)" % (self.accessTokenExpiresAt, timeago(self.accessTokenExpiresAt)))
else:
logger.info("No accessToken cookie found when logging in to ControlIQApi. Triggering AndroidApi auth")
from .android import AndroidApi
android = AndroidApi(email, password)
self.accessToken = android.accessToken
self.accessTokenExpiresAt = android.accessTokenExpiresAt
logger.info("Logged in to AndroidApi successfully via accessToken param (expiration: %s, %s)" % (self.accessTokenExpiresAt, timeago(self.accessTokenExpiresAt)))
self.loginSession = s
return True
def _build_login_data(self, email, password, soup):
try:
version = soup.select_one("#footer_version").text.strip()
self.tconnect_software_ver = version
logger.info("Reported tconnect software version: %s" % version)
if version != self.LAST_CONFIRMED_SOFTWARE_VERSION:
logger.warning("Newer API version than last confirmed working. Saw %s and expected %s" % (version, self.LAST_CONFIRMED_SOFTWARE_VERSION))
logger.warning("If you experience any issues, please report them to https://github.com/jwoglom/tconnectsync")
except Exception:
logger.warning("Unable to find tconnect software version.")
contents = "<unknown>"
if soup:
contents = "%s" % soup.encode_contents()
if len(contents) > 1000:
contents = "%s[SNIP]%s" % (contents[:500], contents[-500:])
logger.info("BeautifulSoup parsed contents: %s" % contents)
pass
if not soup.select_one("#__VIEWSTATE"):
enc_contents = str(soup.encode_contents())
if "Web Page Blocked!" in enc_contents or "Attack ID:" in enc_contents:
logger.warn("Being ratelimited/blocked by web application firewall. Sleeping for 30 minutes before retrying.")
logger.info("BeautifulSoup parsed contents: %s" % enc_contents)
time.sleep(1800)
exit(1)
return {
"__LASTFOCUS": "",
"__EVENTTARGET": "ctl00$ContentBody$LoginControl$linkLogin",
"__EVENTARGUMENT": "",
"__VIEWSTATE": soup.select_one("#__VIEWSTATE")["value"],
"__VIEWSTATEGENERATOR": soup.select_one("#__VIEWSTATEGENERATOR")["value"],
"__EVENTVALIDATION": soup.select_one("#__EVENTVALIDATION")["value"],
"ctl00$ContentBody$LoginControl$txtLoginEmailAddress": email,
"txtLoginEmailAddress_ClientState": '{"enabled":true,"emptyMessage":"","validationText":"%s","valueAsString":"%s","lastSetTextBoxValue":"%s"}' % (email, email, email),
"ctl00$ContentBody$LoginControl$txtLoginPassword": password,
"txtLoginPassword_ClientState": '{"enabled":true,"emptyMessage":"","validationText":"%s","valueAsString":"%s","lastSetTextBoxValue":"%s"}' % (password, password, password)
}
def _find_login_error(self, text):
try:
soup = BeautifulSoup(text, features='lxml')
notice_error = soup.select_one(".notice_error").text.strip()
return notice_error
except Exception:
return None
def needs_relogin(self):
if not self.accessTokenExpiresAt:
return False
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')
return {
'Authorization': 'Bearer %s' % self.accessToken,
'Origin': 'https://tconnect.tandemdiabetes.com',
'Referer': 'https://tconnect.tandemdiabetes.com/',
**base_headers()
}
def _get(self, endpoint, query):
r = base_session().get(self.BASE_URL + endpoint, data=query, headers=self.api_headers())
if r.status_code != 200:
raise ApiException(r.status_code, "ControlIQ API HTTP %s response: %s" % (str(r.status_code), r.text))
return r.json()
def get(self, endpoint, query, tries=0):
try:
return self._get(endpoint, query)
except ApiException as e:
logger.warning("Received ApiException in ControlIQApi with endpoint '%s' (tries %d): %s" % (endpoint, tries, e))
if tries > 0:
raise ApiException(e.status_code, "ControlIQ API HTTP %d on retry #%d: %s", e.status_code, tries, e)
# Trigger automatic re-login, and try again once
if e.status_code == 401:
logger.info("Performing automatic re-login after HTTP 401 for ControlIQApi")
self.accessTokenExpiresAt = time.time()
self.login(self._email, self._password)
return self.get(endpoint, query, tries=tries+1)
if e.status_code == 500:
return self.get(endpoint, query, tries=tries+1)
raise e
"""
Returns detailed basal event information and reasons for delivery suspension.
End-date inclusive: Returns data from 00:00 on start date to 23:59 on end date.
"""
def therapy_timeline(self, start=None, end=None):
startDate = parse_date(start)
endDate = parse_date(end)
# Microsoft-Azure-Application-Gateway/v2 WAF error message appears
# if startDate and endDate are not specified in exactly this order.
return self.get('tconnect/controliq/api/therapytimeline/users/%s?startDate=%s&endDate=%s' % (self.userGuid, startDate, endDate), {})
"""
Returns a summary of pump and cgm activity.
{'averageReading': <integer>, 'timeInUseMinutes': <integer>, 'controlIqSetToOffMinutes': <integer>,
'cgmInactiveMinutes': <integer>, 'pumpInactiveMinutes': <integer>, 'averageDailySleepMinutes': <integer>,
'weeklyExerciseEvents': <integer>, 'timeInUsePercent': <integer>, 'controlIqOffPercent': <integer>,
'cgmInactivePercent': <integer>, 'pumpInactivePercent': <integer>, 'totalDays': <integer>}
"""
def dashboard_summary(self, start, end):
startDate = parse_date(start)
endDate = parse_date(end)
return self.get('tconnect/controliq/api/summary/users/%s?startDate=%s&endDate=%s' % (self.userGuid, startDate, endDate), {})
"""
Returns active account features, including the date when ControlIQ was enabled.
[{"serialNumber": "11111111", "features": {"controlIQ": {"feature": 1, "dateTimeFirstDetected": "YYYY-MM-DD:THH:MM:SS", "unixTimestamp": 1111111111}}}]
"""
def pumpfeatures(self):
return self.get('tconnect/controliq/api/pumpfeatures/users/%s' % self.userGuid, {})
"""
Returns therapy events, used by the webui Therapy Timeline.
{'event': [
{'type': 'Basal', 'basalRate': ...},
{'type': 'Bolus', 'standard': ...},
{'type': 'CGM', 'egv': ...}
]}
"""
def therapy_events(self, start_date=None, end_date=None):
startDate = parse_date(start_date)
endDate = parse_date(end_date)
return self.get('tconnect/therapyevents/api/TherapyEvents/%s/%s/false?userId=%s' % (startDate, endDate, self.userGuid), {})
-296
View File
@@ -1,296 +0,0 @@
from typing import Dict, List, Tuple
import requests
import urllib
import datetime
import arrow
import time
import logging
from bs4 import BeautifulSoup
from tconnectsync.domain.device_settings import Device, Profile, ProfileSegment, DeviceSettings
from tconnectsync.util import removesuffix, removeprefix
from tconnectsync.util.constants import MMOLL_TO_MGDL
from .common import base_headers, ApiException
logger = logging.getLogger(__name__)
"""
WebUIScraper contains data that is scraped from the t:connect Web UI and is
not accessible via any known API.
"""
class WebUIScraper:
BASE_URL = "https://tconnect.tandemdiabetes.com/"
def __init__(self, controliq):
self.controliq = controliq
def needs_relogin(self):
return self.controliq.needs_relogin()
def _get(self, endpoint):
r = self.controliq.loginSession.get(self.BASE_URL + endpoint, headers=base_headers(), allow_redirects=True)
if r.status_code != 200:
raise ApiException(r.status_code, "WebUIScraper HTTP %s response: %s" % (str(r.status_code), r.text))
if 'login.aspx' in r.url:
raise ApiException(401, "WebUIScraper HTTP %s response for login page, returning 401: %s" % (str(r.status_code), r.url))
return r
def get(self, endpoint, tries=0):
try:
return self._get(endpoint)
except ApiException as e:
logger.warning("Received ApiException in WebUIScraper with endpoint '%s' (tries %d): %s" % (endpoint, tries, e))
if tries > 0:
raise ApiException(e.status_code, "WebUIScraper HTTP %d on retry #%d: %s", e.status_code, tries, e)
# Trigger automatic re-login, and try again once
if e.status_code == 401:
logger.info("Performing automatic re-login to ControlIQApi after HTTP 401 for ControlIQApi")
self.controliq.accessTokenExpiresAt = time.time()
self.controliq.login(self.controliq._email, self.controliq._password)
return self.get(endpoint, tries=tries+1)
if e.status_code == 500:
return self.get(endpoint, tries=tries+1)
raise e
def strip(self, txt):
# Remove errant whitespace between litearl newlines (and literal &nbsp;)
sep = '\r\n'
if sep not in txt and '\n' in txt:
sep = '\n'
return ' '.join([i.replace('\xa0',' ').strip() for i in txt.strip().split(sep)])
"""
Returns a mapping between pump/device IDs and information about that device,
including the GUID used for obtaining pump settings.
"""
def my_devices(self) -> Dict[str, DeviceSettings]:
devices = {}
r = self.get('myaccount/my_devices.aspx')
soup = BeautifulSoup(r.content, features='lxml')
for device in soup.select('#content > div.box'):
device_name = self.strip(device.select_one('.subTitle').text)
def find_label_value(lbl):
label = device.find(text=lbl)
if label:
tds = label.parent.parent.parent.select('td')
if len(tds) > 1:
return self.strip(tds[1].text)
return None
serial_number = find_label_value('Serial #')
model_number = find_label_value('Model #')
status = find_label_value('Status')
settings_span = device.find(text='View Settings')
settings_guid = None
if settings_span:
settings_a = settings_span.parent.parent
settings_guid = settings_a.attrs['href'].split('?guid=')[1]
if serial_number:
devices[serial_number] = Device(
name=device_name,
model_number=model_number,
status=status,
guid=settings_guid)
return devices
"""
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
"""
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)
soup = BeautifulSoup(r.content, features='lxml')
settings["upload_date"] = self.strip(soup.select_one('#lblUploadDate').text)
divxml = soup.select_one('#divXML')
divxmlDiv = divxml.findChild('div')
for tbl in divxmlDiv.findChildren('table', recursive=False):
setting_bg = tbl.select_one('.setting_bg')
if setting_bg and self.strip(setting_bg.text) == 'Profile':
profiles.append(self._parse_profile_tbl(tbl))
else:
settings.update(self._parse_settings_tbl(tbl))
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 = {}
profile["title"] = self.strip(tbl.select_one('.setting_title').text)
profile["active"] = bool(tbl.find(text='Active at the time of upload'))
profile["segments"] = []
def parse_basal_rate(rate) -> float:
return float(removesuffix(rate, ' u/hr'))
def parse_factor(ratio) -> int:
return parse_bg_mgdl(removeprefix(ratio, '1u:'))
def parse_ratio(ratio) -> float:
return float(removesuffix(removeprefix(ratio, '1u:'), ' g'))
def parse_bg_mgdl(bg) -> int:
if bg.endswith(' mg/dL'):
return float(removesuffix(bg, ' mg/dL'))
elif bg.endswith(' mmol/L'):
return float(removesuffix(bg, ' mmol/L')) * MMOLL_TO_MGDL
raise ValueError(bg)
def hours_to_mins(text) -> int:
hrmin = removesuffix(text, " hours")
hr, min = hrmin.split(":", 1)
return int(min) + int(hr)*60
for tr in tbl.select('tr'):
# Skip header rows
if tr.select_one('.setting_bg'):
continue
if tr.find(text='Start Time'):
continue
tds = tr.select('td')
def is_time_row(td):
txt = self.strip(td.select_one('strong').text)
return " AM" in txt or " PM" in txt or txt in ("Midnight", "Noon")
if len(tds) > 0 and is_time_row(tds[0]):
display_time = self.strip(tds[0].text)
t = display_time
if display_time == "Midnight":
t = "12:00 AM"
elif display_time == "Noon":
t = "12:00 PM"
segment = {
"display_time": display_time,
"time": t,
"basal_rate": parse_basal_rate(self.strip(tds[1].text)),
"correction_factor": parse_factor(self.strip(tds[2].text)),
"carb_ratio": parse_ratio(self.strip(tds[3].text)),
"target_bg_mgdl": parse_bg_mgdl(self.strip(tds[4].text))
}
profile["segments"].append(ProfileSegment(**segment))
continue
if tr.find(text='Calculated Total Daily Basal'):
profile["calculated_total_daily_basal"] = float(removesuffix(self.strip(tds[1].text), " units"))
continue
# Last row
if tr.find(text='Duration of Insulin:'):
lastrow = self.strip(tr.text)
for part in lastrow.split(' |'):
if len(part) < 1:
continue
key, val = part.split(': ')
key = self.strip(key)
val = self.strip(val)
if key == 'Duration of Insulin':
profile["insulin_duration_min"] = hours_to_mins(val)
elif key == 'Carbohydrates':
profile["carbs_enabled"] = self.strip(val.lower()) == "on"
return Profile(**profile)
def _parse_settings_tbl(self, tbl):
outer_tr = tbl.select('tr')[2]
settings = {}
def loop(td, subhead):
settings[subhead] = {}
for tr in td.select('.settingstable > tr'):
if not tr.select_one('strong'):
continue
key = self.strip(tr.select_one('strong').text)
tds = tr.select('td')
if len(tds) == 1:
subhead = key
settings[subhead] = {}
continue
val_text = self.strip(tds[1].text)
val = {}
if tds[1].find(text=' - '):
val['value'] = False
elif tds[1].find(text='Off'):
val['value'] = False
val_text = self.strip(val_text.split('Off', 1)[1])
elif tds[1].find(text='On'):
val['value'] = True
val_text = self.strip(val_text.split('On', 1)[1])
val['text'] = val_text
settings[subhead][key] = val
children = outer_tr.findChildren('td', recursive=False)
loop(children[0], 'Alerts')
loop(children[1], 'Pump Settings')
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.
"""
def device_settings(self, pump_serial: str) -> Tuple[List[Profile], DeviceSettings]:
devices = self.my_devices()
if str(pump_serial) in devices.keys():
dev = devices[str(pump_serial)]
return self.device_settings_from_guid(dev['guid'])
raise RuntimeError('Unable to find pump with serial number: %s. Known devices: %s' % (pump_serial, devices))
-174
View File
@@ -1,174 +0,0 @@
import requests
import datetime
import csv
import logging
import time
import json
from .common import base_session, parse_date, parsed_date_to_arrow, base_headers, days_between, split_days_range, ApiException
logger = logging.getLogger(__name__)
class WS2Api:
BASE_URL = 'https://tconnectws2.tandemdiabetes.com/'
MAX_RETRIES = 2
SLEEP_SECONDS_INCREMENT = 60
userGuid = None
def __init__(self, userGuid):
self.userGuid = userGuid
self.session = base_session()
def get(self, endpoint, **kwargs):
r = self.session.get(self.BASE_URL + endpoint, headers=base_headers(), **kwargs)
if r.status_code != 200:
raise ApiException(r.status_code, "WS2 API HTTP %s response: %s" % (str(r.status_code), r.text))
return r.text
def get_jsonp(self, endpoint, **kwargs):
r = self.session.get(self.BASE_URL + endpoint + '?callback=cb', headers=base_headers(), **kwargs)
if r.status_code != 200:
raise ApiException(r.status_code, "WS2 API HTTP %s response: %s" % (str(r.status_code), r.text))
t = r.text.strip()
if t.startswith('cb('):
t = t[3:]
if t.endswith(')'):
t = t[:-1]
return json.loads(t)
def _split_empty_sections(self, text):
sections = [[]]
sectionIndex = 0
for line in text.splitlines():
if len(line.strip()) > 0:
sections[sectionIndex].append(line)
else:
sections.append([])
sectionIndex += 1
return sections + [None] * (4 - len(sections))
def _csv_to_dict(self, rawdata):
data = []
if not rawdata or len(rawdata) == 0:
return data
headers = rawdata[0].split(",")
for row in csv.reader(rawdata[1:]):
data.append({headers[i]: row[i] for i in range(len(row)) if i < len(headers)})
return data
"""
Returns information on therapy, displayed in the therapy timeline on the
t:connect website.
Contains BG reading (CGM), IOB, basal, and bolus data.
Basal data does NOT appear for the specified time range if using Control-IQ.
The ControlIQ API endpoints must be used for basal data instead.
However, all other fields are still accessed via this endpoint.
This has its own built-in retry logic because Tandem's frontend serving
the API returns 500s when its backend times out.
"""
MAX_THERAPY_TIMELINE_DAYS = 2
def therapy_timeline_csv(self, start=None, end=None, tries=0):
startDate = parse_date(start)
endDate = parse_date(end)
pStart = parsed_date_to_arrow(startDate)
pEnd = parsed_date_to_arrow(endDate)
if days_between(pStart, pEnd) > self.MAX_THERAPY_TIMELINE_DAYS:
ranges = split_days_range(pStart, pEnd, self.MAX_THERAPY_TIMELINE_DAYS)
logger.debug("Splitting call to therapy_timeline_csv(%s, %s) into: %s", start, end, ranges)
outputs = []
for rng in ranges:
rStart, rEnd = rng
logger.debug("split therapy_timeline_csv(%s, %s)", rStart, rEnd)
output = self.therapy_timeline_csv(rStart, rEnd, tries=tries)
logger.debug("split therapy_timeline_csv(%s, %s) = %s", rStart, rEnd, ["%s: %s items" % (key, len(val)) for key, val in output.items()])
outputs.append(output)
full = {}
for o in outputs:
for key, val in o.items():
if key not in full:
full[key] = val
elif val is not None:
full[key] += val
logger.debug("therapy_timeline_csv merge: %s", ["%s: %s items" % (key, len(val)) for key, val in full.items()])
return full
try:
req_text = self.get('therapytimeline2csv/%s/%s/%s?format=csv' % (self.userGuid, startDate, endDate), timeout=10)
except ApiException as e:
# This seems to occur as some kind of soft rate-limit.
logger.warning("Received ApiException in therapy_timeline_csv: (retry count %d) %s" % (tries, e))
if e.status_code == 500:
sleep_seconds = (tries+1) * self.SLEEP_SECONDS_INCREMENT
logger.error("Retrying in %d seconds after HTTP 500 in therapy_timeline_csv (retry count %d): %s" % (sleep_seconds, tries, e))
time.sleep(sleep_seconds)
if tries < self.MAX_RETRIES:
return self.therapy_timeline_csv(start, end, tries+1)
raise e
logger.debug('req_text: %s', req_text)
sections = self._split_empty_sections(req_text)
readingData = None
iobData = None
basalData = None
bolusData = None
for s in sections:
if s and len(s) > 2:
firstrow = s[1].replace('"', '').strip()
if firstrow.startswith("t:slim X2 Insulin Pump"):
readingData = s
elif firstrow.startswith("IOB"):
iobData = s
elif firstrow.startswith("Basal"):
basalData = s
elif firstrow.startswith("Bolus"):
bolusData = s
return {
"readingData": self._csv_to_dict(readingData),
"iobData": self._csv_to_dict(iobData),
"basalData": self._csv_to_dict(basalData),
"bolusData": self._csv_to_dict(bolusData)
}
"""
Returns information on basal suspension. The filterbasal option only returns site/cartridge changes.
SuspendReason values are:
- "site-cart"
- "basal-profile"
- "manual"
- "previous"
- "alarm"
End-date inclusive: Returns data from 00:00 on start date to 23:59 on end date.
{"BasalSuspension":[{"EventDateTime":"/Date(EPOCH_MS-0000)/", "SuspendReason": "reason"}]}
"""
def basalsuspension(self, start=None, end=None, filterbasal=False):
startDate = parse_date(start)
endDate = parse_date(end)
arg = "filterbasal/1" if filterbasal else ""
return self.get_jsonp('basalsuspension/%s/%s/%s/%s' % (self.userGuid, startDate, endDate, arg), timeout=10)
"""
Returns info on BasalIQ in JSONP format.
"""
def basaliqtech(self, start=None, end=None):
startDate = parse_date(start)
endDate = parse_date(end)
return self.get_jsonp('basaliqtech/%s/%s/%s' % (self.userGuid, startDate, endDate), timeout=10)
-1
View File
@@ -14,7 +14,6 @@ else:
from .nightscout import NightscoutApi
from .parser.nightscout import BASAL_EVENTTYPE, BOLUS_EVENTTYPE
from .parser.tconnect import TConnectEntry
from .domain.tandemsource.event_class import EventClass
from .sync.tandemsource.choose_device import ChooseDevice
-25
View File
@@ -1,25 +0,0 @@
from dataclasses import dataclass, asdict
@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
def to_dict(self):
return asdict(self)
@property
def is_extended_bolus(self):
return self.extended_bolus == "1"
-44
View File
@@ -1,44 +0,0 @@
from dataclasses import dataclass, replace
from typing import List, Optional
@dataclass
class Device:
name: str
model_number: str
status: str
guid: Optional[str]
@dataclass
class ProfileSegment:
display_time: str # Identical to time except written out as Midnight or Noon
time: str
basal_rate: float # _ u/hr
correction_factor: int # 1u: _ mg/dL
carb_ratio: float # 1u: _ g
target_bg_mgdl: int
@dataclass
class Profile:
title: str
active: bool
segments: List[ProfileSegment]
calculated_total_daily_basal: float # in units
insulin_duration_min: int
carbs_enabled: bool
def activeProfile(self):
p = self.copy()
p.active = True
return p
def copy(self):
p = replace(self)
p.segments = [replace(s) for s in p.segments]
return p
# 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
-483
View File
@@ -1,483 +0,0 @@
import arrow
from tconnectsync.domain.bolus import Bolus
from ..secret import TIMEZONE_NAME
def _datetime_parse(date):
# consistent format with ws2 endpoint
return arrow.get(date, tzinfo=TIMEZONE_NAME).format("YYYY-MM-DD HH:mm:ssZZ")
class TherapyEvent:
type = None
eventDateTime = None
sourceRecId = None
def parse(self, json):
self.type = json['type']
self.eventDateTime = json['eventDateTime']
self.sourceRecId = json['sourceRecId']
self.rawJson = json
def __str__(self):
return "%s(%s)" % (self.type, self.rawJson)
class CGMTherapyEvent(TherapyEvent):
eventID = None
egv = None
"""
{
"eventDateTime": "2022-07-21T00:00:08",
"eventID": 256,
"requestDateTime": "0001-01-01T00:00:00",
"type": "CGM",
"description": "EGV",
"sourceRecId": 0,
"eventTypeId": 0,
"deviceType": "t:slim X2 Insulin Pump",
"serialNumber": "xxx",
"indexId": 0,
"uploadId": 0,
"interactive": 0,
"tempRateId": 0,
"tempRateCompleted": 0,
"tempRateActivated": 0,
"egv": {
"estimatedGlucoseValue": 174,
"hypo": 0,
"belowTarget": 0,
"withinTarget": 1,
"aboveTarget": 0,
"hyper": 0
}
},
"""
@classmethod
def parse(_, json):
self = CGMTherapyEvent()
TherapyEvent.parse(self, json)
self.eventID = json['eventID']
self.egv = json['egv']['estimatedGlucoseValue']
return self
class BGTherapyEvent(TherapyEvent):
eventID = None
egv = None
"""
{
'bg': 160, # note in EGV
'cgmCalibration': 1, # not in EGV
'description': 'BG',
'deviceType': 't:slim X2 Insulin Pump',
'eventDateTime': '2022-08-20T07:25:24',
'eventTypeId': 16,
'indexId': 844955,
'interactive': 0,
'iob': 0.75,
'note': { 'active': False,
'eventId': 0, # different location than EGV
'eventTypeId': 16,
'id': 0,
'indexId': '',
'sourceRecordId': 0},
'requestDateTime': '0001-01-01T00:00:00',
'serialNumber': 'xxx',
'sourceRecId': 793549667,
'tempRateActivated': 0,
'tempRateCompleted': 0,
'tempRateId': 0,
'type': 'BG',
'uploadId': 748700213}
"""
@classmethod
def parse(_, json):
self = BGTherapyEvent()
TherapyEvent.parse(self, json)
self.eventID = json['note']['eventId']
# This is probably not how we want to provide CGM calibrations to Nightscout,
# but will just include it as egv data for now to keep the thing from crashing :)
self.egv = json['bg']
return self
class BolusTherapyEvent(TherapyEvent):
bolusRequestOptions = None
REQUEST_AUTOMATIC = "Automatic Bolus/Correction"
REQUEST_STANDARD = "Standard"
bolusType = None
TYPE_AUTOMATIC = "Automatic Correction"
TYPE_CARB = "Carb"
carbSize = None
correctionBolusSize = None
foodBolusSize = None
insulinDelivered = None
insulinRequested = None
completionDateTime = None
completionStatus = None
STATUS_COMPLETED = "Completed"
eventHistoryReportDetails = None
standardPercent = None
sourceRecId = None
@classmethod
def parse(_, json):
self = BolusTherapyEvent()
TherapyEvent.parse(self, json)
self.description = json.get("description")
self.complete = json.get("standard", {}).get("bolusIsComplete")
self.completion = json.get("standard", {}).get("completionStatusDesc")
self.request_time = json.get("requestDateTime")
self.completion_time = json.get("standard", {}).get("insulinDelivered", {}).get("completionDateTime")
# TODO: separate extended vs standard bolus into separate fields
self.insulin = json.get("standard", {}).get("insulinDelivered", {}).get("value")
self.requested_insulin = json.get("standard", {}).get("insulinRequested")
self.carbs = json.get("carbSize")
self.bg = json.get("bg")
self.user_override = json.get("userOverride")
self.extended_bolus = json.get("bolusRequestOptions") == "Extended"
if self.extended_bolus and self.complete:
# TODO(https://github.com/jwoglom/tconnectsync/issues/19): read more extended bolus info
self.complete = json.get("bolex", {}).get("extendedBolusIsComplete")
self.completion = json.get("bolex", {}).get("completionStatusDesc")
self.bolex_completion_time = json.get("bolex", {}).get("insulinDelivered", {}).get("completionDateTime")
self.bolex_start_time = json.get("bolex", {}).get("bolexStartDateTime")
else:
self.bolex_completion_time = ""
self.bolex_start_time = ""
return self
def to_bolus(self):
return Bolus(
description=self.description,
complete="1" if self.complete else "0",
completion=self.completion or "",
request_time=_datetime_parse(self.request_time),
completion_time=_datetime_parse(self.completion_time),
insulin=str(self.insulin),
requested_insulin=str(self.requested_insulin),
carbs=str(self.carbs or "0"), # Nightscout expects non-empty carbs
bg=str(self.bg or ""),
user_override=str(self.user_override),
extended_bolus="1" if self.extended_bolus else "0",
bolex_completion_time=_datetime_parse(self.bolex_completion_time) if self.bolex_completion_time else "",
bolex_start_time=_datetime_parse(self.bolex_start_time) if self.bolex_start_time else ""
)
"""
Correction:
{
"actualTotalBolusRequested": 2.9,
"bg": 254,
"bolusRequestOptions": "Automatic Bolus/Correction",
"bolusType": "Automatic Correction",
"carbSize": 0,
"correctionBolusSize": 2.9,
"correctionFactor": 30,
"declinedCorrection": 0,
"duration": 0,
"eventDateTime": "2022-07-21T11:53:08",
"eventHistoryReportDetails": "CF 1:30 - Carb Ratio 1:0 - Target BG 110",
"eventHistoryReportEventDesc": "Correction Bolus",
"foodBolusSize": 0,
"isQuickBolus": 0,
"note": {
"id": 0,
"indexId": "572946",
"eventTypeId": 64,
"sourceRecordId": 0,
"eventId": 0,
"active": false
},
"requestDateTime": "2022-07-21T11:53:08",
"standard": {
"insulinDelivered": {
"completionDateTime": "2022-07-21T11:55:24",
"value": 2.9
},
"foodDelivered": 0,
"correctionDelivered": 2.9,
"insulinRequested": 2.9,
"completionStatusId": 3,
"completionStatusDesc": "Completed",
"bolusIsComplete": 1,
"bolusRequestId": 3361,
"bolusCompletionId": 3361
},
"standardPercent": 100,
"targetBG": 110,
"userOverride": 0,
"type": "Bolus",
"description": "Automatic Bolus/Correction",
"sourceRecId": 1171791787,
"eventTypeId": 0,
"indexId": 0,
"uploadId": 0,
"interactive": 0,
"tempRateId": 0,
"tempRateCompleted": 0,
"tempRateActivated": 0
},
Standard:
{
"actualTotalBolusRequested": 4.17,
"bolusRequestOptions": "Standard",
"bolusType": "Carb",
"carbSize": 25,
"correctionBolusSize": 0,
"correctionFactor": 30,
"declinedCorrection": 0,
"duration": 0,
"eventDateTime": "2022-07-21T12:27:36",
"eventHistoryReportDetails": "CF 1:30 - Carb Ratio 1:6 - Target BG 110",
"eventHistoryReportEventDesc": "Food Bolus",
"foodBolusSize": 4.17,
"iob": 2.62,
"isQuickBolus": 0,
"note": {
"id": 0,
"indexId": "573042",
"eventTypeId": 64,
"sourceRecordId": 0,
"eventId": 0,
"active": false
},
"requestDateTime": "2022-07-21T12:27:36",
"standard": {
"insulinDelivered": {
"completionDateTime": "2022-07-21T12:29:21",
"value": 4.17
},
"foodDelivered": 4.17,
"correctionDelivered": 0,
"insulinRequested": 4.17,
"completionStatusId": 3,
"completionStatusDesc": "Completed",
"bolusIsComplete": 1,
"bolusRequestId": 3362,
"bolusCompletionId": 3362
},
"standardPercent": 100,
"targetBG": 110,
"userOverride": 0,
"type": "Bolus",
"description": "Standard",
"sourceRecId": 1171853319,
"eventTypeId": 0,
"indexId": 0,
"uploadId": 0,
"interactive": 0,
"tempRateId": 0,
"tempRateCompleted": 0,
"tempRateActivated": 0
},
Extended bolus incomplete:
{
"actualTotalBolusRequested": 0.4,
"bg": 131,
"bolex": {
"size": 0.2,
"bolexStartDateTime": "2022-08-09T23:20:04",
"iob": 0,
"completionStatusId": 0,
"extendedBolusIsComplete": 0,
"insulinRequested": 0,
"bolexCompletionId": 0
},
"bolusRequestOptions": "Extended",
"bolusType": "Carb",
"carbSize": 0,
"correctionBolusSize": 0.0,
"correctionFactor": 30.0,
"declinedCorrection": 0,
"duration": 15,
"eventDateTime": "2022-08-09T23:19:15",
"eventHistoryReportDetails": "CF 1:30 - Carb Ratio 1:6 - Target BG 110<br/>Override: Pump calculated Bolus = 0.0 units",
"eventHistoryReportEventDesc": "Food Bolus: 50&#37; Extended 15 mins",
"foodBolusSize": 0.0,
"iob": 5.87,
"isQuickBolus": 0,
"note": {
"id": 0,
"indexId": "631597",
"eventTypeId": 64,
"sourceRecordId": 0,
"eventId": 0,
"active": false
},
"requestDateTime": "2022-08-09T23:19:15",
"standard": {
"insulinDelivered": {
"completionDateTime": "2022-08-09T23:20:04",
"value": 0.2
},
"foodDelivered": 0.0,
"correctionDelivered": 0.0,
"insulinRequested": 0.2,
"completionStatusId": 3,
"completionStatusDesc": "Completed",
"bolusIsComplete": 1,
"bolusRequestId": 3636.0,
"bolusCompletionId": 3636.0
},
"standardPercent": 50.0,
"targetBG": 110,
"userOverride": 1,
"type": "Bolus",
"description": "Extended 50.00%/0.00",
"sourceRecId": 1209631944,
"eventTypeId": 0,
"indexId": 0,
"uploadId": 0,
"interactive": 0,
"tempRateId": 0,
"tempRateCompleted": 0,
"tempRateActivated": 0
}
Extended bolus (complete):
{
"actualTotalBolusRequested": 0.4,
"bg": 131,
"bolex": {
"size": 0.2,
"bolexStartDateTime": "2022-08-09T23:20:04",
"insulinDelivered": {
"completionDateTime": "2022-08-09T23:35:03",
"value": 0.2
},
"iob": 5.7,
"completionStatusId": 3.0,
"completionStatusDesc": "Completed",
"extendedBolusIsComplete": 1,
"insulinRequested": 0.2,
"bolexCompletionId": 16757133
},
"bolusRequestOptions": "Extended",
"bolusType": "Carb",
"carbSize": 0,
"correctionBolusSize": 0.0,
"correctionFactor": 30.0,
"declinedCorrection": 0,
"duration": 15,
"eventDateTime": "2022-08-09T23:19:15",
"eventHistoryReportDetails": "CF 1:30 - Carb Ratio 1:6 - Target BG 110<br/>Override: Pump calculated Bolus = 0.0 units",
"eventHistoryReportEventDesc": "Food Bolus: 50&#37; Extended 15 mins",
"foodBolusSize": 0.0,
"iob": 5.87,
"isQuickBolus": 0,
"note": {
"id": 0,
"indexId": "631597",
"eventTypeId": 64,
"sourceRecordId": 0,
"eventId": 0,
"active": false
},
"requestDateTime": "2022-08-09T23:19:15",
"standard": {
"insulinDelivered": {
"completionDateTime": "2022-08-09T23:20:04",
"value": 0.2
},
"foodDelivered": 0.0,
"correctionDelivered": 0.0,
"insulinRequested": 0.2,
"completionStatusId": 3,
"completionStatusDesc": "Completed",
"bolusIsComplete": 1,
"bolusRequestId": 3636.0,
"bolusCompletionId": 3636.0
},
"standardPercent": 50.0,
"targetBG": 110,
"userOverride": 1,
"type": "Bolus",
"description": "Extended 50.00%/0.00",
"sourceRecId": 1209631944,
"eventTypeId": 0,
"indexId": 0,
"uploadId": 0,
"interactive": 0,
"tempRateId": 0,
"tempRateCompleted": 0,
"tempRateActivated": 0
}
CGM Calibration (Therapy Event Type BG):
{ 'bg': 160,
'cgmCalibration': 1,
'description': 'BG',
'deviceType': 't:slim X2 Insulin Pump',
'eventDateTime': '2022-08-20T07:25:24',
'eventTypeId': 16,
'indexId': 844955,
'interactive': 0,
'iob': 0.75,
'note': { 'active': False,
'eventId': 0,
'eventTypeId': 16,
'id': 0,
'indexId': '',
'sourceRecordId': 0},
'requestDateTime': '0001-01-01T00:00:00',
'serialNumber': 'xxx',
'sourceRecId': 793549667,
'tempRateActivated': 0,
'tempRateCompleted': 0,
'tempRateId': 0,
'type': 'BG',
'uploadId': 0}
"""
class BasalTherapyEvent(TherapyEvent):
"""
{
'basalRate': {
'duration': 0,
'percent': 0,
'value': 0.0
},
'displayInHistory': 0,
'eventDateTime': '2022-12-02T00:00:00',
'note': {
'id': 0,
'indexId': '16403',
'eventTypeId': 90,
'sourceRecordId': 0,
'eventId': 0,
'active': False
},
'noteDate': {},
'requestDateTime': '0001-01-01T00:00:00',
'type': 'Basal',
'description': 'NDE',
'sourceRecId': xxx,
'eventTypeId': 0,
'indexId': 0,
'uploadId': 0,
'interactive': 1,
'tempRateId': 0,
'tempRateCompleted': 0,
'tempRateActivated': 0
}
"""
basalRateValue = None
basalRatePercent = None
basalRateDuration = None
eventTime = None
@classmethod
def parse(_, json):
self = CGMTherapyEvent()
TherapyEvent.parse(self, json)
if 'basalRate' in json:
self.basalRateValue = json['basalRate']['value']
self.basalRatePercent = json['basalRate']['percent']
self.basalRateDuration = json['basalRate']['duration']
self.eventTime = json['eventDateTime']
return self
-24
View File
@@ -1,24 +0,0 @@
#!/usr/bin/env python3
class Time:
def __init__(self, hour: int, min: int):
self.hour = hour
self.min = min
@classmethod
def parse(cls, input):
if ' ' not in input:
raise ValueError('unable to parse time: %s' % input)
hrmin, ampm = input.split(' ')
hr, min = hrmin.split(':')
hr = int(hr)
min = int(min)
if ampm.lower() == 'pm':
hr += 12
elif ampm.lower() != 'am':
raise ValueError('unable to parse time: %s' % input)
return cls(hr, min)
-27
View File
@@ -1,27 +0,0 @@
from tconnectsync.domain.therapy_event import BolusTherapyEvent, CGMTherapyEvent, BGTherapyEvent, BasalTherapyEvent
from tconnectsync.parser.tconnect import TConnectEntry
import logging
logger = logging.getLogger(__name__)
def split_therapy_events(ciqTherapyEvents):
bolusEvents = []
cgmEvents = []
bgEvents = []
basalEvents = []
for e in ciqTherapyEvents['event']:
event = TConnectEntry.parse_therapy_event(e)
if isinstance(event, BolusTherapyEvent):
bolusEvents.append(event)
elif isinstance(event, CGMTherapyEvent):
cgmEvents.append(event)
elif isinstance(event, BGTherapyEvent):
bgEvents.append(event)
elif isinstance(event, BasalTherapyEvent):
basalEvents.append(event)
logger.debug("split_therapy_events: %d bolus, %d CGM, %d BG, %d basal" % (len(bolusEvents), len(cgmEvents), len(bgEvents), len(basalEvents)))
# TODO: BG events (CGM Calibration) values are not currently returned from ciq_therapy_events.py
return bolusEvents, cgmEvents
-73
View File
@@ -1,6 +1,5 @@
import arrow
from ..domain.device_settings import Profile, DeviceSettings
from ..domain.tandemsource.pump_settings import PumpProfile, PumpSettings
from ..secret import TIMEZONE_NAME, NIGHTSCOUT_PROFILE_CARBS_HR_VALUE, NIGHTSCOUT_PROFILE_DELAY_VALUE
@@ -210,57 +209,6 @@ class NightscoutEntry:
"pump_event_id": pump_event_id
}
# Tandem-scraped profile to Nightscout profile store entry
@staticmethod
def profile_store(profile: Profile, device_settings: DeviceSettings) -> dict:
return {
# insulin duration in hours; Nightscout JS bug requires all top-level fields to be strings
"dia": "%s" % (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"
}
# TandemSource profile to Nightscout profile store entry
@staticmethod
def tandemsource_profile_store(profile: PumpProfile, pump_settings: PumpSettings) -> dict:
@@ -313,24 +261,6 @@ class NightscoutEntry:
"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)
def minutes_to_ns_time(minutes_time: int) -> str:
hr = minutes_time // 60
mn = minutes_time % 60
@@ -339,6 +269,3 @@ def minutes_to_ns_time(minutes_time: int) -> str:
class InvalidBolusTypeException(RuntimeError):
pass
class InvalidTimeException(RuntimeError):
pass
-221
View File
@@ -1,221 +0,0 @@
from os import stat
import sys
import arrow
from tconnectsync.domain.bolus import Bolus
from tconnectsync.domain.therapy_event import BolusTherapyEvent, CGMTherapyEvent, BGTherapyEvent, BasalTherapyEvent
try:
from ..secret import TIMEZONE_NAME
except Exception:
print('Unable to import parser secrets from secret.py')
sys.exit(1)
"""
Conversion methods for parsing raw t:connect data into
a more digestable format, which is used internally.
"""
class TConnectEntry:
BASAL_EVENTS = { 0: "Suspension", 1: "Profile", 2: "TempRate", 3: "Algorithm" }
@staticmethod
def _epoch_parse(x):
# data["x"] is an integer epoch timestamp which, when read as an equivalent timestamp
# stored in Pacific time (America/Los_Angeles), contains the user's local time, but
# with the wrong timezone data attached.
#
# For example, data["x"] references UTC timestamp 2020-09-01T13:00:00+00:00,
# which when read in Pacific time is equivalent to 2020-09-01T06:00:00-07:00.
# However, the user's timezone is Eastern time, so the timezone of America/Los_Angeles
# is overwritten with America/New_York, resulting in 2020-09-01T06:00:00-04:00, the
# correct timestamp.
return arrow.get(x, tzinfo="America/Los_Angeles").replace(tzinfo=TIMEZONE_NAME)
@staticmethod
def _jsonp_epoch_parse(x):
return TConnectEntry._epoch_parse(int(x.replace('/Date(', '').replace('-0000)/', '')))
@staticmethod
def parse_ciq_basal_entry(data, delivery_type=""):
time = TConnectEntry._epoch_parse(data["x"])
duration_mins = data["duration"] / 60
basal_rate = data["y"]
return {
"time": time.format(),
"delivery_type": delivery_type,
"duration_mins": duration_mins,
"basal_rate": basal_rate,
}
@staticmethod
def manual_suspension_to_basal_entry(parsedSuspension, seconds):
duration_mins = seconds / 60
return {
"time": parsedSuspension["time"],
"delivery_type": "%s suspension" % parsedSuspension["suspendReason"],
"duration_mins": duration_mins,
"basal_rate": 0.0
}
@staticmethod
def parse_suspension_entry(data):
time = TConnectEntry._epoch_parse(data["x"])
return {
"time": time.format(),
"continuation": data["continuation"],
"suspendReason": data["suspendReason"],
}
@staticmethod
def _datetime_parse(date):
return arrow.get(date, tzinfo=TIMEZONE_NAME)
@staticmethod
def parse_cgm_entry(data):
# EventDateTime is stored in the user's timezone.
return {
"time": TConnectEntry._datetime_parse(data["EventDateTime"]).format(),
"reading": data["Readings (CGM / BGM)"],
"reading_type": data["Description"],
}
@staticmethod
def parse_iob_entry(data):
# EventDateTime is stored in the user's timezone.
return {
"time": TConnectEntry._datetime_parse(data["EventDateTime"]).format(),
"iob": data["IOB"],
"event_id": data["EventID"],
}
@staticmethod
def parse_csv_basal_entry(data, duration_mins=None):
# EventDateTime is stored in the user's timezone.
return {
"time": TConnectEntry._datetime_parse(data["EventDateTime"]).format(),
"delivery_type": "Unknown",
"duration_mins": duration_mins,
"basal_rate": data["BasalRate"],
}
@staticmethod
def parse_bolus_entry(data):
# All DateTime's are stored in the user's timezone.
def is_complete(s):
return s and int(s) == 1
complete = is_complete(data["ExtendedBolusIsComplete"]) or is_complete(data["BolusIsComplete"])
extended_bolus = ("extended" in data["Description"].lower())
return Bolus(**{
"description": data["Description"],
"complete": "1" if complete else "",
"completion": data["CompletionStatusDesc"] if not extended_bolus else data["BolexCompletionStatusDesc"],
"request_time": TConnectEntry._datetime_parse(data["RequestDateTime"]).format() if not extended_bolus else None,
"completion_time": TConnectEntry._datetime_parse(data["CompletionDateTime"]).format() if not extended_bolus else None,
"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 "",
# 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):
return {
"time": TConnectEntry._datetime_parse(data["EventDateTime"]).format(),
"bg": data["Readings (CGM / BGM)"],
"type": data["Description"]
}
ACTIVITY_EVENTS = { 1: "Sleep", 2: "Exercise", 3: "AutoBolus", 4: "CarbOnly" }
@staticmethod
def parse_ciq_activity_event(data):
if data["eventType"] not in TConnectEntry.ACTIVITY_EVENTS.keys():
raise UnknownCIQActivityEventException(data)
time = TConnectEntry._epoch_parse(data["x"])
return {
"time": time.format(),
"duration_mins": data["duration"] / 60,
"event_type": TConnectEntry.ACTIVITY_EVENTS[data["eventType"]]
}
BASALSUSPENSION_EVENTS = {
# site-cart corresponds to a Site or Cartridge change,
# specifically a Tubing Filled: Norm AND a Cannula Filled: Norm alert.
# (This means that a typical changing of a cartridge and then a site
# will result in two consecutive events of this type.)
"site-cart": "Site/Cartridge Change",
# alarm corresponds to one of the following:
# - an Empty Cartridge alarm
# - a Pump shutdown
"alarm": "Empty Cartridge/Pump Shutdown",
# manual corresponds to a Pumping Suspended by User event
"manual": "User Suspended",
# temp-profile corresponds to a Basal Rate Change event to 0u/hr
"temp-profile": "Basal Rate Change"
}
BASALSUSPENSION_SKIPPED_EVENTS = {
# basal-profile events are not very useful; with ControlIQ enabled,
# Tandem does not show them in the tconnect UI.
"basal-profile",
# If an event continues to occur after the date switches over to the next
# day, then the pump generates a "previous" event. This isn't useful to
# us, so we skip them.
"previous",
}
@staticmethod
def parse_basalsuspension_event(data):
if not data or "SuspendReason" not in data:
return None
if data["SuspendReason"] in TConnectEntry.BASALSUSPENSION_SKIPPED_EVENTS:
return None
if data["SuspendReason"] not in TConnectEntry.BASALSUSPENSION_EVENTS.keys():
raise UnknownBasalSuspensionEventException(data)
time = TConnectEntry._jsonp_epoch_parse(data["EventDateTime"])
return {
"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)
elif data["type"] == "BG":
return BGTherapyEvent.parse(data)
elif data["type"] == "Basal":
return BasalTherapyEvent.parse(data)
raise UnknownTherapyEventException(data)
class UnknownCIQActivityEventException(Exception):
def __init__(self, data):
super().__init__("Unknown CIQ activity event type: %s" % data)
class UnknownBasalSuspensionEventException(Exception):
def __init__(self, data):
super().__init__("Unknown basal suspension event type: %s" % data)
class UnknownTherapyEventException(Exception):
def __init__(self, data):
typ = data["type"]
super().__init__(f"Unknown therapy event type: {typ} in {data}")
-202
View File
@@ -1,202 +0,0 @@
import logging
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 (
process_ciq_basal_events,
add_csv_basal_events,
ns_write_basal_events
)
from .sync.bolus import (
process_bolus_events,
ns_write_bolus_events
)
from .sync.iob import (
process_iob_events,
ns_write_iob_events
)
from .sync.cgm import (
process_cgm_events,
ns_write_cgm_events
)
from .sync.pump_events import (
process_ciq_activity_events,
process_basalsuspension_events,
ns_write_pump_events
)
from .sync.profile import process_profiles
from .parser.tconnect import TConnectEntry
from .features import BASAL, BOLUS, IOB, BOLUS_BG, CGM, DEFAULT_FEATURES, PUMP_EVENTS, PROFILES, PUMP_EVENTS_BASAL_SUSPENSION
from tconnectsync.sync import basal
logger = logging.getLogger(__name__)
"""
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, nightscout, time_start, time_end, pretend, features=DEFAULT_FEATURES):
ciqTherapyTimelineData = None
if BASAL in features or PUMP_EVENTS in features:
logger.info("Downloading t:connect ControlIQ data")
try:
ciqTherapyTimelineData = tconnect.controliq.therapy_timeline(time_start, time_end)
except ApiException as e:
# The ControlIQ API returns a 404 if the user did not have a ControlIQ enabled
# device in the time range which is queried. Since it launched in early 2020,
# ignore 404's before February.
if e.status_code == 404 and time_start.date() < datetime.date(2020, 2, 1):
logger.warning("Ignoring HTTP 404 for ControlIQ API request before Feb 2020")
ciqTherapyTimelineData = None
else:
raise e
csvReadingData = None
csvIobData = None
csvBasalData = None
csvBolusData = None
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 ciqBolusData and len(ciqBolusData) > 0:
lastBolus = ciqBolusData[-1].eventDateTime
lastReading = TConnectEntry._datetime_parse(lastBolus)
logger.debug(ciqBolusData[-1].to_bolus())
logger.info("Last bolus from t:connect CIQ: %s (%s)" % (lastBolus, timeago(lastBolus)))
bolusFallingBack = (BOLUS in features and not ciqBolusData)
ciqFallingBack = (CGM in features and not ciqReadingData)
if bolusFallingBack or \
ciqFallingBack or \
BOLUS_BG in features or \
IOB in features:
logger.warning("Downloading t:connect CSV data")
if bolusFallingBack:
logger.warning("Falling back on WS2 CSV data source because BOLUS is an enabled feature and CIQ bolus data was empty!!")
if ciqFallingBack:
logger.warning("Falling back on WS2 CSV data source because CGM is an enabled feature and CIQ cgm data was empty!!")
if BOLUS_BG in features:
logger.warning("Falling back on WS2 CSV data source because BOLUS_BG is an enabled feature. " +
"Please consider disabling this feature to improve synchronization reliability.")
if IOB in features:
logger.warning("Falling back on WS2 CSV data source because IOB is an enabled feature. " +
"Please consider disabling this feature to improve synchronization reliability.")
logger.warning("<!!> The WS2 data source is unreliable and may prevent timely synchronization")
csvdata = tconnect.ws2.therapy_timeline_csv(time_start, time_end)
csvReadingData = csvdata["readingData"]
csvIobData = csvdata["iobData"]
csvBasalData = csvdata["basalData"]
csvBolusData = csvdata["bolusData"]
if csvReadingData and len(csvReadingData) > 0:
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 CSV: %s (%s)" % (lastReading, timeago(lastReading)))
else:
logger.warning("No last CGM reading is able to be determined from CSV")
added = 0
if csvReadingData:
cgmData = None
if CGM in features or BOLUS_BG in features:
logger.debug("Processing CGM events")
cgmData = process_cgm_events(csvReadingData)
if CGM in features:
logger.debug("Writing CGM events")
added += ns_write_cgm_events(nightscout, cgmData, pretend, time_start=time_start, time_end=time_end)
logger.debug("Finished writing CGM events")
if BASAL in features:
basalEvents = process_ciq_basal_events(ciqTherapyTimelineData)
if csvBasalData:
logger.debug("CSV basal data found: processing it")
add_csv_basal_events(basalEvents, csvBasalData)
else:
logger.debug("No CSV basal data found")
if basalEvents and len(basalEvents) > 0:
logger.info("Last basal event from CIQ: %s" % basalEvents[-1])
logger.debug("Writing basal events")
added += ns_write_basal_events(nightscout, basalEvents, pretend=pretend, time_start=time_start, time_end=time_end)
logger.debug("Finished writing basal events")
if PUMP_EVENTS_BASAL_SUSPENSION in features:
logger.warning("Using WS2 data source for basalsuspension because PUMP_EVENTS_BASAL_SUSPENSION is an enabled feature")
logger.warning("<!!> The WS2 data source is unreliable and may prevent timely synchronization")
ws2BasalSuspension = tconnect.ws2.basalsuspension(time_start, time_end)
bsPumpEvents = process_basalsuspension_events(ws2BasalSuspension)
logger.debug("basalsuspension events: %s" % bsPumpEvents)
logger.debug("Writing pump basalsuspension events")
added += ns_write_pump_events(nightscout, bsPumpEvents, pretend=pretend, time_start=time_start, time_end=time_end)
logger.debug("Finished writing basal events")
if PUMP_EVENTS in features:
pumpEvents = process_ciq_activity_events(ciqTherapyTimelineData)
logger.debug("CIQ activity events: %s" % pumpEvents)
logger.debug("Writing pump events")
added += ns_write_pump_events(nightscout, pumpEvents, pretend=pretend, time_start=time_start, time_end=time_end)
logger.debug("Finished writing basal events")
if BOLUS in features:
bolusEvents = []
if ciqBolusData:
logger.info("Processing ciqBolusData (%d entries)" % len(ciqBolusData))
bolusEvents = process_bolus_events(ciqBolusData, source="ciq")
if csvBolusData and not bolusEvents:
logger.warning("Falling back on non-CIQ csvBolusData")
bolusEvents = process_bolus_events(csvBolusData, source="csv")
logger.debug("ciq bolusEvents: %s" % bolusEvents)
logger.info("finalized bolusEvents: %s" % bolusEvents)
logger.debug("Writing bolus events")
added += ns_write_bolus_events(nightscout, bolusEvents, pretend=pretend, include_bg=(BOLUS_BG in features), time_start=time_start, time_end=time_end)
logger.debug("Finished writing bolus events")
if csvIobData:
if IOB in features:
iobEvents = process_iob_events(csvIobData)
logger.debug("Writing iob events")
added += ns_write_iob_events(nightscout, iobEvents, pretend=pretend)
logger.debug("Finished writing iob events")
if PROFILES in features:
logger.debug("Running profiles feature")
if process_profiles(tconnect, nightscout, pretend=pretend):
added += 1
if pretend:
logger.info("Would have written %d events to Nightscout this process cycle (in pretend mode)" % added)
else:
logger.info("Wrote %d events to Nightscout this process cycle" % added)
return added
+1 -53
View File
@@ -1,54 +1,5 @@
import tconnectsync.api
import requests
class ControlIQApi(tconnectsync.api.controliq.ControlIQApi):
def __init__(self):
self.BASE_URL = 'invalid://'
self.LOGIN_URL = 'invalid://'
self.session = requests.Session() # mocked in tests
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://'
self.SLEEP_SECONDS_INCREMENT = 0.01
def get(self, endpoint):
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 WebUIScraper(tconnectsync.api.webui.WebUIScraper):
def __init__(self, controliq):
self.controliq = controliq
def my_devices(self):
raise NotImplementedError
def device_settings(self, pump_guid):
raise NotImplementedError
class TConnectApi(tconnectsync.api.TConnectApi):
def __init__(self, email=None, password=None):
@@ -57,7 +8,4 @@ class TConnectApi(tconnectsync.api.TConnectApi):
else:
self.with_credentials = False
_ciq = ControlIQApi()
_ws2 = WS2Api()
_android = AndroidApi()
_webui = WebUIScraper(_ciq)
_tandemsource = None
-90
View File
@@ -1,90 +0,0 @@
#!/usr/bin/env python3
import unittest
import itertools
import datetime
from .fake import AndroidApi
from tconnectsync.api.common import ApiException
class TestAndroidApi(unittest.TestCase):
def fake_get_with_http_code(self, http_code, expected_endpoint, num_times):
tries = 0
def fake_get(endpoint, query):
nonlocal http_code, expected_endpoint, num_times, tries
if endpoint.endswith(expected_endpoint):
if tries < num_times:
tries += 1
raise ApiException(http_code, "fake HTTP %d" % http_code)
return {"faked_json": True}
raise NotImplementedError
return fake_get
def test_last_event_uploaded_works_after_single_http_500(self):
android = AndroidApi()
android._get = self.fake_get_with_http_code(500, "cloud/upload/getlasteventuploaded?sn=1111111", 1)
self.assertEqual(
android.last_event_uploaded(1111111),
{
"faked_json": True
})
def test_last_event_uploaded_fails_after_two_http_500s(self):
android = AndroidApi()
android._get = self.fake_get_with_http_code(500, "cloud/upload/getlasteventuploaded?sn=1111111", 2)
self.assertRaises(ApiException, android.last_event_uploaded, 1111111)
def test_last_event_uploaded_triggers_relogin_after_single_http_401(self):
android = AndroidApi()
android._email = 'email'
android._password = 'password'
hit_login = []
def stub_login(email, password):
nonlocal hit_login
hit_login.append((email, password))
android.login = stub_login
android._get = self.fake_get_with_http_code(401, "cloud/upload/getlasteventuploaded?sn=1111111", 1)
self.assertEqual(
android.last_event_uploaded(1111111),
{
"faked_json": True
})
self.assertListEqual(hit_login, [
('email', 'password')
])
def test_last_event_uploaded_fails_after_two_http_401s(self):
android = AndroidApi()
android._email = 'email'
android._password = 'password'
hit_login = []
def stub_login(email, password):
nonlocal hit_login
hit_login.append((email, password))
android.login = stub_login
android._get = self.fake_get_with_http_code(401, "cloud/upload/getlasteventuploaded?sn=1111111", 2)
self.assertRaises(ApiException, android.last_event_uploaded, 1111111)
self.assertListEqual(hit_login, [
('email', 'password')
])
if __name__ == '__main__':
unittest.main()
-293
View File
@@ -1,293 +0,0 @@
#!/usr/bin/env python3
import unittest
import itertools
import datetime
import json
import requests_mock
from bs4 import BeautifulSoup
from .fake import ControlIQApi
from tconnectsync.api.controliq import ControlIQApi as RealControlIQApi
from tconnectsync.api.common import ApiException, ApiLoginException, base_headers
class TestControlIQApi(unittest.TestCase):
LOGIN_HTML = """
<html>
<body>
<form method="post" action="./login.aspx?ReturnUrl=%2f" onsubmit="javascript:return WebForm_OnSubmit();" id="form1">
<div class="aspNetHidden">
<input type="hidden" name="__LASTFOCUS" id="__LASTFOCUS" value="" />
<input type="hidden" name="__EVENTTARGET" id="__EVENTTARGET" value="" />
<input type="hidden" name="__EVENTARGUMENT" id="__EVENTARGUMENT" value="" />
<input type="hidden" name="__VIEWSTATE" id="__VIEWSTATE" value="AAAAA" />
</div>
<div class="aspNetHidden">
<input type="hidden" name="__VIEWSTATEGENERATOR" id="__VIEWSTATEGENERATOR" value="BBBBB" />
<input type="hidden" name="__EVENTVALIDATION" id="__EVENTVALIDATION" value="CCCCC" />
</div>
</form>
</body>
</html>
"""
LOGIN_POST_DATA = {
"__LASTFOCUS": "",
"__EVENTTARGET": "ctl00$ContentBody$LoginControl$linkLogin",
"__EVENTARGUMENT": "",
"__VIEWSTATE": "AAAAA",
"__VIEWSTATEGENERATOR": "BBBBB",
"__EVENTVALIDATION": "CCCCC",
"ctl00$ContentBody$LoginControl$txtLoginEmailAddress": "email@email.com",
"txtLoginEmailAddress_ClientState": '{"enabled":true,"emptyMessage":"","validationText":"%s","valueAsString":"%s","lastSetTextBoxValue":"%s"}' % ("email@email.com", "email@email.com", "email@email.com"),
"ctl00$ContentBody$LoginControl$txtLoginPassword": "password",
"txtLoginPassword_ClientState": '{"enabled":true,"emptyMessage":"","validationText":"%s","valueAsString":"%s","lastSetTextBoxValue":"%s"}' % ("password", "password", "password")
}
def test_build_login_data(self):
ciq = ControlIQApi()
soup = BeautifulSoup(self.LOGIN_HTML, features='lxml')
self.assertDictEqual(
ciq._build_login_data('email@email.com', 'password', soup),
self.LOGIN_POST_DATA)
def test_login_successful(self):
ciq = ControlIQApi()
ciq.LOGIN_URL = RealControlIQApi.LOGIN_URL
ciq.login = lambda email, password: RealControlIQApi.login(ciq, email, password)
with requests_mock.Mocker() as m:
m.get('https://tconnect.tandemdiabetes.com/login.aspx?ReturnUrl=%2f',
request_headers=base_headers(),
text=self.LOGIN_HTML)
def post_callback(request, context):
context.status_code = 302
context.headers['Location'] = '/newlocation'
context.cookies['UserGUID'] = 'user_guid'
context.cookies['accessToken'] = 'access_tok'
context.cookies['accessTokenExpiresAt'] = '2021-05-04T11:18:08.381Z'
return ''
m.post('https://tconnect.tandemdiabetes.com/login.aspx?ReturnUrl=%2f',
request_headers={'Referer': ciq.LOGIN_URL, **base_headers()},
text=post_callback)
m.post('https://tconnect.tandemdiabetes.com/newlocation',
cookies={'cookie': 'value'},
headers=base_headers(),
status_code=200)
self.assertTrue(ciq.login('email@email.com', 'password'))
self.assertEqual(ciq.userGuid, 'user_guid')
self.assertEqual(ciq.accessToken, 'access_tok')
self.assertEqual(ciq.accessTokenExpiresAt, '2021-05-04T11:18:08.381Z')
def test_login_invalid_credentials(self):
ciq = ControlIQApi()
ciq.LOGIN_URL = RealControlIQApi.LOGIN_URL
ciq.login = lambda email, password: RealControlIQApi.login(ciq, email, password)
with requests_mock.Mocker() as m:
m.get('https://tconnect.tandemdiabetes.com/login.aspx?ReturnUrl=%2f',
request_headers=base_headers(),
text=self.LOGIN_HTML)
def post_callback(request, context):
context.status_code = 200
return '<html><body>...</body></html>'
m.post('https://tconnect.tandemdiabetes.com/login.aspx?ReturnUrl=%2f',
request_headers={'Referer': ciq.LOGIN_URL, **base_headers()},
text=post_callback)
self.assertRaisesRegex(ApiLoginException, 'Error logging in to t:connect: Check your login credentials.', ciq.login, 'email@email.com', 'password')
self.assertIsNone(ciq.userGuid)
self.assertIsNone(ciq.accessToken)
self.assertIsNone(ciq.accessTokenExpiresAt)
def test_login_invalid_credentials_parsed_message(self):
ciq = ControlIQApi()
ciq.LOGIN_URL = RealControlIQApi.LOGIN_URL
ciq.login = lambda email, password: RealControlIQApi.login(ciq, email, password)
with requests_mock.Mocker() as m:
m.get('https://tconnect.tandemdiabetes.com/login.aspx?ReturnUrl=%2f',
request_headers=base_headers(),
text=self.LOGIN_HTML)
def post_callback(request, context):
context.status_code = 200
return '<html><body><div class="notice_error" id="literalMessage" style="">The email address or password you entered is invalid. Please re-enter and try again.</div></body></html>'
m.post('https://tconnect.tandemdiabetes.com/login.aspx?ReturnUrl=%2f',
request_headers={'Referer': ciq.LOGIN_URL, **base_headers()},
text=post_callback)
self.assertRaisesRegex(ApiLoginException, 'Error logging in to t:connect: The email address or password you entered is invalid. Please re-enter and try again.', ciq.login, 'email@email.com', 'password')
self.assertIsNone(ciq.userGuid)
self.assertIsNone(ciq.accessToken)
self.assertIsNone(ciq.accessTokenExpiresAt)
def test_login_unexpected_http_code(self):
ciq = ControlIQApi()
ciq.LOGIN_URL = RealControlIQApi.LOGIN_URL
ciq.login = lambda email, password: RealControlIQApi.login(ciq, email, password)
with requests_mock.Mocker() as m:
m.get('https://tconnect.tandemdiabetes.com/login.aspx?ReturnUrl=%2f',
request_headers=base_headers(),
text=self.LOGIN_HTML)
def post_callback(request, context):
context.status_code = 500
return '<html><body>...</body></html>'
m.post('https://tconnect.tandemdiabetes.com/login.aspx?ReturnUrl=%2f',
request_headers={'Referer': ciq.LOGIN_URL, **base_headers()},
text=post_callback)
self.assertRaisesRegex(ApiLoginException, 'Error logging in to t:connect \(HTTP 500\)', ciq.login, 'email@email.com', 'password')
self.assertIsNone(ciq.userGuid)
self.assertIsNone(ciq.accessToken)
self.assertIsNone(ciq.accessTokenExpiresAt)
def fake_get_with_http_code(self, http_code, expected_endpoint, num_times):
tries = 0
def fake_get(endpoint, query):
nonlocal http_code, expected_endpoint, num_times, tries
if endpoint.split("?")[0].endswith(expected_endpoint):
if tries < num_times:
tries += 1
raise ApiException(http_code, "fake HTTP %d" % http_code)
return {"faked_json": True}
raise NotImplementedError
return fake_get
def test_therapy_timeline_works_after_single_http_500(self):
ciq = ControlIQApi()
ciq.userGuid = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
ciq._get = self.fake_get_with_http_code(500, "therapytimeline/users/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", 1)
self.assertEqual(
ciq.therapy_timeline('2021-04-01', '2021-04-02'),
{
"faked_json": True
})
def test_therapy_timeline_fails_after_two_http_500s(self):
ciq = ControlIQApi()
ciq.userGuid = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
ciq._get = self.fake_get_with_http_code(500, "therapytimeline/users/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", 2)
self.assertRaises(ApiException, ciq.therapy_timeline, '2021-04-01', '2021-04-02')
def test_therapy_timeline_triggers_relogin_after_single_http_401(self):
ciq = ControlIQApi()
ciq.userGuid = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
ciq._email = 'email'
ciq._password = 'password'
hit_login = []
def stub_login(email, password):
nonlocal hit_login
hit_login.append((email, password))
ciq.login = stub_login
ciq._get = self.fake_get_with_http_code(401, "therapytimeline/users/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", 1)
self.assertEqual(
ciq.therapy_timeline('2021-04-01', '2021-04-02'),
{
"faked_json": True
})
self.assertListEqual(hit_login, [
('email', 'password')
])
def test_therapy_timeline_fails_after_two_http_401s(self):
ciq = ControlIQApi()
ciq.userGuid = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
ciq._email = 'email'
ciq._password = 'password'
hit_login = []
def stub_login(email, password):
nonlocal hit_login
hit_login.append((email, password))
ciq.login = stub_login
ciq._get = self.fake_get_with_http_code(401, "therapytimeline/users/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", 2)
self.assertRaises(ApiException, ciq.therapy_timeline, '2021-04-01', '2021-04-02')
self.assertListEqual(hit_login, [
('email', 'password')
])
def test_therapy_timeline_parses_date(self):
ciq = ControlIQApi()
ciq.userGuid = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
def fake_get(raw_endpoint, ignored_query):
endpoint, query = raw_endpoint.split("?")
self.assertTrue(endpoint.endswith("therapytimeline/users/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"))
self.assertEqual(query, "startDate=04-01-2021&endDate=04-02-2021")
return {"faked_json": True}
ciq._get = fake_get
self.assertEqual(
ciq.therapy_timeline(datetime.date(2021, 4, 1), datetime.date(2021, 4, 2)),
{
"faked_json": True
})
def test_dashboard_summary_parses_date(self):
ciq = ControlIQApi()
ciq.userGuid = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
def fake_get(raw_endpoint, ignored_query):
endpoint, query = raw_endpoint.split("?")
self.assertTrue(endpoint.endswith("summary/users/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"))
self.assertEqual(query, "startDate=04-01-2021&endDate=04-02-2021")
return {"faked_json": True}
ciq._get = fake_get
self.assertEqual(
ciq.dashboard_summary(datetime.date(2021, 4, 1), datetime.date(2021, 4, 2)),
{
"faked_json": True
})
if __name__ == '__main__':
unittest.main()
File diff suppressed because it is too large Load Diff
-179
View File
@@ -1,179 +0,0 @@
#!/usr/bin/env python3
import unittest
import itertools
import copy
from .fake import WS2Api
from tconnectsync.api.common import ApiException
class TestWS2Api(unittest.TestCase):
def fake_get_with_http_500(self, num_times):
tries = 0
def fake_get(endpoint, **kwargs):
nonlocal tries, num_times
if "therapytimeline2csv" in endpoint:
if tries < num_times:
tries += 1
raise ApiException(500, "fake HTTP 500")
return ""
raise NotImplementedError
return fake_get
def test_therapy_timeline_csv_works_after_two_retries(self):
ws2 = WS2Api()
ws2.get = self.fake_get_with_http_500(2)
self.assertEqual(
ws2.therapy_timeline_csv('04-01-2021', '04-02-2021'),
{
"readingData": [],
"iobData": [],
"basalData": [],
"bolusData": []
})
def test_therapy_timeline_csv_fails_after_three_retries(self):
ws2 = WS2Api()
ws2.get = self.fake_get_with_http_500(3)
self.assertRaises(ApiException, ws2.therapy_timeline_csv, '04-01-2021', '04-02-2021')
RAW_DATA_HEADER = """Tandem Diabetes Care Inc.
t:connect Therapy Timeline Data Export
Patient Name, Sample Name
Patient DOB, 1/1/1990
Report Generated On, 4/24/2021 7:50:04 PM
"""
RAW_DATA_CGM = """DeviceType,SerialNumber,Description,EventDateTime,Readings (CGM / BGM)
"t:slim X2 Insulin Pump","11111111","EGV","2021-04-01T00:01:33","235",
"t:slim X2 Insulin Pump","11111111","EGV","2021-04-01T00:06:33","230",
"t:slim X2 Insulin Pump","11111111","EGV","2021-04-02T23:31:36","181",
"""
RAW_DATA_IOB = """Type,EventID,EventDateTime,IOB
"IOB","81","2021-04-01T00:00:19","13.24"
"IOB","9","2021-04-01T00:03:12","12.80"
"IOB","81","2021-04-02T23:58:19","4.25"
"""
RAW_DATA_BOLUS = """Type,Description,BG,IOB,BolusRequestID,BolusCompletionID,CompletionDateTime,InsulinDelivered,FoodDelivered,CorrectionDelivered,CompletionStatusID,CompletionStatusDesc,BolusIsComplete,BolexCompletionID,BolexSize,BolexStartDateTime,BolexCompletionDateTime,BolexInsulinDelivered,BolexIOB,BolexCompletionStatusID,BolexCompletionStatusDesc,ExtendedBolusIsComplete,EventDateTime,RequestDateTime,BolusType,BolusRequestOptions,StandardPercent,Duration,CarbSize,UserOverride,TargetBG,CorrectionFactor,FoodBolusSize,CorrectionBolusSize,ActualTotalBolusRequested,IsQuickBolus,EventHistoryReportEventDesc,EventHistoryReportDetails,NoteID,IndexID,Note
"Bolus","Standard/Correction","141",,"7001.000","7001.000","2021-04-01T12:58:26","13.53","12.50","1.03","3","Completed","1",,,,,,,,,,"2021-04-01T12:53:36","2021-04-01T12:53:36","Carb","Standard/Correction","100.00","0","75","0","110","30.00","12.50","1.03","13.53","0","0","Correction & Food Bolus","CF 1:30 - Carb Ratio 1:6 - Target BG 110","0","1181649","",
"Bolus","Standard","131","0.71","7003.000","7003.000","2021-04-01T16:03:25","1.50","0.00","0.00","3","Completed","1",,,,,,,,,,"2021-04-01T16:02:04","2021-04-01T16:02:04","Carb","Standard","100.00","0","0","1","110","30.00","0.00","0.00","1.50","0","0","Food Bolus","CF 1:30 - Carb Ratio 1:6 - Target BG 110 | Override: Pump calculated Bolus = 0.0 units","0","1182026","",
"Bolus","Standard/Correction","168","1.71","7004.000","7004.000","2021-04-01T16:24:08","2.00","0.00","0.00","3","Completed","1",,,,,,,,,,"2021-04-01T16:22:21","2021-04-01T16:22:21","Carb","Standard/Correction","100.00","0","0","1","110","30.00","0.00","0.22","2.00","0","0","Correction & Food Bolus","CF 1:30 - Carb Ratio 1:6 - Target BG 110 | Override: Pump calculated Bolus = 0.2 units","0","1182082","",
"Bolus","Standard","220","3.98","7032.000","7032.000","2021-04-02T23:16:24","2.50","0.00","0.00","3","Completed","1",,,,,,,,,,"2021-04-02T23:14:33","2021-04-02T23:14:33","Carb","Standard","100.00","0","0","1","110","30.00","0.00","0.00","2.50","0","0","Food Bolus","CF 1:30 - Carb Ratio 1:6 - Target BG 110 | Override: Pump calculated Bolus = 0.0 units","0","1185846","",
"""
RAW_DATA_FULL = RAW_DATA_HEADER + "\n" + RAW_DATA_CGM + "\n" + RAW_DATA_IOB + "\n" + RAW_DATA_BOLUS
PARSED_DATA = {
'readingData': [
{"DeviceType": "t:slim X2 Insulin Pump", "SerialNumber": "11111111", "Description": "EGV", "EventDateTime": "2021-04-01T00:01:33", "Readings (CGM / BGM)": "235"},
{"DeviceType": "t:slim X2 Insulin Pump", "SerialNumber": "11111111", "Description": "EGV", "EventDateTime": "2021-04-01T00:06:33", "Readings (CGM / BGM)": "230"},
{"DeviceType": "t:slim X2 Insulin Pump", "SerialNumber": "11111111", "Description": "EGV", "EventDateTime": "2021-04-02T23:31:36", "Readings (CGM / BGM)": "181"}
],
'iobData': [
{"Type": "IOB", "EventID": "81", "EventDateTime": "2021-04-01T00:00:19", "IOB": "13.24"},
{"Type": "IOB", "EventID": "9", "EventDateTime": "2021-04-01T00:03:12", "IOB": "12.80"},
{"Type": "IOB", "EventID": "81", "EventDateTime": "2021-04-02T23:58:19", "IOB": "4.25"},
],
'basalData': [],
'bolusData': [
{"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"},
{"Type": "Bolus", "Description": "Standard", "BG": "131", "IOB": "0.71", "BolusRequestID": "7003.000", "BolusCompletionID": "7003.000", "CompletionDateTime": "2021-04-01T16:03:25", "InsulinDelivered": "1.50", "FoodDelivered": "0.00", "CorrectionDelivered": "0.00", "CompletionStatusID": "3", "CompletionStatusDesc": "Completed", "BolusIsComplete": "1", "BolexCompletionID": "", "BolexSize": "", "BolexStartDateTime": "", "BolexCompletionDateTime": "", "BolexInsulinDelivered": "", "BolexIOB": "", "BolexCompletionStatusID": "", "BolexCompletionStatusDesc": "", "ExtendedBolusIsComplete": "", "EventDateTime": "2021-04-01T16:02:04", "RequestDateTime": "2021-04-01T16:02:04", "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.50", "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": "1182026"},
{"Type": "Bolus", "Description": "Standard/Correction", "BG": "168", "IOB": "1.71", "BolusRequestID": "7004.000", "BolusCompletionID": "7004.000", "CompletionDateTime": "2021-04-01T16:24:08", "InsulinDelivered": "2.00", "FoodDelivered": "0.00", "CorrectionDelivered": "0.00", "CompletionStatusID": "3", "CompletionStatusDesc": "Completed", "BolusIsComplete": "1", "BolexCompletionID": "", "BolexSize": "", "BolexStartDateTime": "", "BolexCompletionDateTime": "", "BolexInsulinDelivered": "", "BolexIOB": "", "BolexCompletionStatusID": "", "BolexCompletionStatusDesc": "", "ExtendedBolusIsComplete": "", "EventDateTime": "2021-04-01T16:22:21", "RequestDateTime": "2021-04-01T16:22:21", "BolusType": "Carb", "BolusRequestOptions": "Standard/Correction", "StandardPercent": "100.00", "Duration": "0", "CarbSize": "0", "UserOverride": "1", "TargetBG": "110", "CorrectionFactor": "30.00", "FoodBolusSize": "0.00", "CorrectionBolusSize": "0.22", "ActualTotalBolusRequested": "2.00", "IsQuickBolus": "0", "EventHistoryReportEventDesc": "0", "EventHistoryReportDetails": "Correction & Food Bolus", "NoteID": "CF 1:30 - Carb Ratio 1:6 - Target BG 110 | Override: Pump calculated Bolus = 0.2 units", "IndexID": "0", "Note": "1182082"},
{"Type": "Bolus", "Description": "Standard", "BG": "220", "IOB": "3.98", "BolusRequestID": "7032.000", "BolusCompletionID": "7032.000", "CompletionDateTime": "2021-04-02T23:16:24", "InsulinDelivered": "2.50", "FoodDelivered": "0.00", "CorrectionDelivered": "0.00", "CompletionStatusID": "3", "CompletionStatusDesc": "Completed", "BolusIsComplete": "1", "BolexCompletionID": "", "BolexSize": "", "BolexStartDateTime": "", "BolexCompletionDateTime": "", "BolexInsulinDelivered": "", "BolexIOB": "", "BolexCompletionStatusID": "", "BolexCompletionStatusDesc": "", "ExtendedBolusIsComplete": "", "EventDateTime": "2021-04-02T23:14:33", "RequestDateTime": "2021-04-02T23:14:33", "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": "2.50", "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": "1185846"}
]
}
def test_therapy_timeline_csv_parses_full(self):
ws2 = WS2Api()
ws2.userGuid = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
rawData = self.RAW_DATA_FULL
def fake_get(endpoint, **kwargs):
nonlocal rawData
if endpoint == 'therapytimeline2csv/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee/04-01-2021/04-02-2021?format=csv':
return rawData
ws2.get = fake_get
tt = ws2.therapy_timeline_csv('04-01-2021', '04-02-2021')
self.assertDictEqual(tt, self.PARSED_DATA)
def test_therapy_timeline_csv_parses_random_order(self):
ws2 = WS2Api()
ws2.userGuid = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
rawData = ""
def fake_get(endpoint, **kwargs):
nonlocal rawData
if endpoint == 'therapytimeline2csv/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee/04-01-2021/04-02-2021?format=csv':
return rawData
ws2.get = fake_get
# Randomize the order of all sections
for i in itertools.permutations([self.RAW_DATA_HEADER, self.RAW_DATA_CGM, self.RAW_DATA_IOB, self.RAW_DATA_BOLUS], 4):
rawData = "\n".join(i)
tt = ws2.therapy_timeline_csv('04-01-2021', '04-02-2021')
self.assertDictEqual(tt, self.PARSED_DATA)
def test_therapy_timeline_csv_split_past_max_days(self):
ws2 = WS2Api()
ws2.userGuid = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
def replace_str(raw, one, two):
return raw.replace('04-01-2021', one).replace('04-02-2021', two)
rawData1 = self.RAW_DATA_FULL
rawData2 = replace_str(self.RAW_DATA_FULL, '04-03-2021', '04-04-2021')
rawData3 = replace_str(self.RAW_DATA_FULL, '04-05-2021', '04-06-2021')
rawData4 = replace_str(self.RAW_DATA_FULL, '04-07-2021', '04-07-2021')
def replace_parsed(one, two):
parsedData = copy.deepcopy(self.PARSED_DATA)
for typ in parsedData.keys():
for i in range(len(parsedData[typ])):
for f in parsedData[typ][i].keys():
if 'datetime' in f.lower():
parsedData[typ][i][f] = replace_str(parsedData[typ][i][f], one, two)
return parsedData
parsedData1 = self.PARSED_DATA
parsedData2 = replace_parsed('04-03-2021', '04-04-2021')
parsedData3 = replace_parsed('04-05-2021', '04-06-2021')
parsedData4 = replace_parsed('04-07-2021', '04-07-2021')
fullParsedData = parsedData1
for d in [parsedData2, parsedData3, parsedData4]:
for typ in d.keys():
fullParsedData[typ] += d[typ]
def fake_get(endpoint, **kwargs):
nonlocal rawData1, rawData2, rawData3, rawData4
print('fake_get call %s' % endpoint)
if endpoint == 'therapytimeline2csv/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee/04-01-2021/04-02-2021?format=csv':
return rawData1
elif endpoint == 'therapytimeline2csv/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee/04-03-2021/04-04-2021?format=csv':
return rawData2
elif endpoint == 'therapytimeline2csv/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee/04-05-2021/04-06-2021?format=csv':
return rawData3
elif endpoint == 'therapytimeline2csv/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee/04-07-2021/04-07-2021?format=csv':
return rawData4
ws2.get = fake_get
tt = ws2.therapy_timeline_csv('04-01-2021', '04-07-2021')
self.assertDictEqual(tt, fullParsedData)
if __name__ == '__main__':
unittest.main()
-367
View File
@@ -1,367 +0,0 @@
import dataclasses
import unittest
from tconnectsync.domain.bolus import Bolus
from tconnectsync.domain.therapy_event import BolusTherapyEvent, CGMTherapyEvent
class TestCGMTherapyEvent(unittest.TestCase):
maxDiff = None
sampleJson = {
"eventDateTime": "2022-07-21T00:00:08",
"eventID": 256,
"requestDateTime": "0001-01-01T00:00:00",
"type": "CGM",
"description": "EGV",
"sourceRecId": 0,
"eventTypeId": 0,
"deviceType": "t:slim X2 Insulin Pump",
"serialNumber": "xxx",
"indexId": 0,
"uploadId": 0,
"interactive": 0,
"tempRateId": 0,
"tempRateCompleted": 0,
"tempRateActivated": 0,
"egv": {
"estimatedGlucoseValue": 174,
"hypo": 0,
"belowTarget": 0,
"withinTarget": 1,
"aboveTarget": 0,
"hyper": 0
}
}
def test_parse_cgm(self):
e = CGMTherapyEvent.parse(self.sampleJson)
self.assertEqual(e.type, "CGM")
self.assertEqual(e.eventDateTime, "2022-07-21T00:00:08")
self.assertEqual(e.sourceRecId, 0)
self.assertEqual(e.eventID, 256)
self.assertEqual(e.egv, 174)
class TestBolusTherapyEvent(unittest.TestCase):
maxDiff = None
standardJson = {
"actualTotalBolusRequested": 4.17,
"bolusRequestOptions": "Standard",
"bolusType": "Carb",
"carbSize": 25,
"correctionBolusSize": 0,
"correctionFactor": 30,
"declinedCorrection": 0,
"duration": 0,
"eventDateTime": "2022-07-21T12:27:36",
"eventHistoryReportDetails": "CF 1:30 - Carb Ratio 1:6 - Target BG 110",
"eventHistoryReportEventDesc": "Food Bolus",
"foodBolusSize": 4.17,
"iob": 2.62,
"isQuickBolus": 0,
"note": {
"id": 0,
"indexId": "573042",
"eventTypeId": 64,
"sourceRecordId": 0,
"eventId": 0,
"active": False
},
"requestDateTime": "2022-07-21T12:27:36",
"standard": {
"insulinDelivered": {
"completionDateTime": "2022-07-21T12:29:21",
"value": 4.17
},
"foodDelivered": 4.17,
"correctionDelivered": 0,
"insulinRequested": 4.17,
"completionStatusId": 3,
"completionStatusDesc": "Completed",
"bolusIsComplete": 1,
"bolusRequestId": 3362,
"bolusCompletionId": 3362
},
"standardPercent": 100,
"targetBG": 110,
"userOverride": 0,
"type": "Bolus",
"description": "Standard",
"sourceRecId": 1171853319,
"eventTypeId": 0,
"indexId": 0,
"uploadId": 0,
"interactive": 0,
"tempRateId": 0,
"tempRateCompleted": 0,
"tempRateActivated": 0
}
def test_standard_to_bolus(self):
e = BolusTherapyEvent.parse(self.standardJson)
self.assertIsNotNone(e)
b = e.to_bolus()
self.assertEqual(dataclasses.asdict(b), dataclasses.asdict(Bolus(
description="Standard",
complete="1",
completion="Completed",
request_time="2022-07-21 12:27:36-04:00",
completion_time="2022-07-21 12:29:21-04:00",
insulin="4.17",
requested_insulin="4.17",
carbs="25",
bg="",
user_override="0",
extended_bolus="0",
bolex_completion_time="",
bolex_start_time=""
)))
correctionJson = {
"actualTotalBolusRequested": 2.9,
"bg": 254,
"bolusRequestOptions": "Automatic Bolus/Correction",
"bolusType": "Automatic Correction",
"carbSize": 0,
"correctionBolusSize": 2.9,
"correctionFactor": 30,
"declinedCorrection": 0,
"duration": 0,
"eventDateTime": "2022-07-21T11:53:08",
"eventHistoryReportDetails": "CF 1:30 - Carb Ratio 1:0 - Target BG 110",
"eventHistoryReportEventDesc": "Correction Bolus",
"foodBolusSize": 0,
"isQuickBolus": 0,
"note": {
"id": 0,
"indexId": "572946",
"eventTypeId": 64,
"sourceRecordId": 0,
"eventId": 0,
"active": False
},
"requestDateTime": "2022-07-21T11:53:08",
"standard": {
"insulinDelivered": {
"completionDateTime": "2022-07-21T11:55:24",
"value": 2.9
},
"foodDelivered": 0,
"correctionDelivered": 2.9,
"insulinRequested": 2.9,
"completionStatusId": 3,
"completionStatusDesc": "Completed",
"bolusIsComplete": 1,
"bolusRequestId": 3361,
"bolusCompletionId": 3361
},
"standardPercent": 100,
"targetBG": 110,
"userOverride": 0,
"type": "Bolus",
"description": "Automatic Bolus/Correction",
"sourceRecId": 1171791787,
"eventTypeId": 0,
"indexId": 0,
"uploadId": 0,
"interactive": 0,
"tempRateId": 0,
"tempRateCompleted": 0,
"tempRateActivated": 0
}
def test_correction_to_bolus(self):
e = BolusTherapyEvent.parse(self.correctionJson)
self.assertIsNotNone(e)
b = e.to_bolus()
self.assertEqual(dataclasses.asdict(b), dataclasses.asdict(Bolus(
description="Automatic Bolus/Correction",
complete="1",
completion="Completed",
request_time="2022-07-21 11:53:08-04:00",
completion_time="2022-07-21 11:55:24-04:00",
insulin="2.9",
requested_insulin="2.9",
carbs="0",
bg="254",
user_override="0",
extended_bolus="0",
bolex_completion_time="",
bolex_start_time=""
)))
extendedBolusIncompleteJson = {
"actualTotalBolusRequested": 0.4,
"bg": 131,
"bolex": {
"size": 0.2,
"bolexStartDateTime": "2022-08-09T23:20:04",
"iob": 0,
"completionStatusId": 0,
"extendedBolusIsComplete": 0,
"insulinRequested": 0,
"bolexCompletionId": 0
},
"bolusRequestOptions": "Extended",
"bolusType": "Carb",
"carbSize": 0,
"correctionBolusSize": 0.0,
"correctionFactor": 30.0,
"declinedCorrection": 0,
"duration": 15,
"eventDateTime": "2022-08-09T23:19:15",
"eventHistoryReportDetails": "CF 1:30 - Carb Ratio 1:6 - Target BG 110<br/>Override: Pump calculated Bolus = 0.0 units",
"eventHistoryReportEventDesc": "Food Bolus: 50&#37; Extended 15 mins",
"foodBolusSize": 0.0,
"iob": 5.87,
"isQuickBolus": 0,
"note": {
"id": 0,
"indexId": "631597",
"eventTypeId": 64,
"sourceRecordId": 0,
"eventId": 0,
"active": False
},
"requestDateTime": "2022-08-09T23:19:15",
"standard": {
"insulinDelivered": {
"completionDateTime": "2022-08-09T23:20:04",
"value": 0.2
},
"foodDelivered": 0.0,
"correctionDelivered": 0.0,
"insulinRequested": 0.2,
"completionStatusId": 3,
"completionStatusDesc": "Completed",
"bolusIsComplete": 1,
"bolusRequestId": 3636.0,
"bolusCompletionId": 3636.0
},
"standardPercent": 50.0,
"targetBG": 110,
"userOverride": 1,
"type": "Bolus",
"description": "Extended 50.00%/0.00",
"sourceRecId": 1209631944,
"eventTypeId": 0,
"indexId": 0,
"uploadId": 0,
"interactive": 0,
"tempRateId": 0,
"tempRateCompleted": 0,
"tempRateActivated": 0
}
def test_extended_bolus_incomplete_to_bolus(self):
e = BolusTherapyEvent.parse(self.extendedBolusIncompleteJson)
self.assertIsNotNone(e)
b = e.to_bolus()
self.assertEqual(dataclasses.asdict(b), dataclasses.asdict(Bolus(
description="Extended 50.00%/0.00",
complete="0",
completion="",
request_time="2022-08-09 23:19:15-04:00",
completion_time="2022-08-09 23:20:04-04:00",
insulin="0.2",
requested_insulin="0.2",
carbs="0",
bg="131",
user_override="1",
extended_bolus="1",
bolex_completion_time="",
bolex_start_time="2022-08-09 23:20:04-04:00"
)))
extendedBolusJson = {
"actualTotalBolusRequested": 0.4,
"bg": 131,
"bolex": {
"size": 0.2,
"bolexStartDateTime": "2022-08-09T23:20:04",
"insulinDelivered": {
"completionDateTime": "2022-08-09T23:35:03",
"value": 0.2
},
"iob": 5.7,
"completionStatusId": 3.0,
"completionStatusDesc": "Completed",
"extendedBolusIsComplete": 1,
"insulinRequested": 0.2,
"bolexCompletionId": 16757133
},
"bolusRequestOptions": "Extended",
"bolusType": "Carb",
"carbSize": 0,
"correctionBolusSize": 0.0,
"correctionFactor": 30.0,
"declinedCorrection": 0,
"duration": 15,
"eventDateTime": "2022-08-09T23:19:15",
"eventHistoryReportDetails": "CF 1:30 - Carb Ratio 1:6 - Target BG 110<br/>Override: Pump calculated Bolus = 0.0 units",
"eventHistoryReportEventDesc": "Food Bolus: 50&#37; Extended 15 mins",
"foodBolusSize": 0.0,
"iob": 5.87,
"isQuickBolus": 0,
"note": {
"id": 0,
"indexId": "631597",
"eventTypeId": 64,
"sourceRecordId": 0,
"eventId": 0,
"active": False
},
"requestDateTime": "2022-08-09T23:19:15",
"standard": {
"insulinDelivered": {
"completionDateTime": "2022-08-09T23:20:04",
"value": 0.2
},
"foodDelivered": 0.0,
"correctionDelivered": 0.0,
"insulinRequested": 0.2,
"completionStatusId": 3,
"completionStatusDesc": "Completed",
"bolusIsComplete": 1,
"bolusRequestId": 3636.0,
"bolusCompletionId": 3636.0
},
"standardPercent": 50.0,
"targetBG": 110,
"userOverride": 1,
"type": "Bolus",
"description": "Extended 50.00%/0.00",
"sourceRecId": 1209631944,
"eventTypeId": 0,
"indexId": 0,
"uploadId": 0,
"interactive": 0,
"tempRateId": 0,
"tempRateCompleted": 0,
"tempRateActivated": 0
}
def test_extended_bolus_complete_to_bolus(self):
e = BolusTherapyEvent.parse(self.extendedBolusJson)
self.assertIsNotNone(e)
b = e.to_bolus()
self.assertEqual(dataclasses.asdict(b), dataclasses.asdict(Bolus(
description="Extended 50.00%/0.00",
complete="1",
completion="Completed",
request_time="2022-08-09 23:19:15-04:00",
completion_time="2022-08-09 23:20:04-04:00",
insulin="0.2",
requested_insulin="0.2",
carbs="0",
bg="131",
user_override="1",
extended_bolus="1",
bolex_completion_time="2022-08-09 23:35:03-04:00",
bolex_start_time="2022-08-09 23:20:04-04:00"
)))
BOLUS_FULL_EXAMPLES = [
TestBolusTherapyEvent.standardJson,
TestBolusTherapyEvent.correctionJson,
TestBolusTherapyEvent.extendedBolusJson
]
+1 -36
View File
@@ -1,9 +1,7 @@
#!/usr/bin/env python3
import unittest
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 .test_profile_data import DEVICE_PROFILE_A, DEVICE_SETTINGS, NS_PROFILE_A
from tconnectsync.parser.nightscout import NightscoutEntry, InvalidBolusTypeException
class TestNightscoutEntry(unittest.TestCase):
maxDiff = None
@@ -189,38 +187,5 @@ class TestNightscoutEntry(unittest.TestCase):
)
def test_profile_store(self):
self.assertEqual(
NightscoutEntry.profile_store(
profile=DEVICE_PROFILE_A,
device_settings=DEVICE_SETTINGS
),
NS_PROFILE_A
)
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()
-270
View File
@@ -1,270 +0,0 @@
from tconnectsync.domain.device_settings import Profile, ProfileSegment, DeviceSettings
from tconnectsync.secret import NIGHTSCOUT_PROFILE_CARBS_HR_VALUE, NIGHTSCOUT_PROFILE_DELAY_VALUE, TIMEZONE_NAME
DEVICE_PROFILE_A = Profile(
title='A',
active=False,
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_PROFILE_B = Profile(
title='B',
active=False,
segments=[
ProfileSegment(
display_time='Midnight',
time='12:00 AM',
basal_rate=0.8,
correction_factor=30.0,
carb_ratio=12.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=12.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=12.0,
target_bg_mgdl=110.0),
ProfileSegment(
display_time='Noon',
time='12:00 PM',
basal_rate=0.9,
correction_factor=30.0,
carb_ratio=12.0,
target_bg_mgdl=110.0)
],
calculated_total_daily_basal=22.85,
insulin_duration_min=300,
carbs_enabled=True
)
DEVICE_SETTINGS = DeviceSettings(
low_bg_threshold=80,
high_bg_threshold=200,
raw_settings={}
)
NS_PROFILE_A = {
"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"
}
NS_PROFILE_B = {
"dia": "5.0",
"carbratio": [
{
"time": "00:00",
"timeAsSeconds": 0,
"value": 12.0
},
{
"time": "06:00",
"timeAsSeconds": 6*60*60,
"value": 12.0
},
{
"time": "11:00",
"timeAsSeconds": 11*60*60,
"value": 12.0
},
{
"time": "12:00",
"timeAsSeconds": 12*60*60,
"value": 12.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.9
}
],
"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"
}
NS_PROFILE_STORE = {
'A': NS_PROFILE_A,
'B': NS_PROFILE_B
}
-704
View File
@@ -1,704 +0,0 @@
#!/usr/bin/env python3
import unittest
from tconnectsync.domain.bolus import Bolus
from tconnectsync.parser.tconnect import TConnectEntry, UnknownBasalSuspensionEventException, UnknownCIQActivityEventException
class TestTConnectEntryBasal(unittest.TestCase):
def test_parse_ciq_basal_entry(self):
self.assertEqual(
TConnectEntry.parse_ciq_basal_entry({
"y": 0.8,
"duration": 1221,
"x": 1615878000
}),
{
"time": "2021-03-16 00:00:00-04:00",
"delivery_type": "",
"duration_mins": 1221/60,
"basal_rate": 0.8,
}
)
self.assertEqual(
TConnectEntry.parse_ciq_basal_entry({
"y": 0.797,
"duration": 300,
"x": 1615879521
}, delivery_type="algorithmDelivery"),
{
"time": "2021-03-16 00:25:21-04:00",
"delivery_type": "algorithmDelivery",
"duration_mins": 5,
"basal_rate": 0.797,
}
)
class TestTConnectEntrySuspension(unittest.TestCase):
def test_parse_suspension_entry(self):
self.assertEqual(
TConnectEntry.parse_suspension_entry({
"suspendReason": "control-iq",
"continuation": None,
"x": 1615879821
}),
{
"time": "2021-03-16 00:30:21-04:00",
"continuation": None,
"suspendReason": "control-iq"
}
)
self.assertEqual(
TConnectEntry.parse_suspension_entry({
"suspendReason": "control-iq",
"continuation": "previous",
"x": 1634022000
}),
{
"time": "2021-10-12 00:00:00-04:00",
"continuation": "previous",
"suspendReason": "control-iq"
}
)
class TestTConnectEntrySuspensionToBasal(unittest.TestCase):
def test_manual_suspension_to_basal_entry(self):
suspension = {
"time": "2021-03-16 00:30:21-04:00",
"continuation": None,
"suspendReason": "manual"
}
self.assertEqual(
TConnectEntry.manual_suspension_to_basal_entry(
suspension,
seconds=300
), {
"time": "2021-03-16 00:30:21-04:00",
"delivery_type": "manual suspension",
"duration_mins": 5.0,
"basal_rate": 0.0
}
)
class TestTConnectEntryCGM(unittest.TestCase):
def test_parse_cgm_entry(self):
self.assertEqual(
TConnectEntry.parse_cgm_entry({
"DeviceType": "t:slim X2 Insulin Pump",
"SerialNumber": "11111111",
"Description": "EGV",
"EventDateTime": "2021-10-12T00:01:12",
"Readings (CGM / BGM)": "131"
}),
{
"time": "2021-10-12 00:01:12-04:00",
"reading": "131",
"reading_type": "EGV"
}
)
class TestTConnectEntryIOB(unittest.TestCase):
entry1 = {
"Type": "IOB",
"EventID": "81",
"EventDateTime": "2021-10-12T00:00:30",
"IOB": "6.91"
}
def test_parse_iob_entry1(self):
self.assertEqual(
TConnectEntry.parse_iob_entry(self.entry1),
{
"time": "2021-10-12 00:00:30-04:00",
"iob": "6.91",
"event_id": "81"
}
)
entry2 = {
"Type": "IOB",
"EventID": "9",
"EventDateTime": "2021-10-12T00:10:30",
"IOB": "6.80"
}
def test_parse_iob_entry2(self):
self.assertEqual(
TConnectEntry.parse_iob_entry(self.entry2),
{
"time": "2021-10-12 00:10:30-04:00",
"iob": "6.80",
"event_id": "9"
}
)
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),
Bolus(**{
"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",
"requested_insulin": "13.53",
"carbs": "75",
"bg": "141",
"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),
Bolus(**{
"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",
"requested_insulin": "1.25",
"carbs": "0",
"bg": "159",
"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),
Bolus(**{
"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",
"requested_insulin": "1.70",
"carbs": "0",
"bg": "",
"user_override": "0",
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
}))
entryStdIncompleteZero = {
"Type": "Bolus",
"Description": "Standard",
"BG": "144",
"IOB": "1.20",
"BolusRequestID": "9694.000",
"BolusCompletionID": "9694.000",
"CompletionDateTime": "2021-10-08T15:47:02",
"InsulinDelivered": "0.00",
"FoodDelivered": "0.00",
"CorrectionDelivered": "0.00",
"CompletionStatusID": "0",
"CompletionStatusDesc": "User Aborted",
"BolusIsComplete": "0",
"BolexCompletionID": "",
"BolexSize": "",
"BolexStartDateTime": "",
"BolexCompletionDateTime": "",
"BolexInsulinDelivered": "",
"BolexIOB": "",
"BolexCompletionStatusID": "",
"BolexCompletionStatusDesc": "",
"ExtendedBolusIsComplete": "",
"EventDateTime": "2021-10-08T15:46:56",
"RequestDateTime": "2021-10-08T15:46:56",
"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": "0.50",
"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": "1669328"
}
def test_parse_bolus_entry_std_incomplete_zero(self):
self.assertEqual(
TConnectEntry.parse_bolus_entry(self.entryStdIncompleteZero),
Bolus(**{
"description": "Standard",
"complete": "",
"completion": "User Aborted",
"request_time": "2021-10-08 15:46:56-04:00",
"completion_time": "2021-10-08 15:47:02-04:00",
"insulin": "0.00",
"requested_insulin": "0.50",
"carbs": "0",
"bg": "144",
"user_override": "1",
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
}))
entryStdIncompletePartial = {
"Type": "Bolus",
"Description": "Standard/Correction",
"BG": "189",
"IOB": "",
"BolusRequestID": "9261.000",
"BolusCompletionID": "9261.000",
"CompletionDateTime": "2021-09-06T12:24:47",
"InsulinDelivered": "1.82",
"FoodDelivered": "0.00",
"CorrectionDelivered": "1.82",
"CompletionStatusID": "1",
"CompletionStatusDesc": "Terminated by Alarm",
"BolusIsComplete": "0",
"BolexCompletionID": "",
"BolexSize": "",
"BolexStartDateTime": "",
"BolexCompletionDateTime": "",
"BolexInsulinDelivered": "",
"BolexIOB": "",
"BolexCompletionStatusID": "",
"BolexCompletionStatusDesc": "",
"ExtendedBolusIsComplete": "",
"EventDateTime": "2021-09-06T12:23:23",
"RequestDateTime": "2021-09-06T12:23:23",
"BolusType": "Carb",
"BolusRequestOptions": "Standard/Correction",
"StandardPercent": "100.00",
"Duration": "0",
"CarbSize": "0",
"UserOverride": "0",
"TargetBG": "110",
"CorrectionFactor": "30.00",
"FoodBolusSize": "0.00",
"CorrectionBolusSize": "2.63",
"ActualTotalBolusRequested": "2.63",
"IsQuickBolus": "0",
"EventHistoryReportEventDesc": "0",
"EventHistoryReportDetails": "Correction & Food Bolus",
"NoteID": "CF 1:30 - Carb Ratio 1:6 - Target BG 110",
"IndexID": "0",
"Note": "1589227"
}
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",
"request_time": "2021-09-06 12:23:23-04:00",
"completion_time": "2021-09-06 12:24:47-04:00",
"insulin": "1.82",
"requested_insulin": "2.63",
"carbs": "0",
"bg": "189",
"user_override": "0",
"extended_bolus": "",
"bolex_completion_time": None,
"bolex_start_time": None
}))
entryExtendedComplete = {
"Type": "Bolus",
"Description": "Extended 50.00%/0.00",
"BG": "131",
"IOB": "5.87",
"BolusRequestID": "3636.000",
"BolusCompletionID": "3636.000",
"CompletionDateTime": "2022-08-09T23:20:04",
"InsulinDelivered": "0.20",
"FoodDelivered": "0.00",
"CorrectionDelivered": "0.00",
"CompletionStatusID": "3",
"CompletionStatusDesc": "Completed",
"BolusIsComplete": "1",
"BolexCompletionID": "16757133",
"BolexSize": "0.20",
"BolexStartDateTime": "2022-08-09T23:20:04",
"BolexCompletionDateTime": "2022-08-09T23:35:03",
"BolexInsulinDelivered": "0.20",
"BolexIOB": "5.7",
"BolexCompletionStatusID": "3.00",
"BolexCompletionStatusDesc": "Completed",
"ExtendedBolusIsComplete": "1",
"EventDateTime": "2022-08-09T23:19:15",
"RequestDateTime": "2022-08-09T23:19:15",
"BolusType": "Carb",
"BolusRequestOptions": "Extended",
"StandardPercent": "50.00",
"Duration": "15",
"CarbSize": "0",
"UserOverride": "1",
"TargetBG": "110",
"CorrectionFactor": "30.00",
"FoodBolusSize": "0.00",
"CorrectionBolusSize": "0.00",
"ActualTotalBolusRequested": "0.40",
"IsQuickBolus": "0",
"EventHistoryReportEventDesc": "0",
"EventHistoryReportDetails": "Food Bolus: 50% Extended 15 mins",
"NoteID": "CF 1:30 - Carb Ratio 1:6 - Target BG 110 | Override: Pump calculated Bolus = 0.0 units",
"IndexID": "0",
"Note": "631597"
}
def test_parse_bolus_entry_extended_complete(self):
self.assertEqual(
TConnectEntry.parse_bolus_entry(self.entryExtendedComplete),
Bolus(**{
"description": "Extended 50.00%/0.00",
"complete": "1",
"completion": "Completed",
"request_time": None,
"completion_time": None,
"insulin": "0.20",
"requested_insulin": "0.40",
"carbs": "0",
"bg": "131",
"user_override": "1",
"extended_bolus": "1",
"bolex_completion_time": "2022-08-09 23:35:03-04:00",
"bolex_start_time": "2022-08-09 23:20:04-04:00"
}))
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"
}
)
class TestTConnectEntryCIQEvent(unittest.TestCase):
def test_parse_ciq_activity_event_sleep(self):
self.assertEqual(
TConnectEntry.parse_ciq_activity_event({
"continuation": None,
"duration": 30661,
"eventType": 1,
"timeZoneId": "America/Los_Angeles",
"x": 1638091836
}),
{
"time": "2021-11-28 01:30:36-05:00",
"duration_mins": (30661 / 60),
"event_type": "Sleep"
}
)
def test_parse_ciq_activity_event_exercise(self):
self.assertEqual(
TConnectEntry.parse_ciq_activity_event({
"duration": 1200,
"eventType": 2,
"continuation": None,
"timeZoneId": "America/Los_Angeles",
"x": 1619901912
}),
{
"time": "2021-05-01 13:45:12-04:00",
"duration_mins": 20,
"event_type": "Exercise"
}
)
def test_parse_ciq_activity_event_unknown_id(self):
self.assertRaises(
UnknownCIQActivityEventException,
TConnectEntry.parse_ciq_activity_event,
{
"duration": 1200,
"eventType": 5,
"continuation": None,
"timeZoneId": "America/Los_Angeles",
"x": 1619901912
}
)
class TestTConnectEntryBasalSuspensionEvent(unittest.TestCase):
def test_parse_basalsuspension_event_sitecart(self):
self.assertEqual(
TConnectEntry.parse_basalsuspension_event({
'EventDateTime': '/Date(1638663490000-0000)/',
'SuspendReason': 'site-cart'
}),
{
"time": "2021-12-04 16:18:10-05:00",
"event_type": "Site/Cartridge Change"
}
)
def test_parse_basalsuspension_event_alarm(self):
self.assertEqual(
TConnectEntry.parse_basalsuspension_event({
'EventDateTime': '/Date(1637863616000-0000)/',
'SuspendReason': 'alarm'
}),
{
"time": "2021-11-25 10:06:56-05:00",
"event_type": "Empty Cartridge/Pump Shutdown"
}
)
def test_parse_basalsuspension_event_manual(self):
self.assertEqual(
TConnectEntry.parse_basalsuspension_event({
'EventDateTime': '/Date(1638662852000-0000)/',
'SuspendReason': 'manual'
}),
{
"time": "2021-12-04 16:07:32-05:00",
"event_type": "User Suspended"
}
)
def test_parse_basalsuspension_event_tempprofile(self):
self.assertEqual(
TConnectEntry.parse_basalsuspension_event({
'EventDateTime': '/Date(1640541521000-0000)/',
'SuspendReason': 'temp-profile'
}),
{
"time": "2021-12-26 09:58:41-05:00",
"event_type": "Basal Rate Change"
}
)
def test_parse_basalsuspension_event_basalprofile_skipped(self):
self.assertIsNone(
TConnectEntry.parse_basalsuspension_event({
'EventDateTime': '/Date(1638659343000-0000)/',
'SuspendReason': 'basal-profile',
})
)
def test_parse_basalsuspension_event_previous_skipped(self):
self.assertIsNone(
TConnectEntry.parse_basalsuspension_event({
'Continuation': 'continuation',
'EventDateTime': '/Date(1638604800000-0000)/',
'SuspendReason': 'previous',
})
)
def test_parse_basalsuspension_event_unknown_suspendreason(self):
self.assertRaises(
UnknownBasalSuspensionEventException,
TConnectEntry.parse_basalsuspension_event,
{
'EventDateTime': '/Date(1638604800000-0000)/',
'SuspendReason': 'unknown',
}
)
if __name__ == '__main__':
unittest.main()