Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Use dataclass properties in upnp #60893

Merged
merged 1 commit into from Dec 3, 2021
Merged

Use dataclass properties in upnp #60893

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
17 changes: 7 additions & 10 deletions homeassistant/components/upnp/__init__.py
View file
Open in desktop
Expand Up @@ -15,7 +15,6 @@
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
Expand Down Expand Up @@ -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()

Expand All @@ -121,9 +120,7 @@ async def device_discovered(headers: Mapping[str, Any], change: SsdpChange) -> N
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:
Expand Down
32 changes: 14 additions & 18 deletions homeassistant/components/upnp/config_flow.py
View file
Open in desktop
Expand Up @@ -10,7 +10,7 @@

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
Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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
Expand All @@ -125,15 +125,13 @@ async def async_step_user(
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.
Expand All @@ -148,7 +146,7 @@ async def async_step_user(
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
)
]

Expand All @@ -160,9 +158,7 @@ async def async_step_user(
{
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
}
),
Expand Down Expand Up @@ -204,7 +200,7 @@ async def async_step_import(self, import_info: Mapping | None) -> Mapping[str, A
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)
Expand Down Expand Up @@ -269,7 +265,7 @@ def async_get_options_flow(

async def _async_create_entry_from_discovery(
self,
discovery: Mapping,
discovery: SsdpServiceInfo,
) -> Mapping[str, Any]:
"""Create an entry from discovery."""
LOGGER.debug(
Expand All @@ -279,9 +275,9 @@ async def _async_create_entry_from_discovery(

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)

Expand Down