Add feature exceptions + add feature: category hydroponic garden (#20)

* Add feature exceptions, add feature: category hydroponic garden

 - To prepare for other device categories, add a DeviceFeature for the 'hydroponic garden' category
 - Introduce LetPotFeatureException for exceptions related to features
 - Annotate device client functions to require certain features

* Cleanup imports

Author: jpelgrom

letpot/converters.py

@@ -1,19 +1,20 @@
 """Python client for LetPot hydroponic gardens."""
 
-from abc import ABC, abstractmethod
-from datetime import time
 import logging
 import math
+from abc import ABC, abstractmethod
+from datetime import time
 from typing import Sequence
+
 from aiomqtt.types import PayloadType
 
 from letpot.exceptions import LetPotException
 from letpot.models import (
     DeviceFeature,
-    LightMode,
-    TemperatureUnit,
     LetPotDeviceErrors,
     LetPotDeviceStatus,
+    LightMode,
+    TemperatureUnit,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -102,7 +103,7 @@ def get_device_model(self) -> tuple[str, str] | None:
             return None
 
     def supported_features(self) -> DeviceFeature:
-        features = DeviceFeature.PUMP_STATUS
+        features = DeviceFeature.CATEGORY_HYDROPONIC_GARDEN | DeviceFeature.PUMP_STATUS
         if self._device_type in ["LPH21", "LPH31"]:
             features |= DeviceFeature.LIGHT_BRIGHTNESS_LOW_HIGH
         return features
@@ -182,7 +183,7 @@ def get_device_model(self) -> tuple[str, str] | None:
             return None
 
     def supported_features(self) -> DeviceFeature:
-        return DeviceFeature(0)
+        return DeviceFeature.CATEGORY_HYDROPONIC_GARDEN
 
     def get_current_status_message(self) -> list[int]:
         return [11, 1]
@@ -246,7 +247,8 @@ def get_device_model(self) -> tuple[str, str] | None:
 
     def supported_features(self) -> DeviceFeature:
         features = (
-            DeviceFeature.LIGHT_BRIGHTNESS_LEVELS
+            DeviceFeature.CATEGORY_HYDROPONIC_GARDEN
+            | DeviceFeature.LIGHT_BRIGHTNESS_LEVELS
             | DeviceFeature.PUMP_AUTO
             | DeviceFeature.TEMPERATURE
             | DeviceFeature.TEMPERATURE_SET_UNIT
@@ -328,7 +330,8 @@ def get_device_model(self) -> tuple[str, str] | None:
 
     def supported_features(self) -> DeviceFeature:
         return (
-            DeviceFeature.LIGHT_BRIGHTNESS_LEVELS
+            DeviceFeature.CATEGORY_HYDROPONIC_GARDEN
+            | DeviceFeature.LIGHT_BRIGHTNESS_LEVELS
             | DeviceFeature.NUTRIENT_BUTTON
             | DeviceFeature.PUMP_AUTO
             | DeviceFeature.PUMP_STATUS

letpot/deviceclient.py

@@ -2,27 +2,32 @@
 
 import asyncio
 import dataclasses
-from datetime import time
-from hashlib import md5, sha256
 import logging
 import os
-import time as systime
 import ssl
-from typing import Callable
+import time as systime
+from collections.abc import Coroutine
+from datetime import time
+from functools import wraps
+from hashlib import md5, sha256
+from typing import Any, Callable, Concatenate
+
 import aiomqtt
 
 from letpot.converters import CONVERTERS, LetPotDeviceConverter
 from letpot.exceptions import (
     LetPotAuthenticationException,
     LetPotConnectionException,
     LetPotException,
+    LetPotFeatureException,
 )
 from letpot.models import (
     AuthenticationInfo,
+    DeviceFeature,
     LetPotDeviceInfo,
+    LetPotDeviceStatus,
     LightMode,
     TemperatureUnit,
-    LetPotDeviceStatus,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -38,6 +43,37 @@ def _create_ssl_context() -> ssl.SSLContext:
 _SSL_CONTEXT = _create_ssl_context()
 
 
+def requires_feature[T: "LetPotDeviceClient", _R, **P](
+    *required_feature: DeviceFeature,
+) -> Callable[
+    [Callable[Concatenate[T, str, P], Coroutine[Any, Any, _R]]],
+    Callable[Concatenate[T, str, P], Coroutine[Any, Any, _R]],
+]:
+    """Decorate the function to require device type support for a specific feature (inferred from serial)."""
+
+    def decorator(
+        func: Callable[Concatenate[T, str, P], Coroutine[Any, Any, _R]],
+    ) -> Callable[Concatenate[T, str, P], Coroutine[Any, Any, _R]]:
+        @wraps(func)
+        async def wrapper(
+            self: T, serial: str, *args: P.args, **kwargs: P.kwargs
+        ) -> _R:
+            exception_message = f"Device missing required feature: {required_feature}"
+            try:
+                supported_features = self._converter(serial).supported_features()
+                if not any(
+                    feature in supported_features for feature in required_feature
+                ):
+                    raise LetPotFeatureException(exception_message)
+            except LetPotException:
+                raise LetPotFeatureException(exception_message)
+            return await func(self, serial, *args, **kwargs)
+
+        return wrapper
+
+    return decorator
+
+
 class LetPotDeviceClient:
     """Client for connecting to LetPot device."""
 
@@ -356,10 +392,13 @@ async def get_current_status(self, serial: str) -> LetPotDeviceStatus | None:
         await status_event.wait()
         return self._device_status_last.get(serial)
 
+    @requires_feature(
+        DeviceFeature.LIGHT_BRIGHTNESS_LOW_HIGH, DeviceFeature.LIGHT_BRIGHTNESS_LEVELS
+    )
     async def set_light_brightness(self, serial: str, level: int) -> None:
         """Set the light brightness for this device (brightness level)."""
         if level not in self.get_light_brightness_levels(serial):
-            raise LetPotException(
+            raise LetPotFeatureException(
                 f"Device doesn't support setting light brightness to {level}"
             )
 
@@ -368,11 +407,13 @@ async def set_light_brightness(self, serial: str, level: int) -> None:
         )
         await self._publish_status(serial, status)
 
+    @requires_feature(DeviceFeature.CATEGORY_HYDROPONIC_GARDEN)
     async def set_light_mode(self, serial: str, mode: LightMode) -> None:
         """Set the light mode for this device (flower/vegetable)."""
         status = dataclasses.replace(self._get_publish_status(serial), light_mode=mode)
         await self._publish_status(serial, status)
 
+    @requires_feature(DeviceFeature.CATEGORY_HYDROPONIC_GARDEN)
     async def set_light_schedule(
         self, serial: str, start: time | None, end: time | None
     ) -> None:
@@ -387,6 +428,7 @@ async def set_light_schedule(
         )
         await self._publish_status(serial, status)
 
+    @requires_feature(DeviceFeature.CATEGORY_HYDROPONIC_GARDEN)
     async def set_plant_days(self, serial: str, days: int) -> None:
         """Set the plant days counter for this device (number of days)."""
         status = dataclasses.replace(self._get_publish_status(serial), plant_days=days)
@@ -404,18 +446,21 @@ async def set_pump_mode(self, serial: str, on: bool) -> None:
         )
         await self._publish_status(serial, status)
 
+    @requires_feature(DeviceFeature.CATEGORY_HYDROPONIC_GARDEN)
     async def set_sound(self, serial: str, on: bool) -> None:
         """Set the alarm sound for this device (on/off)."""
         status = dataclasses.replace(self._get_publish_status(serial), system_sound=on)
         await self._publish_status(serial, status)
 
+    @requires_feature(DeviceFeature.TEMPERATURE_SET_UNIT)
     async def set_temperature_unit(self, serial: str, unit: TemperatureUnit) -> None:
         """Set the temperature unit for this device (Celsius/Fahrenheit)."""
         status = dataclasses.replace(
             self._get_publish_status(serial), temperature_unit=unit
         )
         await self._publish_status(serial, status)
 
+    @requires_feature(DeviceFeature.PUMP_AUTO)
     async def set_water_mode(self, serial: str, on: bool) -> None:
         """Set the automatic water/nutrient mode for this device (on/off)."""
         status = dataclasses.replace(

letpot/exceptions.py

@@ -11,3 +11,7 @@ class LetPotConnectionException(LetPotException):
 
 class LetPotAuthenticationException(LetPotException):
     """LetPot authentication exception."""
+
+
+class LetPotFeatureException(LetPotException):
+    """LetPot device feature exception."""

letpot/models.py

@@ -1,14 +1,17 @@
 """Models for Python client for LetPot hydroponic gardens."""
 
+import time as systime
 from dataclasses import dataclass
 from datetime import time
 from enum import IntEnum, IntFlag, auto
-import time as systime
 
 
 class DeviceFeature(IntFlag):
     """Features that a LetPot device can support."""
 
+    CATEGORY_HYDROPONIC_GARDEN = auto()
+    """Features common to the hydroponic garden device category."""
+
     LIGHT_BRIGHTNESS_LOW_HIGH = auto()
     LIGHT_BRIGHTNESS_LEVELS = auto()
     NUTRIENT_BUTTON = auto()

tests/__init__.py

@@ -1,6 +1,13 @@
 """Tests for Python client for LetPot hydroponic gardens."""
 
-from letpot.models import AuthenticationInfo
+from datetime import time
+
+from letpot.models import (
+    AuthenticationInfo,
+    LetPotDeviceErrors,
+    LetPotDeviceStatus,
+    LightMode,
+)
 
 AUTHENTICATION = AuthenticationInfo(
     access_token="access_token",
@@ -10,3 +17,24 @@
     user_id="a1b2c3d4e5f6a1b2c3d4e5f6",
     email="[email protected]",
 )
+
+
+DEVICE_STATUS = LetPotDeviceStatus(
+    errors=LetPotDeviceErrors(low_water=True),
+    light_brightness=500,
+    light_mode=LightMode.VEGETABLE,
+    light_schedule_end=time(17, 0),
+    light_schedule_start=time(7, 30),
+    online=True,
+    plant_days=0,
+    pump_mode=1,
+    pump_nutrient=None,
+    pump_status=0,
+    raw=[77, 0, 1, 18, 98, 1, 0, 1, 1, 1, 1, 0, 0, 7, 30, 17, 0, 1, 244, 0, 0, 0],
+    system_on=True,
+    system_sound=False,
+    temperature_unit=None,
+    temperature_value=None,
+    water_mode=None,
+    water_level=None,
+)

tests/test_converter.py

@@ -1,13 +1,11 @@
 """Tests for the converters."""
 
-from datetime import time
-
 import pytest
 
-from letpot.converters import CONVERTERS, LPHx1Converter, LetPotDeviceConverter
+from letpot.converters import CONVERTERS, LetPotDeviceConverter, LPHx1Converter
 from letpot.exceptions import LetPotException
-from letpot.models import LetPotDeviceErrors, LetPotDeviceStatus, LightMode
 
+from . import DEVICE_STATUS
 
 SUPPORTED_DEVICE_TYPES = [
     "IGS01",
@@ -22,25 +20,6 @@
     "LPH62",
     "LPH63",
 ]
-DEVICE_STATUS = LetPotDeviceStatus(
-    errors=LetPotDeviceErrors(low_water=True),
-    light_brightness=500,
-    light_mode=LightMode.VEGETABLE,
-    light_schedule_end=time(17, 0),
-    light_schedule_start=time(7, 30),
-    online=True,
-    plant_days=0,
-    pump_mode=1,
-    pump_nutrient=None,
-    pump_status=0,
-    raw=[77, 0, 1, 18, 98, 1, 0, 1, 1, 1, 1, 0, 0, 7, 30, 17, 0, 1, 244, 0, 0, 0],
-    system_on=True,
-    system_sound=False,
-    temperature_unit=None,
-    temperature_value=None,
-    water_mode=None,
-    water_level=None,
-)
 
 
 @pytest.mark.parametrize(