diff --git a/homeassistant/components/fritz/binary_sensor.py b/homeassistant/components/fritz/binary_sensor.py index 7655df6e298..edde8c0c22a 100644 --- a/homeassistant/components/fritz/binary_sensor.py +++ b/homeassistant/components/fritz/binary_sensor.py @@ -1,11 +1,14 @@ """AVM FRITZ!Box connectivity sensor.""" -import logging +from __future__ import annotations -from fritzconnection.core.exceptions import FritzConnectionException +import logging from homeassistant.components.binary_sensor import ( DEVICE_CLASS_CONNECTIVITY, + DEVICE_CLASS_PLUG, + DEVICE_CLASS_UPDATE, BinarySensorEntity, + BinarySensorEntityDescription, ) from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant @@ -17,6 +20,25 @@ from .const import DOMAIN _LOGGER = logging.getLogger(__name__) +SENSOR_TYPES: tuple[BinarySensorEntityDescription, ...] = ( + BinarySensorEntityDescription( + key="is_connected", + name="Connection", + device_class=DEVICE_CLASS_CONNECTIVITY, + ), + BinarySensorEntityDescription( + key="is_linked", + name="Link", + device_class=DEVICE_CLASS_PLUG, + ), + BinarySensorEntityDescription( + key="firmware_update", + name="Firmware Update", + device_class=DEVICE_CLASS_UPDATE, + ), +) + + async def async_setup_entry( hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback ) -> None: @@ -24,72 +46,47 @@ async def async_setup_entry( _LOGGER.debug("Setting up FRITZ!Box binary sensors") fritzbox_tools: FritzBoxTools = hass.data[DOMAIN][entry.entry_id] - if fritzbox_tools.connection and "WANIPConn1" in fritzbox_tools.connection.services: + if ( + not fritzbox_tools.connection + or "WANIPConn1" not in fritzbox_tools.connection.services + ): # Only routers are supported at the moment - async_add_entities( - [FritzBoxConnectivitySensor(fritzbox_tools, entry.title)], True - ) + return + + entities = [ + FritzBoxBinarySensor(fritzbox_tools, entry.title, description) + for description in SENSOR_TYPES + ] + + async_add_entities(entities, True) -class FritzBoxConnectivitySensor(FritzBoxBaseEntity, BinarySensorEntity): +class FritzBoxBinarySensor(FritzBoxBaseEntity, BinarySensorEntity): """Define FRITZ!Box connectivity class.""" def __init__( - self, fritzbox_tools: FritzBoxTools, device_friendly_name: str + self, + fritzbox_tools: FritzBoxTools, + device_friendly_name: str, + description: BinarySensorEntityDescription, ) -> None: """Init FRITZ!Box connectivity class.""" - self._unique_id = f"{fritzbox_tools.unique_id}-connectivity" - self._name = f"{device_friendly_name} Connectivity" - self._is_on = True - self._is_available = True + self.entity_description = description + self._attr_name = f"{device_friendly_name} {description.name}" + self._attr_unique_id = f"{fritzbox_tools.unique_id}-{description.key}" super().__init__(fritzbox_tools, device_friendly_name) - @property - def name(self) -> str: - """Return name.""" - return self._name - - @property - def device_class(self) -> str: - """Return device class.""" - return DEVICE_CLASS_CONNECTIVITY - - @property - def is_on(self) -> bool: - """Return status.""" - return self._is_on - - @property - def unique_id(self) -> str: - """Return unique id.""" - return self._unique_id - - @property - def available(self) -> bool: - """Return availability.""" - return self._is_available - def update(self) -> None: """Update data.""" _LOGGER.debug("Updating FRITZ!Box binary sensors") - self._is_on = True - try: - if ( - self._fritzbox_tools.connection - and "WANCommonInterfaceConfig1" - in self._fritzbox_tools.connection.services - ): - link_props = self._fritzbox_tools.connection.call_action( - "WANCommonInterfaceConfig1", "GetCommonLinkProperties" - ) - is_up = link_props["NewPhysicalLinkStatus"] - self._is_on = is_up == "Up" - else: - if self._fritzbox_tools.fritz_status: - self._is_on = self._fritzbox_tools.fritz_status.is_connected - self._is_available = True - - except FritzConnectionException: - _LOGGER.error("Error getting the state from the FRITZ!Box", exc_info=True) - self._is_available = False + if self.entity_description.key == "is_connected": + self._attr_is_on = bool(self._fritzbox_tools.fritz_status.is_connected) + elif self.entity_description.key == "is_linked": + self._attr_is_on = bool(self._fritzbox_tools.fritz_status.is_linked) + elif self.entity_description.key == "firmware_update": + self._attr_is_on = self._fritzbox_tools.update_available + self._attr_extra_state_attributes = { + "installed_version": self._fritzbox_tools.current_firmware, + "latest_available_version:": self._fritzbox_tools.latest_firmware, + } diff --git a/homeassistant/components/fritz/common.py b/homeassistant/components/fritz/common.py index 6b0f0873c85..acb733709a3 100644 --- a/homeassistant/components/fritz/common.py +++ b/homeassistant/components/fritz/common.py @@ -125,7 +125,9 @@ class FritzBoxTools: self.username = username self._mac: str | None = None self._model: str | None = None - self._sw_version: str | None = None + self._current_firmware: str | None = None + self._latest_firmware: str | None = None + self._update_available: bool = False async def async_setup(self) -> None: """Wrap up FritzboxTools class setup.""" @@ -152,7 +154,9 @@ class FritzBoxTools: self._unique_id = info["NewSerialNumber"] self._model = info.get("NewModelName") - self._sw_version = info.get("NewSoftwareVersion") + self._current_firmware = info.get("NewSoftwareVersion") + + self._update_available, self._latest_firmware = self._update_device_info() async def async_start(self, options: MappingProxyType[str, Any]) -> None: """Start FritzHosts connection.""" @@ -187,11 +191,21 @@ class FritzBoxTools: return self._model @property - def sw_version(self) -> str: - """Return SW version.""" - if not self._sw_version: + def current_firmware(self) -> str: + """Return current SW version.""" + if not self._current_firmware: raise ClassSetupMissing() - return self._sw_version + return self._current_firmware + + @property + def latest_firmware(self) -> str | None: + """Return latest SW version.""" + return self._latest_firmware + + @property + def update_available(self) -> bool: + """Return if new SW version is available.""" + return self._update_available @property def mac(self) -> str: @@ -215,10 +229,17 @@ class FritzBoxTools: """Event specific per FRITZ!Box entry to signal updates in devices.""" return f"{DOMAIN}-device-update-{self._unique_id}" - def _update_info(self) -> list[HostInfo]: - """Retrieve latest information from the FRITZ!Box.""" + def _update_hosts_info(self) -> list[HostInfo]: + """Retrieve latest hosts information from the FRITZ!Box.""" return self.fritz_hosts.get_hosts_info() # type: ignore [no-any-return] + def _update_device_info(self) -> tuple[bool, str | None]: + """Retrieve latest device information from the FRITZ!Box.""" + userinterface = self.connection.call_action("UserInterface1", "GetInfo") + return userinterface.get("NewUpgradeAvailable"), userinterface.get( + "NewX_AVM-DE_Version" + ) + def scan_devices(self, now: datetime | None = None) -> None: """Scan for new devices and return a list of found device ids.""" _LOGGER.debug("Checking devices for FRITZ!Box router %s", self.host) @@ -232,7 +253,7 @@ class FritzBoxTools: consider_home = _default_consider_home new_device = False - for known_host in self._update_info(): + for known_host in self._update_hosts_info(): if not known_host.get("mac"): continue @@ -255,6 +276,9 @@ class FritzBoxTools: if new_device: dispatcher_send(self.hass, self.signal_device_new) + _LOGGER.debug("Checking host info for FRITZ!Box router %s", self.host) + self._update_available, self._latest_firmware = self._update_device_info() + async def service_fritzbox(self, service: str) -> None: """Define FRITZ!Box services.""" _LOGGER.debug("FRITZ!Box router: %s", service) @@ -460,5 +484,5 @@ class FritzBoxBaseEntity: "name": self._device_name, "manufacturer": "AVM", "model": self._fritzbox_tools.model, - "sw_version": self._fritzbox_tools.sw_version, + "sw_version": self._fritzbox_tools.current_firmware, } diff --git a/homeassistant/components/fritz/const.py b/homeassistant/components/fritz/const.py index 4ae8314113f..3ed4e705730 100644 --- a/homeassistant/components/fritz/const.py +++ b/homeassistant/components/fritz/const.py @@ -1,12 +1,14 @@ """Constants for the FRITZ!Box Tools integration.""" +from typing import Literal + DOMAIN = "fritz" PLATFORMS = ["binary_sensor", "device_tracker", "sensor", "switch"] DATA_FRITZ = "fritz_data" -DSL_CONNECTION = "dsl" +DSL_CONNECTION: Literal["dsl"] = "dsl" DEFAULT_DEVICE_NAME = "Unknown device" DEFAULT_HOST = "192.168.178.1" diff --git a/homeassistant/components/fritz/sensor.py b/homeassistant/components/fritz/sensor.py index 15aed604ffc..fd82d245b9a 100644 --- a/homeassistant/components/fritz/sensor.py +++ b/homeassistant/components/fritz/sensor.py @@ -1,10 +1,10 @@ """AVM FRITZ!Box binary sensors.""" from __future__ import annotations -from collections.abc import Callable +from dataclasses import dataclass import datetime import logging -from typing import TypedDict +from typing import Any, Callable, Literal from fritzconnection.core.exceptions import ( FritzActionError, @@ -19,6 +19,7 @@ from homeassistant.components.sensor import ( STATE_CLASS_MEASUREMENT, STATE_CLASS_TOTAL_INCREASING, SensorEntity, + SensorEntityDescription, ) from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( @@ -139,117 +140,134 @@ def _retrieve_link_attenuation_received_state( return status.attenuation[1] / 10 # type: ignore[no-any-return] -class SensorData(TypedDict, total=False): - """Sensor data class.""" +@dataclass +class FritzRequireKeysMixin: + """Fritz sensor data class.""" - name: str - device_class: str | None - state_class: str | None - unit_of_measurement: str | None - icon: str | None - state_provider: Callable - connection_type: str | None + value_fn: Callable[[FritzStatus, Any], Any] -SENSOR_DATA = { - "external_ip": SensorData( +@dataclass +class FritzSensorEntityDescription(SensorEntityDescription, FritzRequireKeysMixin): + """Describes Fritz sensor entity.""" + + connection_type: Literal["dsl"] | None = None + + +SENSOR_TYPES: tuple[FritzSensorEntityDescription, ...] = ( + FritzSensorEntityDescription( + key="external_ip", name="External IP", icon="mdi:earth", - state_provider=_retrieve_external_ip_state, + value_fn=_retrieve_external_ip_state, ), - "device_uptime": SensorData( + FritzSensorEntityDescription( + key="device_uptime", name="Device Uptime", device_class=DEVICE_CLASS_TIMESTAMP, - state_provider=_retrieve_device_uptime_state, + value_fn=_retrieve_device_uptime_state, ), - "connection_uptime": SensorData( + FritzSensorEntityDescription( + key="connection_uptime", name="Connection Uptime", device_class=DEVICE_CLASS_TIMESTAMP, - state_provider=_retrieve_connection_uptime_state, + value_fn=_retrieve_connection_uptime_state, ), - "kb_s_sent": SensorData( + FritzSensorEntityDescription( + key="kb_s_sent", name="Upload Throughput", state_class=STATE_CLASS_MEASUREMENT, - unit_of_measurement=DATA_RATE_KILOBYTES_PER_SECOND, + native_unit_of_measurement=DATA_RATE_KILOBYTES_PER_SECOND, icon="mdi:upload", - state_provider=_retrieve_kb_s_sent_state, + value_fn=_retrieve_kb_s_sent_state, ), - "kb_s_received": SensorData( + FritzSensorEntityDescription( + key="kb_s_received", name="Download Throughput", state_class=STATE_CLASS_MEASUREMENT, - unit_of_measurement=DATA_RATE_KILOBYTES_PER_SECOND, + native_unit_of_measurement=DATA_RATE_KILOBYTES_PER_SECOND, icon="mdi:download", - state_provider=_retrieve_kb_s_received_state, + value_fn=_retrieve_kb_s_received_state, ), - "max_kb_s_sent": SensorData( + FritzSensorEntityDescription( + key="max_kb_s_sent", name="Max Connection Upload Throughput", - unit_of_measurement=DATA_RATE_KILOBITS_PER_SECOND, + native_unit_of_measurement=DATA_RATE_KILOBITS_PER_SECOND, icon="mdi:upload", - state_provider=_retrieve_max_kb_s_sent_state, + value_fn=_retrieve_max_kb_s_sent_state, ), - "max_kb_s_received": SensorData( + FritzSensorEntityDescription( + key="max_kb_s_received", name="Max Connection Download Throughput", - unit_of_measurement=DATA_RATE_KILOBITS_PER_SECOND, + native_unit_of_measurement=DATA_RATE_KILOBITS_PER_SECOND, icon="mdi:download", - state_provider=_retrieve_max_kb_s_received_state, + value_fn=_retrieve_max_kb_s_received_state, ), - "gb_sent": SensorData( + FritzSensorEntityDescription( + key="gb_sent", name="GB sent", state_class=STATE_CLASS_TOTAL_INCREASING, - unit_of_measurement=DATA_GIGABYTES, + native_unit_of_measurement=DATA_GIGABYTES, icon="mdi:upload", - state_provider=_retrieve_gb_sent_state, + value_fn=_retrieve_gb_sent_state, ), - "gb_received": SensorData( + FritzSensorEntityDescription( + key="gb_received", name="GB received", state_class=STATE_CLASS_TOTAL_INCREASING, - unit_of_measurement=DATA_GIGABYTES, + native_unit_of_measurement=DATA_GIGABYTES, icon="mdi:download", - state_provider=_retrieve_gb_received_state, + value_fn=_retrieve_gb_received_state, ), - "link_kb_s_sent": SensorData( + FritzSensorEntityDescription( + key="link_kb_s_sent", name="Link Upload Throughput", - unit_of_measurement=DATA_RATE_KILOBITS_PER_SECOND, + native_unit_of_measurement=DATA_RATE_KILOBITS_PER_SECOND, icon="mdi:upload", - state_provider=_retrieve_link_kb_s_sent_state, + value_fn=_retrieve_link_kb_s_sent_state, connection_type=DSL_CONNECTION, ), - "link_kb_s_received": SensorData( + FritzSensorEntityDescription( + key="link_kb_s_received", name="Link Download Throughput", - unit_of_measurement=DATA_RATE_KILOBITS_PER_SECOND, + native_unit_of_measurement=DATA_RATE_KILOBITS_PER_SECOND, icon="mdi:download", - state_provider=_retrieve_link_kb_s_received_state, + value_fn=_retrieve_link_kb_s_received_state, connection_type=DSL_CONNECTION, ), - "link_noise_margin_sent": SensorData( + FritzSensorEntityDescription( + key="link_noise_margin_sent", name="Link Upload Noise Margin", - unit_of_measurement=SIGNAL_STRENGTH_DECIBELS, + native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS, icon="mdi:upload", - state_provider=_retrieve_link_noise_margin_sent_state, + value_fn=_retrieve_link_noise_margin_sent_state, connection_type=DSL_CONNECTION, ), - "link_noise_margin_received": SensorData( + FritzSensorEntityDescription( + key="link_noise_margin_received", name="Link Download Noise Margin", - unit_of_measurement=SIGNAL_STRENGTH_DECIBELS, + native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS, icon="mdi:download", - state_provider=_retrieve_link_noise_margin_received_state, + value_fn=_retrieve_link_noise_margin_received_state, connection_type=DSL_CONNECTION, ), - "link_attenuation_sent": SensorData( + FritzSensorEntityDescription( + key="link_attenuation_sent", name="Link Upload Power Attenuation", - unit_of_measurement=SIGNAL_STRENGTH_DECIBELS, + native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS, icon="mdi:upload", - state_provider=_retrieve_link_attenuation_sent_state, + value_fn=_retrieve_link_attenuation_sent_state, connection_type=DSL_CONNECTION, ), - "link_attenuation_received": SensorData( + FritzSensorEntityDescription( + key="link_attenuation_received", name="Link Download Power Attenuation", - unit_of_measurement=SIGNAL_STRENGTH_DECIBELS, + native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS, icon="mdi:download", - state_provider=_retrieve_link_attenuation_received_state, + value_fn=_retrieve_link_attenuation_received_state, connection_type=DSL_CONNECTION, ), -} +) async def async_setup_entry( @@ -266,7 +284,6 @@ async def async_setup_entry( # Only routers are supported at the moment return - entities = [] dsl: bool = False try: dslinterface = await hass.async_add_executor_job( @@ -283,40 +300,34 @@ async def async_setup_entry( ): pass - for sensor_type, sensor_data in SENSOR_DATA.items(): - if not dsl and sensor_data.get("connection_type") == DSL_CONNECTION: - continue - entities.append(FritzBoxSensor(fritzbox_tools, entry.title, sensor_type)) + entities = [ + FritzBoxSensor(fritzbox_tools, entry.title, description) + for description in SENSOR_TYPES + if dsl or description.connection_type != DSL_CONNECTION + ] - if entities: - async_add_entities(entities, True) + async_add_entities(entities, True) class FritzBoxSensor(FritzBoxBaseEntity, SensorEntity): """Define FRITZ!Box connectivity class.""" + entity_description: FritzSensorEntityDescription + def __init__( - self, fritzbox_tools: FritzBoxTools, device_friendly_name: str, sensor_type: str + self, + fritzbox_tools: FritzBoxTools, + device_friendly_name: str, + description: FritzSensorEntityDescription, ) -> None: """Init FRITZ!Box connectivity class.""" - self._sensor_data: SensorData = SENSOR_DATA[sensor_type] + self.entity_description = description self._last_device_value: str | None = None self._attr_available = True - self._attr_device_class = self._sensor_data.get("device_class") - self._attr_icon = self._sensor_data.get("icon") - self._attr_name = f"{device_friendly_name} {self._sensor_data['name']}" - self._attr_state_class = self._sensor_data.get("state_class") - self._attr_native_unit_of_measurement = self._sensor_data.get( - "unit_of_measurement" - ) - self._attr_unique_id = f"{fritzbox_tools.unique_id}-{sensor_type}" + self._attr_name = f"{device_friendly_name} {description.name}" + self._attr_unique_id = f"{fritzbox_tools.unique_id}-{description.key}" super().__init__(fritzbox_tools, device_friendly_name) - @property - def _state_provider(self) -> Callable: - """Return the state provider for the binary sensor.""" - return self._sensor_data["state_provider"] - def update(self) -> None: """Update data.""" _LOGGER.debug("Updating FRITZ!Box sensors") @@ -329,6 +340,6 @@ class FritzBoxSensor(FritzBoxBaseEntity, SensorEntity): self._attr_available = False return - self._attr_native_value = self._last_device_value = self._state_provider( - status, self._last_device_value - ) + self._attr_native_value = ( + self._last_device_value + ) = self.entity_description.value_fn(status, self._last_device_value) diff --git a/tests/components/fritz/test_config_flow.py b/tests/components/fritz/test_config_flow.py index 1b2a89f0450..0aecefedf0d 100644 --- a/tests/components/fritz/test_config_flow.py +++ b/tests/components/fritz/test_config_flow.py @@ -42,7 +42,7 @@ ATTR_NEW_SERIAL_NUMBER = "NewSerialNumber" MOCK_HOST = "fake_host" MOCK_SERIAL_NUMBER = "fake_serial_number" - +MOCK_FIRMWARE_INFO = [True, "1.1.1"] MOCK_USER_DATA = MOCK_CONFIG[DOMAIN][CONF_DEVICES][0] MOCK_DEVICE_INFO = { @@ -73,6 +73,9 @@ async def test_user(hass: HomeAssistant, fc_class_mock, mock_get_source_ip): "homeassistant.components.fritz.common.FritzConnection", side_effect=fc_class_mock, ), patch("homeassistant.components.fritz.common.FritzStatus"), patch( + "homeassistant.components.fritz.common.FritzBoxTools._update_device_info", + return_value=MOCK_FIRMWARE_INFO, + ), patch( "homeassistant.components.fritz.async_setup_entry" ) as mock_setup_entry, patch( "requests.get" @@ -120,6 +123,9 @@ async def test_user_already_configured( "homeassistant.components.fritz.common.FritzConnection", side_effect=fc_class_mock, ), patch("homeassistant.components.fritz.common.FritzStatus"), patch( + "homeassistant.components.fritz.common.FritzBoxTools._update_device_info", + return_value=MOCK_FIRMWARE_INFO, + ), patch( "requests.get" ) as mock_request_get, patch( "requests.post" @@ -225,6 +231,9 @@ async def test_reauth_successful( "homeassistant.components.fritz.common.FritzConnection", side_effect=fc_class_mock, ), patch("homeassistant.components.fritz.common.FritzStatus"), patch( + "homeassistant.components.fritz.common.FritzBoxTools._update_device_info", + return_value=MOCK_FIRMWARE_INFO, + ), patch( "homeassistant.components.fritz.async_setup_entry" ) as mock_setup_entry, patch( "requests.get" @@ -397,6 +406,9 @@ async def test_ssdp(hass: HomeAssistant, fc_class_mock, mock_get_source_ip): "homeassistant.components.fritz.common.FritzConnection", side_effect=fc_class_mock, ), patch("homeassistant.components.fritz.common.FritzStatus"), patch( + "homeassistant.components.fritz.common.FritzBoxTools._update_device_info", + return_value=MOCK_FIRMWARE_INFO, + ), patch( "homeassistant.components.fritz.async_setup_entry" ) as mock_setup_entry, patch( "requests.get" @@ -462,6 +474,9 @@ async def test_import(hass: HomeAssistant, fc_class_mock, mock_get_source_ip): "homeassistant.components.fritz.common.FritzConnection", side_effect=fc_class_mock, ), patch("homeassistant.components.fritz.common.FritzStatus"), patch( + "homeassistant.components.fritz.common.FritzBoxTools._update_device_info", + return_value=MOCK_FIRMWARE_INFO, + ), patch( "homeassistant.components.fritz.async_setup_entry" ) as mock_setup_entry, patch( "requests.get"