"""Python client for LetPot hydroponic gardens.""" import logging import math from abc import ABC, abstractmethod from datetime import datetime, time, timedelta from typing import Sequence from aiomqtt.types import PayloadType 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_DI2 = ("LetPot Automatic Watering System 2.0", "DI-2") MODEL_DI3 = ("LetPot Automatic Watering System 2.0", "DI-3") MODEL_MAX = ("LetPot Max", "LPH-MAX") MODEL_MINI = ("LetPot Mini", "LPH-MINI") MODEL_PRO = ("LetPot Pro", "LPH-PRO") MODEL_SE = ("LetPot Senior", "LPH-SE") class LetPotDeviceConverter(ABC): """Base class for converters and info for device types.""" _device_type: str def __init__(self, device_type: str) -> None: """Initialize converter.""" if not self.supports_type(device_type): raise LetPotException("Initializing converter with unsupported device type") self._device_type = device_type @staticmethod @abstractmethod def supports_type(device_type: str) -> bool: """Returns if the converter supports the supplied type.""" pass @abstractmethod def get_device_model(self) -> tuple[str, str] | None: """Returns the device model name and code.""" @abstractmethod def supported_features(self) -> DeviceFeature: """Returns the device supported feature(s).""" @abstractmethod def get_current_status_message(self) -> list[int]: """Returns the message content for getting the current device status.""" pass @abstractmethod def convert_hex_to_status(self, message: PayloadType) -> LetPotDeviceStatus | None: """Converts a hexadecimal bytes status message to a status dataclass.""" pass @abstractmethod def get_update_status_message(self, status: LetPotDeviceStatus) -> list[int]: """Returns the message content for updating the device status.""" pass @abstractmethod def get_light_brightness_levels(self) -> list[int]: """Returns the brightness steps supported by the device for this converter.""" pass def _hex_bytes_to_int_array(self, hex_message: PayloadType) -> list[int] | None: """Converts a hexadecimal bytes message to a list of integers.""" if not isinstance(hex_message, bytes): return None try: decoded_hex = hex_message.decode("utf-8") integers = [] for n in range(0, len(decoded_hex), 2): integers.append(int(decoded_hex[n : n + 2], 16)) return integers except Exception as exception: raise LetPotException("Unable to convert from hex") from exception class LPHx1Converter(LetPotDeviceConverter): """Converters and info for device type LPH11 (Mini), LPH21, LPH22 (Air), LPH31, LPH32 (SE).""" @staticmethod def supports_type(device_type: str) -> bool: return device_type in ["LPH11", "LPH21", "LPH22", "LPH31", "LPH32"] def get_device_model(self) -> tuple[str, str] | None: if self._device_type == "LPH11": return MODEL_MINI elif self._device_type in ["LPH21", "LPH22"]: return MODEL_AIR elif self._device_type in ["LPH31", "LPH32"]: return MODEL_SE else: return None def supported_features(self) -> DeviceFeature: features = DeviceFeature.CATEGORY_HYDROPONIC_GARDEN | DeviceFeature.PUMP_STATUS if self._device_type in ["LPH21", "LPH22", "LPH31", "LPH32"]: features |= DeviceFeature.LIGHT_BRIGHTNESS_LOW_HIGH return features 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, 1 if status.system_on else 0, status.pump_mode, status.light_mode.value, math.floor(status.plant_days / 256), status.plant_days % 256, status.light_schedule_start.hour, status.light_schedule_start.minute, status.light_schedule_end.hour, status.light_schedule_end.minute, math.floor(status.light_brightness / 256) if status.light_brightness is not None else 0, status.light_brightness % 256 if status.light_brightness is not None else 0, 1 if status.system_sound is True else 0, ] 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] != 98 or data[5] != 1: _LOGGER.debug("Invalid message received, ignoring: %s", message) return None if self._device_type in ["LPH21", "LPH22"]: error_pump_malfunction = None else: error_pump_malfunction = True if data[7] & 2 else False return LetPotGardenStatus( raw=data, light_brightness=256 * data[17] + data[18], light_mode=LightMode(data[10]), light_schedule_end=time(hour=data[15], minute=data[16]), light_schedule_start=time(hour=data[13], minute=data[14]), online=data[6] == 0, plant_days=256 * data[11] + data[12], pump_mode=data[9], pump_nutrient=None, pump_status=data[19], system_on=data[8] == 1, system_sound=data[20] == 1 if data[20] is not None else None, errors=LetPotDeviceErrors( low_water=True if data[7] & 1 else False, pump_malfunction=error_pump_malfunction, ), ) def get_light_brightness_levels(self) -> list[int]: return ( [500, 1000] if self._device_type in ["LPH21", "LPH22", "LPH31", "LPH32"] else [] ) class IGSorAltConverter(LetPotDeviceConverter): """Converters and info for device type IGS01 (Pro), LPH27, LPH37 (SE), LPH39 (Mini).""" @staticmethod def supports_type(device_type: str) -> bool: return device_type in ["IGS01", "LPH27", "LPH37", "LPH39"] def get_device_model(self) -> tuple[str, str] | None: if self._device_type == "IGS01": return MODEL_PRO elif self._device_type in ["LPH27", "LPH37"]: return MODEL_SE elif self._device_type == "LPH39": return MODEL_MINI else: return None def supported_features(self) -> DeviceFeature: return DeviceFeature.CATEGORY_HYDROPONIC_GARDEN 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, 1 if status.system_on else 0, status.pump_mode, status.light_mode.value, math.floor(status.plant_days / 256), status.plant_days % 256, status.light_schedule_start.hour, status.light_schedule_start.minute, status.light_schedule_end.hour, status.light_schedule_end.minute, 1 if status.system_sound is True else 0, ] 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] != 12 or data[5] != 1: _LOGGER.debug("Invalid message received, ignoring: %s", message) return None if self._device_type == "IGS01": error_low_water = None else: error_low_water = True if data[7] & 1 else False return LetPotGardenStatus( raw=data, light_brightness=None, light_mode=LightMode(data[10]), light_schedule_end=time(hour=data[15], minute=data[16]), light_schedule_start=time(hour=data[13], minute=data[14]), online=data[6] == 0, plant_days=256 * data[11] + data[12], pump_mode=data[9], pump_nutrient=None, pump_status=None, system_on=data[8] == 1, system_sound=data[17] == 1 if data[17] is not None else None, errors=LetPotDeviceErrors(low_water=error_low_water), ) def get_light_brightness_levels(self) -> list[int]: return [] class LPHMaxLowerConverter(LetPotDeviceConverter): """Converters and info for device type LPH60, LPH61, LPH62 (Max).""" @staticmethod def supports_type(device_type: str) -> bool: return device_type in ["LPH60", "LPH61", "LPH62"] def get_device_model(self) -> tuple[str, str] | None: return MODEL_MAX def supported_features(self) -> DeviceFeature: features = ( DeviceFeature.CATEGORY_HYDROPONIC_GARDEN | DeviceFeature.LIGHT_BRIGHTNESS_LEVELS | DeviceFeature.PUMP_AUTO | DeviceFeature.TEMPERATURE | DeviceFeature.TEMPERATURE_SET_UNIT | DeviceFeature.WATER_LEVEL ) if self._device_type != "LPH60": features |= DeviceFeature.NUTRIENT_BUTTON return features 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, 1 if status.system_on else 0, status.pump_mode, status.light_mode.value, math.floor(status.plant_days / 256), status.plant_days % 256, status.light_schedule_start.hour, status.light_schedule_start.minute, status.light_schedule_end.hour, status.light_schedule_end.minute, status.water_mode if status.water_mode is not None else 0, math.floor(status.light_brightness / 256) if status.light_brightness is not None else 0, status.light_brightness % 256 if status.light_brightness is not None else 0, 1 if status.temperature_unit is TemperatureUnit.CELSIUS else 0, 1 if status.system_sound is True else 0, 1 if status.pump_nutrient is True else 0, ] 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] != 14 or data[5] != 1: _LOGGER.debug("Invalid message received, ignoring: %s", message) return None return LetPotGardenStatus( raw=data, light_brightness=256 * data[18] + data[19], light_mode=LightMode(data[10]), light_schedule_end=time(hour=data[15], minute=data[16]), light_schedule_start=time(hour=data[13], minute=data[14]), online=data[6] == 0, plant_days=256 * data[11] + data[12], pump_mode=data[9], pump_nutrient=data[26] == 1, pump_status=None, system_on=data[8] == 1, system_sound=data[25] == 1 if data[25] is not None else None, errors=LetPotDeviceErrors( low_water=True if data[7] & 2 else False, low_nutrients=True if data[7] & 1 else False, refill_error=True if data[7] & 4 else False, ), temperature_unit=TemperatureUnit(data[24]), temperature_value=256 * data[22] + data[23], water_level=256 * data[20] + data[21], water_mode=data[17], ) def get_light_brightness_levels(self) -> list[int]: return [0, 125, 250, 375, 500, 625, 750, 875, 1000] class LPHMaxHigherConverter(LetPotDeviceConverter): """Converters and info for device type LPH63, LPH64 (Max).""" @staticmethod def supports_type(device_type: str) -> bool: return device_type in ["LPH63", "LPH64"] def get_device_model(self) -> tuple[str, str] | None: return MODEL_MAX def supported_features(self) -> DeviceFeature: return ( DeviceFeature.CATEGORY_HYDROPONIC_GARDEN | DeviceFeature.LIGHT_BRIGHTNESS_LEVELS | DeviceFeature.NUTRIENT_BUTTON | DeviceFeature.PUMP_AUTO | DeviceFeature.PUMP_STATUS | DeviceFeature.TEMPERATURE | DeviceFeature.WATER_LEVEL ) 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, 1 if status.system_on else 0, status.pump_mode, 0 if status.light_mode is LightMode.FLOWER else 1, math.floor(status.plant_days / 256), status.plant_days % 256, status.light_schedule_start.hour, status.light_schedule_start.minute, status.light_schedule_end.hour, status.light_schedule_end.minute, status.water_mode if status.water_mode is not None else 0, math.floor(status.light_brightness / 256) if status.light_brightness is not None else 0, status.light_brightness % 256 if status.light_brightness is not None else 0, ] 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] != 102 or data[5] != 1: _LOGGER.debug("Invalid message received, ignoring: %s", message) return None return LetPotGardenStatus( raw=data, light_brightness=256 * data[18] + data[19], light_mode=LightMode(data[10]), light_schedule_end=time(hour=data[15], minute=data[16]), light_schedule_start=time(hour=data[13], minute=data[14]), online=data[6] == 0, plant_days=256 * data[11] + data[12], pump_mode=data[9], pump_nutrient=None, pump_status=data[26], system_on=data[8] == 1, system_sound=None, errors=LetPotDeviceErrors( low_water=True if data[7] & 2 else False, low_nutrients=True if data[7] & 1 else False, refill_error=True if data[7] & 4 else False, ), temperature_unit=TemperatureUnit(data[24]), temperature_value=256 * data[22] + data[23], water_level=256 * data[20] + data[21], water_mode=data[17], ) 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 2.0).""" @staticmethod def supports_type(device_type: str) -> bool: return device_type in ["ISE05", "ISE06"] def get_device_model(self) -> tuple[str, str] | None: if self._device_type == "ISE05": return MODEL_DI2 elif self._device_type == "ISE06": return MODEL_DI3 else: return None 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 [] GARDEN_CONVERTERS: Sequence[type[LetPotDeviceConverter]] = [ LPHx1Converter, IGSorAltConverter, LPHMaxLowerConverter, LPHMaxHigherConverter, ] """Converters for hydroponic gardens providing `letpot.models.LetPotGardenStatus` as their status.""" WATERING_SYSTEM_CONVERTERS: Sequence[type[LetPotDeviceConverter]] = [ ISEConverter, ] """Converters for watering systems providing `letpot.models.LetPotWateringSystemStatus` as their status.""" CONVERTERS: Sequence[type[LetPotDeviceConverter]] = [ *GARDEN_CONVERTERS, *WATERING_SYSTEM_CONVERTERS, ] """Converters for all supported device types."""