From dd9a9ea72cbb10c5c07a33452a0361d44b8c5816 Mon Sep 17 00:00:00 2001 From: James Woglom Date: Mon, 16 Jan 2023 16:00:50 -0500 Subject: [PATCH] tests for profile synchronization --- tconnectsync/domain/device_settings.py | 12 +- tconnectsync/sync/profile.py | 102 +++++- tests/parser/test_nightscout.py | 136 +------ tests/sync/test_profile.py | 471 +++++++++++++++++++++++++ 4 files changed, 580 insertions(+), 141 deletions(-) create mode 100644 tests/sync/test_profile.py diff --git a/tconnectsync/domain/device_settings.py b/tconnectsync/domain/device_settings.py index 0d0ce67..0a8eb3e 100644 --- a/tconnectsync/domain/device_settings.py +++ b/tconnectsync/domain/device_settings.py @@ -1,4 +1,4 @@ -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import List, Optional @dataclass @@ -26,6 +26,16 @@ class Profile: 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: diff --git a/tconnectsync/sync/profile.py b/tconnectsync/sync/profile.py index 3bbe801..62ecba6 100644 --- a/tconnectsync/sync/profile.py +++ b/tconnectsync/sync/profile.py @@ -1,19 +1,25 @@ from typing import List, Tuple import logging +import json +import copy from ..api import TConnectApi -from ..domain.device_settings import Profile +from ..domain.device_settings import Profile, DeviceSettings +from ..parser.nightscout import NightscoutEntry from ..secret import PUMP_SERIAL_NUMBER logger = logging.getLogger(__name__) -def get_pump_profiles(tconnect: TConnectApi) -> List[Profile]: +def get_pump_profiles(tconnect: TConnectApi, serial_number: int = None) -> Tuple[List[Profile], DeviceSettings]: all_devices = tconnect.webui.my_devices() - if str(PUMP_SERIAL_NUMBER) not in all_devices: - logger.warn("Could not find entry for provided pump serial number in t:connect device list: %s, received: %s", PUMP_SERIAL_NUMBER, all_devices) + if serial_number is None: + serial_number = PUMP_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 [] - device = all_devices[str(PUMP_SERIAL_NUMBER)] + 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) @@ -22,7 +28,7 @@ def get_pump_profiles(tconnect: TConnectApi) -> List[Profile]: logger.info("Found pump profiles: %s", ["%s%s" % (profile.title, " (active)" if profile.active else "") for profile in device_profiles]) - return device_profiles + return device_profiles, device_settings """ @@ -32,14 +38,94 @@ currently in Nightscout. ns_profile_obj is the output from NightscoutApi.profiles() 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], ns_profile_obj: dict): +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 profile in Nightscout: %s: %s", profile_name, device.get(profile_name)) + 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.info("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.info("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 + + 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 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) + + configured_str = json.loads(json.dumps(configured)) + map_nested_dicts_modify(configured_str, lambda x: str(x)) + translated_str = json.loads(json.dumps(translated)) + map_nested_dicts_modify(translated_str, lambda x: str(x)) + + 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 diff --git a/tests/parser/test_nightscout.py b/tests/parser/test_nightscout.py index ae250be..789a06b 100644 --- a/tests/parser/test_nightscout.py +++ b/tests/parser/test_nightscout.py @@ -3,8 +3,8 @@ 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 tconnectsync.secret import NIGHTSCOUT_PROFILE_CARBS_HR_VALUE, NIGHTSCOUT_PROFILE_DELAY_VALUE, TIMEZONE_NAME +from ..sync.test_profile import DEVICE_PROFILE_A, DEVICE_SETTINGS, NS_PROFILE_A class TestNightscoutEntry(unittest.TestCase): maxDiff = None def test_basal(self): @@ -198,138 +198,10 @@ class TestNightscoutEntry(unittest.TestCase): def test_profile_store(self): self.assertEqual( NightscoutEntry.profile_store( - profile=Profile( - title='A', - active=True, - segments=[ - ProfileSegment( - display_time='Midnight', - time='12:00 AM', - basal_rate=0.8, - correction_factor=30.0, - carb_ratio=6.0, - target_bg_mgdl=110.0), - ProfileSegment( - display_time='6:00 AM', - time='6:00 AM', - basal_rate=1.25, - correction_factor=30.0, - carb_ratio=6.0, - target_bg_mgdl=110.0), - ProfileSegment( - display_time='11:00 AM', - time='11:00 AM', - basal_rate=1.0, - correction_factor=30.0, - carb_ratio=6.0, - target_bg_mgdl=110.0), - ProfileSegment( - display_time='Noon', - time='12:00 PM', - basal_rate=0.8, - correction_factor=30.0, - carb_ratio=6.0, - target_bg_mgdl=110.0) - ], - calculated_total_daily_basal=21.65, - insulin_duration_min=300, - carbs_enabled=True - ), - device_settings=DeviceSettings( - low_bg_threshold=80, - high_bg_threshold=200, - raw_settings={} - ) + profile=DEVICE_PROFILE_A, + device_settings=DEVICE_SETTINGS ), - { - "dia": 5.0, - "carbratio": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 6.0 - }, - { - "time": "06:00", - "timeAsSeconds": 6*60*60, - "value": 6.0 - }, - { - "time": "11:00", - "timeAsSeconds": 11*60*60, - "value": 6.0 - }, - { - "time": "12:00", - "timeAsSeconds": 12*60*60, - "value": 6.0 - } - ], - - "carbs_hr": NIGHTSCOUT_PROFILE_CARBS_HR_VALUE, - "delay": NIGHTSCOUT_PROFILE_DELAY_VALUE, - "sens": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 30.0 - }, - { - "time": "06:00", - "timeAsSeconds": 6*60*60, - "value": 30.0 - }, - { - "time": "11:00", - "timeAsSeconds": 11*60*60, - "value": 30.0 - }, - { - "time": "12:00", - "timeAsSeconds": 12*60*60, - "value": 30.0 - } - ], - "basal": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 0.8 - }, - { - "time": "06:00", - "timeAsSeconds": 6*60*60, - "value": 1.25 - }, - { - "time": "11:00", - "timeAsSeconds": 11*60*60, - "value": 1.0 - }, - { - "time": "12:00", - "timeAsSeconds": 12*60*60, - "value": 0.8 - } - ], - "target_low": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 80 - } - ], - "target_high": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 200 - } - ], - "timezone": TIMEZONE_NAME, - "startDate": "1970-01-01T00:00:00.000Z", - "units": "mg/dl" - } + NS_PROFILE_A ) diff --git a/tests/sync/test_profile.py b/tests/sync/test_profile.py new file mode 100644 index 0000000..18b267f --- /dev/null +++ b/tests/sync/test_profile.py @@ -0,0 +1,471 @@ +#!/usr/bin/env python3 + +import unittest +from typing import Dict +import copy + +from tconnectsync.sync.profile import get_pump_profiles, compare_profiles +from tconnectsync.domain.device_settings import Profile, ProfileSegment, DeviceSettings +from tconnectsync.secret import NIGHTSCOUT_PROFILE_CARBS_HR_VALUE, NIGHTSCOUT_PROFILE_DELAY_VALUE, TIMEZONE_NAME + +from ..parser.test_tconnect import TestTConnectEntryReading + +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 +} + +def build_ns_profile(profiles: Dict[str, dict], current_profile: str) -> dict: + return copy.deepcopy({ + "store": profiles, + "defaultProfile": current_profile, + "startDate": "1970-01-01T00:00:00.000Z", + "mills": 0, + "units": "mg/dl", + }) + +class TestCompareProfiles(unittest.TestCase): + maxDiff = None + def test_compare_profiles_identical_a(self): + pump_profiles = [DEVICE_PROFILE_A.activeProfile()] + ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A}, 'A') + + changed, new_profiles = compare_profiles( + pump_profiles, + DEVICE_SETTINGS, + ns_profile_obj + ) + + self.assertFalse(changed) + self.assertDictEqual(ns_profile_obj, new_profiles) + + self.ensure_stabilized(pump_profiles, new_profiles) + + def test_compare_profiles_identical_b(self): + pump_profiles = [DEVICE_PROFILE_B.activeProfile()] + ns_profile_obj = build_ns_profile({'B': NS_PROFILE_B}, 'B') + + changed, new_profiles = compare_profiles( + pump_profiles, + DEVICE_SETTINGS, + ns_profile_obj + ) + + self.assertFalse(changed) + self.assertDictEqual(ns_profile_obj, new_profiles) + + self.ensure_stabilized(pump_profiles, new_profiles) + + def test_compare_profiles_extra_profile_in_nightscout_ignored(self): + pump_profiles = [DEVICE_PROFILE_A.activeProfile()] + ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A, 'B': NS_PROFILE_B}, 'A') + + changed, new_profiles = compare_profiles( + pump_profiles, + DEVICE_SETTINGS, + ns_profile_obj + ) + + self.assertFalse(changed) + self.assertDictEqual(ns_profile_obj, new_profiles) + + self.ensure_stabilized(pump_profiles, new_profiles) + + def test_compare_profiles_current_nightscout_profile_changed(self): + pump_profiles = [DEVICE_PROFILE_A.activeProfile()] + ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A, 'B': NS_PROFILE_B}, 'B') + + changed, new_profiles = compare_profiles( + pump_profiles, + DEVICE_SETTINGS, + ns_profile_obj + ) + + self.assertTrue(changed) + self.assertNotEqual(ns_profile_obj, new_profiles) + + ns_profile_obj['defaultProfile'] = 'A' + self.assertDictEqual(ns_profile_obj, new_profiles) + + self.ensure_stabilized(pump_profiles, new_profiles) + + def test_compare_profiles_new_pump_profile_added(self): + pump_profiles = [DEVICE_PROFILE_A.activeProfile(), DEVICE_PROFILE_B] + ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A}, 'A') + + changed, new_profiles = compare_profiles( + pump_profiles, + DEVICE_SETTINGS, + ns_profile_obj + ) + + self.assertTrue(changed) + self.assertNotEqual(ns_profile_obj, new_profiles) + + ns_profile_obj['store']['B'] = new_profiles['store']['B'] + self.assertDictEqual(ns_profile_obj, new_profiles) + + self.ensure_stabilized(pump_profiles, new_profiles) + + def test_compare_profiles_new_pump_profile_added_and_active(self): + pump_profiles = [DEVICE_PROFILE_A, DEVICE_PROFILE_B.activeProfile()] + ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A}, 'A') + + changed, new_profiles = compare_profiles( + pump_profiles, + DEVICE_SETTINGS, + ns_profile_obj + ) + + self.assertTrue(changed) + self.assertNotEqual(ns_profile_obj, new_profiles) + + ns_profile_obj['store']['B'] = new_profiles['store']['B'] + ns_profile_obj['defaultProfile'] = 'B' + self.assertDictEqual(ns_profile_obj, new_profiles) + + self.ensure_stabilized(pump_profiles, new_profiles) + + def test_compare_profiles_existing_profile_edited_basal(self): + pump_profiles = [DEVICE_PROFILE_A.activeProfile()] + pump_profiles[0].segments[0].basal_rate = 0.1 + + ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A}, 'A') + + changed, new_profiles = compare_profiles( + pump_profiles, + DEVICE_SETTINGS, + ns_profile_obj + ) + + self.assertTrue(changed) + self.assertNotEqual(ns_profile_obj, new_profiles) + + ns_profile_obj['store']['A']['basal'][0]['value'] = 0.1 + self.assertDictEqual(ns_profile_obj, new_profiles) + + self.ensure_stabilized(pump_profiles, new_profiles) + + def test_compare_profiles_existing_profile_edited_new_chunk(self): + pump_profiles = [DEVICE_PROFILE_A.activeProfile()] + pump_profiles[0].segments.append(ProfileSegment( + display_time='06:00 PM', + time='06:00 PM', + basal_rate=0.7, + correction_factor=30.0, + carb_ratio=6.0, + target_bg_mgdl=110.0) + ) + + ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A}, 'A') + + changed, new_profiles = compare_profiles( + pump_profiles, + DEVICE_SETTINGS, + ns_profile_obj + ) + + self.assertTrue(changed) + self.assertNotEqual(ns_profile_obj, new_profiles) + + ns_profile_obj['store']['A']['basal'].append({'time': '18:00', 'timeAsSeconds': 18*60*60, 'value': 0.7}) + ns_profile_obj['store']['A']['carbratio'].append({'time': '18:00', 'timeAsSeconds': 18*60*60, 'value': 6}) + ns_profile_obj['store']['A']['sens'].append({'time': '18:00', 'timeAsSeconds': 18*60*60, 'value': 30}) + self.assertDictEqual(ns_profile_obj, new_profiles) + + self.ensure_stabilized(pump_profiles, new_profiles) + + def test_compare_profiles_existing_profile_edited_removed_chunk(self): + pump_profiles = [DEVICE_PROFILE_A.activeProfile()] + pump_profiles[0].segments.pop() + + ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A}, 'A') + + changed, new_profiles = compare_profiles( + pump_profiles, + DEVICE_SETTINGS, + ns_profile_obj + ) + + self.assertTrue(changed) + self.assertNotEqual(ns_profile_obj, new_profiles) + + ns_profile_obj['store']['A']['basal'].pop() + ns_profile_obj['store']['A']['carbratio'].pop() + ns_profile_obj['store']['A']['sens'].pop() + self.assertDictEqual(ns_profile_obj, new_profiles) + + self.ensure_stabilized(pump_profiles, new_profiles) + + def ensure_stabilized(self, pump_profiles, new_profiles): + changed, _ = compare_profiles( + pump_profiles, + DEVICE_SETTINGS, + new_profiles + ) + self.assertFalse(changed, 'did not stabilize') + +if __name__ == '__main__': + unittest.main() \ No newline at end of file