jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-05-03 05:29:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use DataUpdateCoordinator in Vallox (#56966)

Browse Source
This commit is contained in:
Andre Richter 2021-10-24 23:21:35 +02:00 committed by GitHub
parent 02372cd65a
commit 6c01ed8d97
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 176 additions and 264 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

162
homeassistant/components/vallox/__init__.py
View File

@ -1,7 +1,7 @@
"""Support for Vallox ventilation units.""" """Support for Vallox ventilation units."""
from __future__ import annotations from __future__ import annotations
from datetime import datetime from dataclasses import dataclass, field
import ipaddress import ipaddress
import logging import logging
from typing import Any from typing import Any
@ -11,13 +11,12 @@ from vallox_websocket_api.constants import vlxDevConstants
from vallox_websocket_api.exceptions import ValloxApiException from vallox_websocket_api.exceptions import ValloxApiException
import voluptuous as vol import voluptuous as vol
from homeassistant.const import CONF_HOST, CONF_NAME from homeassistant.const import CONF_HOST, CONF_NAME, EVENT_HOMEASSISTANT_STARTED
from homeassistant.core import HomeAssistant, ServiceCall from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import config_validation as cv from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.discovery import async_load_platform from homeassistant.helpers.discovery import async_load_platform
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.typing import ConfigType, StateType from homeassistant.helpers.typing import ConfigType, StateType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import ( from .const import (
DEFAULT_FAN_SPEED_AWAY, DEFAULT_FAN_SPEED_AWAY,
@ -28,8 +27,7 @@ from .const import (
METRIC_KEY_PROFILE_FAN_SPEED_AWAY, METRIC_KEY_PROFILE_FAN_SPEED_AWAY,
METRIC_KEY_PROFILE_FAN_SPEED_BOOST, METRIC_KEY_PROFILE_FAN_SPEED_BOOST,
METRIC_KEY_PROFILE_FAN_SPEED_HOME, METRIC_KEY_PROFILE_FAN_SPEED_HOME,
SIGNAL_VALLOX_STATE_UPDATE, STATE_SCAN_INTERVAL,
STATE_PROXY_SCAN_INTERVAL,
STR_TO_VALLOX_PROFILE_SETTABLE, STR_TO_VALLOX_PROFILE_SETTABLE,
) )
@ -91,6 +89,40 @@ SERVICE_TO_METHOD = {
} }
@dataclass
class ValloxState:
"""Describes the current state of the unit."""
metric_cache: dict[str, Any] = field(default_factory=dict)
profile: VALLOX_PROFILE = VALLOX_PROFILE.NONE
def get_metric(self, metric_key: str) -> StateType:
"""Return cached state value."""
_LOGGER.debug("Fetching metric key: %s", metric_key)
if metric_key not in vlxDevConstants.__dict__:
_LOGGER.debug("Metric key invalid: %s", metric_key)
if (value := self.metric_cache.get(metric_key)) is None:
return None
if not isinstance(value, (str, int, float)):
_LOGGER.debug(
"Return value of metric %s has unexpected type %s",
metric_key,
type(value),
)
return None
return value
class ValloxDataUpdateCoordinator(DataUpdateCoordinator):
"""The DataUpdateCoordinator for Vallox."""
data: ValloxState
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the client and boot the platforms.""" """Set up the client and boot the platforms."""
conf = config[DOMAIN] conf = config[DOMAIN]
@ -98,102 +130,74 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
name = conf.get(CONF_NAME) name = conf.get(CONF_NAME)
client = Vallox(host) client = Vallox(host)
state_proxy = ValloxStateProxy(hass, client)
service_handler = ValloxServiceHandler(client, state_proxy)
hass.data[DOMAIN] = {"client": client, "state_proxy": state_proxy, "name": name} async def async_update_data() -> ValloxState:
"""Fetch state update."""
_LOGGER.debug("Updating Vallox state cache")
try:
metric_cache = await client.fetch_metrics()
profile = await client.get_profile()
except (OSError, ValloxApiException) as err:
raise UpdateFailed("Error during state cache update") from err
return ValloxState(metric_cache, profile)
coordinator = ValloxDataUpdateCoordinator(
hass,
_LOGGER,
name=f"{name} DataUpdateCoordinator",
update_interval=STATE_SCAN_INTERVAL,
update_method=async_update_data,
)
service_handler = ValloxServiceHandler(client, coordinator)
for vallox_service, method in SERVICE_TO_METHOD.items(): for vallox_service, method in SERVICE_TO_METHOD.items():
schema = method["schema"] schema = method["schema"]
hass.services.async_register( hass.services.async_register(
DOMAIN, vallox_service, service_handler.async_handle, schema=schema DOMAIN, vallox_service, service_handler.async_handle, schema=schema
) )
# The vallox hardware expects quite strict timings for websocket requests. Timings that machines hass.data[DOMAIN] = {"client": client, "coordinator": coordinator, "name": name}
# with less processing power, like Raspberries, cannot live up to during the busy start phase of
# Home Asssistant. Hence, async_add_entities() for fan and sensor in respective code will be
# called with update_before_add=False to intentionally delay the first request, increasing
# chance that it is issued only when the machine is less busy again.
hass.async_create_task(async_load_platform(hass, "sensor", DOMAIN, {}, config))
hass.async_create_task(async_load_platform(hass, "fan", DOMAIN, {}, config))
async_track_time_interval(hass, state_proxy.async_update, STATE_PROXY_SCAN_INTERVAL) async def _async_load_platform_delayed(*_: Any) -> None:
await coordinator.async_refresh()
hass.async_create_task(async_load_platform(hass, "sensor", DOMAIN, {}, config))
hass.async_create_task(async_load_platform(hass, "fan", DOMAIN, {}, config))
# The Vallox hardware expects quite strict timings for websocket requests. Timings that machines
# with less processing power, like a Raspberry Pi, cannot live up to during the busy start phase
# of Home Asssistant.
#
# Hence, wait for the started event before doing a first data refresh and loading the platforms,
# because it usually means the system is less busy after the event and can now meet the
# websocket timing requirements.
hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STARTED, _async_load_platform_delayed
)
return True return True
class ValloxStateProxy:
"""Helper class to reduce websocket API calls."""
def __init__(self, hass: HomeAssistant, client: Vallox) -> None:
"""Initialize the proxy."""
self._hass = hass
self._client = client
self._metric_cache: dict[str, Any] = {}
self._profile = VALLOX_PROFILE.NONE
self._valid = False
def fetch_metric(self, metric_key: str) -> StateType:
"""Return cached state value."""
_LOGGER.debug("Fetching metric key: %s", metric_key)
if not self._valid:
raise OSError("Device state out of sync.")
if metric_key not in vlxDevConstants.__dict__:
raise KeyError(f"Unknown metric key: {metric_key}")
if (value := self._metric_cache[metric_key]) is None:
return None
if not isinstance(value, (str, int, float)):
raise TypeError(
f"Return value of metric {metric_key} has unexpected type {type(value)}"
)
return value
def get_profile(self) -> VALLOX_PROFILE:
"""Return cached profile value."""
_LOGGER.debug("Returning profile")
if not self._valid:
raise OSError("Device state out of sync.")
return self._profile
async def async_update(self, time: datetime | None = None) -> None:
"""Fetch state update."""
_LOGGER.debug("Updating Vallox state cache")
try:
self._metric_cache = await self._client.fetch_metrics()
self._profile = await self._client.get_profile()
except (OSError, ValloxApiException) as err:
self._valid = False
_LOGGER.error("Error during state cache update: %s", err)
return
self._valid = True
async_dispatcher_send(self._hass, SIGNAL_VALLOX_STATE_UPDATE)
class ValloxServiceHandler: class ValloxServiceHandler:
"""Services implementation.""" """Services implementation."""
def __init__(self, client: Vallox, state_proxy: ValloxStateProxy) -> None: def __init__(
self, client: Vallox, coordinator: DataUpdateCoordinator[ValloxState]
) -> None:
"""Initialize the proxy.""" """Initialize the proxy."""
self._client = client self._client = client
self._state_proxy = state_proxy self._coordinator = coordinator
async def async_set_profile(self, profile: str = "Home") -> bool: async def async_set_profile(self, profile: str = "Home") -> bool:
"""Set the ventilation profile.""" """Set the ventilation profile."""
_LOGGER.debug("Setting ventilation profile to: %s", profile) _LOGGER.debug("Setting ventilation profile to: %s", profile)
_LOGGER.warning( _LOGGER.warning(
"Attention: The service 'vallox.set_profile' is superseded by the 'fan.set_preset_mode' service." "Attention: The service 'vallox.set_profile' is superseded by the "
"It will be removed in the future, please migrate to 'fan.set_preset_mode' to prevent breakage" "'fan.set_preset_mode' service. It will be removed in the future, please migrate to "
"'fan.set_preset_mode' to prevent breakage"
) )
try: try:
@ -269,4 +273,4 @@ class ValloxServiceHandler:
# This state change affects other entities like sensors. Force an immediate update that can # This state change affects other entities like sensors. Force an immediate update that can
# be observed by all parties involved. # be observed by all parties involved.
if result: if result:
await self._state_proxy.async_update() await self._coordinator.async_request_refresh()

3
homeassistant/components/vallox/const.py
View File

@ -7,8 +7,7 @@ from vallox_websocket_api import PROFILE as VALLOX_PROFILE
DOMAIN = "vallox" DOMAIN = "vallox"
DEFAULT_NAME = "Vallox" DEFAULT_NAME = "Vallox"
SIGNAL_VALLOX_STATE_UPDATE = "vallox_state_update" STATE_SCAN_INTERVAL = timedelta(seconds=60)
STATE_PROXY_SCAN_INTERVAL = timedelta(seconds=60)
# Common metric keys and (default) values. # Common metric keys and (default) values.
METRIC_KEY_MODE = "A_CYC_MODE" METRIC_KEY_MODE = "A_CYC_MODE"

142
homeassistant/components/vallox/fan.py
View File

@ -3,7 +3,7 @@ from __future__ import annotations
from collections.abc import Mapping from collections.abc import Mapping
import logging import logging
from typing import Any from typing import Any, NamedTuple
from vallox_websocket_api import Vallox from vallox_websocket_api import Vallox
from vallox_websocket_api.exceptions import ValloxApiException from vallox_websocket_api.exceptions import ValloxApiException
@ -13,12 +13,12 @@ from homeassistant.components.fan import (
FanEntity, FanEntity,
NotValidPresetModeError, NotValidPresetModeError,
) )
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import ValloxStateProxy from . import ValloxDataUpdateCoordinator
from .const import ( from .const import (
DOMAIN, DOMAIN,
METRIC_KEY_MODE, METRIC_KEY_MODE,
@ -27,25 +27,38 @@ from .const import (
METRIC_KEY_PROFILE_FAN_SPEED_HOME, METRIC_KEY_PROFILE_FAN_SPEED_HOME,
MODE_OFF, MODE_OFF,
MODE_ON, MODE_ON,
SIGNAL_VALLOX_STATE_UPDATE,
STR_TO_VALLOX_PROFILE_SETTABLE, STR_TO_VALLOX_PROFILE_SETTABLE,
VALLOX_PROFILE_TO_STR_SETTABLE, VALLOX_PROFILE_TO_STR_SETTABLE,
) )
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
ATTR_PROFILE_FAN_SPEED_HOME = {
"description": "fan_speed_home", class ExtraStateAttributeDetails(NamedTuple):
"metric_key": METRIC_KEY_PROFILE_FAN_SPEED_HOME, """Extra state attribute properties."""
}
ATTR_PROFILE_FAN_SPEED_AWAY = { description: str
"description": "fan_speed_away", metric_key: str
"metric_key": METRIC_KEY_PROFILE_FAN_SPEED_AWAY,
}
ATTR_PROFILE_FAN_SPEED_BOOST = { EXTRA_STATE_ATTRIBUTES = (
"description": "fan_speed_boost", ExtraStateAttributeDetails(
"metric_key": METRIC_KEY_PROFILE_FAN_SPEED_BOOST, description="fan_speed_home", metric_key=METRIC_KEY_PROFILE_FAN_SPEED_HOME
} ),
ExtraStateAttributeDetails(
description="fan_speed_away", metric_key=METRIC_KEY_PROFILE_FAN_SPEED_AWAY
),
ExtraStateAttributeDetails(
description="fan_speed_boost", metric_key=METRIC_KEY_PROFILE_FAN_SPEED_BOOST
),
)
def _convert_fan_speed_value(value: StateType) -> int | None:
if isinstance(value, (int, float)):
return int(value)
return None
async def async_setup_platform( async def async_setup_platform(
@ -62,31 +75,29 @@ async def async_setup_platform(
client.set_settable_address(METRIC_KEY_MODE, int) client.set_settable_address(METRIC_KEY_MODE, int)
device = ValloxFan( device = ValloxFan(
hass.data[DOMAIN]["name"], client, hass.data[DOMAIN]["state_proxy"] hass.data[DOMAIN]["name"], client, hass.data[DOMAIN]["coordinator"]
) )
async_add_entities([device], update_before_add=False) async_add_entities([device])
class ValloxFan(FanEntity): class ValloxFan(CoordinatorEntity, FanEntity):
"""Representation of the fan.""" """Representation of the fan."""
_attr_should_poll = False coordinator: ValloxDataUpdateCoordinator
def __init__( def __init__(
self, name: str, client: Vallox, state_proxy: ValloxStateProxy self,
name: str,
client: Vallox,
coordinator: ValloxDataUpdateCoordinator,
) -> None: ) -> None:
"""Initialize the fan.""" """Initialize the fan."""
super().__init__(coordinator)
self._client = client self._client = client
self._state_proxy = state_proxy
self._is_on = False
self._preset_mode: str | None = None
self._fan_speed_home: int | None = None
self._fan_speed_away: int | None = None
self._fan_speed_boost: int | None = None
self._attr_name = name self._attr_name = name
self._attr_available = False
@property @property
def supported_features(self) -> int: def supported_features(self) -> int:
@ -102,73 +113,24 @@ class ValloxFan(FanEntity):
@property @property
def is_on(self) -> bool: def is_on(self) -> bool:
"""Return if device is on.""" """Return if device is on."""
return self._is_on return self.coordinator.data.get_metric(METRIC_KEY_MODE) == MODE_ON
@property @property
def preset_mode(self) -> str | None: def preset_mode(self) -> str | None:
"""Return the current preset mode.""" """Return the current preset mode."""
return self._preset_mode vallox_profile = self.coordinator.data.profile
return VALLOX_PROFILE_TO_STR_SETTABLE.get(vallox_profile)
@property @property
def extra_state_attributes(self) -> Mapping[str, int | None]: def extra_state_attributes(self) -> Mapping[str, int | None]:
"""Return device specific state attributes.""" """Return device specific state attributes."""
data = self.coordinator.data
return { return {
ATTR_PROFILE_FAN_SPEED_HOME["description"]: self._fan_speed_home, attr.description: _convert_fan_speed_value(data.get_metric(attr.metric_key))
ATTR_PROFILE_FAN_SPEED_AWAY["description"]: self._fan_speed_away, for attr in EXTRA_STATE_ATTRIBUTES
ATTR_PROFILE_FAN_SPEED_BOOST["description"]: self._fan_speed_boost,
} }
async def async_added_to_hass(self) -> None:
"""Call to update."""
self.async_on_remove(
async_dispatcher_connect(
self.hass, SIGNAL_VALLOX_STATE_UPDATE, self._update_callback
)
)
@callback
def _update_callback(self) -> None:
"""Call update method."""
self.async_schedule_update_ha_state(True)
async def async_update(self) -> None:
"""Fetch state from the device."""
try:
# Fetch if the whole device is in regular operation state.
self._is_on = self._state_proxy.fetch_metric(METRIC_KEY_MODE) == MODE_ON
vallox_profile = self._state_proxy.get_profile()
# Fetch the profile fan speeds.
fan_speed_home = self._state_proxy.fetch_metric(
ATTR_PROFILE_FAN_SPEED_HOME["metric_key"]
)
fan_speed_away = self._state_proxy.fetch_metric(
ATTR_PROFILE_FAN_SPEED_AWAY["metric_key"]
)
fan_speed_boost = self._state_proxy.fetch_metric(
ATTR_PROFILE_FAN_SPEED_BOOST["metric_key"]
)
except (OSError, KeyError, TypeError) as err:
self._attr_available = False
_LOGGER.error("Error updating fan: %s", err)
return
self._preset_mode = VALLOX_PROFILE_TO_STR_SETTABLE.get(vallox_profile)
self._fan_speed_home = (
int(fan_speed_home) if isinstance(fan_speed_home, (int, float)) else None
)
self._fan_speed_away = (
int(fan_speed_away) if isinstance(fan_speed_away, (int, float)) else None
)
self._fan_speed_boost = (
int(fan_speed_boost) if isinstance(fan_speed_boost, (int, float)) else None
)
self._attr_available = True
async def _async_set_preset_mode_internal(self, preset_mode: str) -> bool: async def _async_set_preset_mode_internal(self, preset_mode: str) -> bool:
""" """
Set new preset mode. Set new preset mode.
@ -201,7 +163,7 @@ class ValloxFan(FanEntity):
if update_needed: if update_needed:
# This state change affects other entities like sensors. Force an immediate update that # This state change affects other entities like sensors. Force an immediate update that
# can be observed by all parties involved. # can be observed by all parties involved.
await self._state_proxy.async_update() await self.coordinator.async_request_refresh()
async def async_turn_on( async def async_turn_on(
self, self,
@ -211,7 +173,7 @@ class ValloxFan(FanEntity):
**kwargs: Any, **kwargs: Any,
) -> None: ) -> None:
"""Turn the device on.""" """Turn the device on."""
_LOGGER.debug("Turn on: %s", speed) _LOGGER.debug("Turn on")
update_needed = False update_needed = False
@ -231,7 +193,7 @@ class ValloxFan(FanEntity):
if update_needed: if update_needed:
# This state change affects other entities like sensors. Force an immediate update that # This state change affects other entities like sensors. Force an immediate update that
# can be observed by all parties involved. # can be observed by all parties involved.
await self._state_proxy.async_update() await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs: Any) -> None: async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the device off.""" """Turn the device off."""
@ -246,4 +208,4 @@ class ValloxFan(FanEntity):
return return
# Same as for turn_on method. # Same as for turn_on method.
await self._state_proxy.async_update() await self.coordinator.async_request_refresh()

133
homeassistant/components/vallox/sensor.py
View File

@ -19,143 +19,91 @@ from homeassistant.const import (
PERCENTAGE, PERCENTAGE,
TEMP_CELSIUS, TEMP_CELSIUS,
) )
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import ValloxStateProxy from . import ValloxDataUpdateCoordinator
from .const import ( from .const import DOMAIN, METRIC_KEY_MODE, MODE_ON, VALLOX_PROFILE_TO_STR_REPORTABLE
DOMAIN,
METRIC_KEY_MODE,
MODE_ON,
SIGNAL_VALLOX_STATE_UPDATE,
VALLOX_PROFILE_TO_STR_REPORTABLE,
)
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
class ValloxSensor(SensorEntity): class ValloxSensor(CoordinatorEntity, SensorEntity):
"""Representation of a Vallox sensor.""" """Representation of a Vallox sensor."""
_attr_should_poll = False
entity_description: ValloxSensorEntityDescription entity_description: ValloxSensorEntityDescription
coordinator: ValloxDataUpdateCoordinator
def __init__( def __init__(
self, self,
name: str, name: str,
state_proxy: ValloxStateProxy, coordinator: ValloxDataUpdateCoordinator,
description: ValloxSensorEntityDescription, description: ValloxSensorEntityDescription,
) -> None: ) -> None:
"""Initialize the Vallox sensor.""" """Initialize the Vallox sensor."""
self._state_proxy = state_proxy super().__init__(coordinator)
self.entity_description = description self.entity_description = description
self._attr_name = f"{name} {description.name}" self._attr_name = f"{name} {description.name}"
self._attr_available = False
async def async_added_to_hass(self) -> None: @property
"""Call to update.""" def native_value(self) -> StateType:
self.async_on_remove( """Return the value reported by the sensor."""
async_dispatcher_connect(
self.hass, SIGNAL_VALLOX_STATE_UPDATE, self._update_callback
)
)
@callback
def _update_callback(self) -> None:
"""Call update method."""
self.async_schedule_update_ha_state(True)
async def async_update(self) -> None:
"""Fetch state from the ventilation unit."""
if (metric_key := self.entity_description.metric_key) is None: if (metric_key := self.entity_description.metric_key) is None:
self._attr_available = False _LOGGER.debug("Error updating sensor. Empty metric key")
_LOGGER.error("Error updating sensor. Empty metric key") return None
return
try: return self.coordinator.data.get_metric(metric_key)
self._attr_native_value = self._state_proxy.fetch_metric(metric_key)
except (OSError, KeyError, TypeError) as err:
self._attr_available = False
_LOGGER.error("Error updating sensor: %s", err)
return
self._attr_available = True
class ValloxProfileSensor(ValloxSensor): class ValloxProfileSensor(ValloxSensor):
"""Child class for profile reporting.""" """Child class for profile reporting."""
async def async_update(self) -> None: @property
"""Fetch state from the ventilation unit.""" def native_value(self) -> StateType:
try: """Return the value reported by the sensor."""
vallox_profile = self._state_proxy.get_profile() vallox_profile = self.coordinator.data.profile
return VALLOX_PROFILE_TO_STR_REPORTABLE.get(vallox_profile)
except OSError as err:
self._attr_available = False
_LOGGER.error("Error updating sensor: %s", err)
return
self._attr_native_value = VALLOX_PROFILE_TO_STR_REPORTABLE.get(vallox_profile)
self._attr_available = True
# There seems to be a quirk with respect to the fan speed reporting. The device keeps on reporting # There is a quirk with respect to the fan speed reporting. The device keeps on reporting the last
# the last valid fan speed from when the device was in regular operation mode, even if it left that # valid fan speed from when the device was in regular operation mode, even if it left that state and
# state and has been shut off in the meantime. # has been shut off in the meantime.
# #
# Therefore, first query the overall state of the device, and report zero percent fan speed in case # Therefore, first query the overall state of the device, and report zero percent fan speed in case
# it is not in regular operation mode. # it is not in regular operation mode.
class ValloxFanSpeedSensor(ValloxSensor): class ValloxFanSpeedSensor(ValloxSensor):
"""Child class for fan speed reporting.""" """Child class for fan speed reporting."""
async def async_update(self) -> None: @property
"""Fetch state from the ventilation unit.""" def native_value(self) -> StateType:
try: """Return the value reported by the sensor."""
fan_on = self._state_proxy.fetch_metric(METRIC_KEY_MODE) == MODE_ON fan_is_on = self.coordinator.data.get_metric(METRIC_KEY_MODE) == MODE_ON
return super().native_value if fan_is_on else 0
except (OSError, KeyError, TypeError) as err:
self._attr_available = False
_LOGGER.error("Error updating sensor: %s", err)
return
if fan_on:
await super().async_update()
else:
# Report zero percent otherwise.
self._attr_native_value = 0
self._attr_available = True
class ValloxFilterRemainingSensor(ValloxSensor): class ValloxFilterRemainingSensor(ValloxSensor):
"""Child class for filter remaining time reporting.""" """Child class for filter remaining time reporting."""
async def async_update(self) -> None: @property
"""Fetch state from the ventilation unit.""" def native_value(self) -> StateType:
await super().async_update() """Return the value reported by the sensor."""
super_native_value = super().native_value
# Check if the update in the super call was a success. if not isinstance(super_native_value, (int, float)):
if not self._attr_available: _LOGGER.debug("Value has unexpected type: %s", type(super_native_value))
return return None
if not isinstance(self._attr_native_value, (int, float)):
self._attr_available = False
_LOGGER.error(
"Value has unexpected type: %s", type(self._attr_native_value)
)
return
# Since only a delta of days is received from the device, fix the time so the timestamp does # Since only a delta of days is received from the device, fix the time so the timestamp does
# not change with every update. # not change with every update.
days_remaining = float(self._attr_native_value) days_remaining = float(super_native_value)
days_remaining_delta = timedelta(days=days_remaining) days_remaining_delta = timedelta(days=days_remaining)
now = datetime.utcnow().replace(hour=13, minute=0, second=0, microsecond=0) now = datetime.utcnow().replace(hour=13, minute=0, second=0, microsecond=0)
self._attr_native_value = (now + days_remaining_delta).isoformat() return (now + days_remaining_delta).isoformat()
@dataclass @dataclass
@ -259,12 +207,11 @@ async def async_setup_platform(
return return
name = hass.data[DOMAIN]["name"] name = hass.data[DOMAIN]["name"]
state_proxy = hass.data[DOMAIN]["state_proxy"] coordinator = hass.data[DOMAIN]["coordinator"]
async_add_entities( async_add_entities(
[ [
description.sensor_type(name, state_proxy, description) description.sensor_type(name, coordinator, description)
for description in SENSORS for description in SENSORS
], ]
update_before_add=False,
) )