Add type annotations to TandemSource sync processors

Annotate the 11 event processors, the ProcessTimeRange orchestrator, and
UpdateProfiles. The shared processor interface is now typed:
__init__(tconnect, nightscout, tconnect_device_id, pretend, features),
enabled() -> bool, process(events, time_start, time_end) -> List[dict],
write(ns_entries) -> int. Converter helpers return Optional[dict] since
they fall through to None on type mismatch.

ProcessTimeRange.process() returns Tuple[int, Optional[int]] and its
tconnectDevice param reuses the PumpEventMetadata TypedDict. Client
params (TConnectApi/NightscoutApi/BaseEvent) use TYPE_CHECKING-guarded
imports with string forward refs to avoid import cycles at runtime.

tconnectsync/sync/tandemsource/process.py

@@ -2,6 +2,13 @@ import logging
 import collections
 import arrow
 
+from types import ModuleType
+from typing import List, Optional, Tuple, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
+    from ...api.tandemsource import PumpEventMetadata
+
 from ...features import DEVICE_STATUS, DEFAULT_FEATURES
 from ...eventparser import events as eventtypes
 from ...domain.tandemsource.event_class import EventClass
@@ -21,7 +28,7 @@ from .update_profiles import UpdateProfiles
 logger = logging.getLogger(__name__)
 
 class ProcessTimeRange:
-    def __init__(self, tconnect, nightscout, tconnectDevice, pretend, secret, features=DEFAULT_FEATURES):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnectDevice: "PumpEventMetadata", pretend: bool, secret: ModuleType, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnectDevice['tconnectDeviceId']
@@ -48,7 +55,7 @@ class ProcessTimeRange:
         UpdateProfiles
     ]
 
-    def process(self, time_start, time_end):
+    def process(self, time_start: arrow.Arrow, time_end: arrow.Arrow) -> Tuple[int, Optional[int]]:
         fetch_all_event_types = self.secret.FETCH_ALL_EVENT_TYPES or DEVICE_STATUS in self.features
 
         logger.info(f"ProcessTimeRange time_start={time_start} time_end={time_end} tconnect_device_id={self.tconnect_device_id} features={self.features} fetch_all_event_types={fetch_all_event_types}")

tconnectsync/sync/tandemsource/process_alarm.py

@@ -1,6 +1,12 @@
 import logging
 import arrow
 
+from typing import Iterable, List, Optional, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
+    from ...eventparser.raw_event import BaseEvent
+
 from ...features import DEFAULT_FEATURES
 from ... import features
 from ...eventparser.generic import Events, decode_raw_events, EVENT_LEN
@@ -15,17 +21,17 @@ from ...parser.nightscout import (
 logger = logging.getLogger(__name__)
 
 class ProcessAlarm:
-    def __init__(self, tconnect, nightscout, tconnect_device_id, pretend, features=DEFAULT_FEATURES):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnect_device_id
         self.pretend = pretend
         self.features = features
 
-    def enabled(self):
+    def enabled(self) -> bool:
         return features.PUMP_EVENTS in self.features
 
-    def process(self, events, time_start, time_end):
+    def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arrow) -> List[dict]:
         logger.debug("ProcessAlarm: querying for last uploaded alarm")
         last_upload = self.nightscout.last_uploaded_entry(ALARM_EVENTTYPE, time_start=time_start, time_end=time_end)
         last_upload_time = None
@@ -48,13 +54,13 @@ class ProcessAlarm:
 
         return ns_entries
 
-    def skip_event(self, event):
+    def skip_event(self, event: "BaseEvent") -> bool:
         return event.alarmid in (
             eventtypes.LidAlarmActivated.AlarmidEnum.ResumePumpAlarm,
             eventtypes.LidAlarmActivated.AlarmidEnum.ResumePumpAlarm2
         )
 
-    def write(self, ns_entries):
+    def write(self, ns_entries: List[dict]) -> int:
         count = 0
         for entry in ns_entries:
             if self.pretend:
@@ -67,7 +73,7 @@ class ProcessAlarm:
         return count
 
 
-    def alarm_to_nsentry(self, event):
+    def alarm_to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
         if type(event) == eventtypes.LidAlarmActivated:
             return NightscoutEntry.alarm(
                 created_at = event.eventTimestamp.format(),

tconnectsync/sync/tandemsource/process_basal.py

@@ -1,3 +1,4 @@
+import datetime
 import logging
 import arrow
 
@@ -14,20 +15,26 @@ from ...parser.nightscout import (
     NightscoutEntry
 )
 
+from typing import Iterable, List, Optional, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
+    from ...eventparser.raw_event import BaseEvent
+
 logger = logging.getLogger(__name__)
 
 class ProcessBasal:
-    def __init__(self, tconnect, nightscout, tconnect_device_id, pretend, features=DEFAULT_FEATURES):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnect_device_id
         self.pretend = pretend
         self.features = features
 
-    def enabled(self):
+    def enabled(self) -> bool:
         return features.BASAL in self.features
 
-    def process(self, events, time_start, time_end):
+    def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arrow) -> List[dict]:
         logger.debug("ProcessBasal: querying for last uploaded entry")
         last_upload = self.nightscout.last_uploaded_entry(BASAL_EVENTTYPE, time_start=time_start, time_end=time_end)
         last_upload_time = None
@@ -61,7 +68,7 @@ class ProcessBasal:
 
         return ns_entries
 
-    def write(self, ns_entries):
+    def write(self, ns_entries: List[dict]) -> int:
         count = 0
         for entry in ns_entries:
             if self.pretend:
@@ -74,7 +81,7 @@ class ProcessBasal:
         return count
 
 
-    def basal_to_nsentry(self, start, duration, event):
+    def basal_to_nsentry(self, start: arrow.Arrow, duration: datetime.timedelta, event: "BaseEvent") -> Optional[dict]:
         if type(event) == eventtypes.LidBasalRateChange:
             value = insulin_float_round(event.commandedbasalrate)
             if IGNORE_ZERO_UNIT_BASAL and value < 0.01:

tconnectsync/sync/tandemsource/process_basal_resume.py

@@ -1,6 +1,12 @@
 import logging
 import arrow
 
+from typing import Iterable, List, Optional, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
+    from ...eventparser.raw_event import BaseEvent
+
 from ...features import DEFAULT_FEATURES
 from ... import features
 from ...eventparser.generic import Events, decode_raw_events, EVENT_LEN
@@ -15,17 +21,17 @@ from ...parser.nightscout import (
 logger = logging.getLogger(__name__)
 
 class ProcessBasalResume:
-    def __init__(self, tconnect, nightscout, tconnect_device_id, pretend, features=DEFAULT_FEATURES):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnect_device_id
         self.pretend = pretend
         self.features = features
 
-    def enabled(self):
+    def enabled(self) -> bool:
         return features.PUMP_EVENTS in self.features
 
-    def process(self, events, time_start, time_end):
+    def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arrow) -> List[dict]:
         logger.debug("ProcessBasalResume: querying for last uploaded resume-suspension")
         last_upload = self.nightscout.last_uploaded_entry(BASALRESUME_EVENTTYPE, time_start=time_start, time_end=time_end)
         last_upload_time = None
@@ -45,7 +51,7 @@ class ProcessBasalResume:
 
         return ns_entries
 
-    def write(self, ns_entries):
+    def write(self, ns_entries: List[dict]) -> int:
         count = 0
         for entry in ns_entries:
             if self.pretend:
@@ -58,7 +64,7 @@ class ProcessBasalResume:
         return count
 
 
-    def resume_to_nsentry(self, event):
+    def resume_to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
         if type(event) == eventtypes.LidPumpingResumed:
             return NightscoutEntry.basalresume(
                 created_at = event.eventTimestamp.format(),

tconnectsync/sync/tandemsource/process_basal_suspension.py

@@ -1,6 +1,12 @@
 import logging
 import arrow
 
+from typing import Iterable, List, Optional, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
+    from ...eventparser.raw_event import BaseEvent
+
 from ...features import DEFAULT_FEATURES
 from ... import features
 from ...eventparser.generic import Events, decode_raw_events, EVENT_LEN
@@ -15,17 +21,17 @@ from ...parser.nightscout import (
 logger = logging.getLogger(__name__)
 
 class ProcessBasalSuspension:
-    def __init__(self, tconnect, nightscout, tconnect_device_id, pretend, features=DEFAULT_FEATURES):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnect_device_id
         self.pretend = pretend
         self.features = features
 
-    def enabled(self):
+    def enabled(self) -> bool:
         return features.PUMP_EVENTS in self.features or features.BASAL in self.features
 
-    def process(self, events, time_start, time_end):
+    def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arrow) -> List[dict]:
         logger.debug("ProcessBasalSuspension: querying for last uploaded suspension")
         last_upload = self.nightscout.last_uploaded_entry(BASALSUSPENSION_EVENTTYPE, time_start=time_start, time_end=time_end)
         last_upload_time = None
@@ -45,7 +51,7 @@ class ProcessBasalSuspension:
 
         return ns_entries
 
-    def write(self, ns_entries):
+    def write(self, ns_entries: List[dict]) -> int:
         count = 0
         for entry in ns_entries:
             if self.pretend:
@@ -58,7 +64,7 @@ class ProcessBasalSuspension:
         return count
 
 
-    def suspension_to_nsentry(self, event):
+    def suspension_to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
         if type(event) == eventtypes.LidPumpingSuspended:
             return NightscoutEntry.basalsuspension(
                 created_at = event.eventTimestamp.format(),

tconnectsync/sync/tandemsource/process_bolus.py

@@ -13,20 +13,26 @@ from ...parser.nightscout import (
     NightscoutEntry
 )
 
+from typing import Iterable, List, Optional, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
+    from ...eventparser.raw_event import BaseEvent
+
 logger = logging.getLogger(__name__)
 
 class ProcessBolus:
-    def __init__(self, tconnect, nightscout, tconnect_device_id, pretend, features=DEFAULT_FEATURES):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnect_device_id
         self.pretend = pretend
         self.features = features
 
-    def enabled(self):
+    def enabled(self) -> bool:
         return features.BOLUS in self.features
 
-    def process(self, events, time_start, time_end):
+    def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arrow) -> List[dict]:
         logger.debug("ProcessBolus: querying for last uploaded entry")
         last_upload = self.nightscout.last_uploaded_entry(BOLUS_EVENTTYPE, time_start=time_start, time_end=time_end)
         last_upload_time = None
@@ -68,7 +74,7 @@ class ProcessBolus:
 
         return ns_entries
 
-    def write(self, ns_entries):
+    def write(self, ns_entries: List[dict]) -> int:
         count = 0
         for entry in ns_entries:
             if self.pretend:
@@ -81,7 +87,7 @@ class ProcessBolus:
         return count
 
 
-    def bolus_to_nsentry(self, bolusCompleted, bolusRequested1, bolusRequested2, bolusRequested3):
+    def bolus_to_nsentry(self, bolusCompleted: "BaseEvent", bolusRequested1: "BaseEvent", bolusRequested2: "BaseEvent", bolusRequested3: "BaseEvent") -> Optional[dict]:
         suffixes = []
         if bolusRequested2 and bolusRequested2.useroverride == eventtypes.LidBolusRequestedMsg2.UseroverrideEnum.Yes:
             suffixes.append('(Override)')

tconnectsync/sync/tandemsource/process_cartridge.py

@@ -12,20 +12,26 @@ from ...parser.nightscout import (
     NightscoutEntry
 )
 
+from typing import Iterable, List, Optional, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
+    from ...eventparser.raw_event import BaseEvent
+
 logger = logging.getLogger(__name__)
 
 class ProcessCartridge:
-    def __init__(self, tconnect, nightscout, tconnect_device_id, pretend, features=DEFAULT_FEATURES):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnect_device_id
         self.pretend = pretend
         self.features = features
 
-    def enabled(self):
+    def enabled(self) -> bool:
         return features.PUMP_EVENTS in self.features
 
-    def process(self, events, time_start, time_end):
+    def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arrow) -> List[dict]:
         logger.debug("ProcessCartridge: querying for last uploaded entry")
         last_upload = self.nightscout.last_uploaded_entry(SITECHANGE_EVENTTYPE, time_start=time_start, time_end=time_end)
         last_upload_time = None
@@ -66,7 +72,7 @@ class ProcessCartridge:
 
         return ns_entries
 
-    def write(self, ns_entries):
+    def write(self, ns_entries: List[dict]) -> int:
         count = 0
         for entry in ns_entries:
             if self.pretend:
@@ -78,21 +84,21 @@ class ProcessCartridge:
 
         return count
 
-    def cart_to_nsentry(self, cartFilled):
+    def cart_to_nsentry(self, cartFilled: "BaseEvent") -> Optional[dict]:
         return NightscoutEntry.sitechange(
             created_at = cartFilled.eventTimestamp.format(),
             reason = "Cartridge Filled" + (" (%du filled)" % round(cartFilled.v2Volume) if cartFilled.v2Volume else ""),
             pump_event_id = "%s" % cartFilled.seqNum
         )
 
-    def cannula_to_nsentry(self, cannulaFilled):
+    def cannula_to_nsentry(self, cannulaFilled: "BaseEvent") -> Optional[dict]:
         return NightscoutEntry.sitechange(
             created_at = cannulaFilled.eventTimestamp.format(),
             reason = "Cannula Filled" + (" (%du primed)" % round(cannulaFilled.primesize, 2) if cannulaFilled.primesize else ""),
             pump_event_id = "%s" % cannulaFilled.seqNum
         )
 
-    def tubing_to_nsentry(self, tubingFilled):
+    def tubing_to_nsentry(self, tubingFilled: "BaseEvent") -> Optional[dict]:
         return NightscoutEntry.sitechange(
             created_at = tubingFilled.eventTimestamp.format(),
             reason = "Tubing Filled" + (" (%du primed)" % round(tubingFilled.primesize) if tubingFilled.primesize else ""),

tconnectsync/sync/tandemsource/process_cgm_alert.py

@@ -12,20 +12,26 @@ from ...parser.nightscout import (
     NightscoutEntry
 )
 
+from typing import Iterable, List, Optional, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
+    from ...eventparser.raw_event import BaseEvent
+
 logger = logging.getLogger(__name__)
 
 class ProcessCGMAlert:
-    def __init__(self, tconnect, nightscout, tconnect_device_id, pretend, features=DEFAULT_FEATURES):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnect_device_id
         self.pretend = pretend
         self.features = features
 
-    def enabled(self):
+    def enabled(self) -> bool:
         return features.CGM_ALERTS in self.features
 
-    def process(self, events, time_start, time_end):
+    def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arrow) -> List[dict]:
         logger.debug("ProcessCGMAlert: querying for last uploaded entry")
         last_upload = self.nightscout.last_uploaded_entry(CGM_ALERT_EVENTTYPE, time_start=time_start, time_end=time_end)
         last_upload_time = None
@@ -52,7 +58,7 @@ class ProcessCGMAlert:
 
         return ns_entries
 
-    def write(self, ns_entries):
+    def write(self, ns_entries: List[dict]) -> int:
         count = 0
         for entry in ns_entries:
             if self.pretend:
@@ -64,7 +70,7 @@ class ProcessCGMAlert:
 
         return count
 
-    def alert_to_nsentry(self, alert):
+    def alert_to_nsentry(self, alert: "BaseEvent") -> Optional[dict]:
         # FSL3 alert codes are defined in eventparser/static_dicts.py:CGM_ALERTS_DICT
         # Alert code meanings are documented in comments there.
         if not alert.dalertid:

tconnectsync/sync/tandemsource/process_cgm_reading.py

@@ -7,11 +7,17 @@ from ... import secret
 from ...eventparser.raw_event import TANDEM_EPOCH
 from ...parser.nightscout import NightscoutEntry
 
+from typing import Iterable, List, Optional, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
+    from ...eventparser.raw_event import BaseEvent
+
 
 logger = logging.getLogger(__name__)
 
 class ProcessCGMReading:
-    def __init__(self, tconnect, nightscout, tconnect_device_id, pretend, features=DEFAULT_FEATURES, timezone=None):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES, timezone: Optional[str] = None) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnect_device_id
@@ -19,10 +25,10 @@ class ProcessCGMReading:
         self.features = features
         self.timezone = timezone or secret.TIMEZONE_NAME
 
-    def enabled(self):
+    def enabled(self) -> bool:
         return features.CGM in self.features
 
-    def process(self, events, time_start, time_end):
+    def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arrow) -> List[dict]:
         logger.debug("ProcessCGMReading: querying for last uploaded entry")
         last_upload = self.nightscout.last_uploaded_bg_entry(time_start=time_start, time_end=time_end)
         last_upload_time = None
@@ -47,7 +53,7 @@ class ProcessCGMReading:
 
         return ns_entries
 
-    def write(self, ns_entries):
+    def write(self, ns_entries: List[dict]) -> int:
         count = 0
         for entry in ns_entries:
             if self.pretend:
@@ -59,12 +65,12 @@ class ProcessCGMReading:
 
         return count
 
-    def timestamp_for(self, event):
+    def timestamp_for(self, event: "BaseEvent") -> arrow.Arrow:
         # For backfills the time the event was added to the pump's event store
         # might not be the time it actually occurred, so we use the egvTimestamp
         return arrow.get(TANDEM_EPOCH + event.egvTimestamp, tzinfo='UTC').replace(tzinfo=self.timezone)
 
-    def to_nsentry(self, event):
+    def to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
         return NightscoutEntry.entry(
             sgv = event.currentglucosedisplayvalue,
             created_at = self.timestamp_for(event).format(),

tconnectsync/sync/tandemsource/process_cgm_start_join_stop.py

@@ -12,20 +12,26 @@ from ...parser.nightscout import (
     NightscoutEntry
 )
 
+from typing import Iterable, List, Optional, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
+    from ...eventparser.raw_event import BaseEvent
+
 logger = logging.getLogger(__name__)
 
 class ProcessCGMStartJoinStop:
-    def __init__(self, tconnect, nightscout, tconnect_device_id, pretend, features=DEFAULT_FEATURES):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnect_device_id
         self.pretend = pretend
         self.features = features
 
-    def enabled(self):
+    def enabled(self) -> bool:
         return features.PUMP_EVENTS in self.features or features.CGM_ALERTS in self.features
 
-    def process(self, events, time_start, time_end):
+    def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arrow) -> List[dict]:
         last_upload = None
         last_upload_time = None
         for eventtype in [CGM_START_EVENTTYPE, CGM_JOIN_EVENTTYPE, CGM_STOP_EVENTTYPE]:
@@ -61,7 +67,7 @@ class ProcessCGMStartJoinStop:
 
         return ns_entries
 
-    def write(self, ns_entries):
+    def write(self, ns_entries: List[dict]) -> int:
         count = 0
         for entry in ns_entries:
             if self.pretend:
@@ -74,7 +80,7 @@ class ProcessCGMStartJoinStop:
         return count
 
 
-    def to_nsentry(self, event):
+    def to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
         if type(event) in EventClass._CGM_START:
             return NightscoutEntry.cgm_start(
                 created_at = format_datetime(event.eventTimestamp),

tconnectsync/sync/tandemsource/process_device_status.py

@@ -13,20 +13,26 @@ from ...parser.nightscout import (
     NightscoutEntry
 )
 
+from typing import Iterable, List, Optional, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
+    from ...eventparser.raw_event import BaseEvent
+
 logger = logging.getLogger(__name__)
 
 class ProcessDeviceStatus:
-    def __init__(self, tconnect, nightscout, tconnect_device_id, pretend, features=DEFAULT_FEATURES):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnect_device_id
         self.pretend = pretend
         self.features = features
 
-    def enabled(self):
+    def enabled(self) -> bool:
         return features.DEVICE_STATUS in self.features
 
-    def process(self, events, time_start, time_end):
+    def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arrow) -> List[dict]:
         logger.debug("ProcessDeviceStatus: querying for last uploaded devicestatus")
         last_upload = self.nightscout.last_uploaded_devicestatus(time_start=time_start, time_end=time_end)
         last_upload_time = None
@@ -56,7 +62,7 @@ class ProcessDeviceStatus:
         ns_entries.append(self.daily_basal_to_nsentry(last_daily_basal_event))
         return ns_entries
 
-    def daily_basal_to_nsentry(self, event):
+    def daily_basal_to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
         return NightscoutEntry.devicestatus(
             created_at=event.eventTimestamp.format(),
             batteryVoltage=(float(event.batterylipomillivolts or 0)/1000),
@@ -65,7 +71,7 @@ class ProcessDeviceStatus:
         )
 
 
-    def write(self, ns_entries):
+    def write(self, ns_entries: List[dict]) -> int:
         count = 0
         for entry in ns_entries:
             if self.pretend:

tconnectsync/sync/tandemsource/process_user_mode.py

@@ -1,6 +1,12 @@
 import logging
 import arrow
 
+from typing import Iterable, List, Optional, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
+    from ...eventparser.raw_event import BaseEvent
+
 from ...features import DEFAULT_FEATURES
 from ... import features
 from ...eventparser.generic import Events, decode_raw_events, EVENT_LEN
@@ -18,17 +24,17 @@ NOT_ENDED = "Not Ended"
 logger = logging.getLogger(__name__)
 
 class ProcessUserMode:
-    def __init__(self, tconnect, nightscout, tconnect_device_id, pretend, features=DEFAULT_FEATURES):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnect_device_id
         self.pretend = pretend
         self.features = features
 
-    def enabled(self):
+    def enabled(self) -> bool:
         return features.PUMP_EVENTS in self.features
 
-    def process(self, events, time_start, time_end):
+    def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arrow) -> List[dict]:
         logger.debug("ProcessUserMode: querying for last uploaded exercise entry")
         exercise_last_upload = self.nightscout.last_uploaded_entry(EXERCISE_EVENTTYPE, time_start=time_start, time_end=time_end)
         exercise_last_upload_time = None
@@ -119,7 +125,7 @@ class ProcessUserMode:
 
         return ns_entries
 
-    def write(self, ns_entries):
+    def write(self, ns_entries: List[dict]) -> int:
         count = 0
         for entry in ns_entries:
             if self.pretend:
@@ -131,19 +137,19 @@ class ProcessUserMode:
 
         return count
 
-    def is_start_sleep(self, event):
+    def is_start_sleep(self, event: "BaseEvent") -> bool:
         return event.requestedaction == eventtypes.LidAaUserModeChange.RequestedactionEnum.StartSleep
-    def is_stop_sleep(self, event):
+    def is_stop_sleep(self, event: "BaseEvent") -> bool:
         return event.requestedaction == eventtypes.LidAaUserModeChange.RequestedactionEnum.StopSleep or \
                event.requestedaction == eventtypes.LidAaUserModeChange.RequestedactionEnum.StopAll
-    def is_start_exercise(self, event):
+    def is_start_exercise(self, event: "BaseEvent") -> bool:
         return event.requestedaction == eventtypes.LidAaUserModeChange.RequestedactionEnum.StartExercise
-    def is_stop_exercise(self, event):
+    def is_stop_exercise(self, event: "BaseEvent") -> bool:
         return event.requestedaction == eventtypes.LidAaUserModeChange.RequestedactionEnum.StopExercise or \
                event.requestedaction == eventtypes.LidAaUserModeChange.RequestedactionEnum.StopAll
 
 
-    def sleep_to_nsentry(self, start, stop=None, time_end=None):
+    def sleep_to_nsentry(self, start: "BaseEvent", stop: Optional["BaseEvent"] = None, time_end: Optional[arrow.Arrow] = None) -> Optional[dict]:
         if start and stop:
             reason = None
             if start.sleepstartedbygui == eventtypes.LidAaUserModeChange.SleepstartedbyguiEnum.TrueVal:
@@ -176,7 +182,7 @@ class ProcessUserMode:
             )
 
 
-    def exercise_to_nsentry(self, start, stop=None, time_end=None):
+    def exercise_to_nsentry(self, start: "BaseEvent", stop: Optional["BaseEvent"] = None, time_end: Optional[arrow.Arrow] = None) -> Optional[dict]:
         if start and stop:
             reason = "Exercise"
             if start.exercisechoice == eventtypes.LidAaUserModeChange.ExercisechoiceEnum.Timed:
@@ -207,7 +213,7 @@ class ProcessUserMode:
                 pump_event_id = "%s" % start.seqNum
             )
 
-    def process_unended_sleep_stop(self, event, sleep_last_upload):
+    def process_unended_sleep_stop(self, event: "BaseEvent", sleep_last_upload: dict) -> dict:
         logger.info("ProcessUserMode: Deleting old sleep event treatment before pushing update (delete treatments/%s)" % sleep_last_upload["_id"])
         if self.pretend:
             logger.info("ProcessUserMode: Skipping delete in pretend mode")
@@ -223,7 +229,7 @@ class ProcessUserMode:
             pump_event_id="%s,%s" % (sleep_last_upload.get("pump_event_id",""), event.seqNum)
         )
 
-    def process_unended_exercise_stop(self, event, exercise_last_upload):
+    def process_unended_exercise_stop(self, event: "BaseEvent", exercise_last_upload: dict) -> dict:
         logger.info("ProcessUserMode: Deleting old exercise event treatment before pushing update (delete treatments/%s)" % exercise_last_upload["_id"])
         if self.pretend:
             logger.info("ProcessUserMode: Skipping delete in pretend mode")

tconnectsync/sync/tandemsource/update_profiles.py

@@ -2,7 +2,10 @@ import logging
 import arrow
 import copy
 import json
-from typing import Tuple
+from typing import Any, Callable, List, Tuple, TYPE_CHECKING
+if TYPE_CHECKING:
+    from ...api import TConnectApi
+    from ...nightscout import NightscoutApi
 
 from ...features import DEFAULT_FEATURES
 from ... import features
@@ -14,21 +17,21 @@ from ...secret import NIGHTSCOUT_PROFILE_UPLOAD_MODE
 
 logger = logging.getLogger(__name__)
 
-def _get_default_upload_mode():
+def _get_default_upload_mode() -> str:
     return NIGHTSCOUT_PROFILE_UPLOAD_MODE
 
 class UpdateProfiles:
-    def __init__(self, tconnect, nightscout, tconnect_device_id, pretend, features=DEFAULT_FEATURES):
+    def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
         self.nightscout = nightscout
         self.tconnect_device_id = tconnect_device_id
         self.pretend = pretend
         self.features = features
 
-    def enabled(self):
+    def enabled(self) -> bool:
         return features.PROFILES in self.features
 
-    def update(self, pretend):
+    def update(self, pretend: bool) -> bool:
         upload_mode = _get_default_upload_mode()
         logger.debug("UpdateProfiles: getting Tandem Source profile data")
 
@@ -160,7 +163,7 @@ class UpdateProfiles:
             return True
 
         # convert all JSON values into strings
-        def map_nested_dicts_modify(ob, func):
+        def map_nested_dicts_modify(ob: dict, func: Callable) -> None:
             for k, v in ob.items():
                 if isinstance(v, dict):
                     map_nested_dicts_modify(v, func)
@@ -169,7 +172,7 @@ class UpdateProfiles:
                 else:
                     ob[k] = func(v)
 
-        def map_nested_lists_modify(ob, func):
+        def map_nested_lists_modify(ob: list, func: Callable) -> None:
             for i in range(len(ob)):
                 v = ob[i]
                 if isinstance(v, dict):
@@ -179,7 +182,7 @@ class UpdateProfiles:
                 else:
                     ob[i] = func(v)
 
-        def to_numeric(x):
+        def to_numeric(x: Any) -> Any:
             if type(x) in [int, float]:
                 return '%f' % x
             try: