mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 13:17:32 +00:00
Refactor ZHA electrical measurement sensor. (#30130)
This commit is contained in:
parent
834929a14e
commit
8e3dfbd5c9
@ -5,6 +5,7 @@ For more details about this component, please refer to the documentation at
|
||||
https://home-assistant.io/integrations/zha/
|
||||
"""
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
import zigpy.zcl.clusters.homeautomation as homeautomation
|
||||
|
||||
@ -65,6 +66,12 @@ class ElectricalMeasurementChannel(AttributeListeningChannel):
|
||||
|
||||
REPORT_CONFIG = ({"attr": "active_power", "config": REPORT_CONFIG_DEFAULT},)
|
||||
|
||||
def __init__(self, cluster, device):
|
||||
"""Initialize Metering."""
|
||||
super().__init__(cluster, device)
|
||||
self._divisor = None
|
||||
self._multiplier = None
|
||||
|
||||
async def async_update(self):
|
||||
"""Retrieve latest state."""
|
||||
self.debug("async_update")
|
||||
@ -78,8 +85,39 @@ class ElectricalMeasurementChannel(AttributeListeningChannel):
|
||||
async def async_initialize(self, from_cache):
|
||||
"""Initialize channel."""
|
||||
await self.get_attribute_value("active_power", from_cache=from_cache)
|
||||
await self.fetch_config(from_cache)
|
||||
await super().async_initialize(from_cache)
|
||||
|
||||
async def fetch_config(self, from_cache):
|
||||
"""Fetch config from device and updates format specifier."""
|
||||
divisor = await self.get_attribute_value(
|
||||
"ac_power_divisor", from_cache=from_cache
|
||||
)
|
||||
if divisor is None:
|
||||
divisor = await self.get_attribute_value(
|
||||
"power_divisor", from_cache=from_cache
|
||||
)
|
||||
self._divisor = divisor
|
||||
|
||||
mult = await self.get_attribute_value(
|
||||
"ac_power_multiplier", from_cache=from_cache
|
||||
)
|
||||
if mult is None:
|
||||
mult = await self.get_attribute_value(
|
||||
"power_multiplier", from_cache=from_cache
|
||||
)
|
||||
self._multiplier = mult
|
||||
|
||||
@property
|
||||
def divisor(self) -> Optional[int]:
|
||||
"""Return active power divisor."""
|
||||
return self._divisor or 1
|
||||
|
||||
@property
|
||||
def multiplier(self) -> Optional[int]:
|
||||
"""Return active power divisor."""
|
||||
return self._multiplier or 1
|
||||
|
||||
|
||||
@registries.ZIGBEE_CHANNEL_REGISTRY.register(
|
||||
homeautomation.MeterIdentification.cluster_id
|
||||
|
@ -217,6 +217,10 @@ class ElectricalMeasurement(Sensor):
|
||||
"""Return True if HA needs to poll for state changes."""
|
||||
return True
|
||||
|
||||
def formatter(self, value) -> int:
|
||||
"""Return 'normalized' value."""
|
||||
return round(value * self._channel.multiplier / self._channel.divisor)
|
||||
|
||||
|
||||
@STRICT_MATCH(MatchRule(generic_ids={CHANNEL_ST_HUMIDITY_CLUSTER}))
|
||||
@STRICT_MATCH(MatchRule(channel_names={CHANNEL_HUMIDITY}))
|
||||
|
@ -166,7 +166,7 @@ async def async_test_metering(hass, device_info):
|
||||
async def async_test_electrical_measurement(hass, device_info):
|
||||
"""Test electrical measurement sensor."""
|
||||
await send_attribute_report(hass, device_info["cluster"], 1291, 100)
|
||||
assert_state(hass, device_info, "10.0", "W")
|
||||
assert_state(hass, device_info, "100", "W")
|
||||
|
||||
|
||||
async def send_attribute_report(hass, cluster, attrid, value):
|
||||
|
Loading…
x
Reference in New Issue
Block a user