Add support for automatic watering system

letpot/client.py

@@ -1,7 +1,8 @@
 """Python client for LetPot hydroponic gardens."""
 
 import time
-from aiohttp import ClientSession, ClientResponse
+
+from aiohttp import ClientResponse, ClientSession
 
 from letpot.exceptions import LetPotAuthenticationException, LetPotConnectionException
 

letpot/converters.py

@@ -3,23 +3,28 @@
 import logging
 import math
 from abc import ABC, abstractmethod
-from datetime import time
+from datetime import datetime, time, timedelta
 from typing import Sequence
 
 from aiomqtt.types import PayloadType
 
-from letpot.exceptions import LetPotException
+from letpot.exceptions import LetPotDeviceCategoryException, LetPotException
 from letpot.models import (
+    CycleWateringMode,
     DeviceFeature,
     LetPotDeviceErrors,
     LetPotDeviceStatus,
+    LetPotGardenStatus,
+    LetPotWateringSystemStatus,
     LightMode,
     TemperatureUnit,
+    WateringReason,
 )
 
 _LOGGER = logging.getLogger(__name__)
 
 MODEL_AIR = ("LetPot Air", "LPH-AIR")
+MODEL_DI = ("LetPot Automatic Watering System", "DI")
 MODEL_MAX = ("LetPot Max", "LPH-MAX")
 MODEL_MINI = ("LetPot Mini", "LPH-MINI")
 MODEL_PRO = ("LetPot Pro", "LPH-PRO")
@@ -112,6 +117,8 @@ def get_current_status_message(self) -> list[int]:
         return [97, 1]
 
     def get_update_status_message(self, status: LetPotDeviceStatus) -> list[int]:
+        if not isinstance(status, LetPotGardenStatus):
+            raise LetPotDeviceCategoryException()
         return [
             97,
             2,
@@ -142,7 +149,7 @@ def convert_hex_to_status(self, message: PayloadType) -> LetPotDeviceStatus | No
         else:
             error_pump_malfunction = True if data[7] & 2 else False
 
-        return LetPotDeviceStatus(
+        return LetPotGardenStatus(
             raw=data,
             light_brightness=256 * data[17] + data[18],
             light_mode=LightMode(data[10]),
@@ -189,6 +196,8 @@ def get_current_status_message(self) -> list[int]:
         return [11, 1]
 
     def get_update_status_message(self, status: LetPotDeviceStatus) -> list[int]:
+        if not isinstance(status, LetPotGardenStatus):
+            raise LetPotDeviceCategoryException()
         return [
             11,
             2,
@@ -215,7 +224,7 @@ def convert_hex_to_status(self, message: PayloadType) -> LetPotDeviceStatus | No
         else:
             error_low_water = True if data[7] & 1 else False
 
-        return LetPotDeviceStatus(
+        return LetPotGardenStatus(
             raw=data,
             light_brightness=None,
             light_mode=LightMode(data[10]),
@@ -262,6 +271,8 @@ def get_current_status_message(self) -> list[int]:
         return [13, 1]
 
     def get_update_status_message(self, status: LetPotDeviceStatus) -> list[int]:
+        if not isinstance(status, LetPotGardenStatus):
+            raise LetPotDeviceCategoryException()
         return [
             13,
             2,
@@ -290,7 +301,7 @@ def convert_hex_to_status(self, message: PayloadType) -> LetPotDeviceStatus | No
             _LOGGER.debug("Invalid message received, ignoring: %s", message)
             return None
 
-        return LetPotDeviceStatus(
+        return LetPotGardenStatus(
             raw=data,
             light_brightness=256 * data[18] + data[19],
             light_mode=LightMode(data[10]),
@@ -343,6 +354,8 @@ def get_current_status_message(self) -> list[int]:
         return [101, 1]
 
     def get_update_status_message(self, status: LetPotDeviceStatus) -> list[int]:
+        if not isinstance(status, LetPotGardenStatus):
+            raise LetPotDeviceCategoryException()
         return [
             101,
             2,
@@ -368,7 +381,7 @@ def convert_hex_to_status(self, message: PayloadType) -> LetPotDeviceStatus | No
             _LOGGER.debug("Invalid message received, ignoring: %s", message)
             return None
 
-        return LetPotDeviceStatus(
+        return LetPotGardenStatus(
             raw=data,
             light_brightness=256 * data[18] + data[19],
             light_mode=LightMode(data[10]),
@@ -396,9 +409,100 @@ def get_light_brightness_levels(self) -> list[int]:
         return [0, 125, 250, 375, 500, 625, 750, 875, 1000]
 
 
+class ISEConverter(LetPotDeviceConverter):
+    """Converters and info for device type ISE05, ISE06 (Automatic Watering System)."""
+
+    @staticmethod
+    def supports_type(device_type: str) -> bool:
+        return device_type in ["ISE05", "ISE06"]
+
+    def get_device_model(self) -> tuple[str, str] | None:
+        return MODEL_DI
+
+    def supported_features(self) -> DeviceFeature:
+        return DeviceFeature.CATEGORY_WATERING_SYSTEM
+
+    def get_current_status_message(self) -> list[int]:
+        return [65, 1]
+
+    def get_update_status_message(self, status: LetPotDeviceStatus) -> list[int]:
+        if not isinstance(status, LetPotWateringSystemStatus):
+            raise LetPotDeviceCategoryException()
+        return [
+            65,
+            2,
+            1 if status.pump_mode > 0 else 0,
+            1 if status.pump_cycle_on is True else 0,
+            math.floor((status.pump_manual_duration or 0) / 256),
+            (status.pump_manual_duration or 0) % 256,
+            math.floor((status.pump_cycle_frequency or 0) / 256),
+            (status.pump_cycle_frequency or 0) % 256,
+            math.floor((status.pump_cycle_duration or 0) / 256),
+            (status.pump_cycle_duration or 0) % 256,
+            status.pump_cycle_mode or 0,
+            math.floor((status.pump_cycle_workinginterval or 0) / 256),
+            (status.pump_cycle_workinginterval or 0) % 256,
+            math.floor((status.pump_cycle_restinterval or 0) / 256),
+            (status.pump_cycle_restinterval or 0) % 256,
+        ]
+
+    def convert_hex_to_status(self, message: PayloadType) -> LetPotDeviceStatus | None:
+        data = self._hex_bytes_to_int_array(message)
+        if data is None or data[4] != 66 or data[5] != 1:
+            _LOGGER.debug("Invalid message received, ignoring: %s", message)
+            return None
+
+        if self._device_type == "ISE05":
+            pump_cycle_skipwater = None
+        else:
+            pump_cycle_skipwater = math.floor((256 * data[35] + data[36]) / 60)
+
+        now = datetime.now()
+        if (seconds := int.from_bytes(data[12:16], byteorder="big")) == 0:
+            pump_works_end = None
+        else:
+            pump_works_end = now + timedelta(seconds=seconds)
+
+        if (seconds := int.from_bytes(data[27:31], byteorder="big")) == 0:
+            pump_works_latest_time = None
+        else:
+            pump_works_latest_time = now - timedelta(seconds=seconds)
+
+        if (seconds := int.from_bytes(data[31:35], byteorder="big")) == 0:
+            pump_works_next_time = None
+        else:
+            pump_works_next_time = now + timedelta(seconds=seconds)
+
+        return LetPotWateringSystemStatus(
+            raw=data,
+            pump_mode=data[9],
+            errors=LetPotDeviceErrors(
+                low_water=True if data[7] & 1 else False,
+            ),
+            wifi_state=data[6],
+            pump_on=data[8] == 1,
+            pump_manual_duration=256 * data[10] + data[11],
+            pump_cycle_on=data[16] == 1,
+            pump_cycle_frequency=256 * data[17] + data[18],
+            pump_cycle_duration=256 * data[19] + data[20],
+            pump_cycle_mode=CycleWateringMode(data[21]),
+            pump_cycle_workinginterval=256 * data[22] + data[23],
+            pump_cycle_restinterval=256 * data[24] + data[25],
+            pump_works_end=pump_works_end,
+            pump_works_latest_reason=WateringReason(data[26]),
+            pump_works_latest_time=pump_works_latest_time,
+            pump_works_next_time=pump_works_next_time,
+            pump_cycle_skip_water=pump_cycle_skipwater,
+        )
+
+    def get_light_brightness_levels(self) -> list[int]:
+        return []
+
+
 CONVERTERS: Sequence[type[LetPotDeviceConverter]] = [
     LPHx1Converter,
     IGSorAltConverter,
     LPH6xConverter,
     LPH63Converter,
+    ISEConverter,
 ]

letpot/deviceclient.py

@@ -18,6 +18,7 @@
 from letpot.exceptions import (
     LetPotAuthenticationException,
     LetPotConnectionException,
+    LetPotDeviceCategoryException,
     LetPotException,
     LetPotFeatureException,
 )
@@ -26,6 +27,7 @@
     DeviceFeature,
     LetPotDeviceInfo,
     LetPotDeviceStatus,
+    LetPotGardenStatus,
     LightMode,
     TemperatureUnit,
 )
@@ -406,15 +408,19 @@ async def set_light_brightness(self, serial: str, level: int) -> None:
                 f"Device doesn't support setting light brightness to {level}"
             )
 
-        status = dataclasses.replace(
-            self._get_publish_status(serial), light_brightness=level
-        )
+        use_status = self._get_publish_status(serial)
+        if not isinstance(use_status, LetPotGardenStatus):
+            raise LetPotDeviceCategoryException()
+        status = dataclasses.replace(use_status, light_brightness=level)
         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)
+        use_status = self._get_publish_status(serial)
+        if not isinstance(use_status, LetPotGardenStatus):
+            raise LetPotDeviceCategoryException()
+        status = dataclasses.replace(use_status, light_mode=mode)
         await self._publish_status(serial, status)
 
     @requires_feature(DeviceFeature.CATEGORY_HYDROPONIC_GARDEN)
@@ -423,6 +429,8 @@ async def set_light_schedule(
     ) -> None:
         """Set the light schedule for this device (start time and/or end time)."""
         use_status = self._get_publish_status(serial)
+        if not isinstance(use_status, LetPotGardenStatus):
+            raise LetPotDeviceCategoryException()
         start_time = use_status.light_schedule_start if start is None else start
         end_time = use_status.light_schedule_end if end is None else end
         status = dataclasses.replace(
@@ -435,12 +443,19 @@ async def set_light_schedule(
     @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)
+        use_status = self._get_publish_status(serial)
+        if not isinstance(use_status, LetPotGardenStatus):
+            raise LetPotDeviceCategoryException()
+        status = dataclasses.replace(use_status, plant_days=days)
         await self._publish_status(serial, status)
 
+    @requires_feature(DeviceFeature.CATEGORY_HYDROPONIC_GARDEN)
     async def set_power(self, serial: str, on: bool) -> None:
         """Set the general power for this device (on/off)."""
-        status = dataclasses.replace(self._get_publish_status(serial), system_on=on)
+        use_status = self._get_publish_status(serial)
+        if not isinstance(use_status, LetPotGardenStatus):
+            raise LetPotDeviceCategoryException()
+        status = dataclasses.replace(use_status, system_on=on)
         await self._publish_status(serial, status)
 
     async def set_pump_mode(self, serial: str, on: bool) -> None:
@@ -453,23 +468,28 @@ async def set_pump_mode(self, serial: str, on: bool) -> None:
     @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)
+        use_status = self._get_publish_status(serial)
+        if not isinstance(use_status, LetPotGardenStatus):
+            raise LetPotDeviceCategoryException()
+        status = dataclasses.replace(use_status, 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
-        )
+        use_status = self._get_publish_status(serial)
+        if not isinstance(use_status, LetPotGardenStatus):
+            raise LetPotDeviceCategoryException()
+        status = dataclasses.replace(use_status, 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(
-            self._get_publish_status(serial), water_mode=1 if on else 0
-        )
+        use_status = self._get_publish_status(serial)
+        if not isinstance(use_status, LetPotGardenStatus):
+            raise LetPotDeviceCategoryException()
+        status = dataclasses.replace(use_status, water_mode=1 if on else 0)
         await self._publish_status(serial, status)
 
     # endregion

letpot/exceptions.py

@@ -15,3 +15,7 @@ class LetPotAuthenticationException(LetPotException):
 
 class LetPotFeatureException(LetPotException):
     """LetPot device feature exception."""
+
+
+class LetPotDeviceCategoryException(LetPotFeatureException):
+    """LetPot incorrect device type exception."""