Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Continuous improvement #678

Merged
merged 16 commits into from
Dec 22, 2024
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
# Changelog

## Ongoing

- Continuous improvements [#678](https://github.com/plugwise/python-plugwise/pull/678)

## v1.6.4

- Continuous improvements [#662](https://github.com/plugwise/python-plugwise/pull/662)
Expand Down
67 changes: 24 additions & 43 deletions plugwise/helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -233,7 +233,8 @@ def __init__(self) -> None:
self._elga: bool
self._gw_allowed_modes: list[str] = []
self._heater_id: str
self._home_location: str
self._home_loc_id: str
self._home_location: etree
self._is_thermostat: bool
self._last_active: dict[str, str | None]
self._last_modified: dict[str, str] = {}
Expand Down Expand Up @@ -311,10 +312,10 @@ def _all_appliances(self) -> None:
appl.location = None
if (appl_loc := appliance.find("location")) is not None:
appl.location = appl_loc.attrib["id"]
# Don't assign the _home_location to thermostat-devices without a location,
# Don't assign the _home_loc_id to thermostat-devices without a location,
# they are not active
elif appl.pwclass not in THERMOSTAT_CLASSES:
appl.location = self._home_location
appl.location = self._home_loc_id

# Don't show orphaned thermostat-types
if appl.pwclass in THERMOSTAT_CLASSES and appl.location is None:
Expand Down Expand Up @@ -350,22 +351,16 @@ def _get_p1_smartmeter_info(self) -> None:
switched to maintain backward compatibility with existing implementations.
"""
appl = Munch()
loc_id = next(iter(self._loc_data.keys()))
if (
location := self._domain_objects.find(f'./location[@id="{loc_id}"]')
) is None:
return

locator = MODULE_LOCATOR
module_data = self._get_module_data(location, locator)
module_data = self._get_module_data(self._home_location, locator)
if not module_data["contents"]:
LOGGER.error("No module data found for SmartMeter") # pragma: no cover
return # pragma: no cover
appl.available = None
appl.entity_id = self.gateway_id
appl.firmware = module_data["firmware_version"]
appl.hardware = module_data["hardware_version"]
appl.location = loc_id
appl.location = self._home_loc_id
appl.mac = None
appl.model = module_data["vendor_model"]
appl.model_id = None # don't use model_id for SmartMeter
Expand All @@ -375,8 +370,8 @@ def _get_p1_smartmeter_info(self) -> None:
appl.zigbee_mac = None

# Replace the entity_id of the gateway by the smartmeter location_id
self.gw_entities[loc_id] = self.gw_entities.pop(self.gateway_id)
self.gateway_id = loc_id
self.gw_entities[self._home_loc_id] = self.gw_entities.pop(self.gateway_id)
self.gateway_id = self._home_loc_id

self._create_gw_entities(appl)

Expand All @@ -398,10 +393,14 @@ def _all_locations(self) -> None:
for location in locations:
loc.name = location.find("name").text
loc.loc_id = location.attrib["id"]
if loc.name == "Home":
self._home_location = loc.loc_id

self._loc_data[loc.loc_id] = {"name": loc.name}
if loc.name != "Home":
continue

self._home_loc_id = loc.loc_id
self._home_location = self._domain_objects.find(
f"./location[@id='{loc.loc_id}']"
)

def _appliance_info_finder(self, appl: Munch, appliance: etree) -> Munch:
"""Collect info for all appliances found."""
Expand Down Expand Up @@ -532,7 +531,7 @@ def _get_measurement_data(self, entity_id: str) -> GwEntityData:
# !! DON'T CHANGE below two if-lines, will break stuff !!
if self.smile_type == "power":
if entity["dev_class"] == "smartmeter":
data.update(self._power_data_from_location(entity["location"]))
data.update(self._power_data_from_location())

return data

Expand Down Expand Up @@ -574,18 +573,17 @@ def _get_measurement_data(self, entity_id: str) -> GwEntityData:

return data

def _power_data_from_location(self, loc_id: str) -> GwEntityData:
def _power_data_from_location(self) -> GwEntityData:
"""Helper-function for smile.py: _get_entity_data().

Collect the power-data based on Location ID, from LOCATIONS.
Collect the power-data from the Home location.
"""
data: GwEntityData = {"sensors": {}}
loc = Munch()
log_list: list[str] = ["point_log", "cumulative_log", "interval_log"]
t_string = "tariff"

search = self._domain_objects
loc.logs = search.find(f'./location[@id="{loc_id}"]/logs')
loc.logs = self._home_location.find("./logs")
for loc.measurement, loc.attrs in P1_MEASUREMENTS.items():
for loc.log_type in log_list:
self._collect_power_values(data, loc, t_string)
Expand Down Expand Up @@ -764,31 +762,14 @@ def _get_gateway_mode(
self._count += 1

def _get_gateway_outdoor_temp(self, entity_id: str, data: GwEntityData) -> None:
"""Adam & Anna: the Smile outdoor_temperature is present in DOMAIN_OBJECTS and LOCATIONS.

Available under the Home location.
"""
"""Adam & Anna: the Smile outdoor_temperature is present in the Home location."""
if self._is_thermostat and entity_id == self.gateway_id:
outdoor_temperature = self._object_value(
self._home_location, "outdoor_temperature"
)
if outdoor_temperature is not None:
data.update({"sensors": {"outdoor_temperature": outdoor_temperature}})
locator = "./logs/point_log[type='outdoor_temperature']/period/measurement"
if (found := self._home_location.find(locator)) is not None:
value = format_measure(found.text, NONE)
data.update({"sensors": {"outdoor_temperature": value}})
self._count += 1

def _object_value(self, obj_id: str, measurement: str) -> float | int | None:
"""Helper-function for smile.py: _get_entity_data().

Obtain the value/state for the given object from a location in DOMAIN_OBJECTS
"""
val: float | int | None = None
search = self._domain_objects
locator = f'./location[@id="{obj_id}"]/logs/point_log[type="{measurement}"]/period/measurement'
if (found := search.find(locator)) is not None:
val = format_measure(found.text, NONE)

return val

def _process_c_heating_state(self, data: GwEntityData) -> None:
"""Helper-function for _get_measurement_data().

Expand Down
43 changes: 14 additions & 29 deletions plugwise/legacy/helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ def __init__(self) -> None:
self._count: int
self._domain_objects: etree
self._heater_id: str
self._home_location: str
self._home_loc_id: str
bouwew marked this conversation as resolved.
Show resolved Hide resolved
self._is_thermostat: bool
self._last_modified: dict[str, str] = {}
self._loc_data: dict[str, ThermoLoc]
Expand Down Expand Up @@ -115,7 +115,7 @@ def _all_appliances(self) -> None:
):
continue # pragma: no cover

appl.location = self._home_location
appl.location = self._home_loc_id
appl.entity_id = appliance.attrib["id"]
appl.name = appliance.find("name").text
# Extend device_class name when a Circle/Stealth is type heater_central -- Pw-Beta Issue #739
Expand Down Expand Up @@ -161,7 +161,7 @@ def _all_locations(self) -> None:

# Legacy Anna without outdoor_temp and Stretches have no locations, create fake location-data
if not (locations := self._locations.findall("./location")):
self._home_location = FAKE_LOC
self._home_loc_id = FAKE_LOC
self._loc_data[FAKE_LOC] = {"name": "Home"}
return

Expand All @@ -174,11 +174,11 @@ def _all_locations(self) -> None:
continue

if loc.name == "Home":
self._home_location = loc.loc_id
self._home_loc_id = loc.loc_id
# Replace location-name for P1 legacy, can contain privacy-related info
if self.smile_type == "power":
loc.name = "Home"
self._home_location = loc.loc_id
self._home_loc_id = loc.loc_id

self._loc_data[loc.loc_id] = {"name": loc.name}

Expand All @@ -187,15 +187,15 @@ def _create_legacy_gateway(self) -> None:

Use the home_location or FAKE_APPL as entity id.
"""
self.gateway_id = self._home_location
self.gateway_id = self._home_loc_id
if self.smile_type == "power":
self.gateway_id = FAKE_APPL

self.gw_entities[self.gateway_id] = {"dev_class": "gateway"}
self._count += 1
for key, value in {
"firmware": str(self.smile_fw_version),
"location": self._home_location,
"location": self._home_loc_id,
"mac_address": self.smile_mac_address,
"model": self.smile_model,
"name": self.smile_name,
Expand Down Expand Up @@ -272,7 +272,7 @@ def _get_measurement_data(self, entity_id: str) -> GwEntityData:
Collect the appliance-data based on entity_id.
"""
data: GwEntityData = {"binary_sensors": {}, "sensors": {}, "switches": {}}
# Get P1 smartmeter data from LOCATIONS or MODULES
# Get P1 smartmeter data from MODULES
entity = self.gw_entities[entity_id]
# !! DON'T CHANGE below two if-lines, will break stuff !!
if self.smile_type == "power":
Expand All @@ -294,14 +294,13 @@ def _get_measurement_data(self, entity_id: str) -> GwEntityData:
if appliance.find("type").text in ACTUATOR_CLASSES:
self._get_actuator_functionalities(appliance, entity, data)

# 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
# Anna: the Smile outdoor_temperature is present in the Home location
# For some Anna's LOCATIONS is empty, falling back to domain_objects!
if self._is_thermostat and entity_id == self.gateway_id:
outdoor_temperature = self._object_value(
self._home_location, "outdoor_temperature"
)
if outdoor_temperature is not None:
data.update({"sensors": {"outdoor_temperature": outdoor_temperature}})
locator = f"./location[@id='{self._home_loc_id}']/logs/point_log[type='outdoor_temperature']/period/measurement"
if (found := self._domain_objects.find(locator)) is not None:
value = format_measure(found.text, NONE)
data.update({"sensors": {"outdoor_temperature": value}})
self._count += 1

if "c_heating_state" in data:
Expand Down Expand Up @@ -396,20 +395,6 @@ def _get_actuator_functionalities(
act_item = cast(ActuatorType, item)
data[act_item] = temp_dict

def _object_value(self, obj_id: str, measurement: str) -> float | int | None:
"""Helper-function for smile.py: _get_entity_data().

Obtain the value/state for the given object from a location in DOMAIN_OBJECTS
"""
val: float | int | None = None
search = self._domain_objects
locator = f'./location[@id="{obj_id}"]/logs/point_log[type="{measurement}"]/period/measurement'
if (found := search.find(locator)) is not None:
val = format_measure(found.text, NONE)
return val

return val

def _preset(self) -> str | None:
"""Helper-function for smile.py: _climate_data().

Expand Down
37 changes: 16 additions & 21 deletions plugwise/util.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
SPECIAL_FORMAT,
SPECIALS,
SWITCHES,
TEMP_CELSIUS,
UOM,
BinarySensorType,
GwEntityData,
Expand Down Expand Up @@ -154,26 +153,22 @@ def escape_illegal_xml_characters(xmldata: str) -> str:
def format_measure(measure: str, unit: str) -> float | int:
"""Format measure to correct type."""
result: float | int = 0
try:
result = int(measure)
if unit == TEMP_CELSIUS:
result = float(measure)
except ValueError:
float_measure = float(measure)
if unit == PERCENTAGE and 0 < float_measure <= 1:
return int(float_measure * 100)

if unit == ENERGY_KILO_WATT_HOUR:
float_measure = float_measure / 1000

if unit in SPECIAL_FORMAT:
result = float(f"{round(float_measure, 3):.3f}")
elif unit == ELECTRIC_POTENTIAL_VOLT:
result = float(f"{round(float_measure, 1):.1f}")
elif abs(float_measure) < 10:
result = float(f"{round(float_measure, 2):.2f}")
elif abs(float_measure) >= 10:
result = float(f"{round(float_measure, 1):.1f}")

float_measure = float(measure)
if unit == PERCENTAGE and 0 < float_measure <= 1:
return int(float_measure * 100)

if unit == ENERGY_KILO_WATT_HOUR:
float_measure = float_measure / 1000

if unit in SPECIAL_FORMAT:
result = float(f"{round(float_measure, 3):.3f}")
elif unit == ELECTRIC_POTENTIAL_VOLT:
result = float(f"{round(float_measure, 1):.1f}")
elif abs(float_measure) < 10:
result = float(f"{round(float_measure, 2):.2f}")
elif abs(float_measure) >= 10:
result = float(f"{round(float_measure, 1):.1f}")

return result

Expand Down
Loading