From a9c9a7ebd9ebb16c52b341ad418d13678d49eea2 Mon Sep 17 00:00:00 2001 From: James Woglom Date: Tue, 30 Jun 2026 23:29:17 +0000 Subject: [PATCH] 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 | 11 +++++-- .../sync/tandemsource/process_alarm.py | 18 +++++++---- .../sync/tandemsource/process_basal.py | 17 +++++++---- .../sync/tandemsource/process_basal_resume.py | 16 ++++++---- .../tandemsource/process_basal_suspension.py | 16 ++++++---- .../sync/tandemsource/process_bolus.py | 16 ++++++---- .../sync/tandemsource/process_cartridge.py | 20 ++++++++----- .../sync/tandemsource/process_cgm_alert.py | 16 ++++++---- .../sync/tandemsource/process_cgm_reading.py | 18 +++++++---- .../process_cgm_start_join_stop.py | 16 ++++++---- .../tandemsource/process_device_status.py | 16 ++++++---- .../sync/tandemsource/process_user_mode.py | 30 +++++++++++-------- .../sync/tandemsource/update_profiles.py | 19 +++++++----- 13 files changed, 153 insertions(+), 76 deletions(-) diff --git a/tconnectsync/sync/tandemsource/process.py b/tconnectsync/sync/tandemsource/process.py index aae7b52..2e64f2c 100644 --- a/tconnectsync/sync/tandemsource/process.py +++ b/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}") diff --git a/tconnectsync/sync/tandemsource/process_alarm.py b/tconnectsync/sync/tandemsource/process_alarm.py index 62a02b9..727c052 100644 --- a/tconnectsync/sync/tandemsource/process_alarm.py +++ b/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(), diff --git a/tconnectsync/sync/tandemsource/process_basal.py b/tconnectsync/sync/tandemsource/process_basal.py index 6da3e0f..729eeb8 100644 --- a/tconnectsync/sync/tandemsource/process_basal.py +++ b/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: diff --git a/tconnectsync/sync/tandemsource/process_basal_resume.py b/tconnectsync/sync/tandemsource/process_basal_resume.py index edc0205..fa928f7 100644 --- a/tconnectsync/sync/tandemsource/process_basal_resume.py +++ b/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(), diff --git a/tconnectsync/sync/tandemsource/process_basal_suspension.py b/tconnectsync/sync/tandemsource/process_basal_suspension.py index 6ea8df2..c31dfaa 100644 --- a/tconnectsync/sync/tandemsource/process_basal_suspension.py +++ b/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(), diff --git a/tconnectsync/sync/tandemsource/process_bolus.py b/tconnectsync/sync/tandemsource/process_bolus.py index 3c80f74..f36bd96 100644 --- a/tconnectsync/sync/tandemsource/process_bolus.py +++ b/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)') diff --git a/tconnectsync/sync/tandemsource/process_cartridge.py b/tconnectsync/sync/tandemsource/process_cartridge.py index de53db9..ae9c79e 100644 --- a/tconnectsync/sync/tandemsource/process_cartridge.py +++ b/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 ""), diff --git a/tconnectsync/sync/tandemsource/process_cgm_alert.py b/tconnectsync/sync/tandemsource/process_cgm_alert.py index 5790675..e83275d 100644 --- a/tconnectsync/sync/tandemsource/process_cgm_alert.py +++ b/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: diff --git a/tconnectsync/sync/tandemsource/process_cgm_reading.py b/tconnectsync/sync/tandemsource/process_cgm_reading.py index 19c58d3..e3520e1 100644 --- a/tconnectsync/sync/tandemsource/process_cgm_reading.py +++ b/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(), diff --git a/tconnectsync/sync/tandemsource/process_cgm_start_join_stop.py b/tconnectsync/sync/tandemsource/process_cgm_start_join_stop.py index abd1a2d..4406947 100644 --- a/tconnectsync/sync/tandemsource/process_cgm_start_join_stop.py +++ b/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), diff --git a/tconnectsync/sync/tandemsource/process_device_status.py b/tconnectsync/sync/tandemsource/process_device_status.py index df80aa1..86026cd 100644 --- a/tconnectsync/sync/tandemsource/process_device_status.py +++ b/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: diff --git a/tconnectsync/sync/tandemsource/process_user_mode.py b/tconnectsync/sync/tandemsource/process_user_mode.py index 8a6200b..41515ff 100644 --- a/tconnectsync/sync/tandemsource/process_user_mode.py +++ b/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") diff --git a/tconnectsync/sync/tandemsource/update_profiles.py b/tconnectsync/sync/tandemsource/update_profiles.py index 072ec0c..5aa854d 100644 --- a/tconnectsync/sync/tandemsource/update_profiles.py +++ b/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: