feat(air-purifier): add verified filter usage reset (#583)

* docs: document air purifier filter usage reset

* feat(air-purifier): add verified filter usage reset

* test: cover purifier filter reset verification

README.md

@@ -10,6 +10,19 @@ source .venv/bin/activate
 pip install .
 ```
 
+## Air purifier filter reset
+
+After replacing the filter, call `await purifier.reset_filter_usage()` on a
+configured `SwitchbotAirPurifier` instance. The method uses the device's
+encrypted BLE connection, sends the reset command without automatic retries,
+and reads back the used-hours counter. It returns `True` and refreshes the
+cached `filter element working time` only after the device reports zero.
+
+Failed commands or an unconfirmed reset raise `SwitchbotOperationError`. If a
+reply is lost, check the device before retrying: the reset may already have
+been applied. This resets the device counter and does not restore its previous
+value.
+
 ## OAuth account access
 
 pySwitchbot provides helpers for SwitchBot's authorization-code flow. The

switchbot/devices/air_purifier.py

@@ -33,6 +33,8 @@ DEVICE_GET_BASIC_SETTINGS_KEY = "570f4d81"
 COMMAND_SET_PERCENTAGE = f"{COMMAND_HEAD}02{{percentage:02x}}"
 READ_LED_SETTINGS_COMMAND = "570f4d05"
 READ_LED_STATUS_COMMAND = "570f4d07"
+RESET_FILTER_USAGE_COMMAND = f"{COMMAND_HEAD}09"
+READ_FILTER_USAGE_COMMAND = "570f4d09"
 
 
 class SwitchbotAirPurifier(SwitchbotSequenceBaseLight, SwitchbotEncryptedDevice):
@@ -147,6 +149,31 @@ class SwitchbotAirPurifier(SwitchbotSequenceBaseLight, SwitchbotEncryptedDevice)
             return data | {"pm25": pm25}
         return data
 
+    async def reset_filter_usage(self) -> bool:
+        """Reset filter usage, returning True only after a verified zero readback."""
+        result = await self._send_command(RESET_FILTER_USAGE_COMMAND, retry=0)
+        if not self._check_command_result(result, 0, {1}):
+            raise SwitchbotOperationError("Filter usage reset was rejected")
+
+        result = await self._send_command(READ_FILTER_USAGE_COMMAND)
+        if not result or len(result) < 3:
+            raise SwitchbotOperationError(
+                "Reset was sent, but filter usage could not be verified. "
+                "Check the device before retrying"
+            )
+        if not self._check_command_result(result, 0, {1}):
+            raise SwitchbotOperationError("Filter usage read was rejected")
+        used_hours = _UNPACK_UINT16_BE(result, 1)[0]
+        if used_hours != 0:
+            raise SwitchbotOperationError(
+                f"Reset was sent, but the device still reports {used_hours} used hours. "
+                "Check the device before retrying"
+            )
+
+        self._update_parsed_data({"filter element working time": used_hours})
+        self._fire_callbacks()
+        return True
+
     @update_after_operation
     async def set_preset_mode(self, preset_mode: str) -> bool:
         """Send command to set air purifier preset_mode."""

tests/test_air_purifier.py

@@ -1,6 +1,7 @@
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, call, patch
 
 import pytest
+import pytest_asyncio
 from bleak.backends.device import BLEDevice
 
 from switchbot import (
@@ -21,6 +22,133 @@ common_params = [
 ]
 
 
+@pytest_asyncio.fixture(params=common_params, ids=["table_us", "us", "jp", "table_jp"])
+async def filter_reset_device(
+    request: pytest.FixtureRequest,
+) -> air_purifier.SwitchbotAirPurifier:
+    raw_adv, model, model_name = request.param
+    ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
+    device = air_purifier.SwitchbotAirPurifier(
+        ble_device, "ff", "ffffffffffffffffffffffffffffffff", model=model_name
+    )
+    device.update = AsyncMock()
+    device.update_from_advertisement(
+        make_advertisement_data(ble_device, raw_adv, model, model_name)
+    )
+    return device
+
+
+@pytest.mark.asyncio
+async def test_reset_filter_usage(
+    filter_reset_device: air_purifier.SwitchbotAirPurifier,
+) -> None:
+    device = filter_reset_device
+    initial_data = device.parsed_data.copy()
+    listener = MagicMock()
+    unsubscribe = device.subscribe(listener)
+    with patch.object(
+        device, "_send_command", AsyncMock(side_effect=[b"\x01", b"\x01\x00\x00"])
+    ) as send_command:
+        assert await device.reset_filter_usage() is True
+        assert send_command.await_args_list == [
+            call("570f4c09", retry=0),
+            call("570f4d09"),
+        ]
+    assert device.parsed_data == initial_data | {"filter element working time": 0}
+    listener.assert_called_once_with()
+    unsubscribe()
+
+    advertisement = device._sb_adv_data
+    assert advertisement is not None
+    device.update_from_advertisement(
+        make_advertisement_data(
+            advertisement.device,
+            advertisement.data["rawAdvData"],
+            advertisement.data["model"],
+            advertisement.data["modelName"],
+            {"filter element working time": 1},
+        )
+    )
+    assert device.parsed_data["filter element working time"] == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("reset_response", "read_response", "message", "expected_calls"),
+    [
+        pytest.param(None, None, "Sending command failed", 1, id="no_reset_reply"),
+        pytest.param(b"", None, "Sending command failed", 1, id="empty_reset_reply"),
+        pytest.param(b"\x02", None, "reset was rejected", 1, id="reset_rejected"),
+        pytest.param(b"\x01", None, "could not be verified", 2, id="no_read_reply"),
+        pytest.param(b"\x01", b"", "could not be verified", 2, id="empty_read_reply"),
+        pytest.param(
+            b"\x01", b"\x01", "could not be verified", 2, id="short_read_reply"
+        ),
+        pytest.param(
+            b"\x01", b"\x01\x00", "could not be verified", 2, id="partial_counter"
+        ),
+        pytest.param(
+            b"\x01", b"\x02\x00\x00", "read was rejected", 2, id="read_rejected"
+        ),
+        pytest.param(
+            b"\x01", b"\x01\x07\x68", "1896 used hours", 2, id="counter_not_reset"
+        ),
+    ],
+)
+async def test_reset_filter_usage_failure(
+    filter_reset_device: air_purifier.SwitchbotAirPurifier,
+    reset_response: bytes | None,
+    read_response: bytes | None,
+    message: str,
+    expected_calls: int,
+) -> None:
+    device = filter_reset_device
+    initial_data = device.parsed_data.copy()
+    listener = MagicMock()
+    device.subscribe(listener)
+    with patch.object(
+        device, "_send_command", AsyncMock(side_effect=[reset_response, read_response])
+    ) as send_command:
+        with pytest.raises(SwitchbotOperationError, match=message):
+            await device.reset_filter_usage()
+        assert send_command.await_count == expected_calls
+    assert device.parsed_data == initial_data
+    listener.assert_not_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("responses", "expected_calls"),
+    [
+        pytest.param(
+            [SwitchbotOperationError("Connection failed")], 1, id="reset_connection"
+        ),
+        pytest.param(
+            [b"\x01", SwitchbotOperationError("Connection failed")],
+            2,
+            id="read_connection",
+        ),
+    ],
+)
+async def test_reset_filter_usage_transport_failure(
+    filter_reset_device: air_purifier.SwitchbotAirPurifier,
+    responses: list[bytes | SwitchbotOperationError],
+    expected_calls: int,
+) -> None:
+    device = filter_reset_device
+    initial_data = device.parsed_data.copy()
+    listener = MagicMock()
+    device.subscribe(listener)
+    with patch.object(
+        device, "_send_command", AsyncMock(side_effect=responses)
+    ) as send_command:
+        with pytest.raises(SwitchbotOperationError, match="Connection failed"):
+            await device.reset_filter_usage()
+        assert send_command.await_count == expected_calls
+    assert device.parsed_data == initial_data
+    listener.assert_not_called()
+
+
 def create_device_for_command_testing(
     rawAdvData: bytes,
     model: str,