diff --git a/CHANGELOG.md b/CHANGELOG.md index 77321c8ba..3f4e7c731 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,9 +1,10 @@ # Changelog -## Ongoing +## v0.32.2 Continuous improvements, bugfix - Extend p1v4_442_triple userdata to include a Plugwise notification, extending the related fixture which is used in PW-beta and Core Plugwise. - Optimize and rearrange typing-related constants, implement and cleanup. +- Optimize and reorder code, for the Stretch prevent the creation of a switch-group with an orphaned switch. ## v0.32.1 Improve typing, bugfix diff --git a/fixtures/stretch_v31/all_data.json b/fixtures/stretch_v31/all_data.json index c336a9cb9..8604aaae1 100644 --- a/fixtures/stretch_v31/all_data.json +++ b/fixtures/stretch_v31/all_data.json @@ -48,15 +48,6 @@ "vendor": "Plugwise", "zigbee_mac_address": "ABCD012345670A07" }, - "71e1944f2a944b26ad73323e399efef0": { - "dev_class": "switching", - "members": ["5ca521ac179d468e91d772eeeb8a2117"], - "model": "Switchgroup", - "name": "Test", - "switches": { - "relay": true - } - }, "aac7b735042c4832ac9ff33aae4f453b": { "dev_class": "dishwasher", "firmware": "2011-06-27T10:52:18+02:00", diff --git a/plugwise/__init__.py b/plugwise/__init__.py index 6ee19dacb..867c6a1fe 100644 --- a/plugwise/__init__.py +++ b/plugwise/__init__.py @@ -87,26 +87,23 @@ def update_for_cooling(self, device: DeviceData) -> DeviceData: def _all_device_data(self) -> None: """Helper-function for get_all_devices(). - Collect initial data for each device and add to self.gw_data and self.gw_devices. + Collect data for each device and add to self.gw_data and self.gw_devices. """ - for device_id, device in self._appl_data.items(): - self.gw_devices.update({device_id: device}) - + for device_id, device in self.gw_devices.items(): data = self._get_device_data(device_id) + device.update(data) # Add plugwise notification binary_sensor to the relevant gateway if device_id == self.gateway_id and ( self._is_thermostat or (not self._smile_legacy and self.smile_type == "power") ): - data["binary_sensors"]["plugwise_notification"] = False - - self.gw_devices[device_id].update(data) + device["binary_sensors"]["plugwise_notification"] = False # Update for cooling - if self.gw_devices[device_id]["dev_class"] in ZONE_THERMOSTATS: - self.update_for_cooling(self.gw_devices[device_id]) + if device["dev_class"] in ZONE_THERMOSTATS: + self.update_for_cooling(device) - remove_empty_platform_dicts(self.gw_devices[device_id]) + remove_empty_platform_dicts(device) self.gw_data.update( {"smile_name": self.smile_name, "gateway_id": self.gateway_id} @@ -122,69 +119,50 @@ def get_all_devices(self) -> None: Run this functions once to gather the initial device configuration, then regularly run async_update() to refresh the device data. """ - # Start by determining the system capabilities: - # Find the connected heating/cooling device (heater_central), e.g. heat-pump or gas-fired heater + # Gather all the devices and their initial data + self._all_appliances() if self.smile_type == "thermostat": - onoff_boiler: etree = self._domain_objects.find( - "./module/protocols/onoff_boiler" - ) - open_therm_boiler: etree = self._domain_objects.find( - "./module/protocols/open_therm_boiler" - ) - self._on_off_device = onoff_boiler is not None - self._opentherm_device = open_therm_boiler is not None - - # Determine the presence of special features - locator_1 = "./gateway/features/cooling" - locator_2 = "./gateway/features/elga_support" - search = self._domain_objects - if search.find(locator_1) is not None: - self._cooling_present = True - if search.find(locator_2) is not None: - self._elga = True - + self._scan_thermostats() + # Collect a list of thermostats with offset-capability self.therms_with_offset_func = ( self._get_appliances_with_offset_functionality() ) - # Gather all the device and initial data - self._scan_thermostats() + # Collect switching- or pump-group data + if group_data := self._get_group_switches(): + self.gw_devices.update(group_data) - if group_data := self._group_switches(): - self._appl_data.update(group_data) - - # Collect data for each device via helper function + # Collect the remaining data for all device self._all_device_data() def _device_data_switching_group( - self, details: DeviceData, device_data: DeviceData + self, device: DeviceData, device_data: DeviceData ) -> DeviceData: """Helper-function for _get_device_data(). Determine switching group device data. """ - if details["dev_class"] in SWITCH_GROUP_TYPES: - counter = 0 - for member in details["members"]: - member_data = self._get_appliance_data(member) - if member_data["switches"].get("relay"): - counter += 1 - - device_data["switches"]["relay"] = counter != 0 + if device["dev_class"] not in SWITCH_GROUP_TYPES: + return device_data + counter = 0 + for member in device["members"]: + if self.gw_devices[member]["switches"].get("relay"): + counter += 1 + device_data["switches"]["relay"] = counter != 0 return device_data def _device_data_adam( - self, details: DeviceData, device_data: DeviceData + self, device: DeviceData, device_data: DeviceData ) -> DeviceData: """Helper-function for _get_device_data(). - Determine Adam device data. + Determine Adam heating-status for on-off heating via valves. """ # Indicate heating_state based on valves being open in case of city-provided heating if ( self.smile_name == "Adam" - and details.get("dev_class") == "heater_central" + and device.get("dev_class") == "heater_central" and self._on_off_device and self._heating_valves() is not None ): @@ -193,13 +171,13 @@ def _device_data_adam( return device_data def _device_data_climate( - self, details: DeviceData, device_data: DeviceData + self, device: DeviceData, device_data: DeviceData ) -> DeviceData: """Helper-function for _get_device_data(). Determine climate-control device data. """ - loc_id = details["location"] + loc_id = device["location"] # Presets device_data["preset_modes"] = None @@ -246,14 +224,14 @@ def _device_data_climate( return device_data def _check_availability( - self, details: DeviceData, device_data: DeviceData + self, device: DeviceData, device_data: DeviceData ) -> DeviceData: """Helper-function for _get_device_data(). Provide availability status for the wired-commected devices. """ # OpenTherm device - if details["dev_class"] == "heater_central" and details["name"] != "OnOff": + if device["dev_class"] == "heater_central" and device["name"] != "OnOff": device_data["available"] = True for data in self._notifications.values(): for msg in data.values(): @@ -261,7 +239,7 @@ def _check_availability( device_data["available"] = False # Smartmeter - if details["dev_class"] == "smartmeter": + if device["dev_class"] == "smartmeter": device_data["available"] = True for data in self._notifications.values(): for msg in data.values(): @@ -275,14 +253,10 @@ def _get_device_data(self, dev_id: str) -> DeviceData: Provide device-data, based on Location ID (= dev_id), from APPLIANCES. """ - details = self._appl_data[dev_id] - device_data = self._get_appliance_data(dev_id) - # Remove thermostat-dict for thermo_sensors - if details["dev_class"] == "thermo_sensor": - device_data.pop("thermostat") - + device = self.gw_devices[dev_id] + device_data = self._get_measurement_data(dev_id) # Generic - if self.smile_type == "thermostat" and details["dev_class"] == "gateway": + if self.smile_type == "thermostat" and device["dev_class"] == "gateway": # Adam & Anna: the Smile outdoor_temperature is present in DOMAIN_OBJECTS and LOCATIONS - under Home # The outdoor_temperature present in APPLIANCES is a local sensor connected to the active device outdoor_temperature = self._object_value( @@ -296,30 +270,23 @@ def _get_device_data(self, dev_id: str) -> DeviceData: device_data["regulation_modes"] = self._reg_allowed_modes # Show the allowed dhw_modes - if details["dev_class"] == "heater_central" and self._dhw_allowed_modes: + if device["dev_class"] == "heater_central" and self._dhw_allowed_modes: device_data["dhw_modes"] = self._dhw_allowed_modes - # Get P1 smartmeter data from LOCATIONS or MODULES - if details["dev_class"] == "smartmeter": - if not self._smile_legacy: - device_data.update(self._power_data_from_location(details["location"])) - else: - device_data.update(self._power_data_from_modules()) - # Check availability of non-legacy wired-connected devices if not self._smile_legacy: - self._check_availability(details, device_data) + self._check_availability(device, device_data) # Switching groups data - device_data = self._device_data_switching_group(details, device_data) + device_data = self._device_data_switching_group(device, device_data) # Specific, not generic Adam data - device_data = self._device_data_adam(details, device_data) + device_data = self._device_data_adam(device, device_data) # No need to obtain thermostat data when the device is not a thermostat - if details["dev_class"] not in ZONE_THERMOSTATS: + if device["dev_class"] not in ZONE_THERMOSTATS: return device_data # Thermostat data (presets, temperatures etc) - device_data = self._device_data_climate(details, device_data) + device_data = self._device_data_climate(device, device_data) return device_data @@ -484,7 +451,25 @@ async def _smile_detect(self, result: etree, dsmrmain: etree) -> None: self._stretch_v2 = self.smile_version[1].major == 2 self._stretch_v3 = self.smile_version[1].major == 3 - self._is_thermostat = self.smile_type == "thermostat" + if self.smile_type == "thermostat": + self._is_thermostat = True + # For Adam, Anna, determine the system capabilities: + # Find the connected heating/cooling device (heater_central), + # e.g. heat-pump or gas-fired heater + onoff_boiler: etree = result.find("./module/protocols/onoff_boiler") + open_therm_boiler: etree = result.find( + "./module/protocols/open_therm_boiler" + ) + self._on_off_device = onoff_boiler is not None + self._opentherm_device = open_therm_boiler is not None + + # Determine the presence of special features + locator_1 = "./gateway/features/cooling" + locator_2 = "./gateway/features/elga_support" + if result.find(locator_1) is not None: + self._cooling_present = True + if result.find(locator_2) is not None: + self._elga = True async def _full_update_device(self) -> None: """Perform a first fetch of all XML data, needed for initialization.""" @@ -542,7 +527,6 @@ async def async_update(self) -> PlugwiseData: data["binary_sensors"]["plugwise_notification"] = bool( self._notifications ) - device.update(data) # Update for cooling diff --git a/plugwise/helper.py b/plugwise/helper.py index f506e3427..076104043 100644 --- a/plugwise/helper.py +++ b/plugwise/helper.py @@ -308,7 +308,6 @@ class SmileHelper: def __init__(self) -> None: """Set the constructor for this class.""" self._appliances: etree - self._appl_data: dict[str, DeviceData] = {} self._cooling_activation_outdoor_temp: float self._cooling_deactivation_threshold: float self._cooling_present = False @@ -593,7 +592,7 @@ def _p1_smartmeter_info_finder(self, appl: Munch) -> None: location = self._locations.find(f'./location[@id="{loc_id}"]') appl = self._energy_device_info_finder(location, appl) - self._appl_data[appl.dev_id] = {"dev_class": appl.pwclass} + self.gw_devices[appl.dev_id] = {"dev_class": appl.pwclass} for key, value in { "firmware": appl.firmware, @@ -607,7 +606,7 @@ def _p1_smartmeter_info_finder(self, appl: Munch) -> None: }.items(): if value is not None or key == "location": p1_key = cast(ApplianceType, key) - self._appl_data[appl.dev_id][p1_key] = value + self.gw_devices[appl.dev_id][p1_key] = value def _create_legacy_gateway(self) -> None: """Create the (missing) gateway devices for legacy Anna, P1 and Stretch. @@ -618,7 +617,7 @@ def _create_legacy_gateway(self) -> None: if self.smile_type == "power": self.gateway_id = FAKE_APPL - self._appl_data[self.gateway_id] = {"dev_class": "gateway"} + self.gw_devices[self.gateway_id] = {"dev_class": "gateway"} for key, value in { "firmware": self.smile_fw_version, "location": self._home_location, @@ -630,7 +629,7 @@ def _create_legacy_gateway(self) -> None: }.items(): if value is not None: gw_key = cast(ApplianceType, key) - self._appl_data[self.gateway_id][gw_key] = value + self.gw_devices[self.gateway_id][gw_key] = value def _all_appliances(self) -> None: """Collect all appliances with relevant info.""" @@ -692,7 +691,7 @@ def _all_appliances(self) -> None: ): continue - self._appl_data[appl.dev_id] = {"dev_class": appl.pwclass} + self.gw_devices[appl.dev_id] = {"dev_class": appl.pwclass} for key, value in { "firmware": appl.firmware, "hardware": appl.hardware, @@ -705,13 +704,13 @@ def _all_appliances(self) -> None: }.items(): if value is not None or key == "location": appl_key = cast(ApplianceType, key) - self._appl_data[appl.dev_id][appl_key] = value + self.gw_devices[appl.dev_id][appl_key] = value # For non-legacy P1 collect the connected SmartMeter info if self.smile_type == "power": self._p1_smartmeter_info_finder(appl) # P1: for gateway and smartmeter switch device_id - part 2 - for item in self._appl_data: + for item in self.gw_devices: if item != self.gateway_id: self.gateway_id = item # Leave for-loop to avoid a 2nd device_id switch @@ -724,7 +723,7 @@ def _match_locations(self) -> dict[str, ThermoLoc]: """ matched_locations: dict[str, ThermoLoc] = {} for location_id, location_details in self._loc_data.items(): - for appliance_details in self._appl_data.values(): + for appliance_details in self.gw_devices.values(): if appliance_details["location"] == location_id: location_details.update( {"master": None, "master_prio": 0, "slaves": set()} @@ -836,7 +835,7 @@ def _appliance_measurements( data: DeviceData, measurements: dict[str, DATA | UOM], ) -> None: - """Helper-function for _get_appliance_data() - collect appliance measurement data.""" + """Helper-function for _get_measurement_data() - collect appliance measurement data.""" for measurement, attrs in measurements.items(): p_locator = f'.//logs/point_log[type="{measurement}"]/period/measurement' if (appl_p_loc := appliance.find(p_locator)) is not None: @@ -900,7 +899,7 @@ def _appliance_measurements( ) def _wireless_availablity(self, appliance: etree, data: DeviceData) -> None: - """Helper-function for _get_appliance_data(). + """Helper-function for _get_measurement_data(). Collect the availablity-status for wireless connected devices. """ @@ -929,10 +928,16 @@ def _get_appliances_with_offset_functionality(self) -> list[str]: return therm_list - def _get_actuator_functionalities(self, xml: etree, data: DeviceData) -> None: - """Helper-function for _get_appliance_data().""" + def _get_actuator_functionalities( + self, xml: etree, device: DeviceData, data: DeviceData + ) -> None: + """Helper-function for _get_measurement_data().""" for item in ACTIVE_ACTUATORS: - if item == "max_dhw_temperature": + # Skip max_dhw_temperature, not initially valid, + # skip thermostat for thermo_sensors + if item == "max_dhw_temperature" or ( + item == "thermostat" and device["dev_class"] == "thermo_sensor" + ): continue temp_dict: ActuatorData = {} @@ -975,7 +980,7 @@ def _get_actuator_functionalities(self, xml: etree, data: DeviceData) -> None: data[act_item] = temp_dict def _get_regulation_mode(self, appliance: etree, data: DeviceData) -> None: - """Helper-function for _get_appliance_data(). + """Helper-function for _get_measurement_data(). Collect the gateway regulation_mode. """ @@ -985,7 +990,7 @@ def _get_regulation_mode(self, appliance: etree, data: DeviceData) -> None: self._cooling_enabled = search.find("mode").text == "cooling" def _cleanup_data(self, data: DeviceData) -> None: - """Helper-function for _get_appliance_data(). + """Helper-function for _get_measurement_data(). Clean up the data dict. """ @@ -1000,7 +1005,7 @@ def _cleanup_data(self, data: DeviceData) -> None: data.pop("cooling_enabled") # pragma: no cover def _process_c_heating_state(self, data: DeviceData) -> None: - """Helper-function for _get_appliance_data(). + """Helper-function for _get_measurement_data(). Process the central_heating_state value. """ @@ -1024,23 +1029,31 @@ def _process_c_heating_state(self, data: DeviceData) -> None: if self._elga: data["binary_sensors"]["heating_state"] = data["c_heating_state"] - def _get_appliance_data(self, d_id: str) -> DeviceData: + def _get_measurement_data(self, dev_id: str) -> DeviceData: """Helper-function for smile.py: _get_device_data(). Collect the appliance-data based on device id. - Determined from APPLIANCES, for legacy from DOMAIN_OBJECTS. """ data: DeviceData = {"binary_sensors": {}, "sensors": {}, "switches": {}} - # P1 legacy has no APPLIANCES, also not present in DOMAIN_OBJECTS - if self._smile_legacy and self.smile_type == "power": + # Get P1 smartmeter data from LOCATIONS or MODULES + device = self.gw_devices[dev_id] + # !! DON'T CHANGE below two if-lines, will break stuff !! + if self.smile_type == "power": + if device["dev_class"] == "smartmeter": + if not self._smile_legacy: + data.update(self._power_data_from_location(device["location"])) + else: + data.update(self._power_data_from_modules()) + return data + # Get non-p1 data from APPLIANCES, for legacy from DOMAIN_OBJECTS. measurements = DEVICE_MEASUREMENTS - if d_id == self._heater_id: + if dev_id == self._heater_id: measurements = HEATER_CENTRAL_MEASUREMENTS if ( - appliance := self._appliances.find(f'./appliance[@id="{d_id}"]') + appliance := self._appliances.find(f'./appliance[@id="{dev_id}"]') ) is not None: self._appliance_measurements(appliance, data, measurements) self._get_lock_state(appliance, data) @@ -1049,12 +1062,12 @@ def _get_appliance_data(self, d_id: str) -> DeviceData: self._get_toggle_state(appliance, toggle, name, data) if appliance.find("type").text in ACTUATOR_CLASSES: - self._get_actuator_functionalities(appliance, data) + self._get_actuator_functionalities(appliance, device, data) # Collect availability-status for wireless connected devices to Adam self._wireless_availablity(appliance, data) - if d_id == self.gateway_id and self.smile_name == "Adam": + if dev_id == self.gateway_id and self.smile_name == "Adam": self._get_regulation_mode(appliance, data) if "c_heating_state" in data: @@ -1062,7 +1075,7 @@ def _get_appliance_data(self, d_id: str) -> DeviceData: # Remove c_heating_state after processing data.pop("c_heating_state") - if d_id == self._heater_id and self.smile_name == "Smile Anna": + if dev_id == self._heater_id and self.smile_name == "Smile Anna": # Anna+Elga: base cooling_state on the elga-status-code if "elga_status_code" in data: # Determine _cooling_present and _cooling_enabled @@ -1128,10 +1141,6 @@ def _scan_thermostats(self) -> None: Update locations with thermostat ranking results and use the result to update the device_class of slave thermostats. """ - self._all_appliances() - if self.smile_type != "thermostat": - return - self._thermo_locs = self._match_locations() thermo_matching: dict[str, int] = { @@ -1142,15 +1151,15 @@ def _scan_thermostats(self) -> None: } for loc_id in self._thermo_locs: - for appl_id, details in self._appl_data.items(): - self._rank_thermostat(thermo_matching, loc_id, appl_id, details) + for dev_id, device in self.gw_devices.items(): + self._rank_thermostat(thermo_matching, loc_id, dev_id, device) # Update slave thermostat class where needed - for appl_id, details in self._appl_data.items(): - if (loc_id := details["location"]) in self._thermo_locs: + for dev_id, device in self.gw_devices.items(): + if (loc_id := device["location"]) in self._thermo_locs: tl_loc_id = self._thermo_locs[loc_id] - if "slaves" in tl_loc_id and appl_id in tl_loc_id["slaves"]: - details["dev_class"] = "thermo_sensor" + if "slaves" in tl_loc_id and dev_id in tl_loc_id["slaves"]: + device["dev_class"] = "thermo_sensor" def _thermostat_uri_legacy(self) -> str: """Helper-function for _thermostat_uri(). @@ -1175,7 +1184,7 @@ def _thermostat_uri(self, loc_id: str) -> str: return f"{LOCATIONS};id={loc_id}/thermostat;id={thermostat_functionality_id}" - def _group_switches(self) -> dict[str, DeviceData]: + def _get_group_switches(self) -> dict[str, DeviceData]: """Helper-function for smile.py: get_all_devices(). Collect switching- or pump-group info. @@ -1192,9 +1201,11 @@ def _group_switches(self) -> dict[str, DeviceData]: group_type = group.find("type").text group_appliances = group.findall("appliances/appliance") for item in group_appliances: - members.append(item.attrib["id"]) + # Check if members are not orphaned - stretch + if item.attrib["id"] in self.gw_devices: + members.append(item.attrib["id"]) - if group_type in SWITCH_GROUP_TYPES: + if group_type in SWITCH_GROUP_TYPES and members: switch_groups.update( { group_id: { @@ -1515,7 +1526,7 @@ def _object_value(self, obj_id: str, measurement: str) -> float | int | None: return val def _get_lock_state(self, xml: etree, data: DeviceData) -> None: - """Helper-function for _get_appliance_data(). + """Helper-function for _get_measurement_data(). Adam & Stretches: obtain the relay-switch lock state. """ @@ -1532,7 +1543,7 @@ def _get_lock_state(self, xml: etree, data: DeviceData) -> None: def _get_toggle_state( self, xml: etree, toggle: str, name: ToggleNameType, data: DeviceData ) -> None: - """Helper-function for _get_appliance_data(). + """Helper-function for _get_measurement_data(). Obtain the toggle state of a 'toggle' = switch. """ diff --git a/pyproject.toml b/pyproject.toml index 8299659d3..5148045f4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "plugwise" -version = "0.32.1" +version = "0.32.2" license = {file = "LICENSE"} description = "Plugwise Smile (Adam/Anna/P1) and Stretch module for Python 3." readme = "README.md" diff --git a/tests/test_smile.py b/tests/test_smile.py index 749b951f6..90ee45d57 100644 --- a/tests/test_smile.py +++ b/tests/test_smile.py @@ -4628,13 +4628,6 @@ async def test_connect_stretch_v31(self): }, "switches": {"relay": True, "lock": False}, }, - "71e1944f2a944b26ad73323e399efef0": { - "dev_class": "switching", - "model": "Switchgroup", - "name": "Test", - "members": ["5ca521ac179d468e91d772eeeb8a2117"], - "switches": {"relay": True}, - }, "d950b314e9d8499f968e6db8d82ef78c": { "dev_class": "report", "model": "Switchgroup", @@ -4674,7 +4667,7 @@ async def test_connect_stretch_v31(self): await self.device_test(smile, testdata) assert smile.gateway_id == "0000aaaa0000aaaa0000aaaa0000aa00" - assert self.device_items == 88 + assert self.device_items == 83 await smile.close_connection() await self.disconnect(server, client)