From 0187ef5d4a0c70022e449139b4350ba58bba286c Mon Sep 17 00:00:00 2001 From: Bouwe Westerdijk Date: Tue, 18 Feb 2025 19:28:29 +0100 Subject: [PATCH 1/3] Update to etree.Element typing --- plugwise/__init__.py | 12 ++++++------ plugwise/common.py | 20 +++++++++++--------- plugwise/helper.py | 26 +++++++++++++------------- plugwise/legacy/helper.py | 12 ++++++------ plugwise/smile.py | 2 +- plugwise/smilecomm.py | 6 ++++-- plugwise/util.py | 10 +++++----- 7 files changed, 46 insertions(+), 42 deletions(-) diff --git a/plugwise/__init__.py b/plugwise/__init__.py index c12925eca..5674471e6 100644 --- a/plugwise/__init__.py +++ b/plugwise/__init__.py @@ -198,7 +198,9 @@ async def connect(self) -> Version: return self.smile_version - async def _smile_detect(self, result: etree, dsmrmain: etree) -> None: + async def _smile_detect( + self, result: etree.Element, dsmrmain: etree.Element + ) -> None: """Helper-function for connect(). Detect which type of Plugwise Gateway is being connected. @@ -256,10 +258,8 @@ async def _smile_detect(self, result: etree, dsmrmain: etree) -> None: # 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" - ) + onoff_boiler = result.find("./module/protocols/onoff_boiler") + open_therm_boiler = 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 @@ -272,7 +272,7 @@ async def _smile_detect(self, result: etree, dsmrmain: etree) -> None: self._elga = True async def _smile_detect_legacy( - self, result: etree, dsmrmain: etree, model: str + self, result: etree.Element, dsmrmain: etree.Element, model: str ) -> str: """Helper-function for _smile_detect(). diff --git a/plugwise/common.py b/plugwise/common.py index a9270736b..f20f3d30d 100644 --- a/plugwise/common.py +++ b/plugwise/common.py @@ -27,7 +27,9 @@ from munch import Munch -def get_zigbee_data(module: etree, module_data: ModuleData, legacy: bool) -> None: +def get_zigbee_data( + module: etree.Element, module_data: ModuleData, legacy: bool +) -> None: """Helper-function for _get_module_data().""" if legacy: # Stretches @@ -49,7 +51,7 @@ def __init__(self) -> None: """Init.""" self._cooling_present: bool self._count: int - self._domain_objects: etree + self._domain_objects: etree.Element self._heater_id: str = NONE self._on_off_device: bool self.gw_entities: dict[str, GwEntityData] = {} @@ -63,10 +65,10 @@ def smile(self, name: str) -> bool: def _appl_heater_central_info( self, appl: Munch, - xml_1: etree, + xml_1: etree.Element, legacy: bool, - xml_2: etree = None, - xml_3: etree = None, + xml_2: etree.Element = None, + xml_3: etree.Element = None, ) -> Munch: """Helper-function for _appliance_info_finder().""" # Find the valid heater_central @@ -101,7 +103,7 @@ def _appl_heater_central_info( return appl def _appl_thermostat_info( - self, appl: Munch, xml_1: etree, xml_2: etree = None + self, appl: Munch, xml_1: etree.Element, xml_2: etree.Element = None ) -> Munch: """Helper-function for _appliance_info_finder().""" locator = "./logs/point_log[type='thermostat']/thermostat" @@ -190,7 +192,7 @@ def _get_group_switches(self) -> dict[str, GwEntityData]: return switch_groups def _get_lock_state( - self, xml: etree, data: GwEntityData, stretch_v2: bool = False + self, xml: etree.Element, data: GwEntityData, stretch_v2: bool = False ) -> None: """Helper-function for _get_measurement_data(). @@ -209,9 +211,9 @@ def _get_lock_state( def _get_module_data( self, - xml_1: etree, + xml_1: etree.Element, locator: str, - xml_2: etree = None, + xml_2: etree.Element = None, legacy: bool = False, ) -> ModuleData: """Helper-function for _energy_device_info_finder() and _appliance_info_finder(). diff --git a/plugwise/helper.py b/plugwise/helper.py index c053cdcdf..5e1831a3f 100644 --- a/plugwise/helper.py +++ b/plugwise/helper.py @@ -59,7 +59,9 @@ from packaging import version -def search_actuator_functionalities(appliance: etree, actuator: str) -> etree | None: +def search_actuator_functionalities( + appliance: etree.Element, actuator: str +) -> etree.Element | None: """Helper-function for finding the relevant actuator xml-structure.""" locator = f"./actuator_functionalities/{actuator}" if (search := appliance.find(locator)) is not None: @@ -213,7 +215,7 @@ def _all_locations(self) -> None: f"./location[@id='{loc.loc_id}']" ) - def _appliance_info_finder(self, appl: Munch, appliance: etree) -> Munch: + def _appliance_info_finder(self, appl: Munch, appliance: etree.Element) -> Munch: """Collect info for all appliances found.""" match appl.pwclass: case "gateway": @@ -253,7 +255,7 @@ def _appliance_info_finder(self, appl: Munch, appliance: etree) -> Munch: case _: # pragma: no cover return appl - def _appl_gateway_info(self, appl: Munch, appliance: etree) -> Munch: + def _appl_gateway_info(self, appl: Munch, appliance: etree.Element) -> Munch: """Helper-function for _appliance_info_finder().""" self._gateway_id = appliance.attrib["id"] appl.firmware = str(self.smile_version) @@ -286,7 +288,7 @@ def _appl_gateway_info(self, appl: Munch, appliance: etree) -> Munch: return appl def _get_appl_actuator_modes( - self, appliance: etree, actuator_type: str + self, appliance: etree.Element, actuator_type: str ) -> list[str]: """Get allowed modes for the given actuator type.""" mode_list: list[str] = [] @@ -398,7 +400,7 @@ def _power_data_from_location(self) -> GwEntityData: def _appliance_measurements( self, - appliance: etree, + appliance: etree.Element, data: GwEntityData, measurements: dict[str, DATA | UOM], ) -> None: @@ -430,7 +432,7 @@ def _appliance_measurements( self._count = count_data_items(self._count, data) def _get_toggle_state( - self, xml: etree, toggle: str, name: ToggleNameType, data: GwEntityData + self, xml: etree.Element, toggle: str, name: ToggleNameType, data: GwEntityData ) -> None: """Helper-function for _get_measurement_data(). @@ -459,7 +461,7 @@ def _get_plugwise_notifications(self) -> None: ) def _get_actuator_functionalities( - self, xml: etree, entity: GwEntityData, data: GwEntityData + self, xml: etree.Element, entity: GwEntityData, data: GwEntityData ) -> None: """Get and process the actuator_functionalities details for an entity. @@ -521,7 +523,7 @@ def _get_actuator_functionalities( data[act_item] = temp_dict def _get_actuator_mode( - self, appliance: etree, entity_id: str, key: str + self, appliance: etree.Element, entity_id: str, key: str ) -> str | None: """Helper-function for _get_regulation_mode and _get_gateway_mode. @@ -536,7 +538,7 @@ def _get_actuator_mode( return None def _get_regulation_mode( - self, appliance: etree, entity_id: str, data: GwEntityData + self, appliance: etree.Element, entity_id: str, data: GwEntityData ) -> None: """Helper-function for _get_measurement_data(). @@ -552,7 +554,7 @@ def _get_regulation_mode( self._cooling_enabled = mode == "cooling" def _get_gateway_mode( - self, appliance: etree, entity_id: str, data: GwEntityData + self, appliance: etree.Element, entity_id: str, data: GwEntityData ) -> None: """Helper-function for _get_measurement_data(). @@ -838,9 +840,7 @@ def _presets(self, loc_id: str) -> dict[str, list[float]]: return presets # pragma: no cover for rule_id in rule_ids: - directives: etree = self._domain_objects.find( - f'rule[@id="{rule_id}"]/directives' - ) + directives = self._domain_objects.find(f'rule[@id="{rule_id}"]/directives') for directive in directives: preset = directive.find("then").attrib presets[directive.attrib["preset"]] = [ diff --git a/plugwise/legacy/helper.py b/plugwise/legacy/helper.py index 61dbe22ef..5d900aa20 100644 --- a/plugwise/legacy/helper.py +++ b/plugwise/legacy/helper.py @@ -50,7 +50,7 @@ from packaging.version import Version -def etree_to_dict(element: etree) -> dict[str, str]: +def etree_to_dict(element: etree.Element) -> dict[str, str]: """Helper-function translating xml Element to dict.""" node: dict[str, str] = {} if element is not None: @@ -64,11 +64,11 @@ class SmileLegacyHelper(SmileCommon): def __init__(self) -> None: """Set the constructor for this class.""" - self._appliances: etree + self._appliances: etree.Element self._is_thermostat: bool self._loc_data: dict[str, ThermoLoc] - self._locations: etree - self._modules: etree + self._locations: etree.Element + self._modules: etree.Element self._stretch_v2: bool self.gw_entities: dict[str, GwEntityData] = {} self.smile_mac_address: str | None @@ -318,7 +318,7 @@ def _power_data_from_modules(self) -> GwEntityData: def _appliance_measurements( self, - appliance: etree, + appliance: etree.Element, data: GwEntityData, measurements: dict[str, DATA | UOM], ) -> None: @@ -347,7 +347,7 @@ def _appliance_measurements( self._count = count_data_items(self._count, data) def _get_actuator_functionalities( - self, xml: etree, entity: GwEntityData, data: GwEntityData + self, xml: etree.Element, entity: GwEntityData, data: GwEntityData ) -> None: """Helper-function for _get_measurement_data().""" for item in ACTIVE_ACTUATORS: diff --git a/plugwise/smile.py b/plugwise/smile.py index 40df7d333..c2a67546f 100644 --- a/plugwise/smile.py +++ b/plugwise/smile.py @@ -355,7 +355,7 @@ async def set_schedule_state( def determine_contexts( self, loc_id: str, name: str, state: str, sched_id: str - ) -> etree: + ) -> etree.Element: """Helper-function for set_schedule_state().""" locator = f'.//*[@id="{sched_id}"]/contexts' contexts = self._domain_objects.find(locator) diff --git a/plugwise/smilecomm.py b/plugwise/smilecomm.py index 0f6b54384..61617a7e4 100644 --- a/plugwise/smilecomm.py +++ b/plugwise/smilecomm.py @@ -51,7 +51,7 @@ async def _request( retry: int = 3, method: str = "get", data: str | None = None, - ) -> etree: + ) -> etree.Element: """Get/put/delete data from a give URL.""" resp: ClientResponse url = f"{self._endpoint}{command}" @@ -107,7 +107,9 @@ async def _request( return await self._request_validate(resp, method) - async def _request_validate(self, resp: ClientResponse, method: str) -> etree: + async def _request_validate( + self, resp: ClientResponse, method: str + ) -> etree.Element: """Helper-function for _request(): validate the returned data.""" match resp.status: case 200: diff --git a/plugwise/util.py b/plugwise/util.py index d5c559fdb..839ffe3ba 100644 --- a/plugwise/util.py +++ b/plugwise/util.py @@ -74,7 +74,7 @@ def in_alternative_location(loc: Munch, legacy: bool) -> bool: return present -def check_heater_central(xml: etree) -> str: +def check_heater_central(xml: etree.Element) -> str: """Find the valid heater_central, helper-function for _appliance_info_finder(). Solution for Core Issue #104433, @@ -143,7 +143,7 @@ def collect_power_values( def common_match_cases( measurement: str, attrs: DATA | UOM, - location: etree, + location: etree.Element, data: GwEntityData, ) -> None: """Helper-function for common match-case execution.""" @@ -213,7 +213,7 @@ def format_measure(measure: str, unit: str) -> float | int: return result -def get_vendor_name(module: etree, model_data: ModuleData) -> ModuleData: +def get_vendor_name(module: etree.Element, model_data: ModuleData) -> ModuleData: """Helper-function for _get_model_data().""" if (vendor_name := module.find("vendor_name").text) is not None: model_data["vendor_name"] = vendor_name @@ -302,12 +302,12 @@ def remove_empty_platform_dicts(data: GwEntityData) -> None: data.pop("switches") -def return_valid(value: etree | None, default: etree) -> etree: +def return_valid(value: etree.Element | None, default: etree.Element) -> etree.Element: """Return default when value is None.""" return value if value is not None else default -def skip_obsolete_measurements(xml: etree, measurement: str) -> bool: +def skip_obsolete_measurements(xml: etree.Element, measurement: str) -> bool: """Skipping known obsolete measurements.""" locator = f".//logs/point_log[type='{measurement}']/updated_date" if ( From 7087529b88d7c285830e14827201b433c9b65343 Mon Sep 17 00:00:00 2001 From: Bouwe Westerdijk Date: Wed, 19 Feb 2025 11:04:00 +0100 Subject: [PATCH 2/3] Implement review suggestion --- plugwise/smile.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugwise/smile.py b/plugwise/smile.py index c2a67546f..0b88c8e38 100644 --- a/plugwise/smile.py +++ b/plugwise/smile.py @@ -355,7 +355,7 @@ async def set_schedule_state( def determine_contexts( self, loc_id: str, name: str, state: str, sched_id: str - ) -> etree.Element: + ) -> str: """Helper-function for set_schedule_state().""" locator = f'.//*[@id="{sched_id}"]/contexts' contexts = self._domain_objects.find(locator) From 91beb350ba2b26bfeb84db2eb08bf97a810a5cb5 Mon Sep 17 00:00:00 2001 From: Bouwe Westerdijk Date: Wed, 19 Feb 2025 11:35:37 +0100 Subject: [PATCH 3/3] Force string --- plugwise/smile.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugwise/smile.py b/plugwise/smile.py index 0b88c8e38..ece464b25 100644 --- a/plugwise/smile.py +++ b/plugwise/smile.py @@ -370,7 +370,7 @@ def determine_contexts( if state == "on": contexts.append(subject) - return etree.tostring(contexts, encoding="unicode").rstrip() + return str(etree.tostring(contexts, encoding="unicode").rstrip()) async def set_switch_state( self, appl_id: str, members: list[str] | None, model: str, state: str