diff --git a/homeassistant/components/upnp/__init__.py b/homeassistant/components/upnp/__init__.py index 36b6278d968..da0585df987 100644 --- a/homeassistant/components/upnp/__init__.py +++ b/homeassistant/components/upnp/__init__.py @@ -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: diff --git a/homeassistant/components/upnp/config_flow.py b/homeassistant/components/upnp/config_flow.py index fd4ed9d4051..74acb88983b 100644 --- a/homeassistant/components/upnp/config_flow.py +++ b/homeassistant/components/upnp/config_flow.py @@ -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)