From 4310a7d814977ef5a64fcfe7d1821d12c488ac44 Mon Sep 17 00:00:00 2001 From: ehendrix23 Date: Fri, 3 Sep 2021 09:15:28 -0600 Subject: [PATCH] Add upnp sensor for IP, Status, and Uptime (#54780) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Joakim Sørensen --- homeassistant/components/upnp/__init__.py | 35 ++- .../components/upnp/binary_sensor.py | 35 +-- homeassistant/components/upnp/const.py | 9 +- homeassistant/components/upnp/device.py | 14 +- homeassistant/components/upnp/sensor.py | 224 ++++++++++-------- tests/components/upnp/mock_upnp_device.py | 12 +- 6 files changed, 191 insertions(+), 138 deletions(-) diff --git a/homeassistant/components/upnp/__init__.py b/homeassistant/components/upnp/__init__.py index 80a7753ec8c..9541331fe0b 100644 --- a/homeassistant/components/upnp/__init__.py +++ b/homeassistant/components/upnp/__init__.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio from collections.abc import Mapping +from dataclasses import dataclass from datetime import timedelta from ipaddress import ip_address from typing import Any @@ -11,8 +12,10 @@ import voluptuous as vol from homeassistant import config_entries from homeassistant.components import ssdp +from homeassistant.components.binary_sensor import BinarySensorEntityDescription from homeassistant.components.network import async_get_source_ip from homeassistant.components.network.const import PUBLIC_TARGET_IP +from homeassistant.components.sensor import SensorEntityDescription from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, callback from homeassistant.exceptions import ConfigEntryNotReady @@ -191,6 +194,20 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> return await hass.config_entries.async_unload_platforms(config_entry, PLATFORMS) +@dataclass +class UpnpBinarySensorEntityDescription(BinarySensorEntityDescription): + """A class that describes UPnP entities.""" + + format: str = "s" + + +@dataclass +class UpnpSensorEntityDescription(SensorEntityDescription): + """A class that describes a sensor UPnP entities.""" + + format: str = "s" + + class UpnpDataUpdateCoordinator(DataUpdateCoordinator): """Define an object to update data from UPNP device.""" @@ -221,14 +238,30 @@ class UpnpEntity(CoordinatorEntity): """Base class for UPnP/IGD entities.""" coordinator: UpnpDataUpdateCoordinator + entity_description: UpnpSensorEntityDescription | UpnpBinarySensorEntityDescription - def __init__(self, coordinator: UpnpDataUpdateCoordinator) -> None: + def __init__( + self, + coordinator: UpnpDataUpdateCoordinator, + entity_description: UpnpSensorEntityDescription + | UpnpBinarySensorEntityDescription, + ) -> None: """Initialize the base entities.""" super().__init__(coordinator) self._device = coordinator.device + self.entity_description = entity_description + self._attr_name = f"{coordinator.device.name} {entity_description.name}" + self._attr_unique_id = f"{coordinator.device.udn}_{entity_description.key}" self._attr_device_info = { "connections": {(dr.CONNECTION_UPNP, coordinator.device.udn)}, "name": coordinator.device.name, "manufacturer": coordinator.device.manufacturer, "model": coordinator.device.model_name, } + + @property + def available(self) -> bool: + """Return if entity is available.""" + return super().available and ( + self.coordinator.data.get(self.entity_description.key) or False + ) diff --git a/homeassistant/components/upnp/binary_sensor.py b/homeassistant/components/upnp/binary_sensor.py index 2f2f0af0e96..3bf9635c78b 100644 --- a/homeassistant/components/upnp/binary_sensor.py +++ b/homeassistant/components/upnp/binary_sensor.py @@ -9,8 +9,15 @@ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback -from . import UpnpDataUpdateCoordinator, UpnpEntity -from .const import DOMAIN, LOGGER, WANSTATUS +from . import UpnpBinarySensorEntityDescription, UpnpDataUpdateCoordinator, UpnpEntity +from .const import DOMAIN, LOGGER, WAN_STATUS + +BINARYSENSOR_ENTITY_DESCRIPTIONS: tuple[UpnpBinarySensorEntityDescription, ...] = ( + UpnpBinarySensorEntityDescription( + key=WAN_STATUS, + name="wan status", + ), +) async def async_setup_entry( @@ -23,10 +30,14 @@ async def async_setup_entry( LOGGER.debug("Adding binary sensor") - sensors = [ - UpnpStatusBinarySensor(coordinator), - ] - async_add_entities(sensors) + async_add_entities( + UpnpStatusBinarySensor( + coordinator=coordinator, + entity_description=entity_description, + ) + for entity_description in BINARYSENSOR_ENTITY_DESCRIPTIONS + if coordinator.data.get(entity_description.key) is not None + ) class UpnpStatusBinarySensor(UpnpEntity, BinarySensorEntity): @@ -37,18 +48,12 @@ class UpnpStatusBinarySensor(UpnpEntity, BinarySensorEntity): def __init__( self, coordinator: UpnpDataUpdateCoordinator, + entity_description: UpnpBinarySensorEntityDescription, ) -> None: """Initialize the base sensor.""" - super().__init__(coordinator) - self._attr_name = f"{coordinator.device.name} wan status" - self._attr_unique_id = f"{coordinator.device.udn}_wanstatus" - - @property - def available(self) -> bool: - """Return if entity is available.""" - return super().available and self.coordinator.data.get(WANSTATUS) + super().__init__(coordinator=coordinator, entity_description=entity_description) @property def is_on(self) -> bool: """Return true if the binary sensor is on.""" - return self.coordinator.data[WANSTATUS] == "Connected" + return self.coordinator.data[self.entity_description.key] == "Connected" diff --git a/homeassistant/components/upnp/const.py b/homeassistant/components/upnp/const.py index 769e398c5a4..142c00ad27f 100644 --- a/homeassistant/components/upnp/const.py +++ b/homeassistant/components/upnp/const.py @@ -18,9 +18,9 @@ PACKETS_SENT = "packets_sent" TIMESTAMP = "timestamp" DATA_PACKETS = "packets" DATA_RATE_PACKETS_PER_SECOND = f"{DATA_PACKETS}/{TIME_SECONDS}" -WANSTATUS = "wan_status" -WANIP = "wan_ip" -UPTIME = "uptime" +WAN_STATUS = "wan_status" +ROUTER_IP = "ip" +ROUTER_UPTIME = "uptime" KIBIBYTE = 1024 UPDATE_INTERVAL = timedelta(seconds=30) CONFIG_ENTRY_SCAN_INTERVAL = "scan_interval" @@ -31,3 +31,6 @@ DEFAULT_SCAN_INTERVAL = timedelta(seconds=30).total_seconds() ST_IGD_V1 = "urn:schemas-upnp-org:device:InternetGatewayDevice:1" ST_IGD_V2 = "urn:schemas-upnp-org:device:InternetGatewayDevice:2" SSDP_SEARCH_TIMEOUT = 4 + +RAW_SENSOR = "raw_sensor" +DERIVED_SENSOR = "derived_sensor" diff --git a/homeassistant/components/upnp/device.py b/homeassistant/components/upnp/device.py index ca06f501405..a1040816629 100644 --- a/homeassistant/components/upnp/device.py +++ b/homeassistant/components/upnp/device.py @@ -14,7 +14,6 @@ from async_upnp_client.profiles.igd import IgdDevice from homeassistant.core import HomeAssistant from homeassistant.helpers.aiohttp_client import async_get_clientsession -from homeassistant.helpers.update_coordinator import DataUpdateCoordinator import homeassistant.util.dt as dt_util from .const import ( @@ -26,10 +25,10 @@ from .const import ( LOGGER as _LOGGER, PACKETS_RECEIVED, PACKETS_SENT, + ROUTER_IP, + ROUTER_UPTIME, TIMESTAMP, - UPTIME, - WANIP, - WANSTATUS, + WAN_STATUS, ) @@ -49,7 +48,6 @@ class Device: """Initialize UPnP/IGD device.""" self._igd_device = igd_device self._device_updater = device_updater - self.coordinator: DataUpdateCoordinator = None @classmethod async def async_create_device( @@ -168,7 +166,7 @@ class Device: ) return { - WANSTATUS: values[0][0] if values[0] is not None else None, - UPTIME: values[0][2] if values[0] is not None else None, - WANIP: values[1], + WAN_STATUS: values[0][0] if values[0] is not None else None, + ROUTER_UPTIME: values[0][2] if values[0] is not None else None, + ROUTER_IP: values[1], } diff --git a/homeassistant/components/upnp/sensor.py b/homeassistant/components/upnp/sensor.py index 185d3ecac6d..bebb8e3e957 100644 --- a/homeassistant/components/upnp/sensor.py +++ b/homeassistant/components/upnp/sensor.py @@ -3,73 +3,111 @@ from __future__ import annotations from homeassistant.components.sensor import SensorEntity from homeassistant.config_entries import ConfigEntry -from homeassistant.const import DATA_BYTES, DATA_RATE_KIBIBYTES_PER_SECOND +from homeassistant.const import DATA_BYTES, DATA_RATE_KIBIBYTES_PER_SECOND, TIME_SECONDS from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback -from . import UpnpDataUpdateCoordinator, UpnpEntity +from . import UpnpDataUpdateCoordinator, UpnpEntity, UpnpSensorEntityDescription from .const import ( BYTES_RECEIVED, BYTES_SENT, DATA_PACKETS, DATA_RATE_PACKETS_PER_SECOND, + DERIVED_SENSOR, DOMAIN, KIBIBYTE, LOGGER, PACKETS_RECEIVED, PACKETS_SENT, + RAW_SENSOR, + ROUTER_IP, + ROUTER_UPTIME, TIMESTAMP, + WAN_STATUS, ) -SENSOR_TYPES = { - BYTES_RECEIVED: { - "device_value_key": BYTES_RECEIVED, - "name": f"{DATA_BYTES} received", - "unit": DATA_BYTES, - "unique_id": BYTES_RECEIVED, - "derived_name": f"{DATA_RATE_KIBIBYTES_PER_SECOND} received", - "derived_unit": DATA_RATE_KIBIBYTES_PER_SECOND, - "derived_unique_id": "KiB/sec_received", - }, - BYTES_SENT: { - "device_value_key": BYTES_SENT, - "name": f"{DATA_BYTES} sent", - "unit": DATA_BYTES, - "unique_id": BYTES_SENT, - "derived_name": f"{DATA_RATE_KIBIBYTES_PER_SECOND} sent", - "derived_unit": DATA_RATE_KIBIBYTES_PER_SECOND, - "derived_unique_id": "KiB/sec_sent", - }, - PACKETS_RECEIVED: { - "device_value_key": PACKETS_RECEIVED, - "name": f"{DATA_PACKETS} received", - "unit": DATA_PACKETS, - "unique_id": PACKETS_RECEIVED, - "derived_name": f"{DATA_RATE_PACKETS_PER_SECOND} received", - "derived_unit": DATA_RATE_PACKETS_PER_SECOND, - "derived_unique_id": "packets/sec_received", - }, - PACKETS_SENT: { - "device_value_key": PACKETS_SENT, - "name": f"{DATA_PACKETS} sent", - "unit": DATA_PACKETS, - "unique_id": PACKETS_SENT, - "derived_name": f"{DATA_RATE_PACKETS_PER_SECOND} sent", - "derived_unit": DATA_RATE_PACKETS_PER_SECOND, - "derived_unique_id": "packets/sec_sent", - }, +SENSOR_ENTITY_DESCRIPTIONS: dict[str, tuple[UpnpSensorEntityDescription, ...]] = { + RAW_SENSOR: ( + UpnpSensorEntityDescription( + key=BYTES_RECEIVED, + name=f"{DATA_BYTES} received", + icon="mdi:server-network", + native_unit_of_measurement=DATA_BYTES, + format="d", + ), + UpnpSensorEntityDescription( + key=BYTES_SENT, + name=f"{DATA_BYTES} sent", + icon="mdi:server-network", + native_unit_of_measurement=DATA_BYTES, + format="d", + ), + UpnpSensorEntityDescription( + key=PACKETS_RECEIVED, + name=f"{DATA_PACKETS} received", + icon="mdi:server-network", + native_unit_of_measurement=DATA_PACKETS, + format="d", + ), + UpnpSensorEntityDescription( + key=PACKETS_SENT, + name=f"{DATA_PACKETS} sent", + icon="mdi:server-network", + native_unit_of_measurement=DATA_PACKETS, + format="d", + ), + UpnpSensorEntityDescription( + key=ROUTER_IP, + name="External IP", + icon="mdi:server-network", + ), + UpnpSensorEntityDescription( + key=ROUTER_UPTIME, + name="Uptime", + icon="mdi:server-network", + native_unit_of_measurement=TIME_SECONDS, + entity_registry_enabled_default=False, + format="d", + ), + UpnpSensorEntityDescription( + key=WAN_STATUS, + name="wan status", + icon="mdi:server-network", + ), + ), + DERIVED_SENSOR: ( + UpnpSensorEntityDescription( + key="KiB/sec_received", + name=f"{DATA_RATE_KIBIBYTES_PER_SECOND} received", + icon="mdi:server-network", + native_unit_of_measurement=DATA_RATE_KIBIBYTES_PER_SECOND, + format=".1f", + ), + UpnpSensorEntityDescription( + key="KiB/sent", + name=f"{DATA_RATE_KIBIBYTES_PER_SECOND} sent", + icon="mdi:server-network", + native_unit_of_measurement=DATA_RATE_KIBIBYTES_PER_SECOND, + format=".1f", + ), + UpnpSensorEntityDescription( + key="packets/sec_received", + name=f"{DATA_RATE_PACKETS_PER_SECOND} received", + icon="mdi:server-network", + native_unit_of_measurement=DATA_RATE_PACKETS_PER_SECOND, + format=".1f", + ), + UpnpSensorEntityDescription( + key="packets/sent", + name=f"{DATA_RATE_PACKETS_PER_SECOND} sent", + icon="mdi:server-network", + native_unit_of_measurement=DATA_RATE_PACKETS_PER_SECOND, + format=".1f", + ), + ), } -async def async_setup_platform( - hass: HomeAssistant, config, async_add_entities, discovery_info=None -) -> None: - """Old way of setting up UPnP/IGD sensors.""" - LOGGER.debug( - "async_setup_platform: config: %s, discovery: %s", config, discovery_info - ) - - async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, @@ -80,50 +118,31 @@ async def async_setup_entry( LOGGER.debug("Adding sensors") - sensors = [ - RawUpnpSensor(coordinator, SENSOR_TYPES[BYTES_RECEIVED]), - RawUpnpSensor(coordinator, SENSOR_TYPES[BYTES_SENT]), - RawUpnpSensor(coordinator, SENSOR_TYPES[PACKETS_RECEIVED]), - RawUpnpSensor(coordinator, SENSOR_TYPES[PACKETS_SENT]), - DerivedUpnpSensor(coordinator, SENSOR_TYPES[BYTES_RECEIVED]), - DerivedUpnpSensor(coordinator, SENSOR_TYPES[BYTES_SENT]), - DerivedUpnpSensor(coordinator, SENSOR_TYPES[PACKETS_RECEIVED]), - DerivedUpnpSensor(coordinator, SENSOR_TYPES[PACKETS_SENT]), - ] - async_add_entities(sensors) + entities = [] + entities.append( + RawUpnpSensor( + coordinator=coordinator, + entity_description=entity_description, + ) + for entity_description in SENSOR_ENTITY_DESCRIPTIONS[RAW_SENSOR] + if coordinator.data.get(entity_description.key) is not None + ) + + entities.append( + DerivedUpnpSensor( + coordinator=coordinator, + entity_description=entity_description, + ) + for entity_description in SENSOR_ENTITY_DESCRIPTIONS[DERIVED_SENSOR] + if coordinator.data.get(entity_description.key) is not None + ) + + async_add_entities(entities) class UpnpSensor(UpnpEntity, SensorEntity): """Base class for UPnP/IGD sensors.""" - def __init__( - self, - coordinator: UpnpDataUpdateCoordinator, - sensor_type: dict[str, str], - ) -> None: - """Initialize the base sensor.""" - super().__init__(coordinator) - self._sensor_type = sensor_type - self._attr_name = f"{coordinator.device.name} {sensor_type['name']}" - self._attr_unique_id = f"{coordinator.device.udn}_{sensor_type['unique_id']}" - - @property - def icon(self) -> str: - """Icon to use in the frontend, if any.""" - return "mdi:server-network" - - @property - def available(self) -> bool: - """Return if entity is available.""" - return super().available and self.coordinator.data.get( - self._sensor_type["device_value_key"] - ) - - @property - def native_unit_of_measurement(self) -> str: - """Return the unit of measurement of this entity, if any.""" - return self._sensor_type["unit"] - class RawUpnpSensor(UpnpSensor): """Representation of a UPnP/IGD sensor.""" @@ -131,30 +150,26 @@ class RawUpnpSensor(UpnpSensor): @property def native_value(self) -> str | None: """Return the state of the device.""" - device_value_key = self._sensor_type["device_value_key"] - value = self.coordinator.data[device_value_key] + value = self.coordinator.data[self.entity_description.key] if value is None: return None - return format(value, "d") + return format(value, self.entity_description.format) class DerivedUpnpSensor(UpnpSensor): """Representation of a UNIT Sent/Received per second sensor.""" - def __init__(self, coordinator: UpnpDataUpdateCoordinator, sensor_type) -> None: + entity_description: UpnpSensorEntityDescription + + def __init__( + self, + coordinator: UpnpDataUpdateCoordinator, + entity_description: UpnpSensorEntityDescription, + ) -> None: """Initialize sensor.""" - super().__init__(coordinator, sensor_type) + super().__init__(coordinator=coordinator, entity_description=entity_description) self._last_value = None self._last_timestamp = None - self._attr_name = f"{coordinator.device.name} {sensor_type['derived_name']}" - self._attr_unique_id = ( - f"{coordinator.device.udn}_{sensor_type['derived_unique_id']}" - ) - - @property - def native_unit_of_measurement(self) -> str: - """Return the unit of measurement of this entity, if any.""" - return self._sensor_type["derived_unit"] def _has_overflowed(self, current_value) -> bool: """Check if value has overflowed.""" @@ -164,8 +179,7 @@ class DerivedUpnpSensor(UpnpSensor): def native_value(self) -> str | None: """Return the state of the device.""" # Can't calculate any derivative if we have only one value. - device_value_key = self._sensor_type["device_value_key"] - current_value = self.coordinator.data[device_value_key] + current_value = self.coordinator.data[self.entity_description.key] if current_value is None: return None current_timestamp = self.coordinator.data[TIMESTAMP] @@ -176,7 +190,7 @@ class DerivedUpnpSensor(UpnpSensor): # Calculate derivative. delta_value = current_value - self._last_value - if self._sensor_type["unit"] == DATA_BYTES: + if self.entity_description.native_unit_of_measurement == DATA_BYTES: delta_value /= KIBIBYTE delta_time = current_timestamp - self._last_timestamp if delta_time.total_seconds() == 0: @@ -188,4 +202,4 @@ class DerivedUpnpSensor(UpnpSensor): self._last_value = current_value self._last_timestamp = current_timestamp - return format(derived, ".1f") + return format(derived, self.entity_description.format) diff --git a/tests/components/upnp/mock_upnp_device.py b/tests/components/upnp/mock_upnp_device.py index 42c9291f30f..230fd480cb1 100644 --- a/tests/components/upnp/mock_upnp_device.py +++ b/tests/components/upnp/mock_upnp_device.py @@ -10,10 +10,10 @@ from homeassistant.components.upnp.const import ( BYTES_SENT, PACKETS_RECEIVED, PACKETS_SENT, + ROUTER_IP, + ROUTER_UPTIME, TIMESTAMP, - UPTIME, - WANIP, - WANSTATUS, + WAN_STATUS, ) from homeassistant.components.upnp.device import Device from homeassistant.util import dt @@ -83,9 +83,9 @@ class MockDevice(Device): """Get connection status, uptime, and external IP.""" self.status_times_polled += 1 return { - WANSTATUS: "Connected", - UPTIME: 0, - WANIP: "192.168.0.1", + WAN_STATUS: "Connected", + ROUTER_UPTIME: 0, + ROUTER_IP: "192.168.0.1", } async def async_start(self) -> None: