From 9027ee7828c122dd06f1f1ce5a64de5f06321740 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Wed, 24 Nov 2021 10:14:48 +0100 Subject: [PATCH] Revert "Add type hints to SSDP (#59840)" (#60270) --- homeassistant/components/ssdp/__init__.py | 78 ++++------------------- 1 file changed, 13 insertions(+), 65 deletions(-) diff --git a/homeassistant/components/ssdp/__init__.py b/homeassistant/components/ssdp/__init__.py index a84b2e690da..c937f210368 100644 --- a/homeassistant/components/ssdp/__init__.py +++ b/homeassistant/components/ssdp/__init__.py @@ -7,7 +7,7 @@ from datetime import timedelta from enum import Enum from ipaddress import IPv4Address, IPv6Address import logging -from typing import Any, Callable, Final, Mapping, TypedDict, cast +from typing import Any, Callable, Mapping from async_upnp_client.aiohttp import AiohttpSessionRequester from async_upnp_client.const import DeviceOrServiceType, SsdpHeaders, SsdpSource @@ -32,7 +32,7 @@ SCAN_INTERVAL = timedelta(seconds=60) IPV4_BROADCAST = IPv4Address("255.255.255.255") # Attributes for accessing info from SSDP response -ATTR_SSDP_LOCATION: Final = "ssdp_location" +ATTR_SSDP_LOCATION = "ssdp_location" ATTR_SSDP_ST = "ssdp_st" ATTR_SSDP_NT = "ssdp_nt" ATTR_SSDP_UDN = "ssdp_udn" @@ -56,7 +56,7 @@ ATTR_UPNP_UDN = "UDN" ATTR_UPNP_UPC = "UPC" ATTR_UPNP_PRESENTATION_URL = "presentationURL" # Attributes for accessing info added by Home Assistant -ATTR_HA_MATCHING_DOMAINS: Final = "x_homeassistant_matching_domains" +ATTR_HA_MATCHING_DOMAINS = "x-homeassistant-matching-domains" PRIMARY_MATCH_KEYS = [ATTR_UPNP_MANUFACTURER, "st", ATTR_UPNP_DEVICE_TYPE, "nt"] @@ -85,58 +85,6 @@ SSDP_SOURCE_SSDP_CHANGE_MAPPING: Mapping[SsdpSource, SsdpChange] = { _LOGGER = logging.getLogger(__name__) -class _HaServiceInfoDescription(TypedDict, total=True): - """Keys added by HA.""" - - x_homeassistant_matching_domains: set[str] - - -class _SsdpDescriptionBase(TypedDict, total=True): - """Compulsory keys for SSDP info.""" - - ssdp_usn: str - ssdp_st: str - - -class SsdpDescription(_SsdpDescriptionBase, total=False): - """SSDP info with optional keys.""" - - ssdp_location: str - ssdp_nt: str - ssdp_udn: str - ssdp_ext: str - ssdp_server: str - - -class _UpnpDescriptionBase(TypedDict, total=True): - """Compulsory keys for UPnP info.""" - - deviceType: str - friendlyName: str - manufacturer: str - modelName: str - UDN: str - - -class UpnpDescription(_UpnpDescriptionBase, total=False): - """UPnP info with optional keys.""" - - manufacturerURL: str - modelDescription: str - modelNumber: str - modelURL: str - serialNumber: str - UPC: str - iconList: dict[str, list[dict[str, str]]] - serviceList: dict[str, list[dict[str, str]]] - deviceList: dict[str, Any] - presentationURL: str - - -class SsdpServiceInfo(SsdpDescription, UpnpDescription, _HaServiceInfoDescription): - """Prepared info from ssdp/upnp entries.""" - - @bind_hass async def async_register_callback( hass: HomeAssistant, @@ -154,7 +102,7 @@ async def async_register_callback( @bind_hass async def async_get_discovery_info_by_udn_st( # pylint: disable=invalid-name hass: HomeAssistant, udn: str, st: str -) -> SsdpServiceInfo | None: +) -> dict[str, str] | None: """Fetch the discovery info cache.""" scanner: Scanner = hass.data[DOMAIN] return await scanner.async_get_discovery_info_by_udn_st(udn, st) @@ -163,7 +111,7 @@ async def async_get_discovery_info_by_udn_st( # pylint: disable=invalid-name @bind_hass async def async_get_discovery_info_by_st( # pylint: disable=invalid-name hass: HomeAssistant, st: str -) -> list[SsdpServiceInfo]: +) -> list[dict[str, str]]: """Fetch all the entries matching the st.""" scanner: Scanner = hass.data[DOMAIN] return await scanner.async_get_discovery_info_by_st(st) @@ -172,7 +120,7 @@ async def async_get_discovery_info_by_st( # pylint: disable=invalid-name @bind_hass async def async_get_discovery_info_by_udn( hass: HomeAssistant, udn: str -) -> list[SsdpServiceInfo]: +) -> list[dict[str, str]]: """Fetch all the entries matching the udn.""" scanner: Scanner = hass.data[DOMAIN] return await scanner.async_get_discovery_info_by_udn(udn) @@ -193,7 +141,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: async def _async_process_callbacks( callbacks: list[SsdpCallback], - discovery_info: SsdpServiceInfo, + discovery_info: dict[str, str], ssdp_change: SsdpChange, ) -> None: for callback in callbacks: @@ -479,7 +427,7 @@ class Scanner: async def _async_headers_to_discovery_info( self, headers: Mapping[str, Any] - ) -> SsdpServiceInfo: + ) -> dict[str, Any]: """Combine the headers and description into discovery_info. Building this is a bit expensive so we only do it on demand. @@ -495,7 +443,7 @@ class Scanner: async def async_get_discovery_info_by_udn_st( # pylint: disable=invalid-name self, udn: str, st: str - ) -> SsdpServiceInfo | None: + ) -> dict[str, Any] | None: """Return discovery_info for a udn and st.""" if headers := self._all_headers_from_ssdp_devices.get((udn, st)): return await self._async_headers_to_discovery_info(headers) @@ -503,7 +451,7 @@ class Scanner: async def async_get_discovery_info_by_st( # pylint: disable=invalid-name self, st: str - ) -> list[SsdpServiceInfo]: + ) -> list[dict[str, Any]]: """Return matching discovery_infos for a st.""" return [ await self._async_headers_to_discovery_info(headers) @@ -511,7 +459,7 @@ class Scanner: if udn_st[1] == st ] - async def async_get_discovery_info_by_udn(self, udn: str) -> list[SsdpServiceInfo]: + async def async_get_discovery_info_by_udn(self, udn: str) -> list[dict[str, Any]]: """Return matching discovery_infos for a udn.""" return [ await self._async_headers_to_discovery_info(headers) @@ -522,7 +470,7 @@ class Scanner: def discovery_info_from_headers_and_description( info_with_desc: CaseInsensitiveDict, -) -> SsdpServiceInfo: +) -> dict[str, Any]: """Convert headers and description to discovery_info.""" info = { DISCOVERY_MAPPING.get(k.lower(), k): v @@ -537,7 +485,7 @@ def discovery_info_from_headers_and_description( if ATTR_SSDP_ST not in info and ATTR_SSDP_NT in info: info[ATTR_SSDP_ST] = info[ATTR_SSDP_NT] - return cast(SsdpServiceInfo, info) + return info def _udn_from_usn(usn: str | None) -> str | None: