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,141 @@
+from contextlib import suppress
+from typing import Any, List, Optional
+
+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 getTotalConsumption(self) -> int:
+        return int(self.getProperty("water.consumption.summary")["properties"]["total"]["value"])
+
+    # --- Leak Detection ---
+
+    @property
+    def leakSensors(self) -> List["LeakSensor"]:
+        return [LeakSensor(self, slot) for slot in self.getAvailableLeakSensorSlots()]
+
+    def getAvailableLeakSensorSlots(self) -> List[int]:
+        available: List[int] = []
+        for slot in range(_LEAK_SENSOR_SLOTS):
+            with suppress(PyViCareNotSupportedFeatureError):
+                feature = self.getProperty(f"water.leakDetection.sensors.leakage.{slot}")
+                if feature.get("isEnabled"):
+                    available.append(slot)
+        return available
+
+    @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"])
+
+
+class LeakSensor:
+    """Single wireless leakage sensor connected to a Vitoset Aqua."""
+
+    def __init__(self, device: WaterTreatment, slot: int) -> None:
+        self.service = device.service
+        self.slot = slot
+        self.device = device
+
+    @property
+    def id(self) -> int:
+        return self.slot
+
+    def getProperty(self, property_name: str) -> Any:
+        return self.device.getProperty(property_name)
+
+    @handleNotSupported
+    def getStatus(self) -> str:
+        return str(self.getProperty(f"water.leakDetection.sensors.leakage.{self.slot}")["properties"]["status"]["value"])
+
+    @handleNotSupported
+    def getLeakDetected(self) -> bool:
+        return bool(self.getProperty(f"water.leakDetection.sensors.leakage.{self.slot}")["properties"]["value"]["value"])
+
+    @handleNotSupported
+    def getSensorId(self) -> str:
+        return str(self.getProperty(f"water.leakDetection.sensors.leakage.{self.slot}.id")["properties"]["value"]["value"])
+
+    @handleNotSupported
+    def getName(self) -> str:
+        return str(self.getProperty(f"water.leakDetection.sensors.leakage.{self.slot}.name")["properties"]["name"]["value"])
+
+    @handleNotSupported
+    def getBatteryPercent(self) -> int:
+        return int(self.getProperty(f"water.leakDetection.sensors.leakage.{self.slot}.battery")["properties"]["level"]["value"])
+
+    @handleNotSupported
+    def getRssi(self) -> int:
+        return int(self.getProperty(f"water.leakDetection.sensors.leakage.{self.slot}.rssi")["properties"]["value"]["value"])
+
+    def getHardwareVersion(self) -> Optional[dict[str, int]]:
+        try:
+            feature = self.getProperty(f"water.leakDetection.sensors.leakage.{self.slot}.version.hardware")
+        except PyViCareNotSupportedFeatureError:
+            return None
+        return self._versionFields(feature.get("properties") or {})
+
+    def getSoftwareVersion(self) -> Optional[dict[str, int]]:
+        try:
+            feature = self.getProperty(f"water.leakDetection.sensors.leakage.{self.slot}.version.software")
+        except PyViCareNotSupportedFeatureError:
+            return None
+        return self._versionFields(feature.get("properties") or {})
+
+    @staticmethod
+    def _versionFields(props: dict[str, Any]) -> Optional[dict[str, int]]:
+        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,78 @@
+import unittest
+
+from PyViCare.PyViCareWaterTreatment import LeakSensor, 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_getTotalConsumption(self):
+        self.assertEqual(self.device.getTotalConsumption(), 45510)
+
+    # --- Leak Detection ---
+
+    def test_leakSensors_returns_only_connected(self):
+        self.assertEqual(len(self.device.leakSensors), 1)
+
+    def test_getAvailableLeakSensorSlots(self):
+        self.assertEqual(self.device.getAvailableLeakSensorSlots(), [0])
+
+    def test_leakSensor_fields(self):
+        sensor = self.device.leakSensors[0]
+        self.assertIsInstance(sensor, LeakSensor)
+        self.assertEqual(sensor.slot, 0)
+        self.assertEqual(sensor.id, 0)
+        self.assertEqual(sensor.getStatus(), "connected")
+        self.assertFalse(sensor.getLeakDetected())
+        self.assertEqual(sensor.getSensorId(), "00:00:00:00:00:00:00:00")
+        self.assertEqual(sensor.getName(), "Basement")
+        self.assertEqual(sensor.getBatteryPercent(), 100)
+        self.assertEqual(sensor.getRssi(), 0)
+        for field in ("build", "family", "revision", "version"):
+            self.assertIn(field, sensor.getHardwareVersion())
+            self.assertIn(field, sensor.getSoftwareVersion())
+
+    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()