Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Rewrite parts of upnp component (#33108) 

* Rewrite parts of upnp component

* Linting

* Add SCAN_INTERVAL

* Get values simultaneously

* Move to time related constants, as per #32065

* Linting

* Move constant KIBIBYTE to homeassistant.const

* Simplify code

* Fix tests for #33344

* Changes after review

* Update homeassistant/components/upnp/sensor.py

* Changes after review

* Formatting

* Formatting

* Use ST from discovery info to avoid swapping device_types if device advertises multiple versions

* Linting

* Requirements for upnp + dlna_dmr components

* Linting

* Regen requirements

* Changes after review by @MartinHjelmare

* Changes after review by @MartinHjelmare

* Formatting

* Linting

* Changes after review by @MartinHjelmare

* Changes after review by @MartinHjelmare

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
pull/33997/head
Steven Looman 2020-04-11 00:24:03 +02:00 committed by GitHub
parent de3f5e8d69
commit dfc66b2018
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 334 additions and 289 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/dlna_dmr/manifest.json
View File

@ -2,6 +2,6 @@
"domain": "dlna_dmr",
"name": "DLNA Digital Media Renderer",
"documentation": "https://www.home-assistant.io/integrations/dlna_dmr",
"requirements": ["async-upnp-client==0.14.12"],
"requirements": ["async-upnp-client==0.14.13"],
"codeowners": []
}

51
homeassistant/components/upnp/__init__.py
View File

@ -1,17 +1,15 @@
"""Open ports in your router for Home Assistant and provide statistics."""
from ipaddress import ip_address
from operator import itemgetter
from typing import Mapping
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
dispatcher,
)
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from homeassistant.util import get_local_ip
@ -23,7 +21,6 @@ from .const import (
CONF_PORTS,
DOMAIN,
LOGGER as _LOGGER,
SIGNAL_REMOVE_SENSOR,
)
from .device import Device
@ -37,7 +34,7 @@ CONFIG_SCHEMA = vol.Schema(
vol.Optional(CONF_ENABLE_PORT_MAPPING, default=False): cv.boolean,
vol.Optional(CONF_ENABLE_SENSORS, default=True): cv.boolean,
vol.Optional(CONF_LOCAL_IP): vol.All(ip_address, cv.string),
vol.Optional(CONF_PORTS): vol.Schema(
vol.Optional(CONF_PORTS, default={}): vol.Schema(
{vol.Any(CONF_HASS, cv.port): vol.Any(CONF_HASS, cv.port)}
),
}
@ -47,7 +44,7 @@ CONFIG_SCHEMA = vol.Schema(
)
def _substitute_hass_ports(ports, hass_port=None):
def _substitute_hass_ports(ports: Mapping, hass_port: int = None) -> Mapping:
"""
Substitute 'hass' for the hass_port.
@ -86,8 +83,11 @@ def _substitute_hass_ports(ports, hass_port=None):
return ports
async def async_discover_and_construct(hass, udn=None) -> Device:
async def async_discover_and_construct(
hass: HomeAssistantType, udn: str = None, st: str = None
) -> Device:
"""Discovery devices and construct a Device for one."""
# pylint: disable=invalid-name
discovery_infos = await Device.async_discover(hass)
if not discovery_infos:
_LOGGER.info("No UPnP/IGD devices discovered")
@ -95,7 +95,11 @@ async def async_discover_and_construct(hass, udn=None) -> Device:
if udn:
# get the discovery info with specified UDN
_LOGGER.debug("Discovery_infos: %s", discovery_infos)
filtered = [di for di in discovery_infos if di["udn"] == udn]
if st:
_LOGGER.debug("Filtering on ST: %s", st)
filtered = [di for di in discovery_infos if di["st"] == st]
if not filtered:
_LOGGER.warning(
'Wanted UPnP/IGD device with UDN "%s" not found, ' "aborting", udn
@ -125,8 +129,8 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType):
hass.data[DOMAIN] = {
"config": conf,
"devices": {},
"local_ip": config.get(CONF_LOCAL_IP, local_ip),
"ports": conf.get("ports", {}),
"local_ip": conf.get(CONF_LOCAL_IP, local_ip),
"ports": conf.get(CONF_PORTS),
}
if conf is not None:
@ -139,21 +143,24 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType):
return True
async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry):
async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry) -> bool:
"""Set up UPnP/IGD device from a config entry."""
domain_data = hass.data[DOMAIN]
conf = domain_data["config"]
# discover and construct
device = await async_discover_and_construct(hass, config_entry.data.get("udn"))
udn = config_entry.data.get("udn")
st = config_entry.data.get("st") # pylint: disable=invalid-name
device = await async_discover_and_construct(hass, udn, st)
if not device:
_LOGGER.info("Unable to create UPnP/IGD, aborting")
return False
raise ConfigEntryNotReady
# 'register'/save UDN
# 'register'/save UDN + ST
hass.data[DOMAIN]["devices"][device.udn] = device
hass.config_entries.async_update_entry(
entry=config_entry, data={**config_entry.data, "udn": device.udn}
entry=config_entry,
data={**config_entry.data, "udn": device.udn, "st": device.device_type},
)
# create device registry entry
@ -179,8 +186,8 @@ async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry):
# set up port mapping
if conf.get(CONF_ENABLE_PORT_MAPPING):
_LOGGER.debug("Enabling port mapping")
local_ip = domain_data["local_ip"]
ports = conf.get("ports", {})
local_ip = domain_data[CONF_LOCAL_IP]
ports = conf.get(CONF_PORTS, {})
hass_port = None
if hasattr(hass, "http"):
@ -200,7 +207,9 @@ async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry):
return True
async def async_unload_entry(hass: HomeAssistantType, config_entry: ConfigEntry):
async def async_unload_entry(
hass: HomeAssistantType, config_entry: ConfigEntry
) -> bool:
"""Unload a UPnP/IGD device from a config entry."""
udn = config_entry.data["udn"]
device = hass.data[DOMAIN]["devices"][udn]
@ -211,6 +220,4 @@ async def async_unload_entry(hass: HomeAssistantType, config_entry: ConfigEntry)
# remove sensors
_LOGGER.debug("Deleting sensors")
dispatcher.async_dispatcher_send(hass, SIGNAL_REMOVE_SENSOR, device)
return True
return await hass.config_entries.async_forward_entry_unload(config_entry, "sensor")

13
homeassistant/components/upnp/const.py
View File

@ -1,6 +1,9 @@
"""Constants for the IGD component."""
from datetime import timedelta
import logging
from homeassistant.const import TIME_SECONDS
CONF_ENABLE_PORT_MAPPING = "port_mapping"
CONF_ENABLE_SENSORS = "sensors"
CONF_HASS = "hass"
@ -8,4 +11,12 @@ CONF_LOCAL_IP = "local_ip"
CONF_PORTS = "ports"
DOMAIN = "upnp"
LOGGER = logging.getLogger(__package__)
SIGNAL_REMOVE_SENSOR = "upnp_remove_sensor"
BYTES_RECEIVED = "bytes_received"
BYTES_SENT = "bytes_sent"
PACKETS_RECEIVED = "packets_received"
PACKETS_SENT = "packets_sent"
TIMESTAMP = "timestamp"
DATA_PACKETS = "packets"
DATA_RATE_PACKETS_PER_SECOND = f"{DATA_PACKETS}/{TIME_SECONDS}"
KIBIBYTE = 1024
UPDATE_INTERVAL = timedelta(seconds=30)

95
homeassistant/components/upnp/device.py
View File

@ -1,6 +1,7 @@
"""Home Assistant representation of an UPnP/IGD."""
import asyncio
from ipaddress import IPv4Address
from typing import Mapping
import aiohttp
from async_upnp_client import UpnpError, UpnpFactory
@ -9,8 +10,18 @@ from async_upnp_client.profiles.igd import IgdDevice
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import HomeAssistantType
import homeassistant.util.dt as dt_util
from .const import CONF_LOCAL_IP, DOMAIN, LOGGER as _LOGGER
from .const import (
BYTES_RECEIVED,
BYTES_SENT,
CONF_LOCAL_IP,
DOMAIN,
LOGGER as _LOGGER,
PACKETS_RECEIVED,
PACKETS_SENT,
TIMESTAMP,
)
class Device:
@ -18,7 +29,7 @@ class Device:
def __init__(self, igd_device):
"""Initialize UPnP/IGD device."""
self._igd_device = igd_device
self._igd_device: IgdDevice = igd_device
self._mapped_ports = []
@classmethod
@ -61,26 +72,37 @@ class Device:
return cls(igd_device)
@property
def udn(self):
def udn(self) -> str:
"""Get the UDN."""
return self._igd_device.udn
@property
def name(self):
def name(self) -> str:
"""Get the name."""
return self._igd_device.name
@property
def manufacturer(self):
def manufacturer(self) -> str:
"""Get the manufacturer."""
return self._igd_device.manufacturer
@property
def model_name(self):
def model_name(self) -> str:
"""Get the model name."""
return self._igd_device.model_name
async def async_add_port_mappings(self, ports, local_ip):
@property
def device_type(self) -> str:
"""Get the device type."""
return self._igd_device.device_type
def __str__(self) -> str:
"""Get string representation."""
return f"IGD Device: {self.name}/{self.udn}"
async def async_add_port_mappings(
self, ports: Mapping[int, int], local_ip: str
) -> None:
"""Add port mappings."""
if local_ip == "127.0.0.1":
_LOGGER.error("Could not create port mapping, our IP is 127.0.0.1")
@ -93,7 +115,9 @@ class Device:
await self._async_add_port_mapping(external_port, local_ip, internal_port)
self._mapped_ports.append(external_port)
async def _async_add_port_mapping(self, external_port, local_ip, internal_port):
async def _async_add_port_mapping(
self, external_port: int, local_ip: str, internal_port: int
) -> None:
"""Add a port mapping."""
# create port mapping
_LOGGER.info(
@ -123,12 +147,12 @@ class Device:
internal_port,
)
async def async_delete_port_mappings(self):
"""Remove a port mapping."""
async def async_delete_port_mappings(self) -> None:
"""Remove port mappings."""
for port in self._mapped_ports:
await self._async_delete_port_mapping(port)
async def _async_delete_port_mapping(self, external_port):
async def _async_delete_port_mapping(self, external_port: int) -> None:
"""Remove a port mapping."""
_LOGGER.info("Deleting port mapping %s (TCP)", external_port)
try:
@ -140,30 +164,31 @@ class Device:
except (asyncio.TimeoutError, aiohttp.ClientError, UpnpError):
_LOGGER.error("Could not delete port mapping")
async def async_get_total_bytes_received(self):
"""Get total bytes received."""
try:
return await self._igd_device.async_get_total_bytes_received()
except asyncio.TimeoutError:
_LOGGER.warning("Timeout during get_total_bytes_received")
async def async_get_traffic_data(self) -> Mapping[str, any]:
"""
Get all traffic data in one go.
async def async_get_total_bytes_sent(self):
"""Get total bytes sent."""
try:
return await self._igd_device.async_get_total_bytes_sent()
except asyncio.TimeoutError:
_LOGGER.warning("Timeout during get_total_bytes_sent")
Traffic data consists of:
- total bytes sent
- total bytes received
- total packets sent
- total packats received
async def async_get_total_packets_received(self):
"""Get total packets received."""
try:
return await self._igd_device.async_get_total_packets_received()
except asyncio.TimeoutError:
_LOGGER.warning("Timeout during get_total_packets_received")
Data is timestamped.
"""
_LOGGER.debug("Getting traffic statistics from device: %s", self)
async def async_get_total_packets_sent(self):
"""Get total packets sent."""
try:
return await self._igd_device.async_get_total_packets_sent()
except asyncio.TimeoutError:
_LOGGER.warning("Timeout during get_total_packets_sent")
values = await asyncio.gather(
self._igd_device.async_get_total_bytes_received(),
self._igd_device.async_get_total_bytes_sent(),
self._igd_device.async_get_total_packets_received(),
self._igd_device.async_get_total_packets_sent(),
)
return {
TIMESTAMP: dt_util.utcnow(),
BYTES_RECEIVED: values[0],
BYTES_SENT: values[1],
PACKETS_RECEIVED: values[2],
PACKETS_SENT: values[3],
}

13
homeassistant/components/upnp/manifest.json
View File

@ -3,6 +3,15 @@
"name": "UPnP",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/upnp",
"requirements": ["async-upnp-client==0.14.12"],
"codeowners": ["@StevenLooman"]
"requirements": ["async-upnp-client==0.14.13"],
"dependencies": [],
"codeowners": ["@StevenLooman"],
"ssdp": [
{
"st": "urn:schemas-upnp-org:device:InternetGatewayDevice:1"
},
{
"st": "urn:schemas-upnp-org:device:InternetGatewayDevice:2"
}
]
}

374
homeassistant/components/upnp/sensor.py
View File

@ -1,275 +1,247 @@
"""Support for UPnP/IGD Sensors."""
from datetime import timedelta
import logging
from typing import Mapping
from homeassistant.const import DATA_BYTES, DATA_KIBIBYTES, TIME_SECONDS
from homeassistant.core import callback
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import DATA_BYTES, DATA_RATE_KIBIBYTES_PER_SECOND
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util import Throttle
import homeassistant.util.dt as dt_util
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN as DOMAIN_UPNP, SIGNAL_REMOVE_SENSOR
_LOGGER = logging.getLogger(__name__)
BYTES_RECEIVED = "bytes_received"
BYTES_SENT = "bytes_sent"
PACKETS_RECEIVED = "packets_received"
PACKETS_SENT = "packets_sent"
from .const import (
BYTES_RECEIVED,
BYTES_SENT,
DATA_PACKETS,
DATA_RATE_PACKETS_PER_SECOND,
DOMAIN,
KIBIBYTE,
LOGGER as _LOGGER,
PACKETS_RECEIVED,
PACKETS_SENT,
TIMESTAMP,
UPDATE_INTERVAL,
)
from .device import Device
SENSOR_TYPES = {
BYTES_RECEIVED: {"name": "bytes received", "unit": DATA_BYTES},
BYTES_SENT: {"name": "bytes sent", "unit": DATA_BYTES},
PACKETS_RECEIVED: {"name": "packets received", "unit": "packets"},
PACKETS_SENT: {"name": "packets sent", "unit": "packets"},
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",
},
}
IN = "received"
OUT = "sent"
KIBIBYTE = 1024
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=30)
async def async_setup_platform(
hass: HomeAssistantType, 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, config_entry, async_add_entities):
"""Set up the UPnP/IGD sensor."""
@callback
def async_add_sensor(device):
"""Add sensors from UPnP/IGD device."""
# raw sensors + per-second sensors
sensors = [
RawUPnPIGDSensor(device, name, sensor_type)
for name, sensor_type in SENSOR_TYPES.items()
]
sensors += [
KBytePerSecondUPnPIGDSensor(device, IN),
KBytePerSecondUPnPIGDSensor(device, OUT),
PacketsPerSecondUPnPIGDSensor(device, IN),
PacketsPerSecondUPnPIGDSensor(device, OUT),
]
async_add_entities(sensors, True)
async def async_setup_entry(
hass, config_entry: ConfigEntry, async_add_entities
) -> None:
"""Set up the UPnP/IGD sensors."""
data = config_entry.data
if "udn" in data:
udn = data["udn"]
else:
# any device will do
udn = list(hass.data[DOMAIN_UPNP]["devices"].keys())[0]
udn = list(hass.data[DOMAIN]["devices"].keys())[0]
device = hass.data[DOMAIN_UPNP]["devices"][udn]
async_add_sensor(device)
device: Device = hass.data[DOMAIN]["devices"][udn]
_LOGGER.debug("Adding sensors")
coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name=device.name,
update_method=device.async_get_traffic_data,
update_interval=timedelta(seconds=UPDATE_INTERVAL.seconds),
)
await coordinator.async_refresh()
sensors = [
RawUpnpSensor(coordinator, device, SENSOR_TYPES[BYTES_RECEIVED]),
RawUpnpSensor(coordinator, device, SENSOR_TYPES[BYTES_SENT]),
RawUpnpSensor(coordinator, device, SENSOR_TYPES[PACKETS_RECEIVED]),
RawUpnpSensor(coordinator, device, SENSOR_TYPES[PACKETS_SENT]),
DerivedUpnpSensor(coordinator, device, SENSOR_TYPES[BYTES_RECEIVED]),
DerivedUpnpSensor(coordinator, device, SENSOR_TYPES[BYTES_SENT]),
DerivedUpnpSensor(coordinator, device, SENSOR_TYPES[PACKETS_RECEIVED]),
DerivedUpnpSensor(coordinator, device, SENSOR_TYPES[PACKETS_SENT]),
]
async_add_entities(sensors, True)
class UpnpSensor(Entity):
"""Base class for UPnP/IGD sensors."""
def __init__(self, device):
def __init__(
self,
coordinator: DataUpdateCoordinator,
device: Device,
sensor_type: Mapping[str, str],
) -> None:
"""Initialize the base sensor."""
self._coordinator = coordinator
self._device = device
async def async_added_to_hass(self):
"""Subscribe to sensors events."""
self.async_on_remove(
async_dispatcher_connect(
self.hass, SIGNAL_REMOVE_SENSOR, self._upnp_remove_sensor
)
)
@callback
def _upnp_remove_sensor(self, device):
"""Remove sensor."""
if self._device != device:
# not for us
return
self.hass.async_create_task(self.async_remove())
self._sensor_type = sensor_type
@property
def device_info(self):
def should_poll(self) -> bool:
"""Inform we should not be polled."""
return False
@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."""
device_value_key = self._sensor_type["device_value_key"]
return (
self._coordinator.last_update_success
and device_value_key in self._coordinator.data
)
@property
def name(self) -> str:
"""Return the name of the sensor."""
return f"{self._device.name} {self._sensor_type['name']}"
@property
def unique_id(self) -> str:
"""Return an unique ID."""
return f"{self._device.udn}_{self._sensor_type['unique_id']}"
@property
def unit_of_measurement(self) -> str:
"""Return the unit of measurement of this entity, if any."""
return self._sensor_type["unit"]
@property
def device_info(self) -> Mapping[str, any]:
"""Get device info."""
return {
"connections": {(dr.CONNECTION_UPNP, self._device.udn)},
"identifiers": {(DOMAIN_UPNP, self._device.udn)},
"name": self._device.name,
"manufacturer": self._device.manufacturer,
"model": self._device.model_name,
}
async def async_update(self):
"""Request an update."""
await self._coordinator.async_request_refresh()
class RawUPnPIGDSensor(UpnpSensor):
async def async_added_to_hass(self) -> None:
"""Subscribe to sensors events."""
remove_from_coordinator = self._coordinator.async_add_listener(
self.async_write_ha_state
)
self.async_on_remove(remove_from_coordinator)
class RawUpnpSensor(UpnpSensor):
"""Representation of a UPnP/IGD sensor."""
def __init__(self, device, sensor_type_name, sensor_type):
"""Initialize the UPnP/IGD sensor."""
super().__init__(device)
self._type_name = sensor_type_name
self._type = sensor_type
self._name = "{} {}".format(device.name, sensor_type["name"])
self._state = None
@property
def name(self) -> str:
"""Return the name of the sensor."""
return self._name
@property
def unique_id(self) -> str:
"""Return an unique ID."""
return f"{self._device.udn}_{self._type_name}"
@property
def state(self) -> str:
"""Return the state of the device."""
if self._state is None:
return None
return format(self._state, "d")
@property
def icon(self) -> str:
"""Icon to use in the frontend, if any."""
return "mdi:server-network"
@property
def unit_of_measurement(self) -> str:
"""Return the unit of measurement of this entity, if any."""
return self._type["unit"]
@Throttle(MIN_TIME_BETWEEN_UPDATES)
async def async_update(self):
"""Get the latest information from the IGD."""
if self._type_name == BYTES_RECEIVED:
self._state = await self._device.async_get_total_bytes_received()
elif self._type_name == BYTES_SENT:
self._state = await self._device.async_get_total_bytes_sent()
elif self._type_name == PACKETS_RECEIVED:
self._state = await self._device.async_get_total_packets_received()
elif self._type_name == PACKETS_SENT:
self._state = await self._device.async_get_total_packets_sent()
device_value_key = self._sensor_type["device_value_key"]
value = self._coordinator.data[device_value_key]
return format(value, "d")
class PerSecondUPnPIGDSensor(UpnpSensor):
"""Abstract representation of a X Sent/Received per second sensor."""
class DerivedUpnpSensor(UpnpSensor):
"""Representation of a UNIT Sent/Received per second sensor."""
def __init__(self, device, direction):
def __init__(self, coordinator, device, sensor_type) -> None:
"""Initialize sensor."""
super().__init__(device)
self._direction = direction
self._state = None
super().__init__(coordinator, device, sensor_type)
self._last_value = None
self._last_update_time = None
@property
def unit(self) -> str:
"""Get unit we are measuring in."""
raise NotImplementedError()
async def _async_fetch_value(self):
"""Fetch a value from the IGD."""
raise NotImplementedError()
@property
def unique_id(self) -> str:
"""Return an unique ID."""
return f"{self._device.udn}_{self.unit}/sec_{self._direction}"
self._last_timestamp = None
@property
def name(self) -> str:
"""Return the name of the sensor."""
return f"{self._device.name} {self.unit}/sec {self._direction}"
return f"{self._device.name} {self._sensor_type['derived_name']}"
@property
def icon(self) -> str:
"""Icon to use in the frontend, if any."""
return "mdi:server-network"
def unique_id(self) -> str:
"""Return an unique ID."""
return f"{self._device.udn}_{self._sensor_type['derived_unique_id']}"
@property
def unit_of_measurement(self) -> str:
"""Return the unit of measurement of this entity, if any."""
return f"{self.unit}/{TIME_SECONDS}"
return self._sensor_type["derived_unit"]
def _is_overflowed(self, new_value) -> bool:
def _has_overflowed(self, current_value) -> bool:
"""Check if value has overflowed."""
return new_value < self._last_value
async def async_update(self):
"""Get the latest information from the UPnP/IGD."""
new_value = await self._async_fetch_value()
if self._last_value is None:
self._last_value = new_value
self._last_update_time = dt_util.utcnow()
return
now = dt_util.utcnow()
if self._is_overflowed(new_value):
self._state = None # temporarily report nothing
else:
delta_time = (now - self._last_update_time).seconds
delta_value = new_value - self._last_value
self._state = delta_value / delta_time
self._last_value = new_value
self._last_update_time = now
class KBytePerSecondUPnPIGDSensor(PerSecondUPnPIGDSensor):
"""Representation of a KBytes Sent/Received per second sensor."""
@property
def unit(self) -> str:
"""Get unit we are measuring in."""
return DATA_KIBIBYTES
async def _async_fetch_value(self) -> float:
"""Fetch value from device."""
if self._direction == IN:
return await self._device.async_get_total_bytes_received()
return await self._device.async_get_total_bytes_sent()
return current_value < self._last_value
@property
def state(self) -> str:
"""Return the state of the device."""
if self._state is None:
# 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_timestamp = self._coordinator.data[TIMESTAMP]
if self._last_value is None or self._has_overflowed(current_value):
self._last_value = current_value
self._last_timestamp = current_timestamp
return None
return format(float(self._state / KIBIBYTE), ".1f")
class PacketsPerSecondUPnPIGDSensor(PerSecondUPnPIGDSensor):
"""Representation of a Packets Sent/Received per second sensor."""
@property
def unit(self) -> str:
"""Get unit we are measuring in."""
return "packets"
async def _async_fetch_value(self) -> float:
"""Fetch value from device."""
if self._direction == IN:
return await self._device.async_get_total_packets_received()
return await self._device.async_get_total_packets_sent()
@property
def state(self) -> str:
"""Return the state of the device."""
if self._state is None:
# Calculate derivative.
delta_value = current_value - self._last_value
if self._sensor_type["unit"] == DATA_BYTES:
delta_value /= KIBIBYTE
delta_time = current_timestamp - self._last_timestamp
if delta_time.seconds == 0:
# Prevent division by 0.
return None
derived = delta_value / delta_time.seconds
return format(float(self._state), ".1f")
# Store current values for future use.
self._last_value = current_value
self._last_timestamp = current_timestamp
return format(derived, ".1f")

8
homeassistant/generated/ssdp.py
View File

@ -76,6 +76,14 @@ SSDP = {
"manufacturer": "Synology"
}
],
"upnp": [
{
"st": "urn:schemas-upnp-org:device:InternetGatewayDevice:1"
},
{
"st": "urn:schemas-upnp-org:device:InternetGatewayDevice:2"
}
],
"wemo": [
{
"manufacturer": "Belkin International Inc."

2
requirements_all.txt
View File

@ -269,7 +269,7 @@ asterisk_mbox==0.5.0
# homeassistant.components.dlna_dmr
# homeassistant.components.upnp
async-upnp-client==0.14.12
async-upnp-client==0.14.13
# homeassistant.components.aten_pe
atenpdu==0.3.0

2
requirements_test_all.txt
View File

@ -122,7 +122,7 @@ arcam-fmj==0.4.3
# homeassistant.components.dlna_dmr
# homeassistant.components.upnp
async-upnp-client==0.14.12
async-upnp-client==0.14.13
# homeassistant.components.stream
av==6.1.2

63
tests/components/upnp/test_init.py
View File

@ -1,14 +1,14 @@
"""Test UPnP/IGD setup process."""
from ipaddress import ip_address
from unittest.mock import MagicMock, patch
from ipaddress import IPv4Address
from unittest.mock import patch
from homeassistant.components import upnp
from homeassistant.components.upnp.device import Device
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, MockDependency, mock_coro
from tests.common import MockConfigEntry, mock_coro
class MockDevice(Device):
@ -16,11 +16,8 @@ class MockDevice(Device):
def __init__(self, udn):
"""Initialize mock device."""
device = MagicMock()
device.manufacturer = "mock-manuf"
device.name = "mock-name"
device.model_name = "mock-model-name"
super().__init__(device)
igd_device = object()
super().__init__(igd_device)
self._udn = udn
self.added_port_mappings = []
self.removed_port_mappings = []
@ -31,16 +28,38 @@ class MockDevice(Device):
return cls("UDN")
@property
def udn(self):
def udn(self) -> str:
"""Get the UDN."""
return self._udn
async def _async_add_port_mapping(self, external_port, local_ip, internal_port):
@property
def manufacturer(self) -> str:
"""Get manufacturer."""
return "mock-manufacturer"
@property
def name(self) -> str:
"""Get name."""
return "mock-name"
@property
def model_name(self) -> str:
"""Get the model name."""
return "mock-model-name"
@property
def device_type(self) -> str:
"""Get the device type."""
return "urn:schemas-upnp-org:device:InternetGatewayDevice:1"
async def _async_add_port_mapping(
self, external_port: int, local_ip: str, internal_port: int
) -> None:
"""Add a port mapping."""
entry = [external_port, local_ip, internal_port]
self.added_port_mappings.append(entry)
async def _async_delete_port_mapping(self, external_port):
async def _async_delete_port_mapping(self, external_port: int) -> None:
"""Remove a port mapping."""
entry = external_port
self.removed_port_mappings.append(entry)
@ -52,18 +71,11 @@ async def test_async_setup_entry_default(hass):
entry = MockConfigEntry(domain=upnp.DOMAIN)
config = {
"http": {},
"discovery": {},
# no upnp
}
with MockDependency("netdisco.discovery"), patch(
"homeassistant.components.upnp.get_local_ip", return_value="192.168.1.10"
), patch.object(Device, "async_create_device") as create_device, patch.object(
Device, "async_create_device"
) as create_device, patch.object(
with patch.object(Device, "async_create_device") as create_device, patch.object(
Device, "async_discover", return_value=mock_coro([])
) as async_discover:
await async_setup_component(hass, "http", config)
await async_setup_component(hass, "upnp", config)
await hass.async_block_till_done()
@ -97,12 +109,13 @@ async def test_async_setup_entry_port_mapping(hass):
config = {
"http": {},
"discovery": {},
"upnp": {"port_mapping": True, "ports": {"hass": "hass"}},
"upnp": {
"local_ip": "192.168.1.10",
"port_mapping": True,
"ports": {"hass": "hass"},
},
}
with MockDependency("netdisco.discovery"), patch(
"homeassistant.components.upnp.get_local_ip", return_value="192.168.1.10"
), patch.object(Device, "async_create_device") as create_device, patch.object(
with patch.object(Device, "async_create_device") as create_device, patch.object(
Device, "async_discover", return_value=mock_coro([])
) as async_discover:
await async_setup_component(hass, "http", config)
@ -124,7 +137,7 @@ async def test_async_setup_entry_port_mapping(hass):
# ensure add-port-mapping-methods called
assert mock_device.added_port_mappings == [
[8123, ip_address("192.168.1.10"), 8123]
[8123, IPv4Address("192.168.1.10"), 8123]
]
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)