Fix #156: handle LidMalfunctionActivated in ProcessAlarm

.github/workflows/python-package.yml

@@ -27,7 +27,7 @@ jobs:
         run: |
           python -m venv .venv
           .venv/bin/python -m pip install --upgrade pip
-          .venv/bin/python -m pip install -e . flake8 pytest coverage
+          .venv/bin/python -m pip install -e . flake8 pytest coverage mypy
       # - name: Run pipenv check
       #   run: |
       #     # DDoS attacks in wheel and setuptools packages, not relevant
@@ -53,6 +53,9 @@ jobs:
           .venv/bin/flake8 . --exclude=.venv --count --select=E9,F63,F7,F82 --ignore=F824 --show-source --statistics
           # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
           .venv/bin/flake8 . --exclude=.venv --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+      - name: Type-check annotated modules with mypy
+        run: |
+          .venv/bin/mypy
       - name: Run tconnectsync --help
         run: |
           .venv/bin/tconnectsync --help

Pipfile

@@ -8,6 +8,7 @@ ptpython = "*"
 flake8 = "*"
 pytest = "*"
 coverage = "*"
+mypy = "*"
 
 [packages]
 tconnectsync = {path = "."}
@@ -17,3 +18,4 @@ tconnectsync = "python3 main.py"
 test = "python3 -m unittest discover -vv"
 build_events = "bash -c 'cd tconnectsync/eventparser && python3 build_events.py > events.py'"
 lint = "bash -c 'flake8 . --count --select=E9,F63,F7,F82 && flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 && echo PASS'"
+typecheck = "mypy"

setup.cfg

@@ -47,3 +47,9 @@ exclude =
 [options.entry_points]
 console_scripts =
     tconnectsync = tconnectsync:main
+
+[mypy]
+files =
+    tconnectsync/sync/tandemsource/process_alarm.py
+follow_imports = silent
+ignore_missing_imports = True

tconnectsync/sync/tandemsource/process_alarm.py

@@ -1,11 +1,11 @@
 import logging
 import arrow
 
-from typing import Iterable, List, Optional, TYPE_CHECKING
+from typing import Iterable, List, Union, TYPE_CHECKING
+from typing_extensions import assert_never
 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
@@ -20,6 +20,8 @@
 
 logger = logging.getLogger(__name__)
 
+AlarmOrMalfunction = Union[eventtypes.LidAlarmActivated, eventtypes.LidMalfunctionActivated]
+
 class ProcessAlarm:
     def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
         self.tconnect = tconnect
@@ -54,7 +56,10 @@ def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arr
 
         return ns_entries
 
-    def skip_event(self, event: "BaseEvent") -> bool:
+    def skip_event(self, event: AlarmOrMalfunction) -> bool:
+        if not isinstance(event, eventtypes.LidAlarmActivated):
+            return False
+
         return event.alarmId in (
             eventtypes.LidAlarmActivated.AlarmidEnum.ResumePumpAlarm,
             eventtypes.LidAlarmActivated.AlarmidEnum.ResumePumpAlarm2
@@ -73,16 +78,19 @@ def write(self, ns_entries: List[dict]) -> int:
         return count
 
 
-    def alarm_to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
-        if type(event) == eventtypes.LidAlarmActivated:
+    def alarm_to_nsentry(self, event: AlarmOrMalfunction) -> dict:
+        if isinstance(event, eventtypes.LidAlarmActivated):
+            alarmId = event.alarmId
+            reason = alarmId.name if alarmId is not None else "Alarm%s" % event.alarmIdRaw
             return NightscoutEntry.alarm(
                 created_at = event.eventTimestamp.format(),
-                reason = "%s" % event.alarmId.name,
+                reason = reason,
                 pump_event_id = "%s" % event.seqNum
             )
-        elif type(event) == eventtypes.LidMalfunctionActivated:
+        elif isinstance(event, eventtypes.LidMalfunctionActivated):
             return NightscoutEntry.alarm(
                 created_at = event.eventTimestamp.format(),
                 reason = "Malfunction",
                 pump_event_id = "%s" % event.seqNum
             )
+        assert_never(event)

tests/eventparser/events/test_lidmalfunctionactivated.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+
+import json
+import unittest
+
+from tconnectsync.eventparser.generic import Event
+from tconnectsync.eventparser import events as eventtypes
+from tconnectsync.eventparser.raw_event import RawEvent
+
+
+class TestLidMalfunctionActivated(unittest.TestCase):
+    maxDiff = None
+
+    def setUp(self):
+        self.fixture = {
+            "deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
+            "eventCode": 6,
+            "sequenceGroup": 0,
+            "sequenceNumber": 500123,
+            "pumpDateTime": "2026-05-16T00:07:00",
+            "eventProperties": {"malfId": 7, "faultLocatorData": 8311, "param1": 42, "param2": 0},
+            "estimatedDateTime": "2026-05-16T00:07:00Z",
+        }
+
+    def test_dispatches_to_correct_class(self):
+        self.assertIsInstance(Event(self.fixture), eventtypes.LidMalfunctionActivated)
+        self.assertNotIsInstance(Event(self.fixture), RawEvent)
+
+    def test_has_no_alarmid_attribute(self):
+        ev = Event(self.fixture)
+        self.assertFalse(hasattr(ev, 'alarmId'))
+        self.assertEqual(ev.malfIdRaw, 7)
+
+    def test_envelope_fields(self):
+        ev = Event(self.fixture)
+        self.assertEqual(ev.eventId, 6)
+        self.assertEqual(ev.seqNum, 500123)
+
+    def test_timestamp_preserves_wall_clock(self):
+        ev = Event(self.fixture)
+        self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'), "2026-05-16T00:07:00")
+
+    def test_plain_fields(self):
+        ev = Event(self.fixture)
+        self.assertEqual(ev.faultLocatorData, 8311)
+        self.assertEqual(ev.param1, 42)
+        self.assertEqual(ev.param2, 0)
+
+    def test_todict_is_json_serializable(self):
+        ev = Event(self.fixture)
+        d = ev.todict()
+        json.dumps(d)  # must not raise
+        self.assertEqual(d["id"], 6)
+        self.assertEqual(d["name"], "LID_MALFUNCTION_ACTIVATED")
+        self.assertEqual(d["malfIdRaw"], 7)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/sync/tandemsource/test_process_alarm.py

@@ -92,6 +92,16 @@ def test_empty_cartridge_alarm(self):
     "eventProperties": {"alarmId": 18, "faultLocatorData": 8311, "param1": 5228339, "param2": 0},
 }
 
+MALFUNCTION = {
+    "deviceAssignmentId": "00000000-0000-0000-0000-000000000000",
+    "eventCode": 6,
+    "sequenceGroup": 0,
+    "sequenceNumber": 500123,
+    "pumpDateTime": "2026-05-16T00:07:00",
+    "estimatedDateTime": "2026-05-16T00:07:00Z",
+    "eventProperties": {"malfId": 7, "faultLocatorData": 8311, "param1": 42, "param2": 0},
+}
+
 
 class TestProcessAlarmJson(unittest.TestCase):
     maxDiff = None
@@ -127,6 +137,41 @@ def test_resume_alarm_skipped(self):
         p = self.process.process(list(Events([dict(ALARM_RESUME)])), None, None)
         self.assertEqual(p, [])
 
+    def test_malfunction_alarm_uploaded(self):
+        event = Event(dict(MALFUNCTION))
+        self.assertEqual(type(event), eventtypes.LidMalfunctionActivated)
+        self.assertFalse(hasattr(event, 'alarmId'))
+
+        p = self.process.process([event], None, None)
+
+        self.assertEqual(len(p), 1)
+        self.assertDictEqual(p[0], {
+            'eventType': 'Alarm',
+            'reason': 'Malfunction',
+            'notes': 'Malfunction',
+            'created_at': '2026-05-16 00:07:00-04:00',
+            'enteredBy': 'Pump (tconnectsync)',
+            'pump_event_id': '500123'
+        })
+
+    def test_alarm_and_malfunction_mixed(self):
+        # A batch mixing both ALARM-class event types must not crash and must
+        # emit an entry for each.
+        p = self.process.process(list(Events([dict(ALARM_PUMP_RESET), dict(MALFUNCTION)])), None, None)
+
+        self.assertEqual(len(p), 2)
+        reasons = {entry['reason'] for entry in p}
+        self.assertEqual(reasons, {'PumpResetAlarm', 'Malfunction'})
+
+
+class TestAlarmOrMalfunctionUnion(unittest.TestCase):
+    def test_union_matches_eventclass(self):
+        from typing import get_args
+        from tconnectsync.sync.tandemsource.process_alarm import AlarmOrMalfunction
+        from tconnectsync.domain.tandemsource.event_class import EventClass
+
+        self.assertEqual(set(get_args(AlarmOrMalfunction)), set(EventClass.ALARM))
+
 
 if __name__ == '__main__':
     unittest.main()