feat: add WaterTreatment class for Vitoset Aqua

PyViCare/PyViCareDeviceConfig.py

@@ -17,6 +17,7 @@
 from PyViCare.PyViCareElectricalEnergySystem import ElectricalEnergySystem
 from PyViCare.PyViCareGateway import Gateway
 from PyViCare.PyViCareVentilationDevice import VentilationDevice
+from PyViCare.PyViCareWaterTreatment import WaterTreatment
 
 logger = logging.getLogger(__name__)
 logger.addHandler(logging.NullHandler())
@@ -80,6 +81,9 @@ def asGateway(self):
     def asVentilation(self):
         return VentilationDevice(self.service)
 
+    def asWaterTreatment(self):
+        return WaterTreatment(self.service)
+
     def getConfig(self):
         return self.service.accessor
 
@@ -120,6 +124,7 @@ def asAutoDetectDevice(self):
             (self.asRoomControl, r"E3_RoomControl|Smart_RoomControl", ["type:virtual;smartRoomControl"]),
             (self.asRoomSensor, r"E3_RoomSensor", ["type:climateSensor"]),
             (self.asRepeater, r"E3_Repeater", ["type:repeater"]),
+            (self.asWaterTreatment, r"VitosetAqua", ["type:waterTreatment"]),
             (self.asGateway, r"E3_TCU41_x04", ["type:gateway;TCU100"]),
             (self.asGateway, r"E3_TCU19_x05", ["type:gateway;TCU200"]),
             (self.asGateway, r"E3_TCU10_x07", ["type:gateway;TCU300"]),

PyViCare/PyViCareWaterTreatment.py

@@ -0,0 +1,118 @@
+from typing import Any
+
+from PyViCare.PyViCareDevice import Device
+from PyViCare.PyViCareUtils import PyViCareNotSupportedFeatureError, handleNotSupported
+
+
+_LEAK_SENSOR_SLOTS = 5
+
+
+class WaterTreatment(Device):
+    """Viessmann Vitoset Aqua water treatment station.
+
+    Combines water softener, leak detection (up to 5 sensors), consumption metering
+    and main shutoff valve in one device.
+    """
+
+    # --- Softener ---
+
+    @handleNotSupported
+    def getSaltDaysRemaining(self) -> int:
+        return int(self.getProperty("water.softener.salt.level.days")["properties"]["remaining"]["value"])
+
+    @handleNotSupported
+    def getLowSaltAlertDays(self) -> int:
+        return int(self.getProperty("water.softener.configuration.lowSaltAlert")["properties"]["lowLevelAlertDays"]["value"])
+
+    # --- Consumption ---
+
+    @handleNotSupported
+    def getCurrentFlow(self) -> float:
+        return float(self.getProperty("water.consumption.flow.current")["properties"]["value"]["value"])
+
+    @handleNotSupported
+    def getMaxFlow(self) -> float:
+        return float(self.getProperty("water.consumption.flow.max")["properties"]["value"]["value"])
+
+    @handleNotSupported
+    def getCurrentDayConsumption(self) -> int:
+        return int(self.getProperty("water.consumption.summary")["properties"]["currentDay"]["value"])
+
+    @handleNotSupported
+    def getLastSevenDaysConsumption(self) -> int:
+        return int(self.getProperty("water.consumption.summary")["properties"]["lastSevenDays"]["value"])
+
+    @handleNotSupported
+    def getTotalConsumption(self) -> int:
+        return int(self.getProperty("water.consumption.summary")["properties"]["total"]["value"])
+
+    # --- Leak Detection ---
+
+    def getLeakSensors(self) -> list[dict[str, Any]]:
+        """Return all connected leak sensors.
+
+        Each Vitoset Aqua exposes 5 sensor slots; only slots that report data
+        are returned.
+        """
+        sensors: list[dict[str, Any]] = []
+        for slot in range(_LEAK_SENSOR_SLOTS):
+            base = self._readProperties(f"water.leakDetection.sensors.leakage.{slot}")
+            if not base:
+                continue
+            id_props = self._readProperties(f"water.leakDetection.sensors.leakage.{slot}.id")
+            name_props = self._readProperties(f"water.leakDetection.sensors.leakage.{slot}.name")
+            battery_props = self._readProperties(f"water.leakDetection.sensors.leakage.{slot}.battery")
+            rssi_props = self._readProperties(f"water.leakDetection.sensors.leakage.{slot}.rssi")
+            hw_props = self._readProperties(f"water.leakDetection.sensors.leakage.{slot}.version.hardware")
+            sw_props = self._readProperties(f"water.leakDetection.sensors.leakage.{slot}.version.software")
+            sensors.append({
+                "slot": slot,
+                "status": base.get("status", {}).get("value"),
+                "leak_detected": base.get("value", {}).get("value"),
+                "id": id_props.get("value", {}).get("value"),
+                "name": name_props.get("name", {}).get("value"),
+                "battery_percent": battery_props.get("level", {}).get("value"),
+                "rssi_dbm": rssi_props.get("value", {}).get("value"),
+                "hardware_version": _versionDict(hw_props),
+                "software_version": _versionDict(sw_props),
+            })
+        return sensors
+
+    def _readProperties(self, feature: str) -> dict[str, Any]:
+        try:
+            data = self.getProperty(feature)
+        except PyViCareNotSupportedFeatureError:
+            return {}
+        return data.get("properties") or {}
+
+    @handleNotSupported
+    def getFlowAlertMaxDuration(self) -> int:
+        return int(self.getProperty("water.leakDetection.configuration.flowAlert")["properties"]["maxDuration"]["value"])
+
+    @handleNotSupported
+    def getFlowAlertMaxFlow(self) -> float:
+        return float(self.getProperty("water.leakDetection.configuration.flowAlert")["properties"]["maxFlow"]["value"])
+
+    # --- Shutoff Valve ---
+
+    @handleNotSupported
+    def getShutoffPosition(self) -> str:
+        return str(self.getProperty("water.valves.shutoff.position")["properties"]["value"]["value"])
+
+    @handleNotSupported
+    def getShutoffMotorState(self) -> str:
+        return str(self.getProperty("water.valves.shutoff.motor")["properties"]["state"]["value"])
+
+    @handleNotSupported
+    def getHolidayModeActive(self) -> bool:
+        return bool(self.getProperty("water.valves.shutoff.holiday")["properties"]["active"]["value"])
+
+
+def _versionDict(props: dict[str, Any]) -> dict[str, int] | None:
+    if not props:
+        return None
+    return {
+        field: int(props[field]["value"])
+        for field in ("build", "family", "revision", "version")
+        if field in props
+    }

tests/test_PyViCareDeviceConfig.py

@@ -177,6 +177,22 @@ def test_autoDetect_RoleGateway_asGateway_TCU300(self):
         device_type = c.asAutoDetectDevice()
         self.assertEqual("Gateway", type(device_type).__name__)
 
+    def test_autoDetect_VitosetAqua19D_asWaterTreatment(self):
+        c = PyViCareDeviceConfig(self.service, "0", "VitosetAqua19D", "Online")
+        device_type = c.asAutoDetectDevice()
+        self.assertEqual("WaterTreatment", type(device_type).__name__)
+
+    def test_autoDetect_VitosetAqua42D_asWaterTreatment(self):
+        c = PyViCareDeviceConfig(self.service, "0", "VitosetAqua42D", "Online")
+        device_type = c.asAutoDetectDevice()
+        self.assertEqual("WaterTreatment", type(device_type).__name__)
+
+    def test_autoDetect_RoleWaterTreatment_asWaterTreatment(self):
+        self.service.hasRoles = has_roles(["type:waterTreatment"])
+        c = PyViCareDeviceConfig(self.service, "0", "Unknown", "Online")
+        device_type = c.asAutoDetectDevice()
+        self.assertEqual("WaterTreatment", type(device_type).__name__)
+
     def test_legacy_device(self):
         self.service.hasRoles = has_roles(["type:legacy"])
         c = PyViCareDeviceConfig(self.service, "0", "Unknown", "Online")

tests/test_TestForMissingProperties.py

@@ -349,24 +349,8 @@ def test_missingProperties(self):
             'ventilation.sensors.airQuality',
             'ventilation.operating.programs.forcedLevelFour',
             'ventilation.operating.programs.silent',
-            # Vitoset Aqua water softener (testdata only, no class yet)
+            # Vitoset Aqua water softener: device-level status not yet exposed
             'device.status',
-            'water.consumption.flow.current',
-            'water.consumption.flow.max',
-            'water.consumption.summary',
-            'water.leakDetection.configuration.flowAlert',
-            'water.leakDetection.sensors.leakage.0',
-            'water.leakDetection.sensors.leakage.0.battery',
-            'water.leakDetection.sensors.leakage.0.id',
-            'water.leakDetection.sensors.leakage.0.name',
-            'water.leakDetection.sensors.leakage.0.rssi',
-            'water.leakDetection.sensors.leakage.0.version.hardware',
-            'water.leakDetection.sensors.leakage.0.version.software',
-            'water.softener.configuration.lowSaltAlert',
-            'water.softener.salt.level.days',
-            'water.valves.shutoff.holiday',
-            'water.valves.shutoff.motor',
-            'water.valves.shutoff.position',
         ]
 
         all_features = self.read_all_features()
@@ -415,7 +399,7 @@ def test_unverifiedProperties(self):
                 continue
 
             for match in re.findall(r'getProperty\(\s*?f?"(.*)"\s*?\)', all_python_files[python]):
-                feature_name = re.sub(r'{(self\.)?(circuit|burner|compressor|condensor|evaporator|inverter|room_id)}', '0', match)
+                feature_name = re.sub(r'{(self\.)?(circuit|burner|compressor|condensor|evaporator|inverter|room_id|slot)}', '0', match)
                 feature_name = re.sub(r'{burner}', '0', feature_name)
                 feature_name = re.sub(r'{circuit}', '0', feature_name)  # for local variable in loops
                 feature_name = re.sub(r'\.{(quickmode|mode|program|active_program)}', '', feature_name)

tests/test_VitosetAqua.py

@@ -0,0 +1,82 @@
+import unittest
+
+from PyViCare.PyViCareWaterTreatment import WaterTreatment
+from tests.ViCareServiceMock import ViCareServiceMock
+
+
+class VitosetAquaTest(unittest.TestCase):
+    def setUp(self):
+        self.service = ViCareServiceMock('response/VitosetAqua.json')
+        self.device = WaterTreatment(self.service)
+
+    # --- Softener ---
+
+    def test_getSaltDaysRemaining(self):
+        self.assertEqual(self.device.getSaltDaysRemaining(), 15)
+
+    def test_getLowSaltAlertDays(self):
+        self.assertEqual(self.device.getLowSaltAlertDays(), 14)
+
+    # --- Consumption ---
+
+    def test_getCurrentFlow(self):
+        self.assertEqual(self.device.getCurrentFlow(), 0.0)
+
+    def test_getMaxFlow(self):
+        self.assertAlmostEqual(self.device.getMaxFlow(), 40.2)
+
+    def test_getCurrentDayConsumption(self):
+        self.assertEqual(self.device.getCurrentDayConsumption(), 144)
+
+    def test_getLastSevenDaysConsumption(self):
+        self.assertEqual(self.device.getLastSevenDaysConsumption(), 201)
+
+    def test_getTotalConsumption(self):
+        self.assertEqual(self.device.getTotalConsumption(), 45510)
+
+    # --- Leak Detection ---
+
+    def test_getLeakSensors_returns_only_connected(self):
+        sensors = self.device.getLeakSensors()
+        self.assertEqual(len(sensors), 1)
+
+    def test_getLeakSensors_first_sensor_fields(self):
+        sensor = self.device.getLeakSensors()[0]
+        self.assertEqual(sensor["slot"], 0)
+        self.assertEqual(sensor["status"], "connected")
+        self.assertFalse(sensor["leak_detected"])
+        self.assertEqual(sensor["id"], "00:00:00:00:00:00:00:00")
+        self.assertEqual(sensor["name"], "Basement")
+        self.assertEqual(sensor["battery_percent"], 100)
+        self.assertEqual(sensor["rssi_dbm"], 0)
+        self.assertIsInstance(sensor["hardware_version"], dict)
+        self.assertIsInstance(sensor["software_version"], dict)
+        for field in ("build", "family", "revision", "version"):
+            self.assertIn(field, sensor["hardware_version"])
+            self.assertIn(field, sensor["software_version"])
+
+    def test_getFlowAlertMaxDuration(self):
+        self.assertEqual(self.device.getFlowAlertMaxDuration(), 0)
+
+    def test_getFlowAlertMaxFlow(self):
+        self.assertEqual(self.device.getFlowAlertMaxFlow(), 0.0)
+
+    # --- Shutoff Valve ---
+
+    def test_getShutoffPosition(self):
+        self.assertEqual(self.device.getShutoffPosition(), "open")
+
+    def test_getShutoffMotorState(self):
+        self.assertEqual(self.device.getShutoffMotorState(), "off")
+
+    def test_getHolidayModeActive(self):
+        self.assertFalse(self.device.getHolidayModeActive())
+
+    # --- Inherited from Device ---
+
+    def test_getSerial(self):
+        self.assertEqual(self.device.getSerial(), "################")
+
+
+if __name__ == '__main__':
+    unittest.main()