Add Special Status support (#42)

* Add support for SpecialStatus in DeviceStatus

* SpecialStatus and set status

aioaquarea/__init__.py

@@ -19,6 +19,7 @@
     UpdateOperationMode,
     HolidayTimer,
     PowerfulTime,
+    SpecialStatus
 )
 from .errors import (
     ApiError,
@@ -58,4 +59,5 @@
     "HolidayTimer",
     "PowerfulTime",
     "AquareaEnvironment",
+    "SpecialStatus",
 )

aioaquarea/core.py

@@ -37,10 +37,12 @@
     PowerfulTime,
     QuietMode,
     SensorMode,
+    SpecialStatus,
     Tank,
     TankStatus,
     UpdateOperationMode,
     ZoneSensor,
+    ZoneTemperatureSetUpdate,
 )
 from .errors import (
     ApiError,
@@ -362,6 +364,12 @@ async def get_device_status(self, long_id: str) -> DeviceStatus:
         device = data.get("status")[0]
         operation_mode_value = device.get("operationMode")
 
+        enabled_special_modes = [
+            mode["specialMode"]
+            for mode in device.get("specialStatus", [])
+            if mode.get("operationStatus") == 1
+        ]
+
         device_status = DeviceStatus(
             long_id=long_id,
             operation_status=OperationStatus(device.get("operationStatus")),
@@ -388,15 +396,19 @@ async def get_device_status(self, long_id: str) -> DeviceStatus:
             ],
             zones=[
                 DeviceZoneStatus(
-                    zone_status["zoneId"],
-                    zone_status["temparatureNow"],
-                    OperationStatus(zone_status["operationStatus"]),
-                    zone_status["heatMax"],
-                    zone_status["heatMin"],
-                    zone_status["heatSet"],
-                    zone_status["coolMax"],
-                    zone_status["coolMin"],
-                    zone_status["coolSet"],
+                    zone_id=zone_status["zoneId"],
+                    temperature=zone_status["temparatureNow"],
+                    operation_status=OperationStatus(zone_status["operationStatus"]),
+                    heat_max=zone_status["heatMax"],
+                    heat_min=zone_status["heatMin"],
+                    heat_set=zone_status["heatSet"],
+                    cool_max=zone_status["coolMax"],
+                    cool_min=zone_status["coolMin"],
+                    cool_set=zone_status["coolSet"],
+                    comfort_cool=zone_status["comfortCool"],
+                    comfort_heat=zone_status["comfortHeat"],
+                    eco_cool=zone_status["ecoCool"],
+                    eco_heat=zone_status["ecoHeat"],
                 )
                 for zone_status in device.get("zoneStatus", [])
             ],
@@ -405,6 +417,9 @@ async def get_device_status(self, long_id: str) -> DeviceStatus:
             force_heater=ForceHeater(device.get("forceHeater", 0)),
             holiday_timer=HolidayTimer(device.get("holidayTimer", 0)),
             powerful_time=PowerfulTime(device.get("powerful", 0)),
+            special_status=SpecialStatus(enabled_special_modes[0])
+            if enabled_special_modes
+            else None,
         )
 
         return device_status
@@ -521,7 +536,7 @@ async def post_device_operation_update(
         long_id: str,
         mode: UpdateOperationMode,
         zones: dict[int, OperationStatus],
-        operation_status: OperationStatus.ON,
+        operation_status: OperationStatus,
     ) -> None:
         """Post device operation update."""
         data = {
@@ -549,6 +564,39 @@ async def post_device_operation_update(
             json=data,
         )
 
+    @auth_required
+    async def post_device_set_special_status(
+        self,
+        long_id: str,
+        special_status: SpecialStatus | None,
+        zones: list[ZoneTemperatureSetUpdate]
+    ) -> None:
+        """Post device operation update."""
+        data = {
+            "status": [
+                {
+                    "deviceGuid": long_id,
+                    "specialStatus": special_status.value if special_status else 0,
+                    "zoneStatus": [
+                        {
+                            "zoneId": zone.zone_id,
+                            "heatSet": zone.heat_set,
+                            **({"coolSet": zone.cool_set} if zone.cool_set is not None else {})
+                        }
+                        for zone in zones
+                    ],
+                }
+            ]
+        }
+
+        response = await self.request(
+            "POST",
+            f"{AQUAREA_SERVICE_DEVICES}/{long_id}",
+            referer=f"{self._base_url}{AQUAREA_SERVICE_A2W_STATUS_DISPLAY}",
+            content_type="application/json",
+            json=data,
+        )
+
     async def post_device_zone_heat_temperature(
         self, long_id: str, zone_id: int, temperature: int
     ) -> None:
@@ -916,3 +964,10 @@ async def set_powerful_time(self, powerful_time: PowerfulTime) -> None:
             await self._client.post_device_set_powerful_time(
                 self.long_id, powerful_time
             )
+
+    async def __set_special_status__(self, special_status: SpecialStatus | None, zones: list[ZoneTemperatureSetUpdate]) -> None:
+        """Set the special status.
+        :param special_status: Special status to set
+        :param zones: Zones to set the special status for
+        """
+        await self._client.post_device_set_special_status(self.long_id, special_status, zones)

aioaquarea/data.py

@@ -8,7 +8,7 @@
 
 from .const import PANASONIC
 from .statistics import Consumption, ConsumptionType
-from .util import LimitedSizeDict
+from .util import LimitedSizeDict, limit_range
 
 try:
     from enum import StrEnum
@@ -130,6 +130,17 @@ class PowerfulTime(IntEnum):
     ON_90MIN = 3
 
 
+class SpecialStatus(IntEnum):
+    """Special status"""
+
+    ECO = 1
+    COMFORT = 2
+
+@dataclass
+class TemperatureModifiers:
+    heat: int | None
+    cool: int | None
+
 @dataclass
 class TankStatus:
     """Tank status"""
@@ -173,6 +184,10 @@ class DeviceZoneStatus:
     cool_max: int | None
     cool_min: int | None
     cool_set: int | None
+    comfort_heat: int | None
+    comfort_cool: int | None
+    eco_heat: int | None
+    eco_cool: int | None
 
 
 @dataclass
@@ -190,7 +205,13 @@ class DeviceInfo:
 
 @dataclass()
 class DeviceStatus:
-    """Device status"""
+    """Device status
+
+    Parameters
+    ----------
+    special_status : SpecialStatus  | None
+        Current special status of the device. As of now it only supports one value at a time.
+    """
 
     long_id: str
     operation_status: OperationStatus
@@ -207,7 +228,14 @@ class DeviceStatus:
     force_heater: ForceHeater
     holiday_timer: HolidayTimer
     powerful_time: PowerfulTime
+    special_status: SpecialStatus | None
+
 
+@dataclass
+class ZoneTemperatureSetUpdate:
+    zone_id: int
+    cool_set: int | None
+    heat_set: int | None
 
 @dataclass
 class OperationStatusUpdate:
@@ -235,6 +263,12 @@ def __init__(self, info: DeviceZoneInfo, status: DeviceZoneStatus) -> None:
         self._info = info
         self._status = status
 
+        if self.supports_special_status:
+            self._temperature_modifiers = {
+                SpecialStatus.ECO: TemperatureModifiers(self._status.eco_heat, self._status.eco_cool),
+                SpecialStatus.COMFORT: TemperatureModifiers(self._status.comfort_heat, self._status.comfort_cool)
+            }
+
     @property
     def zone_id(self) -> int:
         """Zone ID"""
@@ -315,6 +349,26 @@ def supports_set_temperature(self) -> bool:
         """Gets if the zone supports setting the temperature"""
         return self.sensor_mode != ZoneSensor.EXTERNAL
 
+    @property
+    def supports_special_status(self) -> bool:
+        """Gets if the zone supports special status"""
+        return self.sensor_mode != ZoneSensor.EXTERNAL
+
+    @property
+    def eco(self) -> TemperatureModifiers:
+        """Gets the eco temperature modifiers for the zone"""
+        return self.temperature_modifiers[SpecialStatus.ECO]
+
+    @property
+    def comfort(self) -> TemperatureModifiers:
+        """Gets the confort temperature modifiers for the zone"""
+        return self.temperature_modifiers[SpecialStatus.COMFORT]
+
+    @property
+    def temperature_modifiers(self) -> dict[SpecialStatus, TemperatureModifiers]:
+        """Gets the temperature modifiers for the zone"""
+        return self._temperature_modifiers
+
 
 class Tank(ABC):
     """Tank"""
@@ -546,11 +600,74 @@ def powerful_time(self) -> PowerfulTime:
         """Specifies if the powerful time is enabled and for how long"""
         return self._status.powerful_time
 
+    @property
+    def special_status(self) -> SpecialStatus | None:
+        """Specifies if the device is in a special status"""
+        return self._status.special_status
+
     def support_cooling(self, zone_id: int = 1) -> bool:
         """True if the device supports cooling in the given zone"""
         zone = self.zones.get(zone_id, None)
         return zone is not None and zone.cool_mode
 
+    @property
+    def support_special_status(self) -> bool:
+        """True if the device supports special status"""
+        return any(zone.supports_special_status for zone in self.zones.values())
+
+    async def set_special_status(self, special_status: SpecialStatus | None) -> None:
+        """Set the special status.
+        :param special_status: Special status to set
+        """
+
+        if not self.support_special_status:
+            raise Exception("Device does not support special status")
+
+        if self.special_status == special_status:
+            return
+
+        zones: list[ZoneTemperatureSetUpdate] = [
+            self.__calculate_zone_special_status_update__(zone, special_status)
+            for zone in self.zones.values()
+        ]
+
+        await self.__set_special_status__(special_status, zones)
+
+    @abstractmethod
+    async def __set_special_status__(special_status: SpecialStatus | None, zones: list[ZoneTemperatureSetUpdate]) -> None:
+        """Set the special status.
+        :param special_status: Special status to set
+        :param zones: Zones to set the special status for
+        """
+
+    def __calculate_zone_special_status_update__(
+        self, zone: DeviceZone, special_status: SpecialStatus | None
+    ) -> ZoneTemperatureSetUpdate:
+        """Calculate the zone temperature set update based on the special status.
+        :param zone: The zone for which to calculate the update
+        :param special_status: The special status to set
+        :return: The zone temperature set update
+        """
+
+        current_status = self.special_status
+        cool_set = zone.cool_target_temperature
+        heat_set = zone.heat_target_temperature
+
+        """If the zone is already on a special status, we need to revert the temperature to normal first"""
+        if current_status is not None:
+            modifiers = zone.temperature_modifiers[current_status]
+            cool_set = limit_range(cool_set - modifiers.cool, zone.cool_min, zone.cool_max) if cool_set is not None else None
+            heat_set = limit_range(heat_set - modifiers.heat, zone.heat_min, zone.heat_max) if heat_set is not None else None
+
+        """If we're setting a special status, we need to apply the modifiers"""
+        if special_status is not None:
+            modifiers = zone.temperature_modifiers[special_status]
+            cool_set = limit_range(cool_set + modifiers.cool, zone.cool_min, zone.cool_max) if cool_set is not None else None
+            heat_set = limit_range(heat_set + modifiers.heat, zone.heat_min, zone.heat_max) if heat_set is not None else None
+
+        return ZoneTemperatureSetUpdate(zone.zone_id, cool_set, heat_set)
+
+
     @abstractmethod
     async def __set_operation_status__(self, status: OperationStatus) -> None:
         """Set the operation status of the device"""

aioaquarea/util.py

@@ -1,6 +1,13 @@
 from collections import OrderedDict
 
-
+def limit_range(value, min_value, max_value):
+        if value < min_value:
+            return min_value
+        elif value > max_value:
+            return max_value
+        else:
+            return value
+
 class LimitedSizeDict(OrderedDict):
     def __init__(self, max_keys: int, *args, **kwds):
         self.size_limit = max_keys