Merge pull request #80 from kzosabe/develop

Release v0.4.0

Author: kzosabe

.github/workflows/main.yml

@@ -17,10 +17,10 @@ jobs:
         python-version: ["3.7", "3.8", "3.9", "3.10"]
     steps:
       - name: Checkout
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
 
       - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v2
+        uses: actions/setup-python@v3
         with:
           python-version: ${{ matrix.python-version }}
 
@@ -32,7 +32,7 @@ jobs:
 
       - name: Load cached venv
         id: cached-poetry-dependencies
-        uses: actions/cache@v2
+        uses: actions/cache@v3.0.2
         with:
           path: .venv
           key: venv-${{ runner.os }}-${{ hashFiles('**/poetry.lock') }}

CHANGELOG.md

@@ -1,3 +1,11 @@
+0.4.0, 2022-05-24
+-------------------------
+
+- Add webhook support(#77, #78)
+  - You can create, get, set, and delete webhook configurations via SwitchBotClient or SwitchBotAPIClient
+- Add pseudo status for AirConditioner(#59)
+- Fix Humidifier behavior when lackWater column is missing(#58)
+
 0.3.2, 2022-02-03
 -------------------------
 

README.md

@@ -164,6 +164,22 @@ SwitchBotCommandResult(status_code=100, message='success', response_body={})
 ```
 The specified scene can be executed immediately.
 
+### Webhooks
+
+```python
+from switchbot_client import SwitchBotClient
+
+client = SwitchBotClient()
+client.set_webhook(url="https://example.com/foo", enable=True)
+print(client.webhooks())
+```
+
+```
+[SwitchBotWebhook(url='https://example.com/foo', enable=True, device_list='ALL', create_time=datetime.datetime(2022, 1, 1, 12, 0, 0, 123456), last_update_time=datetime.datetime(2022, 1, 1, 12, 0, 0, 123456))]
+```
+
+You can handle [webhook](https://github.com/OpenWonderLabs/SwitchBotAPI#webhook) configurations via SwitchBotClient.
+
 ### Raw API interface
 
 Devices and scenes also can be manipulated via the low-level raw API client.

docs/conf.py

@@ -5,7 +5,7 @@
 project = "switchbot-client"
 copyright = "2021, kzosabe"
 author = "kzosabe"
-release = "0.3.2"
+release = "0.4.0"
 extensions = ["sphinx.ext.autodoc", "sphinx.ext.napoleon"]
 templates_path = ["_templates"]
 exclude_patterns = ["_build", ".DS_Store"]

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "switchbot-client"
-version = "0.3.2"
+version = "0.4.0"
 description = "A Python client library for SwitchBot API."
 license = "Apache-2.0 or MIT"
 authors = [
@@ -29,12 +29,12 @@ typing-extensions = ">=3.10,<5.0"
 black = ">=20.8b1"
 flake8 = "^4.0.1"
 isort = "^5.9.3"
-mypy = "^0.931"
-pylint = "^2.12.2"
-pytest = "^6.2.5"
+mypy = "^0.950"
+pylint = "^2.13.8"
+pytest = "^7.1.2"
 pytest-cov = "^3.0.0"
 pytest-mock = "^3.7.0"
-tox = "^3.24.5"
+tox = "^3.25.0"
 Sphinx = "^4.3.2"
 
 [build-system]

pyproject.toml

@@ -1,6 +1,7 @@
 import json
 import os
 from dataclasses import dataclass
+from typing import List
 
 import requests
 import yaml
@@ -101,6 +102,62 @@ def scenes_execute(self, scene_id: str) -> SwitchBotAPIResponse:
         formatted_response: SwitchBotAPIResponse = self._check_api_response(response)
         return formatted_response
 
+    def webhook_setup(self, url: str) -> SwitchBotAPIResponse:
+        payload = {
+            "action": "setupWebhook",
+            "url": url,
+            "deviceList": "ALL",
+        }
+        response: requests.Response = requests.post(
+            self._uri("v1.0/webhook/setupWebhook"),
+            headers=self._headers(),
+            data=json.dumps(payload),
+        )
+        formatted_response: SwitchBotAPIResponse = self._check_api_response(response)
+        return formatted_response
+
+    def webhook_query_url(self) -> SwitchBotAPIResponse:
+        payload = {
+            "action": "queryUrl",
+        }
+        response: requests.Response = requests.post(
+            self._uri("v1.0/webhook/queryWebhook"),
+            headers=self._headers(),
+            data=json.dumps(payload),
+        )
+        formatted_response: SwitchBotAPIResponse = self._check_api_response(response)
+        return formatted_response
+
+    def webhook_query_details(self, urls: List[str]) -> SwitchBotAPIResponse:
+        payload = {"action": "queryDetails", "urls": urls}
+        response: requests.Response = requests.post(
+            self._uri("v1.0/webhook/queryWebhook"),
+            headers=self._headers(),
+            data=json.dumps(payload),
+        )
+        formatted_response: SwitchBotAPIResponse = self._check_api_response(response)
+        return formatted_response
+
+    def webhook_update(self, config: dict) -> SwitchBotAPIResponse:
+        payload = {"action": "updateWebhook", "config": config}
+        response: requests.Response = requests.post(
+            self._uri("v1.0/webhook/updateWebhook"),
+            headers=self._headers(),
+            data=json.dumps(payload),
+        )
+        formatted_response: SwitchBotAPIResponse = self._check_api_response(response)
+        return formatted_response
+
+    def webhook_delete(self, url: str) -> SwitchBotAPIResponse:
+        payload = {"action": "deleteWebhook", "url": url}
+        response: requests.Response = requests.post(
+            self._uri("v1.0/webhook/deleteWebhook"),
+            headers=self._headers(),
+            data=json.dumps(payload),
+        )
+        formatted_response: SwitchBotAPIResponse = self._check_api_response(response)
+        return formatted_response
+
     def config_file_path(self):
         if self.__config_file_path is None:
             return os.path.expanduser(SwitchBotAPIClient.DEFAULT_CONFIG_FILE_PATH)

switchbot_client/api.py

@@ -1,9 +1,11 @@
+from datetime import datetime
 from typing import List, Optional
 
-from switchbot_client.api import SwitchBotAPIClient
+from switchbot_client.api import SwitchBotAPIClient, SwitchBotAPIResponse
 from switchbot_client.devices.base import SwitchBotDevice
 from switchbot_client.devices.factory import SwitchBotDeviceFactory
 from switchbot_client.scenes import SwitchBotScene
+from switchbot_client.webhooks.base import SwitchBotWebhook
 
 
 class SwitchBotClient:
@@ -42,3 +44,30 @@ def scene(self, scene_id: str) -> Optional[SwitchBotScene]:
         if len(filtered) == 0:
             return None
         return filtered[0]
+
+    def webhooks(self) -> List[SwitchBotWebhook]:
+        response_urls = self.api_client.webhook_query_url().body["urls"]
+        response = self.api_client.webhook_query_details(response_urls).body
+        return [
+            SwitchBotWebhook(
+                r["url"],
+                r["enable"],
+                r["deviceList"],
+                datetime.fromtimestamp(r["createTime"] / 1000),
+                datetime.fromtimestamp(r["lastUpdateTime"] / 1000),
+            )
+            for r in response
+        ]
+
+    def create_webhook(self, url: str) -> SwitchBotAPIResponse:
+        return self.api_client.webhook_setup(url)
+
+    def set_webhook(self, url: str, enable: bool) -> SwitchBotAPIResponse:
+        config = {
+            "url": url,
+            "enable": enable,
+        }
+        return self.api_client.webhook_update(config)
+
+    def delete_webhook(self, url: str) -> SwitchBotAPIResponse:
+        return self.api_client.webhook_delete(url)

switchbot_client/client.py

@@ -621,7 +621,7 @@ def status(self) -> HumidifierDeviceStatus:
             is_auto=status.raw_data["auto"],
             is_child_lock=status.raw_data["childLock"],
             is_muted=not status.raw_data["sound"],
-            is_lack_water=status.raw_data["lackWater"],
+            is_lack_water=status.raw_data["lackWater"] if "lackWater" in status.raw_data else False,
         )
 
     def power(self) -> str:

switchbot_client/devices/physical.py

@@ -1,23 +1,25 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Generic, Optional, TypeVar
 
 from switchbot_client.enums import ControlCommand, RemoteType
 from switchbot_client.types import APIRemoteDeviceObject
 
 from .base import SwitchBotCommandResult, SwitchBotDevice
-from .status import PseudoRemoteDeviceStatus
+from .status import PseudoAirConditionerStatus, PseudoRemoteDeviceStatus
 
 if TYPE_CHECKING:
     from switchbot_client import SwitchBotClient
 
+AnyRemoteDeviceStatus = TypeVar("AnyRemoteDeviceStatus", bound=PseudoRemoteDeviceStatus)
 
-class SwitchBotRemoteDevice(SwitchBotDevice):
+
+class SwitchBotRemoteDevice(SwitchBotDevice, Generic[AnyRemoteDeviceStatus]):
     def __init__(
         self,
         client: SwitchBotClient,
         device: APIRemoteDeviceObject,
-        pseudo_status: PseudoRemoteDeviceStatus,
+        pseudo_status: AnyRemoteDeviceStatus,
     ):
         super().__init__(
             client,
@@ -41,7 +43,7 @@ def turn_off(self) -> SwitchBotCommandResult:
         self.pseudo_status.set_power("off")
         return response
 
-    def status(self) -> PseudoRemoteDeviceStatus:
+    def status(self) -> AnyRemoteDeviceStatus:
         return self.pseudo_status
 
     @staticmethod
@@ -116,7 +118,7 @@ def _validate_pseudo_status(self):
             )
 
 
-class AirConditioner(SwitchBotRemoteDevice):
+class AirConditioner(SwitchBotRemoteDevice[PseudoAirConditionerStatus]):
     class Parameters:
         MODE_AUTO = 1
         MODE_COOL = 2
@@ -131,40 +133,46 @@ class Parameters:
         POWER_OFF = "off"
 
     def __init__(self, client: SwitchBotClient, device: APIRemoteDeviceObject):
-        pseudo_status = PseudoRemoteDeviceStatus(
+        pseudo_status = PseudoAirConditionerStatus(
             device_id=device["deviceId"],
             device_type=device["remoteType"],
             device_name=device["deviceName"],
             hub_device_id=device["hubDeviceId"],
             power=None,
             raw_data={},
+            temperature=25.0,
+            mode=AirConditioner.Parameters.MODE_AUTO,
+            fan_speed=AirConditioner.Parameters.FAN_SPEED_AUTO,
         )
         super().__init__(client, device, pseudo_status)
         self._check_remote_type(RemoteType.AIR_CONDITIONER)
 
-        # remote devices don't have status fetch commands
-        # we can only memorize recently applied parameter and use it
-        self.temperature_memory = 25.0
-        self.mode_memory = AirConditioner.Parameters.MODE_AUTO
-        self.fan_speed_memory = AirConditioner.Parameters.FAN_SPEED_AUTO
-
     @staticmethod
     def create_by_id(client: SwitchBotClient, device_id: str) -> AirConditioner:
         device = SwitchBotRemoteDevice.get_device_by_id(client, device_id)
         return AirConditioner(client, device)
 
     def set_all(
-        self, temperature: float, mode: int, fan_speed: int, power: str
+        self,
+        temperature: Optional[float],
+        mode: Optional[int],
+        fan_speed: Optional[int],
+        power: Optional[str],
     ) -> SwitchBotCommandResult:
         """
         temperature: temperature in celsius
         mode(Parameters.MODE_XXX): 1(auto), 2(cool), 3(dry), 4(fan), 5(heat)
         fan_speed(Parameters.FAN_SPEED_XXX): 1(auto), 2(low), 3(medium), 4(high)
         power(Parameters.POWER_XXX): on, off
         """
-        self.temperature_memory = temperature
-        self.mode_memory = mode
-        self.fan_speed_memory = fan_speed
+        if temperature is not None:
+            self.pseudo_status.set_temperature(temperature)
+        if mode is not None:
+            self.pseudo_status.set_mode(mode)
+        if fan_speed is not None:
+            self.pseudo_status.set_fan_speed(fan_speed)
+        if power is not None:
+            self.pseudo_status.set_power(power)
         return self.command(
             ControlCommand.VirtualInfrared.SET_ALL,
             parameter=f"{temperature},{mode},{fan_speed},{power}",
@@ -182,7 +190,9 @@ def set_temperature(self, temperature: float) -> SwitchBotCommandResult:
         It is recommended to turn on AirConditioner with the set_all
         with all values specified before use this function.
         """
-        return self.set_all(temperature, self.mode_memory, self.fan_speed_memory, "on")
+        return self.set_all(
+            temperature, self.pseudo_status.mode, self.pseudo_status.fan_speed, "on"
+        )
 
     def set_mode(self, mode: int) -> SwitchBotCommandResult:
         """
@@ -196,7 +206,9 @@ def set_mode(self, mode: int) -> SwitchBotCommandResult:
         It is recommended to turn on AirConditioner with the set_all
         with all values specified before use this function.
         """
-        return self.set_all(self.temperature_memory, mode, self.fan_speed_memory, "on")
+        return self.set_all(
+            self.pseudo_status.temperature, mode, self.pseudo_status.fan_speed, "on"
+        )
 
     def set_fan_speed(self, fan_speed: int) -> SwitchBotCommandResult:
         """
@@ -210,7 +222,9 @@ def set_fan_speed(self, fan_speed: int) -> SwitchBotCommandResult:
         It is recommended to turn on AirConditioner with the set_all
         with all values specified before use this function.
         """
-        return self.set_all(self.temperature_memory, self.mode_memory, fan_speed, "on")
+        return self.set_all(
+            self.pseudo_status.temperature, self.pseudo_status.mode, fan_speed, "on"
+        )
 
 
 class TV(SwitchBotRemoteDevice):