From 9e49f9272bff90c86271623ab8ea0002528ec918 Mon Sep 17 00:00:00 2001 From: Christian Lackas Date: Thu, 7 May 2026 20:20:05 +0200 Subject: [PATCH] feat: add WaterTreatment class for Vitoset Aqua Adds read-only support for the Vitoset Aqua water treatment station, covering the four functional areas exposed by the device: - Softener: salt days remaining, low-salt alert threshold - Consumption: current/max flow, daily/7-day/total volume - Leak detection: up to 5 sensors (status, leak, battery, RSSI, name, versions) plus configurable flow alert - Shutoff valve: position, motor state, holiday-mode flag Autodetect matches device model "VitosetAqua*" (e.g. VitosetAqua19D, VitosetAqua42D) and the "type:waterTreatment" role. Setters are intentionally left out for a follow-up: the IoT-scope dump contains no command definitions for this device. --- PyViCare/PyViCareDeviceConfig.py | 5 ++ PyViCare/PyViCareWaterTreatment.py | 118 +++++++++++++++++++++++++ tests/test_PyViCareDeviceConfig.py | 16 ++++ tests/test_TestForMissingProperties.py | 20 +---- tests/test_VitosetAqua.py | 82 +++++++++++++++++ 5 files changed, 223 insertions(+), 18 deletions(-) create mode 100644 PyViCare/PyViCareWaterTreatment.py create mode 100644 tests/test_VitosetAqua.py diff --git a/PyViCare/PyViCareDeviceConfig.py b/PyViCare/PyViCareDeviceConfig.py index 364193fa..5ffeabcb 100644 --- a/PyViCare/PyViCareDeviceConfig.py +++ b/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"]), diff --git a/PyViCare/PyViCareWaterTreatment.py b/PyViCare/PyViCareWaterTreatment.py new file mode 100644 index 00000000..064fce80 --- /dev/null +++ b/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 + } diff --git a/tests/test_PyViCareDeviceConfig.py b/tests/test_PyViCareDeviceConfig.py index fc8dc659..0fb4afd3 100644 --- a/tests/test_PyViCareDeviceConfig.py +++ b/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") diff --git a/tests/test_TestForMissingProperties.py b/tests/test_TestForMissingProperties.py index b04959c1..d05b34af 100644 --- a/tests/test_TestForMissingProperties.py +++ b/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) diff --git a/tests/test_VitosetAqua.py b/tests/test_VitosetAqua.py new file mode 100644 index 00000000..74eae58f --- /dev/null +++ b/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()