jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 17:27:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fritz new binary sensor for link and firmware status + code cleanup (#55446)

Browse Source
This commit is contained in:
Simone Chemelli 2021-09-30 11:18:04 +02:00 committed by GitHub
parent a6a3745413
commit 8993ff0377
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 198 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

111
homeassistant/components/fritz/binary_sensor.py
View File

@ -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,
}

44
homeassistant/components/fritz/common.py
View File

@ -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,
}

4
homeassistant/components/fritz/const.py
View File

@ -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"

171
homeassistant/components/fritz/sensor.py
View File

@ -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)

17
tests/components/fritz/test_config_flow.py
View File

@ -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"