mirror of
https://github.com/bckelley/tconnectsync.git
synced 2026-08-24 11:44:12 -05:00
99 lines
4.0 KiB
Python
99 lines
4.0 KiB
Python
import sys
|
|
import arrow
|
|
|
|
try:
|
|
from secret import TIMEZONE_NAME
|
|
except Exception:
|
|
print('Unable to import parser secrets from secret.py')
|
|
sys.exit(1)
|
|
|
|
class TConnectEntry:
|
|
BASAL_EVENTS = { 0: "Suspension", 1: "Profile", 2: "TempRate", 3: "Algorithm" }
|
|
ACTIVITY_EVENTS = { 1: "Sleep", 2: "Exercise", 3: "AutoBolus", 4: "CarbOnly" }
|
|
|
|
@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 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 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.
|
|
complete = (data["ExtendedBolusIsComplete"] or data["BolusIsComplete"])
|
|
extended_bolus = ("extended" in data["Description"].lower())
|
|
|
|
return {
|
|
"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 complete and not extended_bolus else None,
|
|
"completion_time": TConnectEntry._datetime_parse(data["CompletionDateTime"]).format() if complete and not extended_bolus else None,
|
|
"insulin": data["InsulinDelivered"],
|
|
"carbs": data["CarbSize"],
|
|
"user_override": data["UserOverride"],
|
|
"extended_bolus": "1" if extended_bolus else "",
|
|
"bolex_completion_time": TConnectEntry._datetime_parse(data["BolexCompletionDateTime"]).format() if complete and extended_bolus else None,
|
|
"bolex_start_time": TConnectEntry._datetime_parse(data["BolexStartDateTime"]).format() if complete and extended_bolus else None,
|
|
} |