diff --git a/tconnectsync/__init__.py b/tconnectsync/__init__.py index e8e55db..2f2785b 100644 --- a/tconnectsync/__init__.py +++ b/tconnectsync/__init__.py @@ -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.") diff --git a/tconnectsync/autoupdate.py b/tconnectsync/autoupdate.py deleted file mode 100644 index a490927..0000000 --- a/tconnectsync/autoupdate.py +++ /dev/null @@ -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 diff --git a/tconnectsync/sync/basal.py b/tconnectsync/sync/basal.py deleted file mode 100644 index bcad62d..0000000 --- a/tconnectsync/sync/basal.py +++ /dev/null @@ -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 diff --git a/tconnectsync/sync/bolus.py b/tconnectsync/sync/bolus.py deleted file mode 100644 index 56c6b5b..0000000 --- a/tconnectsync/sync/bolus.py +++ /dev/null @@ -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 diff --git a/tconnectsync/sync/cgm.py b/tconnectsync/sync/cgm.py deleted file mode 100644 index dcf9388..0000000 --- a/tconnectsync/sync/cgm.py +++ /dev/null @@ -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 diff --git a/tconnectsync/sync/iob.py b/tconnectsync/sync/iob.py deleted file mode 100644 index d92a8f4..0000000 --- a/tconnectsync/sync/iob.py +++ /dev/null @@ -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 \ No newline at end of file diff --git a/tconnectsync/sync/profile.py b/tconnectsync/sync/profile.py deleted file mode 100644 index ad1de05..0000000 --- a/tconnectsync/sync/profile.py +++ /dev/null @@ -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) \ No newline at end of file diff --git a/tconnectsync/sync/pump_events.py b/tconnectsync/sync/pump_events.py deleted file mode 100644 index dd972f6..0000000 --- a/tconnectsync/sync/pump_events.py +++ /dev/null @@ -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 diff --git a/tests/api/fake.py b/tests/api/fake.py index d15496e..559620c 100644 --- a/tests/api/fake.py +++ b/tests/api/fake.py @@ -46,7 +46,7 @@ class WebUIScraper(tconnectsync.api.webui.WebUIScraper): def my_devices(self): raise NotImplementedError - + def device_settings(self, pump_guid): raise NotImplementedError