Add type hints to media_player (part 1) (#64005)

* Add type hints to media_player (part 1)

* Fix roku to match
This commit is contained in:
epenet 2022-01-13 10:36:44 +01:00 committed by GitHub
parent c021e58ee2
commit 5cd73170de
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 18 additions and 13 deletions

View File

@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio import asyncio
import base64 import base64
import collections import collections
from collections.abc import Callable
from contextlib import suppress from contextlib import suppress
from dataclasses import dataclass from dataclasses import dataclass
import datetime as dt import datetime as dt
@ -12,7 +13,7 @@ import hashlib
from http import HTTPStatus from http import HTTPStatus
import logging import logging
import secrets import secrets
from typing import final from typing import Any, cast, final
from urllib.parse import urlparse from urllib.parse import urlparse
from aiohttp import web from aiohttp import web
@ -63,6 +64,7 @@ from homeassistant.helpers.config_validation import ( # noqa: F401
from homeassistant.helpers.entity import Entity, EntityDescription from homeassistant.helpers.entity import Entity, EntityDescription
from homeassistant.helpers.entity_component import EntityComponent from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.network import get_url from homeassistant.helpers.network import get_url
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import bind_hass from homeassistant.loader import bind_hass
from .const import ( from .const import (
@ -208,7 +210,7 @@ def is_on(hass, entity_id=None):
) )
def _rename_keys(**keys): def _rename_keys(**keys: Any) -> Callable[[dict[str, Any]], dict[str, Any]]:
"""Create validator that renames keys. """Create validator that renames keys.
Necessary because the service schema names do not match the command parameters. Necessary because the service schema names do not match the command parameters.
@ -216,7 +218,7 @@ def _rename_keys(**keys):
Async friendly. Async friendly.
""" """
def rename(value): def rename(value: dict[str, Any]) -> dict[str, Any]:
for to_key, from_key in keys.items(): for to_key, from_key in keys.items():
if from_key in value: if from_key in value:
value[to_key] = value.pop(from_key) value[to_key] = value.pop(from_key)
@ -225,7 +227,7 @@ def _rename_keys(**keys):
return rename return rename
async def async_setup(hass, config): async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Track states and offer events for media_players.""" """Track states and offer events for media_players."""
component = hass.data[DOMAIN] = EntityComponent( component = hass.data[DOMAIN] = EntityComponent(
logging.getLogger(__name__), DOMAIN, hass, SCAN_INTERVAL logging.getLogger(__name__), DOMAIN, hass, SCAN_INTERVAL
@ -508,7 +510,7 @@ class MediaPlayerEntity(Entity):
return None return None
async def async_get_media_image(self): async def async_get_media_image(self) -> tuple[bytes | None, str | None]:
"""Fetch media image of current playing image.""" """Fetch media image of current playing image."""
if (url := self.media_image_url) is None: if (url := self.media_image_url) is None:
return None, None return None, None
@ -520,7 +522,7 @@ class MediaPlayerEntity(Entity):
media_content_type: str, media_content_type: str,
media_content_id: str, media_content_id: str,
media_image_id: str | None = None, media_image_id: str | None = None,
) -> tuple[str | None, str | None]: ) -> tuple[bytes | None, str | None]:
""" """
Optionally fetch internally accessible image for media browser. Optionally fetch internally accessible image for media browser.
@ -965,13 +967,15 @@ class MediaPlayerEntity(Entity):
"""Remove this player from any group.""" """Remove this player from any group."""
await self.hass.async_add_executor_job(self.unjoin_player) await self.hass.async_add_executor_job(self.unjoin_player)
async def _async_fetch_image_from_cache(self, url): async def _async_fetch_image_from_cache(
self, url: str
) -> tuple[bytes | None, str | None]:
"""Fetch image. """Fetch image.
Images are cached in memory (the images are typically 10-100kB in size). Images are cached in memory (the images are typically 10-100kB in size).
""" """
cache_images = ENTITY_IMAGE_CACHE[CACHE_IMAGES] cache_images = cast(collections.OrderedDict, ENTITY_IMAGE_CACHE[CACHE_IMAGES])
cache_maxsize = ENTITY_IMAGE_CACHE[CACHE_MAXSIZE] cache_maxsize = cast(int, ENTITY_IMAGE_CACHE[CACHE_MAXSIZE])
if urlparse(url).hostname is None: if urlparse(url).hostname is None:
url = f"{get_url(self.hass)}{url}" url = f"{get_url(self.hass)}{url}"
@ -981,7 +985,7 @@ class MediaPlayerEntity(Entity):
async with cache_images[url][CACHE_LOCK]: async with cache_images[url][CACHE_LOCK]:
if CACHE_CONTENT in cache_images[url]: if CACHE_CONTENT in cache_images[url]:
return cache_images[url][CACHE_CONTENT] return cache_images[url][CACHE_CONTENT] # type:ignore[no-any-return]
(content, content_type) = await self._async_fetch_image(url) (content, content_type) = await self._async_fetch_image(url)
@ -992,7 +996,7 @@ class MediaPlayerEntity(Entity):
return content, content_type return content, content_type
async def _async_fetch_image(self, url): async def _async_fetch_image(self, url: str) -> tuple[bytes | None, str | None]:
"""Retrieve an image.""" """Retrieve an image."""
content, content_type = (None, None) content, content_type = (None, None)
websession = async_get_clientsession(self.hass) websession = async_get_clientsession(self.hass)
@ -1037,7 +1041,7 @@ class MediaPlayerImageView(HomeAssistantView):
url + "/browse_media/{media_content_type}/{media_content_id}", url + "/browse_media/{media_content_type}/{media_content_id}",
] ]
def __init__(self, component): def __init__(self, component: EntityComponent) -> None:
"""Initialize a media player view.""" """Initialize a media player view."""
self.component = component self.component = component
@ -1057,6 +1061,7 @@ class MediaPlayerImageView(HomeAssistantView):
) )
return web.Response(status=status) return web.Response(status=status)
assert isinstance(player, MediaPlayerEntity)
authenticated = ( authenticated = (
request[KEY_AUTHENTICATED] request[KEY_AUTHENTICATED]
or request.query.get("token") == player.access_token or request.query.get("token") == player.access_token

View File

@ -257,7 +257,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
media_content_type: str, media_content_type: str,
media_content_id: str, media_content_id: str,
media_image_id: str | None = None, media_image_id: str | None = None,
) -> tuple[str | None, str | None]: ) -> tuple[bytes | None, str | None]:
"""Fetch media browser image to serve via proxy.""" """Fetch media browser image to serve via proxy."""
if media_content_type == MEDIA_TYPE_APP and media_content_id: if media_content_type == MEDIA_TYPE_APP and media_content_id:
image_url = self.coordinator.roku.app_icon_url(media_content_id) image_url = self.coordinator.roku.app_icon_url(media_content_id)