Skip to content

Commit

Permalink
Merge pull request #678 from plugwise/improve-241222
Browse files Browse the repository at this point in the history 
Continuous improvement
  • Loading branch information
@bouwew
bouwew authored Dec 22, 2024
2 parents 5806452 + 566f266 commit 9fae322
Show file tree
Hide file tree
Showing 4 changed files with 58 additions and 93 deletions.
4 changes: 4 additions & 0 deletions CHANGELOG.md
View file
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
View file
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
View file
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
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
View file
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

0 comments on commit 9fae322

Please sign in to comment.