From 391d8954260608e7d16e04bbdbd0d08c9a9d5456 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Sat, 17 Sep 2022 21:23:16 +0200 Subject: [PATCH] Enable disallow-any-generics in media-player (#78498) Co-authored-by: Marc Mueller <30130371+cdce8p@users.noreply.github.com> --- .../components/media_player/__init__.py | 50 +++++++++++++------ 1 file changed, 34 insertions(+), 16 deletions(-) diff --git a/homeassistant/components/media_player/__init__.py b/homeassistant/components/media_player/__init__.py index 408f33bbecc..8c89288b41d 100644 --- a/homeassistant/components/media_player/__init__.py +++ b/homeassistant/components/media_player/__init__.py @@ -12,7 +12,7 @@ import hashlib from http import HTTPStatus import logging import secrets -from typing import Any, cast, final +from typing import Any, Final, TypedDict, final from urllib.parse import quote, urlparse from aiohttp import web @@ -136,12 +136,11 @@ _LOGGER = logging.getLogger(__name__) ENTITY_ID_FORMAT = DOMAIN + ".{}" -CACHE_IMAGES = "images" -CACHE_MAXSIZE = "maxsize" -CACHE_LOCK = "lock" -CACHE_URL = "url" -CACHE_CONTENT = "content" -ENTITY_IMAGE_CACHE = {CACHE_IMAGES: collections.OrderedDict(), CACHE_MAXSIZE: 16} +CACHE_IMAGES: Final = "images" +CACHE_MAXSIZE: Final = "maxsize" +CACHE_LOCK: Final = "lock" +CACHE_URL: Final = "url" +CACHE_CONTENT: Final = "content" SCAN_INTERVAL = dt.timedelta(seconds=10) @@ -214,6 +213,25 @@ ATTR_TO_PROPERTY = [ ATTR_MEDIA_REPEAT, ] +# mypy: disallow-any-generics + + +class _CacheImage(TypedDict, total=False): + """Class to hold a cached image.""" + + lock: asyncio.Lock + content: tuple[bytes | None, str | None] + + +class _ImageCache(TypedDict): + """Class to hold a cached image.""" + + images: collections.OrderedDict[str, _CacheImage] + maxsize: int + + +_ENTITY_IMAGE_CACHE = _ImageCache(images=collections.OrderedDict(), maxsize=16) + @bind_hass def is_on(hass: HomeAssistant, entity_id: str | None = None) -> bool: @@ -248,7 +266,7 @@ def _rename_keys(**keys: Any) -> Callable[[dict[str, Any]], dict[str, Any]]: async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Track states and offer events for media_players.""" - component = hass.data[DOMAIN] = EntityComponent( + component = hass.data[DOMAIN] = EntityComponent[MediaPlayerEntity]( logging.getLogger(__name__), DOMAIN, hass, SCAN_INTERVAL ) @@ -420,13 +438,13 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up a config entry.""" - component: EntityComponent = hass.data[DOMAIN] + component: EntityComponent[MediaPlayerEntity] = hass.data[DOMAIN] return await component.async_setup_entry(entry) async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Unload a config entry.""" - component: EntityComponent = hass.data[DOMAIN] + component: EntityComponent[MediaPlayerEntity] = hass.data[DOMAIN] return await component.async_unload_entry(entry) @@ -1060,8 +1078,8 @@ class MediaPlayerEntity(Entity): Images are cached in memory (the images are typically 10-100kB in size). """ - cache_images = cast(collections.OrderedDict, ENTITY_IMAGE_CACHE[CACHE_IMAGES]) - cache_maxsize = cast(int, ENTITY_IMAGE_CACHE[CACHE_MAXSIZE]) + cache_images = _ENTITY_IMAGE_CACHE[CACHE_IMAGES] + cache_maxsize = _ENTITY_IMAGE_CACHE[CACHE_MAXSIZE] if urlparse(url).hostname is None: url = f"{get_url(self.hass)}{url}" @@ -1071,7 +1089,7 @@ class MediaPlayerEntity(Entity): async with cache_images[url][CACHE_LOCK]: if CACHE_CONTENT in cache_images[url]: - return cache_images[url][CACHE_CONTENT] # type:ignore[no-any-return] + return cache_images[url][CACHE_CONTENT] (content, content_type) = await self._async_fetch_image(url) @@ -1120,7 +1138,7 @@ class MediaPlayerImageView(HomeAssistantView): + "/browse_media/{media_content_type}/{media_content_id:.+}", ] - def __init__(self, component: EntityComponent) -> None: + def __init__(self, component: EntityComponent[MediaPlayerEntity]) -> None: """Initialize a media player view.""" self.component = component @@ -1191,8 +1209,8 @@ async def websocket_browse_media( To use, media_player integrations can implement MediaPlayerEntity.async_browse_media() """ - component = hass.data[DOMAIN] - player: MediaPlayerEntity | None = component.get_entity(msg["entity_id"]) + component: EntityComponent[MediaPlayerEntity] = hass.data[DOMAIN] + player = component.get_entity(msg["entity_id"]) if player is None: connection.send_error(msg["id"], "entity_not_found", "Entity not found")