Merge pull request #50 from acces90/main

Added Starkvind device control

README.md

@@ -9,6 +9,7 @@ This repository provides an unofficial Python client for controlling the IKEA Di
 
 - [light control](#controlling-lights)
 - [outlet control](#controlling-outlets)
+- [air purifier control](#controlling-air-purifier)
 - [blinds control](#controlling-blinds)
 - [remote controllers](#remote-controllers) (tested with STYRBAR)
 - [environment sensor](#environment-sensor) (tested with VINDSTYRKA)
@@ -183,6 +184,44 @@ outlet.set_on(outlet_on=True)
 outlet.set_startup_behaviour(behaviour=StartupEnum.START_OFF)
 ```
 
+## [Controlling Air Purifier](./src/dirigera/devices/air_purifier.py)
+
+To get information about the available air purifiers, you can use the `get_air_purifiers()` method:
+
+```python
+air_purifiers = dirigera_hub.get_air_purifiers()
+```
+
+The air purifier object has the following attributes (additional to the core attributes):
+
+```python
+fan_mode: FanModeEnum
+fan_mode_sequence: str
+motor_state: int
+child_lock: bool
+status_light: bool
+motor_runtime: int
+filter_alarm_status: bool
+filter_elapsed_time: int
+filter_lifetime: int
+current_p_m25: int
+```
+
+Available methods for blinds are:
+
+```python
+air_purifier.set_name(name="living room purifier")
+
+air_purifier.set_fan_mode(fan_mode=FanModeEnum.AUTO)
+
+air_purifier.set_motor_state(motor_state=42)
+
+air_purifier.set_child_lock(child_lock=True)
+
+air_purifier.set_status_light(light_state=False)
+```
+
+
 ## [Controlling Blinds](./src/dirigera/devices/blinds.py)
 
 To get information about the available blinds, you can use the `get_blinds()` method:

src/dirigera/devices/air_purifier.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+from enum import Enum
+from typing import Any, Dict
+from .device import Attributes, Device
+from ..hub.abstract_smart_home_hub import AbstractSmartHomeHub
+
+class FanModeEnum(Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    AUTO = "auto"
+
+class AirPurifierAttributes(Attributes):
+    """canReceive"""
+    fan_mode: FanModeEnum
+    fan_mode_sequence: str
+    motor_state: int
+    child_lock: bool
+    status_light: bool
+    """readOnly"""
+    motor_runtime: int
+    filter_alarm_status: bool
+    filter_elapsed_time: int
+    filter_lifetime: int
+    current_p_m25: int
+
+class AirPurifier(Device):
+    dirigera_client: AbstractSmartHomeHub
+    attributes: AirPurifierAttributes
+
+    def reload(self) -> AirPurifier:
+        data = self.dirigera_client.get(route=f"/devices/{self.id}")
+        return AirPurifier(dirigeraClient=self.dirigera_client, **data)
+
+    def set_name(self, name: str) -> None:
+        if "customName" not in self.capabilities.can_receive:
+            raise AssertionError("This airpurifier does not support the set_name function")
+
+        data = [{"attributes": {"customName": name}}]
+        self.dirigera_client.patch(route=f"/devices/{self.id}", data=data)
+        self.attributes.custom_name = name
+
+    def set_fan_mode(self, fan_mode: FanModeEnum) -> None:
+        data = [{"attributes": {"fanMode": fan_mode.value}}]
+        self.dirigera_client.patch(route=f"/devices/{self.id}", data=data)
+        self.attributes.fan_mode = fan_mode
+
+    def set_motor_state(self, motor_state: int) -> None:
+        """
+        Sets the fan behaviour.
+        Values 0 to 50 allowed.
+        0 == off
+        1 == auto
+        """
+        desired_motor_state = int(motor_state)
+        if desired_motor_state < 0 or desired_motor_state > 50:
+            raise ValueError("Motor state must be a value between 0 and 50")
+
+        data = [{"attributes": {"motorState": desired_motor_state}}]
+        self.dirigera_client.patch(route=f"/devices/{self.id}", data=data)
+        self.attributes.motor_state = desired_motor_state
+
+    def set_child_lock(self, child_lock: bool) -> None:
+        if "childLock" not in self.capabilities.can_receive:
+            raise AssertionError("This air-purifier does not support the child lock function")
+
+        data = [{"attributes": {"childLock": child_lock}}]
+        self.dirigera_client.patch(route=f"/devices/{self.id}", data=data)
+        self.attributes.child_lock = child_lock
+
+    def set_status_light(self, light_state: bool) -> None:
+        data = [{"attributes": {"statusLight": light_state}}]
+        self.dirigera_client.patch(route=f"/devices/{self.id}", data=data)
+        self.attributes.status_light = light_state
+
+def dict_to_air_purifier(data: Dict[str, Any], dirigera_client: AbstractSmartHomeHub) -> AirPurifier:
+    return AirPurifier(
+        dirigeraClient=dirigera_client,
+        **data
+    )

src/dirigera/hub/hub.py

@@ -9,6 +9,7 @@
 
 from ..devices.device import Device
 from .abstract_smart_home_hub import AbstractSmartHomeHub
+from ..devices.air_purifier import AirPurifier, dict_to_air_purifier
 from ..devices.light import Light, dict_to_light
 from ..devices.blinds import Blind, dict_to_blind
 from ..devices.controller import Controller, dict_to_controller
@@ -119,6 +120,14 @@ def _get_device_data_by_id(self, id_: str) -> Dict:
                 raise ValueError("Device id not found") from err
             raise err
 
+    def get_air_purifiers(self) -> List[AirPurifier]:
+        """
+        Fetches all air purifiers registered in the Hub
+        """
+        devices = self.get("/devices")
+        airpurifiers = list(filter(lambda x: x["type"] == "airPurifier", devices))
+        return [dict_to_air_purifier(air_p, self) for air_p in airpurifiers]
+
     def get_lights(self) -> List[Light]:
         """
         Fetches all lights registered in the Hub
@@ -268,6 +277,7 @@ def get_all_devices(self) -> List[Device]:
         Fetches all devices registered in the Hub
         """
         devices: List[Device] = []
+        devices.extend(self.get_air_purifiers())
         devices.extend(self.get_blinds())
         devices.extend(self.get_controllers())
         devices.extend(self.get_environment_sensors())

tests/test_air_purifier.py

@@ -0,0 +1,150 @@
+from typing import Dict
+import pytest
+from src.dirigera.hub.abstract_smart_home_hub import FakeDirigeraHub
+from src.dirigera.devices.air_purifier import (
+    AirPurifier,
+    FanModeEnum,
+    dict_to_air_purifier
+)
+
+
+@pytest.fixture(name="fake_client")
+def fixture_fake_client() -> FakeDirigeraHub:
+    return FakeDirigeraHub()
+
+@pytest.fixture(name="purifier_dict")
+def fixture_fake_air_purifier_dict() -> dict:
+    return {
+        "id": "d121f38a-fc37-4bd9-8a3c-f79e4f45fccf_1",
+        "type": "airPurifier",
+        "deviceType": "airPurifier",
+        "createdAt": "2023-08-09T12:31:59.000Z",
+        "isReachable": True,
+        "lastSeen": "2024-02-21T19:55:44.000Z",
+        "attributes": {
+            "customName": "Air Purifier",
+            "firmwareVersion": "1.0.033",
+            "hardwareVersion": "1",
+            "manufacturer": "IKEA of Sweden",
+            "model": "STARKVIND Air purifier",
+            "productCode": "E2007",
+            "serialNumber": "2C1165FFFE89F47C",
+            "fanMode": "auto",
+            "fanModeSequence": "lowMediumHighAuto",
+            "motorRuntime": 106570,
+            "motorState": 15,
+            "filterAlarmStatus": False,
+            "filterElapsedTime": 227980,
+            "filterLifetime": 259200,
+            "childLock": False,
+            "statusLight": True,
+            "currentPM25": 3,
+            "identifyPeriod": 0,
+            "identifyStarted": "2000-01-01T00:00:00.000Z",
+            "permittingJoin": False,
+            "otaPolicy": "autoUpdate",
+            "otaProgress": 0,
+            "otaScheduleEnd": "00:00",
+            "otaScheduleStart": "00:00",
+            "otaState": "readyToCheck",
+            "otaStatus": "updateAvailable",
+        },
+        "capabilities": {
+            "canSend": [],
+            "canReceive": [
+                "customName",
+                "fanMode",
+                "fanModeSequence",
+                "motorState",
+                "childLock",
+                "statusLight",
+            ],
+        },
+        "room": {
+            "id": "1a846fdc-317c-4d94-8722-cb0196256a16",
+            "name": "Livingroom",
+            "color": "ikea_green_no_66",
+            "icon": "rooms_arm_chair",
+        },
+        "deviceSet": [],
+        "remoteLinks": [],
+        "isHidden": False,
+    }
+
+
+@pytest.fixture(name="fake_purifier")
+def fixture_purifier(
+    fake_client: FakeDirigeraHub, purifier_dict: Dict
+) -> AirPurifier:
+    return AirPurifier(dirigeraClient=fake_client, **purifier_dict)
+
+def test_set_name(fake_purifier: AirPurifier, fake_client: FakeDirigeraHub) -> None:
+    new_name = "Luftreiniger"
+    fake_purifier.set_name(new_name)
+    action = fake_client.patch_actions.pop()
+    assert action["route"] == f"/devices/{fake_purifier.id}"
+    assert action["data"] == [{"attributes": {"customName": new_name}}]
+    assert fake_purifier.attributes.custom_name == new_name
+
+def test_set_fan_mode_enum(fake_purifier: AirPurifier, fake_client: FakeDirigeraHub) -> None:
+    new_mode = FanModeEnum.LOW
+    fake_purifier.set_fan_mode(new_mode)
+    action = fake_client.patch_actions.pop()
+    assert action["route"] == f"/devices/{fake_purifier.id}"
+    assert action["data"] == [{"attributes": {"fanMode": new_mode.value}}]
+    assert fake_purifier.attributes.fan_mode == new_mode
+
+def test_set_motor_state(fake_purifier: AirPurifier, fake_client: FakeDirigeraHub) -> None:
+    new_motor_state = 42
+    fake_purifier.set_motor_state(new_motor_state)
+    action = fake_client.patch_actions.pop()
+    assert action["route"] == f"/devices/{fake_purifier.id}"
+    assert action["data"] == [{"attributes": {"motorState": new_motor_state}}]
+    assert fake_purifier.attributes.motor_state == new_motor_state
+
+def test_set_child_lock(fake_purifier: AirPurifier, fake_client: FakeDirigeraHub) -> None:
+    new_child_lock = True
+    fake_purifier.set_child_lock(new_child_lock)
+    action = fake_client.patch_actions.pop()
+    assert action["route"] == f"/devices/{fake_purifier.id}"
+    assert action["data"] == [{"attributes": {"childLock": new_child_lock}}]
+    assert fake_purifier.attributes.child_lock == new_child_lock
+
+def test_status_light(fake_purifier: AirPurifier, fake_client: FakeDirigeraHub) -> None:
+    new_status_light = False
+    fake_purifier.set_status_light(new_status_light)
+    action = fake_client.patch_actions.pop()
+    assert action["route"] == f"/devices/{fake_purifier.id}"
+    assert action["data"] == [{"attributes": {"statusLight": new_status_light}}]
+    assert fake_purifier.attributes.status_light == new_status_light
+
+
+def test_dict_to_purifier(fake_client: FakeDirigeraHub, purifier_dict: Dict) -> None:
+    purifier = dict_to_air_purifier(purifier_dict, fake_client)
+    assert purifier.id == purifier_dict["id"]
+    assert purifier.is_reachable == purifier_dict["isReachable"]
+    assert purifier.attributes.custom_name == purifier_dict["attributes"]["customName"]
+    assert (
+        purifier.attributes.firmware_version
+        == purifier_dict["attributes"]["firmwareVersion"]
+    )
+    assert (
+        purifier.attributes.hardware_version
+        == purifier_dict["attributes"]["hardwareVersion"]
+    )
+    assert purifier.attributes.model == purifier_dict["attributes"]["model"]
+    assert purifier.attributes.serial_number == purifier_dict["attributes"]["serialNumber"]
+    assert purifier.attributes.manufacturer == purifier_dict["attributes"]["manufacturer"]
+    assert purifier.attributes.fan_mode.value == purifier_dict["attributes"]["fanMode"]
+    assert purifier.attributes.fan_mode_sequence == purifier_dict["attributes"]["fanModeSequence"]
+    assert purifier.attributes.motor_state == purifier_dict["attributes"]["motorState"]
+    assert purifier.attributes.child_lock == purifier_dict["attributes"]["childLock"]
+    assert purifier.attributes.status_light == purifier_dict["attributes"]["statusLight"]
+    assert purifier.attributes.motor_runtime == purifier_dict["attributes"]["motorRuntime"]
+    assert purifier.attributes.filter_alarm_status == purifier_dict["attributes"]["filterAlarmStatus"]
+    assert purifier.attributes.filter_elapsed_time == purifier_dict["attributes"]["filterElapsedTime"]
+    assert purifier.attributes.filter_lifetime == purifier_dict["attributes"]["filterLifetime"]
+    assert purifier.attributes.current_p_m25 == purifier_dict["attributes"]["currentPM25"]
+    assert purifier.capabilities.can_receive == purifier_dict["capabilities"]["canReceive"]
+    assert purifier.room.id == purifier_dict["room"]["id"]
+    assert purifier.room.name == purifier_dict["room"]["name"]