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

Add upnp sensor for IP, Status, and Uptime (#54780)

Browse source 
Co-authored-by: Joakim Sørensen <hi@ludeeus.dev>
This commit is contained in:
ehendrix23 2021-09-03 09:15:28 -06:00 • committed by GitHub
parent ae9e3c237a
commit 4310a7d814
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 191 additions and 138 deletions
Download patch file Download diff file Expand all files Collapse all files

224
homeassistant/components/upnp/sensor.py
View file

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