Enable disallow-any-generics in media-player (#78498)
Co-authored-by: Marc Mueller <30130371+cdce8p@users.noreply.github.com>pull/78661/head
parent
b51fd7f688
commit
391d895426
|
@ -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")
|
||||
|
|
Loading…
Reference in New Issue