Add typing to media extractor (#99207)

* Add typing to media extractor

* Add typing to media extractor

* Add typing to media extractor

* Add typing to media extractor
This commit is contained in:
Joost Lekkerkerker 2023-08-28 14:43:22 +02:00 committed by GitHub
parent 1d403a961f
commit 60844954d2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 33 additions and 13 deletions

View File

@ -213,6 +213,7 @@ homeassistant.components.luftdaten.*
homeassistant.components.mailbox.*
homeassistant.components.mastodon.*
homeassistant.components.matter.*
homeassistant.components.media_extractor.*
homeassistant.components.media_player.*
homeassistant.components.media_source.*
homeassistant.components.metoffice.*

View File

@ -1,5 +1,7 @@
"""Decorator service for the media_player.play_media service."""
from collections.abc import Callable
import logging
from typing import Any, cast
import voluptuous as vol
from yt_dlp import YoutubeDL
@ -68,21 +70,26 @@ class MEQueryException(Exception):
class MediaExtractor:
"""Class which encapsulates all extraction logic."""
def __init__(self, hass, component_config, call_data):
def __init__(
self,
hass: HomeAssistant,
component_config: dict[str, Any],
call_data: dict[str, Any],
) -> None:
"""Initialize media extractor."""
self.hass = hass
self.config = component_config
self.call_data = call_data
def get_media_url(self):
def get_media_url(self) -> str:
"""Return media content url."""
return self.call_data.get(ATTR_MEDIA_CONTENT_ID)
return cast(str, self.call_data[ATTR_MEDIA_CONTENT_ID])
def get_entities(self):
def get_entities(self) -> list[str]:
"""Return list of entities."""
return self.call_data.get(ATTR_ENTITY_ID, [])
def extract_and_send(self):
def extract_and_send(self) -> None:
"""Extract exact stream format for each entity_id and play it."""
try:
stream_selector = self.get_stream_selector()
@ -97,7 +104,7 @@ class MediaExtractor:
for entity_id in entities:
self.call_media_player_service(stream_selector, entity_id)
def get_stream_selector(self):
def get_stream_selector(self) -> Callable[[str], str]:
"""Return format selector for the media URL."""
ydl = YoutubeDL({"quiet": True, "logger": _LOGGER})
@ -118,7 +125,7 @@ class MediaExtractor:
else:
selected_media = all_media
def stream_selector(query):
def stream_selector(query: str) -> str:
"""Find stream URL that matches query."""
try:
ydl.params["format"] = query
@ -131,12 +138,14 @@ class MediaExtractor:
best_stream = requested_stream["formats"][
len(requested_stream["formats"]) - 1
]
return best_stream["url"]
return requested_stream["url"]
return str(best_stream["url"])
return str(requested_stream["url"])
return stream_selector
def call_media_player_service(self, stream_selector, entity_id):
def call_media_player_service(
self, stream_selector: Callable[[str], str], entity_id: str | None
) -> None:
"""Call Media player play_media service."""
stream_query = self.get_stream_query_for_entity(entity_id)
@ -156,16 +165,16 @@ class MediaExtractor:
self.hass.services.async_call(MEDIA_PLAYER_DOMAIN, SERVICE_PLAY_MEDIA, data)
)
def get_stream_query_for_entity(self, entity_id):
def get_stream_query_for_entity(self, entity_id: str | None) -> str:
"""Get stream format query for entity."""
default_stream_query = self.config.get(
default_stream_query: str = self.config.get(
CONF_DEFAULT_STREAM_QUERY, DEFAULT_STREAM_QUERY
)
if entity_id:
media_content_type = self.call_data.get(ATTR_MEDIA_CONTENT_TYPE)
return (
return str(
self.config.get(CONF_CUSTOMIZE_ENTITIES, {})
.get(entity_id, {})
.get(media_content_type, default_stream_query)

View File

@ -1892,6 +1892,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.media_extractor.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.media_player.*]
check_untyped_defs = true
disallow_incomplete_defs = true