Energy-device-related improvements

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Ongoing
 
+- Fix typing of P1 sensors, energy-device-related improvements.
 - Rename mode to climate_mode.
 
 ## v1.5.0

plugwise/constants.py

@@ -431,30 +431,30 @@ class SmileSensors(TypedDict, total=False):
  dhw_temperature: float
  domestic_hot_water_setpoint: float
  temperature: float
- electricity_consumed: float
- electricity_consumed_interval: float
+ electricity_consumed: int
+ electricity_consumed_interval:int
  electricity_consumed_off_peak_cumulative: float
  electricity_consumed_off_peak_interval: int
  electricity_consumed_off_peak_point: int
  electricity_consumed_peak_cumulative: float
  electricity_consumed_peak_interval: int
  electricity_consumed_peak_point: int
- electricity_consumed_point: float
- electricity_phase_one_consumed: float
- electricity_phase_two_consumed: float
- electricity_phase_three_consumed: float
- electricity_phase_one_produced: float
- electricity_phase_two_produced: float
- electricity_phase_three_produced: float
- electricity_produced: float
- electricity_produced_interval: float
+ electricity_consumed_point: int
+ electricity_phase_one_consumed: int
+ electricity_phase_two_consumed: int
+ electricity_phase_three_consumed: int
+ electricity_phase_one_produced: int
+ electricity_phase_two_produced: int
+ electricity_phase_three_produced: int
+ electricity_produced: int
+ electricity_produced_interval: int
  electricity_produced_off_peak_cumulative: float
  electricity_produced_off_peak_interval: int
  electricity_produced_off_peak_point: int
  electricity_produced_peak_cumulative: float
  electricity_produced_peak_interval: int
  electricity_produced_peak_point: int
- electricity_produced_point: float
+ electricity_produced_point: int
  gas_consumed_cumulative: float
  gas_consumed_interval: float
  humidity: float

plugwise/helper.py

@@ -269,6 +269,10 @@ def _all_appliances(self) -> None:
  for appliance in self._domain_objects.findall("./appliance"):
  appl = Munch()
  appl.pwclass = appliance.find("type").text
+ # Don't collect data for the OpenThermGateway appliance
+ if appl.pwclass == "open_therm_gateway":
+ continue
+
  # Extend device_class name of Plugs (Plugwise and Aqara) - Pw-Beta Issue #739
  description = appliance.find("description").text
  if description is not None and (
@@ -291,6 +295,10 @@ def _all_appliances(self) -> None:
  elif appl.pwclass not in THERMOSTAT_CLASSES:
  appl.location = self._home_location
 
+ # Don't show orphaned thermostat-types
+ if appl.pwclass in THERMOSTAT_CLASSES and appl.location is None:
+ continue
+
  appl.dev_id = appliance.attrib["id"]
  appl.name = appliance.find("name").text
  appl.model = None
@@ -301,24 +309,15 @@ def _all_appliances(self) -> None:
  appl.zigbee_mac = None
  appl.vendor_name = None
 
- # Determine class for this appliance
- # Skip on heater_central when no active device present
+ # Collect appliance info, skip orphaned/removed devices
  if not (appl := self._appliance_info_finder(appl, appliance)):
  continue
 
- # Skip orphaned heater_central (Core Issue #104433)
- if appl.pwclass == "heater_central" and appl.dev_id != self._heater_id:
- continue
-
  # P1: for gateway and smartmeter switch device_id - part 1
  # This is done to avoid breakage in HA Core
  if appl.pwclass == "gateway" and self.smile_type == "power":
  appl.dev_id = appl.location
 
- # Don't show orphaned thermostat-types or the OpenTherm Gateway.
- if appl.pwclass in THERMOSTAT_CLASSES and appl.location is None:
- continue
-
  self._create_gw_devices(appl)
 
  # For P1 collect the connected SmartMeter info
@@ -354,23 +353,32 @@ def _all_locations(self) -> None:
  self.loc_data[loc.loc_id] = {"name": loc.name}
 
  def _p1_smartmeter_info_finder(self, appl: Munch) -> None:
- """Collect P1 DSMR Smartmeter info."""
+ """Collect P1 DSMR SmartMeter info."""
  loc_id = next(iter(self.loc_data.keys()))
+ location = self._domain_objects.find(f'./location[@id="{loc_id}"]')
+ locator = "./logs/point_log/electricity_point_meter"
+ mod_type = "electricity_point_meter"
+ module_data = self._get_module_data(location, locator, mod_type)
+ if not module_data["contents"]:
+ LOGGER.error("No module data found for SmartMeter") # pragma: no cover
+ return None # pragma: no cover
+
  appl.dev_id = self.gateway_id
+ appl.firmware = module_data["firmware_version"]
+ appl.hardware = module_data["hardware_version"]
  appl.location = loc_id
  appl.mac = None
- appl.model = None
- appl.model_id = None
+ appl.model = module_data["vendor_model"]
+ appl.model_id = None # don't use model_id for SmartMeter
  appl.name = "P1"
  appl.pwclass = "smartmeter"
+ appl.vendor_name = module_data["vendor_name"]
  appl.zigbee_mac = None
- location = self._domain_objects.find(f'./location[@id="{loc_id}"]')
- appl = self._energy_device_info_finder(appl, location)
 
  self._create_gw_devices(appl)
 
  def _appliance_info_finder(self, appl: Munch, appliance: etree) -> Munch:
- """Collect device info (Smile/Stretch, Thermostats, OpenTherm/On-Off): firmware, model and vendor name."""
+ """Collect info for all appliances found."""
  match appl.pwclass:
  case "gateway":
  # Collect gateway device info
@@ -382,45 +390,28 @@ def _appliance_info_finder(self, appl: Munch, appliance: etree) -> Munch:
  # Collect heater_central device info
  self._appl_heater_central_info(appl, appliance, False) # False means non-legacy device
  self._appl_dhw_mode_info(appl, appliance)
+ # Skip orphaned heater_central (Core Issue #104433)
+ if appl.dev_id != self._heater_id:
+ return Munch()
+ return appl
+ case _ as s if s.endswith("_plug"):
+ # Collect info from plug-types (Plug, Aqara Smart Plug)
+ locator = "./logs/interval_log/electricity_interval_meter"
+ mod_type = "electricity_interval_meter"
+ module_data = self._get_module_data(appliance, locator, mod_type)
+ # A plug without module-data is orphaned/ no present
+ if not module_data["contents"]:
+ return Munch()
+
+ appl.firmware = module_data["firmware_version"]
+ appl.hardware = module_data["hardware_version"]
+ appl.model_id = module_data["vendor_model"]
+ appl.vendor_name = module_data["vendor_name"]
+ appl.model = check_model(appl.model_id, appl.vendor_name)
+ appl.zigbee_mac = module_data["zigbee_mac_address"]
+ return appl
+ case _: # pragma: no cover
  return appl
- case _:
- # Collect info from power-related devices (Plug, Aqara Smart Plug)
- return self._energy_device_info_finder(appl, appliance)
-
- def _energy_device_info_finder(self, appl: Munch, appliance: etree) -> Munch:
- """Helper-function for _appliance_info_finder().
-
- Collect energy device info (Smartmeter): firmware, model and vendor name.
- """
- if self.smile_type == "power":
- locator = "./logs/point_log/electricity_point_meter"
- mod_type = "electricity_point_meter"
- module_data = self._get_module_data(appliance, locator, mod_type)
- appl.hardware = module_data["hardware_version"]
- appl.model = module_data["vendor_model"] # don't use model_id for Smartmeter
- appl.vendor_name = module_data["vendor_name"]
- appl.firmware = module_data["firmware_version"]
-
- return appl
-
- if self.smile(ADAM):
- locator = "./logs/interval_log/electricity_interval_meter"
- mod_type = "electricity_interval_meter"
- module_data = self._get_module_data(appliance, locator, mod_type)
- # Filter appliance without zigbee_mac, it's an orphaned device
- appl.zigbee_mac = module_data["zigbee_mac_address"]
- if appl.zigbee_mac is None:
- return None
-
- appl.vendor_name = module_data["vendor_name"]
- appl.model_id = module_data["vendor_model"]
- appl.model = check_model(appl.model_id, appl.vendor_name)
- appl.hardware = module_data["hardware_version"]
- appl.firmware = module_data["firmware_version"]
-
- return appl
-
- return appl # pragma: no cover
 
  def _appl_gateway_info(self, appl: Munch, appliance: etree) -> Munch:
  """Helper-function for _appliance_info_finder()."""
@@ -725,15 +716,18 @@ def _wireless_availability(self, appliance: etree, data: DeviceData) -> None:
  Collect the availability-status for wireless connected devices.
  """
  if self.smile(ADAM):
- # Collect for Plugs
+ # Try collecting for a Plug
  locator = "./logs/interval_log/electricity_interval_meter"
  mod_type = "electricity_interval_meter"
  module_data = self._get_module_data(appliance, locator, mod_type)
- if module_data["reachable"] is None:
- # Collect for wireless thermostats
+ if not module_data["contents"]:
+ # Try collecting for a wireless thermostat
  locator = "./logs/point_log[type='thermostat']/thermostat"
  mod_type = "thermostat"
  module_data = self._get_module_data(appliance, locator, mod_type)
+ if not module_data["contents"]:
+ LOGGER.error("No module data found for Plug or wireless thermostat") # pragma: no cover
+ return None # pragma: no cover
 
  if module_data["reachable"] is not None:
  data["available"] = module_data["reachable"]

plugwise/util.py

@@ -115,7 +115,7 @@ def check_model(name: str | None, vendor_name: str | None) -> str | None:
  if name is not None and "lumi.plug" in name:
  return "Aqara Smart Plug"
 
- return name
+ return name # pragma: no cover
 
 
 def common_match_cases(

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "plugwise"
-version = "1.5.0"
+version = "1.5.1a3"
 license = {file = "LICENSE"}
 description = "Plugwise Smile (Adam/Anna/P1) and Stretch module for Python 3."
 readme = "README.md"