Merge pull request #176 from jbergler/i-175

Handle the case when a lock supports a sensor, but it isn't installed

custom_components/ttlock/api.py

@@ -158,13 +158,15 @@ async def get_lock(self, lock_id: int) -> Lock:
         res = await self.get("lock/detail", lockId=lock_id)
         return Lock.parse_obj(res)
 
-    async def get_sensor(self, lock_id: int) -> Sensor:
+    async def get_sensor(self, lock_id: int) -> Sensor | None:
         """Get the Sensor."""
-        res = await self.get("doorSensor/query", lockId=lock_id)
-        if "errcode" in res and res["errcode"] != 0:
-            _LOGGER.error("Error setting up sensor", lock_id, res["errmsg"])
-            pass
-        return Sensor.parse_obj(res)
+
+        try:
+            res = await self.get("doorSensor/query", lockId=lock_id)
+            return Sensor.parse_obj(res)
+        except RequestFailed:
+            # Janky but the API doesn't return different errors if the sensor is missing or there's some other problem
+            return None
 
     async def get_lock_state(self, lock_id: int) -> LockState:
         """Get the state of a lock."""

custom_components/ttlock/coordinator.py

@@ -30,6 +30,11 @@ class SensorData:
     battery: int | None = None
     last_fetched: datetime | None = None
 
+    @property
+    def present(self) -> bool:
+        """To indicate if a sensor is installed."""
+        return self.battery is not None
+
 
 @dataclass
 class LockState:
@@ -156,17 +161,17 @@ async def _async_update_data(self) -> LockState:
                     or new_data.sensor.last_fetched < dt.now() - timedelta(days=1)
                 ):
                     sensor = await self.api.get_sensor(self.lock_id)
-
-                    new_data.sensor.battery = sensor.battery_level
                     new_data.sensor.last_fetched = dt.now()
+                    if sensor:
+                        new_data.sensor.battery = sensor.battery_level
             else:
                 new_data.sensor = None
 
             if new_data.locked is None:
                 try:
                     state = await self.api.get_lock_state(self.lock_id)
                     new_data.locked = state.locked == State.locked
-                    if new_data.sensor:
+                    if new_data.sensor and new_data.sensor.present:
                         new_data.sensor.opened = state.opened == SensorState.opened
                 except Exception:
                     pass
@@ -210,7 +215,7 @@ def _process_webhook_data(self, event: WebhookEvent):
                 new_data.last_user = event.user
                 new_data.last_reason = event.event.description
 
-        if new_data.sensor and event.sensorState:
+        if new_data.sensor and new_data.sensor.present and event.sensorState:
             if event.sensorState.opened == SensorState.opened:
                 new_data.sensor.opened = True
             if event.sensorState.opened == SensorState.closed:

tests/conftest.py

@@ -132,7 +132,10 @@ def create_mock_data(scenario: str = "default") -> MockApiData:
                 lock=Lock.parse_obj(LOCK_DETAILS_WITH_SENSOR),
                 sensor=Sensor.parse_obj(SENSOR_DETAILS),
                 state=LockState.parse_obj(LOCK_STATE_UNLOCKED),
-                passage_mode=PassageModeConfig.parse_obj(PASSAGE_MODE_6_TO_6_7_DAYS),
+            ),
+            "sensor_not_installed": MockApiData(
+                lock=Lock.parse_obj(LOCK_DETAILS_WITH_SENSOR),
+                state=LockState.parse_obj(LOCK_STATE_UNLOCKED),
             ),
             "locked": MockApiData(
                 lock=Lock.parse_obj(BASIC_LOCK_DETAILS),

tests/test_coordinator.py

@@ -148,6 +148,17 @@ async def test_coordinator_loads_sensor_data(
                 seconds=3
             )
 
+        async def test_coordinator_handles_missing_sensor(
+            self, coordinator: LockUpdateCoordinator, mock_api_responses
+        ):
+            mock_api_responses("sensor_not_installed")
+            await coordinator.async_refresh()
+
+            assert coordinator.data.sensor.present is False
+            assert coordinator.data.sensor.last_fetched > dt.now() - timedelta(
+                seconds=3
+            )
+
         async def test_sensor_data_only_fetched_once(
             self, coordinator: LockUpdateCoordinator, mock_api_responses
         ):