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

Use dataclass properties in upnp (#60893)

Browse source 
Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
epenet 2021-12-03 08:32:42 +01:00 • committed by GitHub
parent e1b4e40ac6
commit 4207d5a85f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 21 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
homeassistant/components/upnp/__init__.py
View file

@ -15,7 +15,6 @@ from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.components.binary_sensor import BinarySensorEntityDescription
from homeassistant.components.sensor import SensorEntityDescription
from homeassistant.components.ssdp import SsdpChange
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
@ -91,16 +90,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# Register device discovered-callback.
device_discovered_event = asyncio.Event()
discovery_info: Mapping[str, Any] | None = None
discovery_info: ssdp.SsdpServiceInfo | None = None
async def device_discovered(headers: Mapping[str, Any], change: SsdpChange) -> None:
if change == SsdpChange.BYEBYE:
async def device_discovered(
headers: ssdp.SsdpServiceInfo, change: ssdp.SsdpChange
) -> None:
if change == ssdp.SsdpChange.BYEBYE:
return
nonlocal discovery_info
LOGGER.debug(
"Device discovered: %s, at: %s", usn, headers[ssdp.ATTR_SSDP_LOCATION]
)
LOGGER.debug("Device discovered: %s, at: %s", usn, headers.ssdp_location)
discovery_info = headers
device_discovered_event.set()
@ -121,9 +120,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
cancel_discovered_callback()
# Create device.
location = discovery_info[ # pylint: disable=unsubscriptable-object
ssdp.ATTR_SSDP_LOCATION
]
location = discovery_info.ssdp_location
try:
device = await Device.async_create_device(hass, location)
except UpnpConnectionError as err:

32
homeassistant/components/upnp/config_flow.py
View file

@ -10,7 +10,7 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.components.ssdp import SsdpChange
from homeassistant.components.ssdp import SsdpChange, SsdpServiceInfo
from homeassistant.const import CONF_SCAN_INTERVAL
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
@ -52,14 +52,14 @@ async def _async_wait_for_discoveries(hass: HomeAssistant) -> bool:
"""Wait for a device to be discovered."""
device_discovered_event = asyncio.Event()
async def device_discovered(info: Mapping[str, Any], change: SsdpChange) -> None:
async def device_discovered(info: SsdpServiceInfo, change: SsdpChange) -> None:
if change == SsdpChange.BYEBYE:
return
LOGGER.info(
"Device discovered: %s, at: %s",
info[ssdp.ATTR_SSDP_USN],
info[ssdp.ATTR_SSDP_LOCATION],
info.ssdp_usn,
info.ssdp_location,
)
device_discovered_event.set()
@ -112,7 +112,7 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize the UPnP/IGD config flow."""
self._discoveries: Mapping = None
self._discoveries: list[SsdpServiceInfo] | None = None
async def async_step_user(
self, user_input: Mapping | None = None
@ -125,15 +125,13 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
matching_discoveries = [
discovery
for discovery in self._discoveries
if discovery[ssdp.ATTR_SSDP_USN] == user_input["unique_id"]
if discovery.ssdp_usn == user_input["unique_id"]
]
if not matching_discoveries:
return self.async_abort(reason="no_devices_found")
discovery = matching_discoveries[0]
await self.async_set_unique_id(
discovery[ssdp.ATTR_SSDP_USN], raise_on_progress=False
)
await self.async_set_unique_id(discovery.ssdp_usn, raise_on_progress=False)
return await self._async_create_entry_from_discovery(discovery)
# Discover devices.
@ -148,7 +146,7 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
for discovery in discoveries
if (
_is_complete_discovery(discovery)
and discovery[ssdp.ATTR_SSDP_USN] not in current_unique_ids
and discovery.ssdp_usn not in current_unique_ids
)
]
@ -160,9 +158,7 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
{
vol.Required("unique_id"): vol.In(
{
discovery[ssdp.ATTR_SSDP_USN]: _friendly_name_from_discovery(
discovery
)
discovery.ssdp_usn: _friendly_name_from_discovery(discovery)
for discovery in self._discoveries
}
),
@ -204,7 +200,7 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_abort(reason="incomplete_discovery")
# Ensure not already configuring/configured.
unique_id = discovery[ssdp.ATTR_SSDP_USN]
unique_id = discovery.ssdp_usn
await self.async_set_unique_id(unique_id)
return await self._async_create_entry_from_discovery(discovery)
@ -269,7 +265,7 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_create_entry_from_discovery(
self,
discovery: Mapping,
discovery: SsdpServiceInfo,
) -> Mapping[str, Any]:
"""Create an entry from discovery."""
LOGGER.debug(
@ -279,9 +275,9 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
title = _friendly_name_from_discovery(discovery)
data = {
CONFIG_ENTRY_UDN: discovery[ssdp.ATTR_UPNP_UDN],
CONFIG_ENTRY_ST: discovery[ssdp.ATTR_SSDP_ST],
CONFIG_ENTRY_HOSTNAME: discovery["_host"],
CONFIG_ENTRY_UDN: discovery.upnp[ssdp.ATTR_UPNP_UDN],
CONFIG_ENTRY_ST: discovery.ssdp_st,
CONFIG_ENTRY_HOSTNAME: discovery.ssdp_headers["_host"],
}
return self.async_create_entry(title=title, data=data)