evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement DataUpdateCoordinator to fritzbox integration (#49611)

Browse source
This commit is contained in:
Michael 2021-04-25 02:40:12 +02:00 • committed by GitHub
parent f1d48ddfe3
commit a352516944
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 372 additions and 323 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

132
homeassistant/components/fritzbox/__init__.py
View file

@ -1,22 +1,43 @@
"""Support for AVM Fritz!Box smarthome devices."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import socket
from pyfritzhome import Fritzhome, LoginError
from pyfritzhome import Fritzhome, FritzhomeDevice, LoginError
import requests
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_NAME,
ATTR_UNIT_OF_MEASUREMENT,
CONF_DEVICES,
CONF_HOST,
CONF_PASSWORD,
CONF_USERNAME,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import CONF_CONNECTIONS, DEFAULT_HOST, DEFAULT_USERNAME, DOMAIN, PLATFORMS
from .const import (
CONF_CONNECTIONS,
CONF_COORDINATOR,
DEFAULT_HOST,
DEFAULT_USERNAME,
DOMAIN,
LOGGER,
PLATFORMS,
)
def ensure_unique_hosts(value):
@ -58,7 +79,7 @@ CONFIG_SCHEMA = vol.Schema(
)
async def async_setup(hass, config):
async def async_setup(hass: HomeAssistant, config: dict[str, str]) -> bool:
"""Set up the AVM Fritz!Box integration."""
if DOMAIN in config:
for entry_config in config[DOMAIN][CONF_DEVICES]:
@ -71,7 +92,7 @@ async def async_setup(hass, config):
return True
async def async_setup_entry(hass, entry):
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up the AVM Fritz!Box platforms."""
fritz = Fritzhome(
host=entry.data[CONF_HOST],
@ -84,8 +105,44 @@ async def async_setup_entry(hass, entry):
except LoginError as err:
raise ConfigEntryAuthFailed from err
hass.data.setdefault(DOMAIN, {CONF_CONNECTIONS: {}, CONF_DEVICES: set()})
hass.data[DOMAIN][CONF_CONNECTIONS][entry.entry_id] = fritz
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = {
CONF_CONNECTIONS: fritz,
}
def _update_fritz_devices() -> dict[str, FritzhomeDevice]:
"""Update all fritzbox device data."""
try:
devices = fritz.get_devices()
except requests.exceptions.HTTPError:
# If the device rebooted, login again
try:
fritz.login()
except requests.exceptions.HTTPError as ex:
raise ConfigEntryAuthFailed from ex
devices = fritz.get_devices()
data = {}
for device in devices:
device.update()
data[device.ain] = device
return data
async def async_update_coordinator():
"""Fetch all device data."""
return await hass.async_add_executor_job(_update_fritz_devices)
hass.data[DOMAIN][entry.entry_id][
CONF_COORDINATOR
] = coordinator = DataUpdateCoordinator(
hass,
LOGGER,
name=f"{entry.entry_id}",
update_method=async_update_coordinator,
update_interval=timedelta(seconds=30),
)
await coordinator.async_config_entry_first_refresh()
for platform in PLATFORMS:
hass.async_create_task(
@ -103,9 +160,9 @@ async def async_setup_entry(hass, entry):
return True
async def async_unload_entry(hass, entry):
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unloading the AVM Fritz!Box platforms."""
fritz = hass.data[DOMAIN][CONF_CONNECTIONS][entry.entry_id]
fritz = hass.data[DOMAIN][entry.entry_id][CONF_CONNECTIONS]
await hass.async_add_executor_job(fritz.logout)
unload_ok = all(
@ -117,6 +174,61 @@ async def async_unload_entry(hass, entry):
)
)
if unload_ok:
hass.data[DOMAIN][CONF_CONNECTIONS].pop(entry.entry_id)
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
class FritzBoxEntity(CoordinatorEntity):
"""Basis FritzBox entity."""
def __init__(
self,
entity_info: dict[str, str],
coordinator: DataUpdateCoordinator,
ain: str,
):
"""Initialize the FritzBox entity."""
super().__init__(coordinator)
self.ain = ain
self._name = entity_info[ATTR_NAME]
self._unique_id = entity_info[ATTR_ENTITY_ID]
self._unit_of_measurement = entity_info[ATTR_UNIT_OF_MEASUREMENT]
self._device_class = entity_info[ATTR_DEVICE_CLASS]
@property
def device(self) -> FritzhomeDevice:
"""Return device object from coordinator."""
return self.coordinator.data[self.ain]
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": self.device.name,
"identifiers": {(DOMAIN, self.ain)},
"manufacturer": self.device.manufacturer,
"model": self.device.productname,
"sw_version": self.device.fw_version,
}
@property
def unique_id(self):
"""Return the unique ID of the device."""
return self._unique_id
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def unit_of_measurement(self):
"""Return the unit of measurement."""
return self._unit_of_measurement
@property
def device_class(self):
"""Return the device class."""
return self._device_class

90
homeassistant/components/fritzbox/binary_sensor.py
View file

@ -1,74 +1,56 @@
"""Support for Fritzbox binary sensors."""
import requests
from typing import Callable
from homeassistant.components.binary_sensor import (
DEVICE_CLASS_WINDOW,
BinarySensorEntity,
)
from homeassistant.const import CONF_DEVICES
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_NAME,
ATTR_UNIT_OF_MEASUREMENT,
)
from homeassistant.core import HomeAssistant
from .const import CONF_CONNECTIONS, DOMAIN as FRITZBOX_DOMAIN, LOGGER
from . import FritzBoxEntity
from .const import CONF_COORDINATOR, DOMAIN as FRITZBOX_DOMAIN
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Fritzbox binary sensor from config_entry."""
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: Callable
) -> None:
"""Set up the Fritzbox binary sensor from ConfigEntry."""
entities = []
devices = hass.data[FRITZBOX_DOMAIN][CONF_DEVICES]
fritz = hass.data[FRITZBOX_DOMAIN][CONF_CONNECTIONS][config_entry.entry_id]
coordinator = hass.data[FRITZBOX_DOMAIN][entry.entry_id][CONF_COORDINATOR]
for device in await hass.async_add_executor_job(fritz.get_devices):
if device.has_alarm and device.ain not in devices:
entities.append(FritzboxBinarySensor(device, fritz))
devices.add(device.ain)
for ain, device in coordinator.data.items():
if not device.has_alarm:
continue
async_add_entities(entities, True)
entities.append(
FritzboxBinarySensor(
{
ATTR_NAME: f"{device.name}",
ATTR_ENTITY_ID: f"{device.ain}",
ATTR_UNIT_OF_MEASUREMENT: None,
ATTR_DEVICE_CLASS: DEVICE_CLASS_WINDOW,
},
coordinator,
ain,
)
)
async_add_entities(entities)
class FritzboxBinarySensor(BinarySensorEntity):
class FritzboxBinarySensor(FritzBoxEntity, BinarySensorEntity):
"""Representation of a binary Fritzbox device."""
def __init__(self, device, fritz):
"""Initialize the Fritzbox binary sensor."""
self._device = device
self._fritz = fritz
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": self.name,
"identifiers": {(FRITZBOX_DOMAIN, self._device.ain)},
"manufacturer": self._device.manufacturer,
"model": self._device.productname,
"sw_version": self._device.fw_version,
}
@property
def unique_id(self):
"""Return the unique ID of the device."""
return self._device.ain
@property
def name(self):
"""Return the name of the entity."""
return self._device.name
@property
def device_class(self):
"""Return the class of this sensor."""
return DEVICE_CLASS_WINDOW
@property
def is_on(self):
"""Return true if sensor is on."""
if not self._device.present:
if not self.device.present:
return False
return self._device.alert_state
def update(self):
"""Get latest data from the Fritzbox."""
try:
self._device.update()
except requests.exceptions.HTTPError as ex:
LOGGER.warning("Connection error: %s", ex)
self._fritz.login()
return self.device.alert_state

149
homeassistant/components/fritzbox/climate.py
View file

@ -1,5 +1,5 @@
"""Support for AVM Fritz!Box smarthome thermostate devices."""
import requests
from typing import Callable
from homeassistant.components.climate import ClimateEntity
from homeassistant.components.climate.const import (
@ -11,14 +11,20 @@ from homeassistant.components.climate.const import (
SUPPORT_PRESET_MODE,
SUPPORT_TARGET_TEMPERATURE,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_BATTERY_LEVEL,
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_NAME,
ATTR_TEMPERATURE,
CONF_DEVICES,
ATTR_UNIT_OF_MEASUREMENT,
PRECISION_HALVES,
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant
from . import FritzBoxEntity
from .const import (
ATTR_STATE_BATTERY_LOW,
ATTR_STATE_DEVICE_LOCKED,
@ -26,9 +32,8 @@ from .const import (
ATTR_STATE_LOCKED,
ATTR_STATE_SUMMER_MODE,
ATTR_STATE_WINDOW_OPEN,
CONF_CONNECTIONS,
CONF_COORDINATOR,
DOMAIN as FRITZBOX_DOMAIN,
LOGGER,
)
SUPPORT_FLAGS = SUPPORT_TARGET_TEMPERATURE | SUPPORT_PRESET_MODE
@ -47,48 +52,36 @@ ON_REPORT_SET_TEMPERATURE = 30.0
OFF_REPORT_SET_TEMPERATURE = 0.0
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Fritzbox smarthome thermostat from config_entry."""
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: Callable
) -> None:
"""Set up the Fritzbox smarthome thermostat from ConfigEntry."""
entities = []
devices = hass.data[FRITZBOX_DOMAIN][CONF_DEVICES]
fritz = hass.data[FRITZBOX_DOMAIN][CONF_CONNECTIONS][config_entry.entry_id]
coordinator = hass.data[FRITZBOX_DOMAIN][entry.entry_id][CONF_COORDINATOR]
for device in await hass.async_add_executor_job(fritz.get_devices):
if device.has_thermostat and device.ain not in devices:
entities.append(FritzboxThermostat(device, fritz))
devices.add(device.ain)
for ain, device in coordinator.data.items():
if not device.has_thermostat:
continue
entities.append(
FritzboxThermostat(
{
ATTR_NAME: f"{device.name}",
ATTR_ENTITY_ID: f"{device.ain}",
ATTR_UNIT_OF_MEASUREMENT: None,
ATTR_DEVICE_CLASS: None,
},
coordinator,
ain,
)
)
async_add_entities(entities)
class FritzboxThermostat(ClimateEntity):
class FritzboxThermostat(FritzBoxEntity, ClimateEntity):
"""The thermostat class for Fritzbox smarthome thermostates."""
def __init__(self, device, fritz):
"""Initialize the thermostat."""
self._device = device
self._fritz = fritz
self._current_temperature = self._device.actual_temperature
self._target_temperature = self._device.target_temperature
self._comfort_temperature = self._device.comfort_temperature
self._eco_temperature = self._device.eco_temperature
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": self.name,
"identifiers": {(FRITZBOX_DOMAIN, self._device.ain)},
"manufacturer": self._device.manufacturer,
"model": self._device.productname,
"sw_version": self._device.fw_version,
}
@property
def unique_id(self):
"""Return the unique ID of the device."""
return self._device.ain
@property
def supported_features(self):
"""Return the list of supported features."""
@ -97,12 +90,7 @@ class FritzboxThermostat(ClimateEntity):
@property
def available(self):
"""Return if thermostat is available."""
return self._device.present
@property
def name(self):
"""Return the name of the device."""
return self._device.name
return self.device.present
@property
def temperature_unit(self):
@ -117,32 +105,35 @@ class FritzboxThermostat(ClimateEntity):
@property
def current_temperature(self):
"""Return the current temperature."""
return self._current_temperature
return self.device.actual_temperature
@property
def target_temperature(self):
"""Return the temperature we try to reach."""
if self._target_temperature == ON_API_TEMPERATURE:
if self.device.target_temperature == ON_API_TEMPERATURE:
return ON_REPORT_SET_TEMPERATURE
if self._target_temperature == OFF_API_TEMPERATURE:
if self.device.target_temperature == OFF_API_TEMPERATURE:
return OFF_REPORT_SET_TEMPERATURE
return self._target_temperature
return self.device.target_temperature
def set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs):
"""Set new target temperature."""
if ATTR_HVAC_MODE in kwargs:
hvac_mode = kwargs.get(ATTR_HVAC_MODE)
self.set_hvac_mode(hvac_mode)
await self.async_set_hvac_mode(hvac_mode)
elif ATTR_TEMPERATURE in kwargs:
temperature = kwargs.get(ATTR_TEMPERATURE)
self._device.set_target_temperature(temperature)
await self.hass.async_add_executor_job(
self.device.set_target_temperature, temperature
)
await self.coordinator.async_refresh()
@property
def hvac_mode(self):
"""Return the current operation mode."""
if (
self._target_temperature == OFF_REPORT_SET_TEMPERATURE
or self._target_temperature == OFF_API_TEMPERATURE
self.device.target_temperature == OFF_REPORT_SET_TEMPERATURE
or self.device.target_temperature == OFF_API_TEMPERATURE
):
return HVAC_MODE_OFF
@ -153,19 +144,21 @@ class FritzboxThermostat(ClimateEntity):
"""Return the list of available operation modes."""
return OPERATION_LIST
def set_hvac_mode(self, hvac_mode):
async def async_set_hvac_mode(self, hvac_mode):
"""Set new operation mode."""
if hvac_mode == HVAC_MODE_OFF:
self.set_temperature(temperature=OFF_REPORT_SET_TEMPERATURE)
await self.async_set_temperature(temperature=OFF_REPORT_SET_TEMPERATURE)
else:
self.set_temperature(temperature=self._comfort_temperature)
await self.async_set_temperature(
temperature=self.device.comfort_temperature
)
@property
def preset_mode(self):
"""Return current preset mode."""
if self._target_temperature == self._comfort_temperature:
if self.device.target_temperature == self.device.comfort_temperature:
return PRESET_COMFORT
if self._target_temperature == self._eco_temperature:
if self.device.target_temperature == self.device.eco_temperature:
return PRESET_ECO
@property
@ -173,12 +166,14 @@ class FritzboxThermostat(ClimateEntity):
"""Return supported preset modes."""
return [PRESET_ECO, PRESET_COMFORT]
def set_preset_mode(self, preset_mode):
async def async_set_preset_mode(self, preset_mode):
"""Set preset mode."""
if preset_mode == PRESET_COMFORT:
self.set_temperature(temperature=self._comfort_temperature)
await self.async_set_temperature(
temperature=self.device.comfort_temperature
)
elif preset_mode == PRESET_ECO:
self.set_temperature(temperature=self._eco_temperature)
await self.async_set_temperature(temperature=self.device.eco_temperature)
@property
def min_temp(self):
@ -194,31 +189,19 @@ class FritzboxThermostat(ClimateEntity):
def extra_state_attributes(self):
"""Return the device specific state attributes."""
attrs = {
ATTR_STATE_BATTERY_LOW: self._device.battery_low,
ATTR_STATE_DEVICE_LOCKED: self._device.device_lock,
ATTR_STATE_LOCKED: self._device.lock,
ATTR_STATE_BATTERY_LOW: self.device.battery_low,
ATTR_STATE_DEVICE_LOCKED: self.device.device_lock,
ATTR_STATE_LOCKED: self.device.lock,
}
# the following attributes are available since fritzos 7
if self._device.battery_level is not None:
attrs[ATTR_BATTERY_LEVEL] = self._device.battery_level
if self._device.holiday_active is not None:
attrs[ATTR_STATE_HOLIDAY_MODE] = self._device.holiday_active
if self._device.summer_active is not None:
attrs[ATTR_STATE_SUMMER_MODE] = self._device.summer_active
if self.device.battery_level is not None:
attrs[ATTR_BATTERY_LEVEL] = self.device.battery_level
if self.device.holiday_active is not None:
attrs[ATTR_STATE_HOLIDAY_MODE] = self.device.holiday_active
if self.device.summer_active is not None:
attrs[ATTR_STATE_SUMMER_MODE] = self.device.summer_active
if ATTR_STATE_WINDOW_OPEN is not None:
attrs[ATTR_STATE_WINDOW_OPEN] = self._device.window_open
attrs[ATTR_STATE_WINDOW_OPEN] = self.device.window_open
return attrs
def update(self):
"""Update the data from the thermostat."""
try:
self._device.update()
self._current_temperature = self._device.actual_temperature
self._target_temperature = self._device.target_temperature
self._comfort_temperature = self._device.comfort_temperature
self._eco_temperature = self._device.eco_temperature
except requests.exceptions.HTTPError as ex:
LOGGER.warning("Fritzbox connection error: %s", ex)
self._fritz.login()

3
homeassistant/components/fritzbox/const.py
View file

@ -14,12 +14,13 @@ ATTR_TOTAL_CONSUMPTION = "total_consumption"
ATTR_TOTAL_CONSUMPTION_UNIT = "total_consumption_unit"
CONF_CONNECTIONS = "connections"
CONF_COORDINATOR = "coordinator"
DEFAULT_HOST = "fritz.box"
DEFAULT_USERNAME = "admin"
DOMAIN = "fritzbox"
LOGGER = logging.getLogger(__package__)
LOGGER: logging.Logger = logging.getLogger(__package__)
PLATFORMS = ["binary_sensor", "climate", "switch", "sensor"]

142
homeassistant/components/fritzbox/sensor.py
View file

@ -1,143 +1,93 @@
"""Support for AVM Fritz!Box smarthome temperature sensor only devices."""
import requests
from typing import Callable
from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_DEVICES,
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_NAME,
ATTR_UNIT_OF_MEASUREMENT,
DEVICE_CLASS_BATTERY,
PERCENTAGE,
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant
from . import FritzBoxEntity
from .const import (
ATTR_STATE_DEVICE_LOCKED,
ATTR_STATE_LOCKED,
CONF_CONNECTIONS,
CONF_COORDINATOR,
DOMAIN as FRITZBOX_DOMAIN,
LOGGER,
)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Fritzbox smarthome sensor from config_entry."""
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: Callable
) -> None:
"""Set up the Fritzbox smarthome sensor from ConfigEntry."""
entities = []
devices = hass.data[FRITZBOX_DOMAIN][CONF_DEVICES]
fritz = hass.data[FRITZBOX_DOMAIN][CONF_CONNECTIONS][config_entry.entry_id]
coordinator = hass.data[FRITZBOX_DOMAIN][entry.entry_id][CONF_COORDINATOR]
for device in await hass.async_add_executor_job(fritz.get_devices):
for ain, device in coordinator.data.items():
if (
device.has_temperature_sensor
and not device.has_switch
and not device.has_thermostat
and device.ain not in devices
):
entities.append(FritzBoxTempSensor(device, fritz))
devices.add(device.ain)
entities.append(
FritzBoxTempSensor(
{
ATTR_NAME: f"{device.name}",
ATTR_ENTITY_ID: f"{device.ain}",
ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS,
ATTR_DEVICE_CLASS: None,
},
coordinator,
ain,
)
)
if device.battery_level is not None:
entities.append(FritzBoxBatterySensor(device, fritz))
devices.add(f"{device.ain}_battery")
entities.append(
FritzBoxBatterySensor(
{
ATTR_NAME: f"{device.name} Battery",
ATTR_ENTITY_ID: f"{device.ain}_battery",
ATTR_UNIT_OF_MEASUREMENT: PERCENTAGE,
ATTR_DEVICE_CLASS: DEVICE_CLASS_BATTERY,
},
coordinator,
ain,
)
)
async_add_entities(entities)
class FritzBoxBatterySensor(SensorEntity):
"""The entity class for Fritzbox battery sensors."""
def __init__(self, device, fritz):
"""Initialize the sensor."""
self._device = device
self._fritz = fritz
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": self.name,
"identifiers": {(FRITZBOX_DOMAIN, self._device.ain)},
"manufacturer": self._device.manufacturer,
"model": self._device.productname,
"sw_version": self._device.fw_version,
}
@property
def unique_id(self):
"""Return the unique ID of the device."""
return f"{self._device.ain}_battery"
@property
def name(self):
"""Return the name of the device."""
return f"{self._device.name} Battery"
class FritzBoxBatterySensor(FritzBoxEntity, SensorEntity):
"""The entity class for Fritzbox sensors."""
@property
def state(self):
"""Return the state of the sensor."""
return self._device.battery_level
@property
def unit_of_measurement(self):
"""Return the unit of measurement."""
return PERCENTAGE
@property
def device_class(self):
"""Return the device class."""
return DEVICE_CLASS_BATTERY
return self.device.battery_level
class FritzBoxTempSensor(SensorEntity):
class FritzBoxTempSensor(FritzBoxEntity, SensorEntity):
"""The entity class for Fritzbox temperature sensors."""
def __init__(self, device, fritz):
"""Initialize the switch."""
self._device = device
self._fritz = fritz
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": self.name,
"identifiers": {(FRITZBOX_DOMAIN, self._device.ain)},
"manufacturer": self._device.manufacturer,
"model": self._device.productname,
"sw_version": self._device.fw_version,
}
@property
def unique_id(self):
"""Return the unique ID of the device."""
return self._device.ain
@property
def name(self):
"""Return the name of the device."""
return self._device.name
@property
def state(self):
"""Return the state of the sensor."""
return self._device.temperature
@property
def unit_of_measurement(self):
"""Return the unit of measurement."""
return TEMP_CELSIUS
def update(self):
"""Get latest data and states from the device."""
try:
self._device.update()
except requests.exceptions.HTTPError as ex:
LOGGER.warning("Fritzhome connection error: %s", ex)
self._fritz.login()
return self.device.temperature
@property
def extra_state_attributes(self):
"""Return the state attributes of the device."""
attrs = {
ATTR_STATE_DEVICE_LOCKED: self._device.device_lock,
ATTR_STATE_LOCKED: self._device.lock,
ATTR_STATE_DEVICE_LOCKED: self.device.device_lock,
ATTR_STATE_LOCKED: self.device.lock,
}
return attrs

106
homeassistant/components/fritzbox/switch.py
View file

@ -1,113 +1,99 @@
"""Support for AVM Fritz!Box smarthome switch devices."""
import requests
from typing import Callable
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_NAME,
ATTR_TEMPERATURE,
CONF_DEVICES,
ATTR_UNIT_OF_MEASUREMENT,
ENERGY_KILO_WATT_HOUR,
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant
from . import FritzBoxEntity
from .const import (
ATTR_STATE_DEVICE_LOCKED,
ATTR_STATE_LOCKED,
ATTR_TEMPERATURE_UNIT,
ATTR_TOTAL_CONSUMPTION,
ATTR_TOTAL_CONSUMPTION_UNIT,
CONF_CONNECTIONS,
CONF_COORDINATOR,
DOMAIN as FRITZBOX_DOMAIN,
LOGGER,
)
ATTR_TOTAL_CONSUMPTION_UNIT_VALUE = ENERGY_KILO_WATT_HOUR
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Fritzbox smarthome switch from config_entry."""
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: Callable
) -> None:
"""Set up the Fritzbox smarthome switch from ConfigEntry."""
entities = []
devices = hass.data[FRITZBOX_DOMAIN][CONF_DEVICES]
fritz = hass.data[FRITZBOX_DOMAIN][CONF_CONNECTIONS][config_entry.entry_id]
coordinator = hass.data[FRITZBOX_DOMAIN][entry.entry_id][CONF_COORDINATOR]
for device in await hass.async_add_executor_job(fritz.get_devices):
if device.has_switch and device.ain not in devices:
entities.append(FritzboxSwitch(device, fritz))
devices.add(device.ain)
for ain, device in coordinator.data.items():
if not device.has_switch:
continue
entities.append(
FritzboxSwitch(
{
ATTR_NAME: f"{device.name}",
ATTR_ENTITY_ID: f"{device.ain}",
ATTR_UNIT_OF_MEASUREMENT: None,
ATTR_DEVICE_CLASS: None,
},
coordinator,
ain,
)
)
async_add_entities(entities)
class FritzboxSwitch(SwitchEntity):
class FritzboxSwitch(FritzBoxEntity, SwitchEntity):
"""The switch class for Fritzbox switches."""
def __init__(self, device, fritz):
"""Initialize the switch."""
self._device = device
self._fritz = fritz
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": self.name,
"identifiers": {(FRITZBOX_DOMAIN, self._device.ain)},
"manufacturer": self._device.manufacturer,
"model": self._device.productname,
"sw_version": self._device.fw_version,
}
@property
def unique_id(self):
"""Return the unique ID of the device."""
return self._device.ain
@property
def available(self):
"""Return if switch is available."""
return self._device.present
@property
def name(self):
"""Return the name of the device."""
return self._device.name
return self.device.present
@property
def is_on(self):
"""Return true if the switch is on."""
return self._device.switch_state
return self.device.switch_state
def turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs):
"""Turn the switch on."""
self._device.set_switch_state_on()
await self.hass.async_add_executor_job(self.device.set_switch_state_on)
await self.coordinator.async_refresh()
def turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs):
"""Turn the switch off."""
self._device.set_switch_state_off()
def update(self):
"""Get latest data and states from the device."""
try:
self._device.update()
except requests.exceptions.HTTPError as ex:
LOGGER.warning("Fritzhome connection error: %s", ex)
self._fritz.login()
await self.hass.async_add_executor_job(self.device.set_switch_state_off)
await self.coordinator.async_refresh()
@property
def extra_state_attributes(self):
"""Return the state attributes of the device."""
attrs = {}
attrs[ATTR_STATE_DEVICE_LOCKED] = self._device.device_lock
attrs[ATTR_STATE_LOCKED] = self._device.lock
attrs[ATTR_STATE_DEVICE_LOCKED] = self.device.device_lock
attrs[ATTR_STATE_LOCKED] = self.device.lock
if self._device.has_powermeter:
if self.device.has_powermeter:
attrs[
ATTR_TOTAL_CONSUMPTION
] = f"{((self._device.energy or 0.0) / 1000):.3f}"
] = f"{((self.device.energy or 0.0) / 1000):.3f}"
attrs[ATTR_TOTAL_CONSUMPTION_UNIT] = ATTR_TOTAL_CONSUMPTION_UNIT_VALUE
if self._device.has_temperature_sensor:
if self.device.has_temperature_sensor:
attrs[ATTR_TEMPERATURE] = str(
self.hass.config.units.temperature(
self._device.temperature, TEMP_CELSIUS
self.device.temperature, TEMP_CELSIUS
)
)
attrs[ATTR_TEMPERATURE_UNIT] = self.hass.config.units.temperature_unit
@ -116,4 +102,4 @@ class FritzboxSwitch(SwitchEntity):
@property
def current_power_w(self):
"""Return the current power usage in W."""
return self._device.power / 1000
return self.device.power / 1000

4
tests/components/fritzbox/test_binary_sensor.py
View file

@ -58,7 +58,7 @@ async def test_is_off(hass: HomeAssistant, fritz: Mock):
async def test_update(hass: HomeAssistant, fritz: Mock):
"""Test update with error."""
"""Test update without error."""
device = FritzDeviceBinarySensorMock()
fritz().get_devices.return_value = [device]
@ -91,4 +91,4 @@ async def test_update_error(hass: HomeAssistant, fritz: Mock):
await hass.async_block_till_done()
assert device.update.call_count == 2
assert fritz().login.call_count == 2
assert fritz().login.call_count == 1

12
tests/components/fritzbox/test_climate.py
View file

@ -105,7 +105,7 @@ async def test_target_temperature_off(hass: HomeAssistant, fritz: Mock):
async def test_update(hass: HomeAssistant, fritz: Mock):
"""Test update with error."""
"""Test update without error."""
device = FritzDeviceClimateMock()
fritz().get_devices.return_value = [device]
@ -126,7 +126,7 @@ async def test_update(hass: HomeAssistant, fritz: Mock):
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert device.update.call_count == 1
assert device.update.call_count == 2
assert state
assert state.attributes[ATTR_CURRENT_TEMPERATURE] == 19
assert state.attributes[ATTR_TEMPERATURE] == 20
@ -139,14 +139,14 @@ async def test_update_error(hass: HomeAssistant, fritz: Mock):
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert device.update.call_count == 0
assert device.update.call_count == 1
assert fritz().login.call_count == 1
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert device.update.call_count == 1
assert device.update.call_count == 2
assert fritz().login.call_count == 2
@ -290,7 +290,7 @@ async def test_preset_mode_update(hass: HomeAssistant, fritz: Mock):
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert device.update.call_count == 1
assert device.update.call_count == 2
assert state
assert state.attributes[ATTR_PRESET_MODE] == PRESET_COMFORT
@ -301,6 +301,6 @@ async def test_preset_mode_update(hass: HomeAssistant, fritz: Mock):
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert device.update.call_count == 2
assert device.update.call_count == 3
assert state
assert state.attributes[ATTR_PRESET_MODE] == PRESET_ECO

37
tests/components/fritzbox/test_init.py
View file

@ -2,6 +2,7 @@
from unittest.mock import Mock, call, patch
from pyfritzhome import LoginError
from requests.exceptions import HTTPError
from homeassistant.components.fritzbox.const import DOMAIN as FB_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
@ -57,6 +58,39 @@ async def test_setup_duplicate_config(hass: HomeAssistant, fritz: Mock, caplog):
assert "duplicate host entries found" in caplog.text
async def test_coordinator_update_after_reboot(hass: HomeAssistant, fritz: Mock):
"""Test coordinator after reboot."""
entry = MockConfigEntry(
domain=FB_DOMAIN,
data=MOCK_CONFIG[FB_DOMAIN][CONF_DEVICES][0],
unique_id="any",
)
entry.add_to_hass(hass)
fritz().get_devices.side_effect = [HTTPError(), ""]
assert await hass.config_entries.async_setup(entry.entry_id)
assert fritz().get_devices.call_count == 2
assert fritz().login.call_count == 2
async def test_coordinator_update_after_password_change(
hass: HomeAssistant, fritz: Mock
):
"""Test coordinator after password change."""
entry = MockConfigEntry(
domain=FB_DOMAIN,
data=MOCK_CONFIG[FB_DOMAIN][CONF_DEVICES][0],
unique_id="any",
)
entry.add_to_hass(hass)
fritz().get_devices.side_effect = HTTPError()
fritz().login.side_effect = ["", HTTPError()]
assert not await hass.config_entries.async_setup(entry.entry_id)
assert fritz().get_devices.call_count == 1
assert fritz().login.call_count == 2
async def test_unload_remove(hass: HomeAssistant, fritz: Mock):
"""Test unload and remove of integration."""
fritz().get_devices.return_value = [FritzDeviceSwitchMock()]
@ -107,9 +141,10 @@ async def test_raise_config_entry_not_ready_when_offline(hass):
with patch(
"homeassistant.components.fritzbox.Fritzhome.login",
side_effect=LoginError("user"),
):
) as mock_login:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
mock_login.assert_called_once()
entries = hass.config_entries.async_entries()
config_entry = entries[0]

10
tests/components/fritzbox/test_sensor.py
View file

@ -57,19 +57,19 @@ async def test_setup(hass: HomeAssistant, fritz: Mock):
async def test_update(hass: HomeAssistant, fritz: Mock):
"""Test update with error."""
"""Test update without error."""
device = FritzDeviceSensorMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert device.update.call_count == 0
assert device.update.call_count == 1
assert fritz().login.call_count == 1
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert device.update.call_count == 1
assert device.update.call_count == 2
assert fritz().login.call_count == 1
@ -80,12 +80,12 @@ async def test_update_error(hass: HomeAssistant, fritz: Mock):
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert device.update.call_count == 0
assert device.update.call_count == 1
assert fritz().login.call_count == 1
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert device.update.call_count == 1
assert device.update.call_count == 2
assert fritz().login.call_count == 2

10
tests/components/fritzbox/test_switch.py
View file

@ -87,19 +87,19 @@ async def test_turn_off(hass: HomeAssistant, fritz: Mock):
async def test_update(hass: HomeAssistant, fritz: Mock):
"""Test update with error."""
"""Test update without error."""
device = FritzDeviceSwitchMock()
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert device.update.call_count == 0
assert device.update.call_count == 1
assert fritz().login.call_count == 1
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert device.update.call_count == 1
assert device.update.call_count == 2
assert fritz().login.call_count == 1
@ -110,12 +110,12 @@ async def test_update_error(hass: HomeAssistant, fritz: Mock):
fritz().get_devices.return_value = [device]
await setup_fritzbox(hass, MOCK_CONFIG)
assert device.update.call_count == 0
assert device.update.call_count == 1
assert fritz().login.call_count == 1
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert device.update.call_count == 1
assert device.update.call_count == 2
assert fritz().login.call_count == 2