Delete dead non-tandem source code

This commit is contained in:
James Woglom
2025-06-07 23:21:18 -04:00
parent 0b4674bae9
commit ffd3c5b1ea
9 changed files with 4 additions and 1041 deletions
+3 -6
View File
@@ -12,8 +12,6 @@ if sys.version_info < (3, 8):
typing.Protocol = typing_extensions.Protocol
from .api import TConnectApi
from .process import process_time_range
from .autoupdate import Autoupdate
from .sync.tandemsource.autoupdate import TandemSourceAutoupdate
from .sync.tandemsource.choose_device import ChooseDevice as TandemSourceChooseDevice
from .sync.tandemsource.process import ProcessTimeRange as TandemSourceProcessTimeRange
@@ -54,7 +52,7 @@ def parse_args(*args, **kwargs):
parser.add_argument('--auto-update', dest='auto_update', action='store_const', const=True, default=False, help='If set, continuously checks for updates from t:connect and syncs with Nightscout.')
parser.add_argument('--check-login', dest='check_login', action='store_const', const=True, default=False, help='If set, checks that the provided t:connect credentials can be used to log in.')
parser.add_argument('--features', dest='features', nargs='+', default=DEFAULT_FEATURES, choices=ALL_FEATURES, help='Specifies what data should be synchronized between tconnect and Nightscout.')
parser.add_argument('--tandem-source', dest='tandem_source', action='store_const', const=True, default=False, help='FOR TESTING: Use Tandem Source')
parser.add_argument('--tandem-source', dest='tandem_source', action='store_const', const=True, default=True, help=argparse.SUPPRESS) # no longer used
parser.add_argument('--region', dest='region', type=str, choices=['US', 'EU'], default=None, help='Tandem t:connect server region (US or EU). If not specified, uses TCONNECT_REGION from configuration or defaults to US.')
return parser.parse_args(*args, **kwargs)
@@ -106,9 +104,8 @@ def main(*args, **kwargs):
nightscout = NightscoutApi(NS_URL, NS_SECRET, skip_verify=NS_SKIP_TLS_VERIFY, ignore_conn_errors=NS_IGNORE_CONN_ERRORS)
# NOT YET MIGRATED
# if args.check_login:
# return check_login(tconnect, time_start, time_end)
if args.check_login:
return check_login(tconnect, time_start, time_end)
logging.warning("THIS VERSION OF TCONNECTSYNC READS DATA FROM TANDEM SOURCE, AND MAY CONTAIN BUGS!")
logging.info("You may notice different behavior compared to older versions which utilized t:connect data sources.")
-207
View File
@@ -1,207 +0,0 @@
import time
import logging
import datetime
import sys
from .process import process_time_range
from .features import DEFAULT_FEATURES
from . import secret
logger = logging.getLogger(__name__)
class Autoupdate:
"""Wrap access to secrets for easier testing."""
def __init__(self, secret):
self.secret = secret
self.autoupdate_invocations = 0
self.last_event_index = None
self.last_event_time = None
self.last_successful_process_time_range = None
self.time_diffs_between_updates = []
self.last_attempt_time = None
self.time_diffs_between_attempts = []
"""
Performs the auto-update functionality. Runs indefinitely in a loop
until stopped (ctrl+c), or a maximum of AUTOUPDATE_MAX_LOOP_INVOCATIONS times.
Stops if AUTOUPDATE_RESTART_ON_FAILURE is set and an error occurs.
"""
def process(self, tconnect, nightscout, time_start, time_end, pretend, features=None):
if features is None:
features = DEFAULT_FEATURES
# Read from android api, find exact interval to cut down on API calls
# Refresh API token. If failure, die, have wrapper script re-run.
self.autoupdate_start = time.time()
while True:
logger.debug("autoupdate loop")
now = time.time()
last_event = tconnect.android.last_event_uploaded(self.secret.PUMP_SERIAL_NUMBER)
if not self.last_event_index or last_event['maxPumpEventIndex'] > self.last_event_index:
logger.info('New reported t:connect data. (event index: %s last: %s)' % (last_event['maxPumpEventIndex'], self.last_event_index))
if pretend:
logger.info('Would update now if not in pretend mode')
else:
added = process_time_range(tconnect, nightscout, time_start, time_end, pretend, features=features)
logger.info('Added %d items from process_time_range' % added)
if added == 0:
# If we've been unable to find new events, but the last_event_index is increasing,
# suggesting there are more events being added, we might be in a bugged
# situation where we can't get any more data without restarting.
# We skip this check on the first process cycle, since we might
# just already be in sync with tconnect's pump data.
if self.last_event_index:
# Find the timestamp of the last time we've successfully obtained data,
# or the time when the autoupdate run started, if we haven't at all.
last_action_or_start = self.last_successful_process_time_range
if not last_action_or_start:
last_action_or_start = self.autoupdate_start
# If it's been AUTOUPDATE_FAILURE_MINUTES in the state of not seeing
# event index changes reflected in the tconnect data we're pulling,
# raise an error and potentially restart.
# This is likely a tconnectsync problem, not a problem with the pump or app
# (we can see the indexes increasing, so we know something's happening!)
if (now - last_action_or_start) >= 60 * self.secret.AUTOUPDATE_FAILURE_MINUTES:
logger.error(AutoupdateFailureError(
("%s: An event index change was recorded, but no new data was found via the API. " % datetime.datetime.now()) +
"The %s was %d minutes ago. This is a problem with tconnectsync." %
("last processed event" if self.last_successful_process_time_range else "start of autoupdate", (now - last_action_or_start)//60)))
if self.secret.AUTOUPDATE_RESTART_ON_FAILURE:
logger.error("Exiting with error code due to AUTOUPDATE_RESTART_ON_FAILURE")
return 1
else:
logger.warning(AutoupdateFailureWarning(("%s: An event index change was recorded, but no new data was found via the API. " % datetime.datetime.now()) +
"The %s was %d minutes ago. Resetting TConnectApi to attempt to solve this problem." %
("last processed event" if self.last_successful_process_time_range else "start of autoupdate", (now - last_action_or_start)//60)))
# As a stop-gap, try to re-initialize TConnectApi (triggering a re-login)
# Use __class__ instead of direct TConnectApi invocation to avoid initializing a real TConnectApi over a fake
tconnect = tconnect.__class__(self.secret.TCONNECT_EMAIL, self.secret.TCONNECT_PASSWORD)
else:
# Mark the last successful time we got data from tconnect
self.last_successful_process_time_range = now
# Track the time it took to find a new event between runs,
# but skip this calculation the first process cycle (since
# we don't know at what exact point the event index changed)
if self.last_event_index:
self.time_diffs_between_updates.append(now - self.last_event_time)
logger.debug('Updating tracking of time since last update: %s' % self.time_diffs_between_updates)
# Mark the last event index uploaded from the pump and timestamp
self.last_event_index = last_event['maxPumpEventIndex']
self.last_event_time = now
self.last_attempt_time = now
self.time_diffs_between_attempts = []
else:
logger.info('No new reported t:connect data. (last event index: %s)' % last_event['maxPumpEventIndex'])
# If we haven't seen the pump event index update in AUTOUPDATE_NO_DATA_FAILURE_MINUTES,
# then trigger an error and potentially restart.
# The most likely case here is that the pump isn't uploading right now.
if self.last_event_time and (now - self.last_event_time) >= 60 * self.secret.AUTOUPDATE_NO_DATA_FAILURE_MINUTES:
logger.error(AutoupdateNoEventIndexesDetectedError(
"%s: No new data event indexes have been detected for %d minutes. " % (datetime.datetime.now(), (now - self.last_event_time)//60) +
"The t:connect app might no longer be functioning."))
# TODO: restarting doesn't really help anything here.
# Should we notify the user?
if self.secret.AUTOUPDATE_RESTART_ON_FAILURE:
logger.error("Exiting with error code due to AUTOUPDATE_RESTART_ON_FAILURE")
return 1
# Similarly, if we HAVE seen pump event indexes update but have not successfully
# found any associated data updates from the tconnect API for AUTOUPDATE_NO_DATA_FAILURE_MINUTES,
# trigger an error and potentially restart. This could either be a tconnectsync problem,
# where we can see the indexes increasing, but it takes us until a period of no index
# update to reach our AUTOUPDATE_FAILURE_MINUTES threshold; or, a side effect of the
# above no indexes warning.
elif self.last_successful_process_time_range and (now - self.last_successful_process_time_range) >= 60 * self.secret.AUTOUPDATE_FAILURE_MINUTES:
logger.error(AutoupdateNoNewDataDetectedError(
"%s: No new data has been detected via the API for %d minutes. " % (datetime.datetime.now(), (now - self.last_successful_process_time_range)//60) +
"tconnectsync might not be functioning properly."))
if self.secret.AUTOUPDATE_RESTART_ON_FAILURE:
logger.error("%s: Exiting with error code due to AUTOUPDATE_RESTART_ON_FAILURE" % datetime.datetime.now())
return 1
# Track how long we've been retrying
if self.last_attempt_time:
self.time_diffs_between_attempts.append(now - self.last_attempt_time)
self.last_attempt_time = now
# If it's been 3 loops since the last time we found new data,
# then we're not in sync with the rate at which pump data is being
# uploaded, so
if len(self.time_diffs_between_attempts) >= 3:
# The pump hasn't sent us data that, based on previous cadence, we were expecting
logger.warning(AutoupdateNoIndexChangeWarning("Sleeping %d seconds after unexpected no index change based on previous cadence. (New data might be delayed.)" %
int(self.secret.AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS)))
logger.debug("Last event time: %s, time diffs between attempts: %s" % (self.last_event_time, self.time_diffs_between_attempts))
time.sleep(self.secret.AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS)
# Since we bail early, update the invocations count and potentially exit after sleeping.
self.autoupdate_invocations += 1
if self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS > 0 and self.autoupdate_invocations >= self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS:
return 0
continue
sleep_secs = self.secret.AUTOUPDATE_DEFAULT_SLEEP_SECONDS
# Sleep for a rolling average of time between updates
if self.secret.AUTOUPDATE_USE_FIXED_SLEEP != 1:
logger.debug("Time diffs between updates: %s" % self.time_diffs_between_updates)
# Only keep the 10 latest time diffs
if len(self.time_diffs_between_updates) > 10:
self.time_diffs_between_updates = self.time_diffs_between_updates[1:]
# If we have less than 3 data points,
if len(self.time_diffs_between_updates) > 2:
sleep_secs = sum(self.time_diffs_between_updates) / len(self.time_diffs_between_updates)
# At minimum, update every AUTOUPDATE_MAX_SLEEP_SECONDS regardless
# of how often we're seeing new data appear
if sleep_secs > self.secret.AUTOUPDATE_MAX_SLEEP_SECONDS:
sleep_secs = self.secret.AUTOUPDATE_MAX_SLEEP_SECONDS
logger.info('Sleeping for %0.01f sec' % sleep_secs)
time.sleep(sleep_secs)
self.autoupdate_invocations += 1
if self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS > 0 and self.autoupdate_invocations >= self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS:
return 0
class AutoupdateError(RuntimeError):
def __str__(self):
return "%s: %s" % (self.__class__.__name__, super().__str__())
class AutoupdateWarning(RuntimeWarning):
def __str__(self):
return "%s: %s" % (self.__class__.__name__, super().__str__())
class AutoupdateFailureError(AutoupdateError):
pass
class AutoupdateFailureWarning(AutoupdateWarning):
pass
class AutoupdateNoEventIndexesDetectedError(AutoupdateError):
pass
class AutoupdateNoNewDataDetectedError(AutoupdateError):
pass
class AutoupdateNoIndexChangeWarning(AutoupdateWarning):
pass
-161
View File
@@ -1,161 +0,0 @@
import arrow
import logging
from ..parser.nightscout import (
BASAL_EVENTTYPE,
NightscoutEntry
)
from ..parser.tconnect import TConnectEntry
from ..secret import SKIP_NS_LAST_UPLOADED_CHECK
logger = logging.getLogger(__name__)
"""
Merges together input from the therapy timeline API
into a digestable format of basal data.
"""
def process_ciq_basal_events(data):
if data is None:
return []
suspensionEvents = {}
for s in data["suspensionDeliveryEvents"]:
entry = TConnectEntry.parse_suspension_entry(s)
suspensionEvents[entry["time"]] = entry
basalEvents = []
for b in data["basal"]["tempDeliveryEvents"]:
basalEvents.append(TConnectEntry.parse_ciq_basal_entry(b, delivery_type="tempDelivery"))
for b in data["basal"]["algorithmDeliveryEvents"]:
basalEvents.append(TConnectEntry.parse_ciq_basal_entry(b, delivery_type="algorithmDelivery"))
for b in data["basal"]["profileDeliveryEvents"]:
basalEvents.append(TConnectEntry.parse_ciq_basal_entry(b, delivery_type="profileDelivery"))
# Suspensions with suspendReason 'control-iq' will match a basal event found above.
for i in basalEvents:
if i["time"] in suspensionEvents:
i["delivery_type"] += " (" + suspensionEvents[i["time"]]["suspendReason"] + " suspension)"
del suspensionEvents[i["time"]]
# Suspensions with suspendReason 'manual' do not have an associated basal event,
# and require extra processing.
basalEvents.sort(key=lambda x: arrow.get(x["time"]))
unprocessedSuspensions = list(suspensionEvents.values())
unprocessedSuspensions.sort(key=lambda x: arrow.get(x["time"]))
# For the remaining suspensions which did not match with an existing basal event,
# add a new event manually. This means we need to calculate the duration of the
# suspension.
newEvents = []
for i in range(len(basalEvents)):
if len(unprocessedSuspensions) == 0:
break
existingTime = arrow.get(basalEvents[i]["time"])
unprocessedTime = arrow.get(unprocessedSuspensions[0]["time"])
# If we've found an event which occurs after the suspension, then the
# difference in their timestamps is the duration of the suspension.
if i > 0 and existingTime > unprocessedTime:
suspension = unprocessedSuspensions.pop(0)
# TConnect's internal duration object tracks the duration in seconds
seconds = (existingTime - unprocessedTime).seconds
newEvent = TConnectEntry.manual_suspension_to_basal_entry(suspension, seconds)
logger.debug("Adding basal event for unprocessed suspension: %s" % newEvent)
newEvents.append(newEvent)
# Any remaining suspensions which have not been processed have not ended,
# which means we do not know their duration; so we will skip them (for now)
# Add any new events and re-sort
if newEvents:
basalEvents += newEvents
basalEvents.sort(key=lambda x: arrow.get(x["time"]))
return basalEvents
"""
Processes basal data input from the therapy timeline CSV (which only
exists for pre Control-IQ data) into a digestable format.
"""
def add_csv_basal_events(basalEvents, data):
last_entry = {}
for row in data:
entry = TConnectEntry.parse_csv_basal_entry(row)
if last_entry:
diff_mins = (arrow.get(entry["time"]) - arrow.get(last_entry["time"])).seconds // 60
entry["duration_mins"] = diff_mins
basalEvents.append(entry)
last_entry = entry
basalEvents.sort(key=lambda x: arrow.get(x["time"]))
return basalEvents
"""
Given processed basal data, adds basal events to Nightscout.
"""
def ns_write_basal_events(nightscout, basalEvents, pretend=False, time_start=None, time_end=None):
logger.debug("ns_write_basal_events: querying for last uploaded entry")
last_upload = nightscout.last_uploaded_entry(BASAL_EVENTTYPE, time_start=time_start, time_end=time_end)
last_upload_time = None
if last_upload:
last_upload_time = arrow.get(last_upload["created_at"])
logger.info("Last Nightscout basal upload: %s" % last_upload_time)
if SKIP_NS_LAST_UPLOADED_CHECK:
logger.warning("Overriding last upload check")
last_upload = None
last_upload_time = None
add_count = 0
for event in basalEvents:
if last_upload_time and arrow.get(event["time"]) < last_upload_time:
if pretend:
logger.info("Skipping basal event before last upload time: %s (time range: %s - %s)" % (event, time_start, time_end))
continue
recent_needs_update = False
if last_upload_time and arrow.get(event["time"]) == last_upload_time:
# If this entry has the same time as the most recent upload, but
# has newer info, then delete and recreate it.
recent_needs_update = (round(last_upload["duration"]) < round(event["duration_mins"]))
# If the timestamps are identical, and the duration is identical,
# then don't upload a duplicate entry of what we already have.
if not recent_needs_update:
continue
reason = event["delivery_type"]
if "suspendReason" in reason:
reason += " (" + reason["suspendReason"] + ")"
entry = NightscoutEntry.basal(
value=event["basal_rate"],
duration_mins=event["duration_mins"],
created_at=event["time"],
reason=reason
)
add_count += 1
logger.info(" Processing basal: %s entry: %s" % (event, entry))
if recent_needs_update:
logger.info("Replacing last uploaded entry: %s" % last_upload)
if not pretend:
entry['_id'] = last_upload['_id']
nightscout.put_entry(entry, entity='treatments')
elif not pretend:
nightscout.upload_entry(entry)
logger.debug("ns_write_basal_events: added %d events" % add_count)
return add_count
-120
View File
@@ -1,120 +0,0 @@
import arrow
import logging
from tconnectsync.domain.bolus import Bolus
from tconnectsync.sync.cgm import find_event_at
from ..parser.nightscout import (
BOLUS_EVENTTYPE,
NightscoutEntry
)
from ..parser.tconnect import TConnectEntry
from ..secret import SKIP_NS_LAST_UPLOADED_CHECK
logger = logging.getLogger(__name__)
"""
Given bolus data input from the therapy timeline CSV, converts it into a digestable format.
"""
def process_bolus_events(bolusdata, cgmEvents=None, source=""):
bolusEvents = []
for b in bolusdata:
parsed = None
if source == "ciq":
parsed = b.to_bolus()
else:
parsed = TConnectEntry.parse_bolus_entry(b)
assert type(parsed) == Bolus
if parsed.completion != "Completed":
if parsed.insulin and float(parsed.insulin) > 0:
# Count non-completed bolus if any insulin was delivered (vs. the amount of insulin requested)
parsed.description += " (%s: requested %s units)" % (parsed.completion, parsed.requested_insulin)
else:
logger.warning("Skipping non-completed %s bolus data (was a bolus in progress?): %s parsed: %s" % (source, b, parsed))
continue
if parsed.is_extended_bolus:
if not parsed.bolex_start_time and not parsed.request_time:
logger.warning("Skipping non-completed %s extended bolus data with no request_time: %s parsed: %s" % (source, b, parsed))
elif not parsed.bolex_start_time and parsed.request_time:
logger.warning("Setting bolex_start_time to request_time for non-completed %s extended bolus: %s parsed: %s" % (source, b, parsed))
parsed.bolex_start_time = parsed.request_time
logger.debug("process_bolus_events for incomplete bolus: %s parsed: %s" % (b, parsed))
elif parsed.is_extended_bolus:
logger.debug("process_bolus_events for complete extended bolus: %s parsed: %s" % (b, parsed))
if parsed.bg and cgmEvents:
requested_at = parsed.request_time if not parsed.extended_bolus else parsed.bolex_start_time
parsed.bg_type = guess_bolus_bg_type(parsed.bg, requested_at, cgmEvents)
bolusEvents.append(parsed)
bolusEvents.sort(key=lambda event: arrow.get(event.request_time if not event.is_extended_bolus else event.bolex_start_time))
return bolusEvents
"""
Determine whether the given BG specified in the bolus is identical to the
most recent CGM reading at that time. If it is, return SENSOR.
Otherwise, return FINGER.
"""
def guess_bolus_bg_type(bg, created_at, cgmEvents):
if not cgmEvents:
return NightscoutEntry.FINGER
event = find_event_at(cgmEvents, created_at)
if event and str(event["bg"]) == str(bg):
return NightscoutEntry.SENSOR
return NightscoutEntry.FINGER
"""
Given processed bolus data, adds bolus events to Nightscout.
"""
def ns_write_bolus_events(nightscout, bolusEvents, pretend=False, include_bg=False, reading_events=None, time_start=None, time_end=None):
logger.debug("ns_write_bolus_events: querying for last uploaded entry")
last_upload = nightscout.last_uploaded_entry(BOLUS_EVENTTYPE, time_start=time_start, time_end=time_end)
last_upload_time = None
if last_upload:
last_upload_time = arrow.get(last_upload["created_at"])
logger.info("Last Nightscout bolus upload: %s" % last_upload_time)
if SKIP_NS_LAST_UPLOADED_CHECK:
logger.warning("Overriding last upload check")
last_upload = None
last_upload_time = None
add_count = 0
for event in bolusEvents:
created_at = event.completion_time if not event.is_extended_bolus else event.bolex_start_time
if last_upload_time and arrow.get(created_at) <= last_upload_time:
if pretend:
logger.info("Skipping basal event before last upload time: %s (time range: %s - %s)" % (event, time_start, time_end))
continue
if include_bg and event.bg:
entry = NightscoutEntry.bolus(
bolus=event.insulin,
carbs=event.carbs,
created_at=created_at,
notes="{}{}{}".format(event.description, " (Override)" if event.user_override == "1" else "", " (Extended)" if event.extended_bolus == "1" else ""),
bg=event.bg,
bg_type=event.bg_type
)
else:
entry = NightscoutEntry.bolus(
bolus=event.insulin,
carbs=event.carbs,
created_at=created_at,
notes="{}{}{}".format(event.description, " (Override)" if event.user_override == "1" else "", " (Extended)" if event.extended_bolus == "1" else "")
)
add_count += 1
logger.info(" Processing bolus: %s entry: %s" % (event, entry))
if not pretend:
nightscout.upload_entry(entry)
return add_count
-69
View File
@@ -1,69 +0,0 @@
import json
import arrow
import logging
from ..parser.tconnect import TConnectEntry
from ..parser.nightscout import NightscoutEntry
logger = logging.getLogger(__name__)
def process_cgm_events(readingData):
data = []
for r in readingData:
data.append(TConnectEntry.parse_reading_entry(r))
return data
"""
Given reading data and a time, finds the BG reading event which would have
been the current one at that time. e.g., it looks before the given time,
not after.
This is a heuristic for checking whether the BG component of a bolus was
manually entered or inferred based on the pump's CGM.
"""
def find_event_at(cgmEvents, find_time):
find_t = arrow.get(find_time)
events = list(map(lambda x: (arrow.get(x["time"]), x), cgmEvents))
events.sort()
closestReading = None
for t, r in events:
if t > find_t:
break
closestReading = r
return closestReading
"""
Given processed CGM data, adds reading entries to Nightscout.
"""
def ns_write_cgm_events(nightscout, cgmEvents, pretend=False, time_start=None, time_end=None):
logger.debug("ns_write_cgm_events: querying for last uploaded entry")
last_upload = nightscout.last_uploaded_bg_entry(time_start=time_start, time_end=time_end)
last_upload_time = None
if last_upload:
last_upload_time = arrow.get(last_upload["dateString"])
logger.info("Last Nightscout CGM upload: %s" % last_upload_time)
add_count = 0
for event in cgmEvents:
created_at = event["time"]
if last_upload_time and arrow.get(created_at) <= last_upload_time:
if pretend:
logger.info("Skipping CGM event before last upload time: %s (time range: %s - %s)" % (event, time_start, time_end))
continue
entry = NightscoutEntry.entry(
sgv=event["bg"],
created_at=created_at
)
add_count += 1
logger.info(" Processing cgm reading: %s entry: %s" % (event, entry))
if not pretend:
nightscout.upload_entry(entry, entity='entries')
return add_count
-59
View File
@@ -1,59 +0,0 @@
import arrow
import logging
from ..parser.nightscout import (
IOB_ACTIVITYTYPE,
NightscoutEntry
)
from ..parser.tconnect import TConnectEntry
logger = logging.getLogger(__name__)
"""
Given IOB data input from the therapy timeline CSV, converts it into a digestable format.
"""
def process_iob_events(iobdata):
iobEvents = []
for d in iobdata:
iobEvents.append(TConnectEntry.parse_iob_entry(d))
iobEvents.sort(key=lambda x: arrow.get(x["time"]))
return iobEvents
"""
Given processed IOB data, creates a single Nightscout activity definition to store IOB.
"""
def ns_write_iob_events(nightscout, iobEvents, pretend=False, time_start=None, time_end=None):
logger.debug("ns_write_iob_events: querying for last uploaded entry")
last_upload = nightscout.last_uploaded_activity(IOB_ACTIVITYTYPE, time_start=time_start, time_end=time_end)
last_upload_time = None
if last_upload:
last_upload_time = arrow.get(last_upload["created_at"])
logger.info("Last Nightscout iob upload: %s" % last_upload_time)
if not iobEvents or len(iobEvents) == 0:
logger.info("No IOB events present from API: skipping")
return 0
event = iobEvents[-1]
if last_upload_time and arrow.get(event["time"]) <= last_upload_time:
logger.info(" Skipping already uploaded iob event: %s" % event)
return 0
entry = NightscoutEntry.iob(
iob=event["iob"],
created_at=event["time"]
)
logger.info(" Processing iob: %s entry: %s" % (event, entry))
if not pretend:
nightscout.upload_entry(entry, entity='activity')
# Delete the previous activity
if last_upload and '_id' in last_upload:
logger.info(" Deleting old iob entry: %s" % last_upload)
if not pretend:
nightscout.delete_entry('activity/{}'.format(last_upload['_id']))
return 1
-200
View File
@@ -1,200 +0,0 @@
from typing import List, Tuple
import logging
import json
import copy
import arrow
from ..api import TConnectApi
from ..domain.device_settings import Profile, DeviceSettings
from ..parser.nightscout import NightscoutEntry
from ..nightscout import NightscoutApi
from ..secret import PUMP_SERIAL_NUMBER, NIGHTSCOUT_PROFILE_UPLOAD_MODE
logger = logging.getLogger(__name__)
def _get_default_serial_number():
return PUMP_SERIAL_NUMBER
def _get_default_upload_mode():
return NIGHTSCOUT_PROFILE_UPLOAD_MODE
def get_pump_profiles(tconnect: TConnectApi, serial_number: int = None) -> Tuple[List[Profile], DeviceSettings]:
all_devices = tconnect.webui.my_devices()
if serial_number is None:
serial_number = _get_default_serial_number()
if str(serial_number) not in all_devices:
logger.warn("Could not find entry for provided pump serial number in t:connect device list: %s, received: %s", serial_number, all_devices)
return [], None
device = all_devices[str(serial_number)]
logger.info("Getting profile settings for %s", device)
device_profiles, device_settings = tconnect.webui.device_settings_from_guid(device.guid)
logger.debug("device_profiles: %s", device_profiles)
logger.debug("device_settings: %s", device_settings)
logger.info("Found pump profiles: %s", ["%s%s" % (profile.title, " (active)" if profile.active else "") for profile in device_profiles])
return device_profiles, device_settings
"""
Compare pump device and Nightscout profiles, and return a final dictionary of
Nightscout profile objects, with the pump profile settings overriding what is
currently in Nightscout.
ns_profile_obj is the output from NightscoutApi.current_profile() and should be the most
recent profile object in mongo.
Returns the new Nightscout profile and whether it was changed.
"""
def compare_profiles(device_profiles: List[Profile], device_settings: DeviceSettings, ns_profile_obj: dict) -> Tuple[bool, dict]:
device = {profile.title: profile for profile in device_profiles}
ns = ns_profile_obj.get('store', {})
logger.info("compare_profiles profile names: device: %s ns: %s", device.keys(), ns.keys())
new_ns_profile = copy.deepcopy(ns_profile_obj)
updated_ns_profile = False
missing_profiles_in_ns = set(device.keys()) - set(ns.keys())
for profile_name in missing_profiles_in_ns:
logger.info("Missing %s profile in Nightscout: %s", profile_name, device.get(profile_name))
pump_configured_profile = device[profile_name]
ns_translated_profile = NightscoutEntry.profile_store(pump_configured_profile, device_settings)
logger.info("Will add %s profile to Nightscout: %s", profile_name, ns_translated_profile)
new_ns_profile['store'][profile_name] = ns_translated_profile
updated_ns_profile = True
existent_profiles_in_ns = set(device.keys()) & set(ns.keys())
for profile_name in existent_profiles_in_ns:
logger.debug("Checking for differences for %s profile between pump and nightscout", profile_name)
pump_configured_profile = device[profile_name]
ns_translated_profile = NightscoutEntry.profile_store(pump_configured_profile, device_settings)
ns_configured_profile = ns[profile_name]
logger.debug("Comparing %s profile from pump: %s to nightscout: %s", profile_name, ns_translated_profile, ns_configured_profile)
if nightscout_profiles_identical(ns_configured_profile, ns_translated_profile):
logger.info("Profile %s identical between pump and nightscout", profile_name)
continue
logger.info("Profile %s needs update in nightscout: %s", profile_name, ns_translated_profile)
new_ns_profile['store'][profile_name] = ns_translated_profile
updated_ns_profile = True
current_pump_profile = None
for profile in device_profiles:
if profile.active:
current_pump_profile = profile.title
if not current_pump_profile:
logger.error('No current pump profile, so skipping profile update: device: %s', device_profiles)
return False, ns_profile_obj
current_ns_profile = ns_profile_obj.get('defaultProfile')
if current_pump_profile != current_ns_profile:
logger.info("Current profile changed: pump: %s nightscout: %s", current_pump_profile, current_ns_profile)
new_ns_profile['defaultProfile'] = current_pump_profile
updated_ns_profile = True
if not updated_ns_profile:
logger.info("No Nightscout profile changes")
return False, ns_profile_obj
logger.info("New Nightscout profile object: %s", new_ns_profile)
return True, new_ns_profile
def nightscout_profiles_identical(configured: dict, translated: dict) -> bool:
if configured == translated:
logger.debug("direct dicts equal")
return True
if json.dumps(configured, sort_keys=True, indent=None) == json.dumps(translated, sort_keys=True, indent=None):
logger.debug("initial JSON dump identical")
return True
# convert all JSON values into strings
def map_nested_dicts_modify(ob, func):
for k, v in ob.items():
if isinstance(v, dict):
map_nested_dicts_modify(v, func)
elif isinstance(v, list):
map_nested_lists_modify(v, func)
else:
ob[k] = func(v)
def map_nested_lists_modify(ob, func):
for i in range(len(ob)):
v = ob[i]
if isinstance(v, dict):
map_nested_dicts_modify(v, func)
elif isinstance(v, list):
map_nested_lists_modify(v, func)
else:
ob[i] = func(v)
def to_numeric(x):
if type(x) in [int, float]:
return '%f' % x
try:
return '%f' % float(x)
except (ValueError, TypeError):
return x
convert_func = lambda x: to_numeric(x)
configured_str = json.loads(json.dumps(configured))
map_nested_dicts_modify(configured_str, convert_func)
translated_str = json.loads(json.dumps(translated))
map_nested_dicts_modify(translated_str, convert_func)
if json.dumps(configured_str, sort_keys=True, indent=None) == json.dumps(translated_str, sort_keys=True, indent=None):
logger.debug("map_nested_dicts JSON dump identical")
return True
logger.debug("profiles not identical")
return False
def setup_new_profile(ns_profile: dict) -> dict:
if '_id' in ns_profile:
del ns_profile['_id']
now = arrow.now().isoformat()
ns_profile['startDate'] = now
ns_profile['created_at'] = now
return ns_profile
def process_profiles(tconnect: TConnectApi, nightscout: NightscoutApi, pretend: bool = False, upload_mode: str = None) -> bool:
if not upload_mode:
upload_mode = _get_default_upload_mode()
logger.debug("Checking for differences between pump and nightscout profiles: %s mode", upload_mode)
ns_profile_obj = nightscout.current_profile()
pump_profiles, pump_settings = get_pump_profiles(tconnect)
diff, ns_profile_new = compare_profiles(pump_profiles, pump_settings, ns_profile_obj)
if not diff:
logger.info("Pump and Nightscout profiles up to date")
return False
if upload_mode == 'add':
profile_to_upload = setup_new_profile(ns_profile_new)
logger.info("Adding new Nightscout profiles object: %s", profile_to_upload)
if not pretend:
nightscout.upload_entry(profile_to_upload, entity='profile')
return True
elif upload_mode == 'replace':
logger.info("Replacing new Nightscout profiles object: %s", ns_profile_new)
if not pretend:
nightscout.put_entry(ns_profile_new, entity='profile')
return True
else:
raise RuntimeError('invalid upload_mode: %s' % upload_mode)
-218
View File
@@ -1,218 +0,0 @@
import arrow
import logging
from ..parser.nightscout import (
SITECHANGE_EVENTTYPE,
BASALSUSPENSION_EVENTTYPE,
EXERCISE_EVENTTYPE,
SLEEP_EVENTTYPE,
ACTIVITY_EVENTTYPE,
NightscoutEntry
)
from ..parser.tconnect import TConnectEntry
from ..secret import SKIP_NS_LAST_UPLOADED_CHECK
logger = logging.getLogger(__name__)
"""
Given a list of "activity events" from the CIQ therapy timeline endpoint,
process it into our internal events format.
These events contain a duration.
"""
def process_ciq_activity_events(data):
events = []
for event in data["events"]:
events.append(TConnectEntry.parse_ciq_activity_event(event))
return events
"""
Given a list of "basal suspension events" from the basalsuspension WS2 endpoint,
process it into our internal events format.
These events do NOT contain a duration.
"""
def process_basalsuspension_events(data):
events = []
for event in data['BasalSuspension']:
parsed = TConnectEntry.parse_basalsuspension_event(event)
if parsed:
events.append(parsed)
return events
"""
Given processed pump event data (of various types), write them to Nightscout
"""
def ns_write_pump_events(nightscout, pumpEvents, pretend=False, time_start=None, time_end=None):
count = 0
siteChangeEvents = []
emptyCartEvents = []
userSuspendedEvents = []
exerciseEvents = []
sleepEvents = []
activityEvents = []
for event in pumpEvents:
if event["event_type"] == TConnectEntry.BASALSUSPENSION_EVENTS["site-cart"]:
siteChangeEvents.append(event)
elif event["event_type"] in TConnectEntry.BASALSUSPENSION_EVENTS["alarm"]:
emptyCartEvents.append(event)
elif event["event_type"] in TConnectEntry.BASALSUSPENSION_EVENTS["manual"]:
userSuspendedEvents.append(event)
elif event["event_type"] == "Exercise":
exerciseEvents.append(event)
elif event["event_type"] == "Sleep":
sleepEvents.append(event)
elif event["event_type"] in TConnectEntry.ACTIVITY_EVENTS.values():
activityEvents.append(event)
logger.debug("siteChangeEvents: %s" % siteChangeEvents)
logger.debug("emptyCartEvents: %s" % emptyCartEvents)
logger.debug("userSuspendedEvents: %s" % userSuspendedEvents)
logger.debug("exerciseEvents: %s" % exerciseEvents)
logger.debug("sleepEvents: %s" % sleepEvents)
logger.debug("activityEvents: %s" % activityEvents)
count += ns_write_pump_sitechange_events(nightscout, siteChangeEvents, pretend=pretend, time_start=time_start, time_end=time_end)
count += ns_write_empty_cart_events(nightscout, emptyCartEvents, pretend=pretend, time_start=time_start, time_end=time_end)
count += ns_write_user_suspended_events(nightscout, userSuspendedEvents, pretend=pretend, time_start=time_start, time_end=time_end)
count += ns_write_exercise_activity_events(nightscout, exerciseEvents, pretend=pretend, time_start=time_start, time_end=time_end)
count += ns_write_sleep_activity_events(nightscout, sleepEvents, pretend=pretend, time_start=time_start, time_end=time_end)
count += ns_write_activity_events(nightscout, activityEvents, pretend=pretend, time_start=time_start, time_end=time_end)
return count
def ns_write_pump_sitechange_events(nightscout, siteChangeEvents, pretend=False, time_start=None, time_end=None):
return _ns_write_pump_events(
nightscout,
siteChangeEvents,
lambda event: NightscoutEntry.sitechange(
created_at=event["time"],
reason=event["event_type"]
),
SITECHANGE_EVENTTYPE,
pretend=pretend,
time_start=time_start,
time_end=time_end)
def ns_write_empty_cart_events(nightscout, emptyCartEvents, pretend=False, time_start=None, time_end=None):
return _ns_write_pump_events(
nightscout,
emptyCartEvents,
lambda event: NightscoutEntry.basalsuspension(
created_at=event["time"],
reason=event["event_type"]
),
BASALSUSPENSION_EVENTTYPE,
pretend=pretend,
time_start=time_start,
time_end=time_end)
def ns_write_user_suspended_events(nightscout, userSuspendedEvents, pretend=False, time_start=None, time_end=None):
return _ns_write_pump_events(
nightscout,
userSuspendedEvents,
lambda event: NightscoutEntry.basalsuspension(
created_at=event["time"],
reason=event["event_type"]
),
BASALSUSPENSION_EVENTTYPE,
pretend=pretend,
time_start=time_start,
time_end=time_end)
def ns_write_exercise_activity_events(nightscout, exerciseEvents, pretend=False, time_start=None, time_end=None):
return _ns_write_pump_events(
nightscout,
exerciseEvents,
lambda event: NightscoutEntry.activity(
created_at=event["time"],
reason=event["event_type"],
duration=event["duration_mins"],
event_type=EXERCISE_EVENTTYPE
),
EXERCISE_EVENTTYPE,
pretend=pretend,
time_start=time_start,
time_end=time_end)
def ns_write_sleep_activity_events(nightscout, sleepEvents, pretend=False, time_start=None, time_end=None):
return _ns_write_pump_events(
nightscout,
sleepEvents,
lambda event: NightscoutEntry.activity(
created_at=event["time"],
reason=event["event_type"],
duration=event["duration_mins"],
event_type=SLEEP_EVENTTYPE
),
SLEEP_EVENTTYPE,
pretend=pretend,
time_start=time_start,
time_end=time_end)
def ns_write_activity_events(nightscout, activityEvents, pretend=False, time_start=None, time_end=None):
return _ns_write_pump_events(
nightscout,
activityEvents,
lambda event: NightscoutEntry.activity(
created_at=event["time"],
reason=event["event_type"],
duration=event["duration_mins"]
),
ACTIVITY_EVENTTYPE,
pretend=pretend,
time_start=time_start,
time_end=time_end)
def _ns_write_pump_events(nightscout, events, buildNsEventFunc, eventType, pretend=False, time_start=None, time_end=None):
if len(events) == 0:
logger.debug("No %s events to process" % eventType)
return 0
logger.debug("ns_write_pump_events: querying for last %s" % eventType)
last_upload = nightscout.last_uploaded_entry(eventType, time_start=time_start, time_end=time_end)
last_upload_time = None
if last_upload:
last_upload_time = arrow.get(last_upload["created_at"])
logger.info("Last Nightscout %s: %s" % (eventType, last_upload_time))
if SKIP_NS_LAST_UPLOADED_CHECK:
logger.warning("Overriding last upload check")
last_upload = None
last_upload_time = None
add_count = 0
for event in events:
created_at = arrow.get(event["time"])
if last_upload_time and created_at <= last_upload_time:
skip = True
if "duration_mins" in event.keys() and "duration" in last_upload.keys():
if created_at == arrow.get(last_upload["created_at"]) and float(event["duration_mins"]) > float(last_upload["duration"]):
logger.info("Latest %s event needs updating: duration has increased from %s to %s: %s" % (eventType, last_upload["duration"], event["duration_mins"], event))
logger.info("Deleting previous %s: %s" % (eventType, last_upload))
nightscout.delete_entry('treatments/%s' % last_upload["_id"])
skip = False
if skip:
if pretend:
logger.info("Skipping %s pump event before last upload time: %s (time range: %s - %s)" % (eventType, event, time_start, time_end))
continue
entry = buildNsEventFunc(event)
add_count += 1
logger.info(" Processing %s: %s entry: %s" % (eventType, event, entry))
if not pretend:
nightscout.upload_entry(entry)
return add_count
+1 -1
View File
@@ -46,7 +46,7 @@ class WebUIScraper(tconnectsync.api.webui.WebUIScraper):
def my_devices(self):
raise NotImplementedError
def device_settings(self, pump_guid):
raise NotImplementedError