diff --git a/homeassistant/components/sonos/media_player.py b/homeassistant/components/sonos/media_player.py index 4f265bc6f56..6d594b906ea 100644 --- a/homeassistant/components/sonos/media_player.py +++ b/homeassistant/components/sonos/media_player.py @@ -1,10 +1,13 @@ """Support to interface with Sonos players.""" +from __future__ import annotations + import asyncio from contextlib import suppress import datetime import functools as ft import logging import socket +from typing import Any, Callable, Coroutine import urllib.parse import async_timeout @@ -16,7 +19,10 @@ from pysonos.core import ( MUSIC_SRC_TV, PLAY_MODE_BY_MEANING, PLAY_MODES, + SoCo, ) +from pysonos.data_structures import DidlFavorite +from pysonos.events_base import Event, SubscriptionBase from pysonos.exceptions import SoCoException, SoCoUPnPException import pysonos.music_library import pysonos.snapshot @@ -51,20 +57,22 @@ from homeassistant.components.media_player.const import ( from homeassistant.components.media_player.errors import BrowseError from homeassistant.components.plex.const import PLEX_URI_SCHEME from homeassistant.components.plex.services import play_on_sonos +from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( ATTR_TIME, + CONF_HOSTS, EVENT_HOMEASSISTANT_STOP, STATE_IDLE, STATE_PAUSED, STATE_PLAYING, ) -from homeassistant.core import ServiceCall, callback +from homeassistant.core import HomeAssistant, ServiceCall, callback from homeassistant.helpers import config_validation as cv, entity_platform, service import homeassistant.helpers.device_registry as dr from homeassistant.helpers.network import is_internal_request from homeassistant.util.dt import utcnow -from . import CONF_ADVERTISE_ADDR, CONF_HOSTS, CONF_INTERFACE_ADDR +from . import CONF_ADVERTISE_ADDR, CONF_INTERFACE_ADDR from .const import ( DATA_SONOS, DOMAIN as SONOS_DOMAIN, @@ -77,6 +85,7 @@ _LOGGER = logging.getLogger(__name__) SCAN_INTERVAL = 10 DISCOVERY_INTERVAL = 60 +SEEN_EXPIRE_TIME = 3.5 * DISCOVERY_INTERVAL SUPPORT_SONOS = ( SUPPORT_BROWSE_MEDIA @@ -139,23 +148,18 @@ UNAVAILABLE_VALUES = {"", "NOT_IMPLEMENTED", None} class SonosData: """Storage class for platform global data.""" - def __init__(self): + def __init__(self) -> None: """Initialize the data.""" - self.entities = [] - self.discovered = [] + self.entities: list[SonosEntity] = [] + self.discovered: list[str] = [] self.topology_condition = asyncio.Condition() self.discovery_thread = None self.hosts_heartbeat = None -async def async_setup_platform(hass, config, async_add_entities, discovery_info=None): - """Set up the Sonos platform. Obsolete.""" - _LOGGER.error( - "Loading Sonos by media_player platform configuration is no longer supported" - ) - - -async def async_setup_entry(hass, config_entry, async_add_entities): +async def async_setup_entry( + hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: Callable +) -> None: """Set up Sonos from a config entry.""" if DATA_SONOS not in hass.data: hass.data[DATA_SONOS] = SonosData() @@ -168,7 +172,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities): if advertise_addr: pysonos.config.EVENT_ADVERTISE_IP = advertise_addr - def _stop_discovery(event): + def _stop_discovery(event: Event) -> None: data = hass.data[DATA_SONOS] if data.discovery_thread: data.discovery_thread.stop() @@ -177,11 +181,11 @@ async def async_setup_entry(hass, config_entry, async_add_entities): data.hosts_heartbeat() data.hosts_heartbeat = None - def _discovery(now=None): + def _discovery(now: datetime.datetime | None = None) -> None: """Discover players from network or configuration.""" hosts = config.get(CONF_HOSTS) - def _discovered_player(soco): + def _discovered_player(soco: SoCo) -> None: """Handle a (re)discovered player.""" try: _LOGGER.debug("Reached _discovered_player, soco=%s", soco) @@ -194,7 +198,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities): entity = _get_entity_from_soco_uid(hass, soco.uid) if entity and (entity.soco == soco or not entity.available): _LOGGER.debug("Seen %s", entity) - hass.add_job(entity.async_seen(soco)) + hass.add_job(entity.async_seen(soco)) # type: ignore except SoCoException as ex: _LOGGER.debug("SoCoException, ex=%s", ex) @@ -234,31 +238,35 @@ async def async_setup_entry(hass, config_entry, async_add_entities): platform = entity_platform.current_platform.get() @service.verify_domain_control(hass, SONOS_DOMAIN) - async def async_service_handle(service_call: ServiceCall): + async def async_service_handle(service_call: ServiceCall) -> None: """Handle dispatched services.""" + assert platform is not None entities = await platform.async_extract_from_service(service_call) if not entities: return + for entity in entities: + assert isinstance(entity, SonosEntity) + if service_call.service == SERVICE_JOIN: master = platform.entities.get(service_call.data[ATTR_MASTER]) if master: - await SonosEntity.join_multi(hass, master, entities) + await SonosEntity.join_multi(hass, master, entities) # type: ignore[arg-type] else: _LOGGER.error( "Invalid master specified for join service: %s", service_call.data[ATTR_MASTER], ) elif service_call.service == SERVICE_UNJOIN: - await SonosEntity.unjoin_multi(hass, entities) + await SonosEntity.unjoin_multi(hass, entities) # type: ignore[arg-type] elif service_call.service == SERVICE_SNAPSHOT: await SonosEntity.snapshot_multi( - hass, entities, service_call.data[ATTR_WITH_GROUP] + hass, entities, service_call.data[ATTR_WITH_GROUP] # type: ignore[arg-type] ) elif service_call.service == SERVICE_RESTORE: await SonosEntity.restore_multi( - hass, entities, service_call.data[ATTR_WITH_GROUP] + hass, entities, service_call.data[ATTR_WITH_GROUP] # type: ignore[arg-type] ) hass.services.async_register( @@ -287,7 +295,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities): SONOS_DOMAIN, SERVICE_RESTORE, async_service_handle, join_unjoin_schema ) - platform.async_register_entity_service( + platform.async_register_entity_service( # type: ignore SERVICE_SET_TIMER, { vol.Required(ATTR_SLEEP_TIME): vol.All( @@ -297,9 +305,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities): "set_sleep_timer", ) - platform.async_register_entity_service(SERVICE_CLEAR_TIMER, {}, "clear_sleep_timer") + platform.async_register_entity_service(SERVICE_CLEAR_TIMER, {}, "clear_sleep_timer") # type: ignore - platform.async_register_entity_service( + platform.async_register_entity_service( # type: ignore SERVICE_UPDATE_ALARM, { vol.Required(ATTR_ALARM_ID): cv.positive_int, @@ -311,7 +319,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities): "set_alarm", ) - platform.async_register_entity_service( + platform.async_register_entity_service( # type: ignore SERVICE_SET_OPTION, { vol.Optional(ATTR_NIGHT_SOUND): cv.boolean, @@ -321,50 +329,36 @@ async def async_setup_entry(hass, config_entry, async_add_entities): "set_option", ) - platform.async_register_entity_service( + platform.async_register_entity_service( # type: ignore SERVICE_PLAY_QUEUE, {vol.Optional(ATTR_QUEUE_POSITION): cv.positive_int}, "play_queue", ) - platform.async_register_entity_service( + platform.async_register_entity_service( # type: ignore SERVICE_REMOVE_FROM_QUEUE, {vol.Optional(ATTR_QUEUE_POSITION): cv.positive_int}, "remove_from_queue", ) -class _ProcessSonosEventQueue: - """Queue like object for dispatching sonos events.""" - - def __init__(self, handler): - """Initialize Sonos event queue.""" - self._handler = handler - - def put(self, item, block=True, timeout=None): - """Process event.""" - try: - self._handler(item) - except SoCoException as ex: - _LOGGER.warning("Error calling %s: %s", self._handler, ex) - - -def _get_entity_from_soco_uid(hass, uid): +def _get_entity_from_soco_uid(hass: HomeAssistant, uid: str) -> SonosEntity | None: """Return SonosEntity from SoCo uid.""" - for entity in hass.data[DATA_SONOS].entities: + entities: list[SonosEntity] = hass.data[DATA_SONOS].entities + for entity in entities: if uid == entity.unique_id: return entity return None -def soco_error(errorcodes=None): +def soco_error(errorcodes: list[str] | None = None) -> Callable: """Filter out specified UPnP errors from logs and avoid exceptions.""" - def decorator(funct): + def decorator(funct: Callable) -> Callable: """Decorate functions.""" @ft.wraps(funct) - def wrapper(*args, **kwargs): + def wrapper(*args: Any, **kwargs: Any) -> Any: """Wrap for all soco UPnP exception.""" try: return funct(*args, **kwargs) @@ -379,11 +373,11 @@ def soco_error(errorcodes=None): return decorator -def soco_coordinator(funct): +def soco_coordinator(funct: Callable) -> Callable: """Call function on coordinator.""" @ft.wraps(funct) - def wrapper(entity, *args, **kwargs): + def wrapper(entity: SonosEntity, *args: Any, **kwargs: Any) -> Any: """Wrap for call to coordinator.""" if entity.is_coordinator: return funct(entity, *args, **kwargs) @@ -392,81 +386,82 @@ def soco_coordinator(funct): return wrapper -def _timespan_secs(timespan): +def _timespan_secs(timespan: str | None) -> None | float: """Parse a time-span into number of seconds.""" if timespan in UNAVAILABLE_VALUES: return None + assert timespan is not None return sum(60 ** x[0] * int(x[1]) for x in enumerate(reversed(timespan.split(":")))) class SonosEntity(MediaPlayerEntity): """Representation of a Sonos entity.""" - def __init__(self, player): + def __init__(self, player: SoCo) -> None: """Initialize the Sonos entity.""" - self._subscriptions = [] - self._poll_timer = None - self._seen_timer = None + self._subscriptions: list[SubscriptionBase] = [] + self._poll_timer: Callable | None = None + self._seen_timer: Callable | None = None self._volume_increment = 2 - self._unique_id = player.uid - self._player = player - self._player_volume = None - self._player_muted = None - self._play_mode = None - self._coordinator = None - self._sonos_group = [self] - self._status = None - self._uri = None + self._unique_id: str = player.uid + self._player: SoCo = player + self._player_volume: int | None = None + self._player_muted: bool | None = None + self._play_mode: str | None = None + self._coordinator: SonosEntity | None = None + self._sonos_group: list[SonosEntity] = [self] + self._status: str | None = None + self._uri: str | None = None self._media_library = pysonos.music_library.MusicLibrary(self.soco) - self._media_duration = None - self._media_position = None - self._media_position_updated_at = None - self._media_image_url = None - self._media_channel = None - self._media_artist = None - self._media_album_name = None - self._media_title = None - self._queue_position = None - self._night_sound = None - self._speech_enhance = None - self._source_name = None - self._favorites = [] - self._soco_snapshot = None - self._snapshot_group = None + self._media_duration: float | None = None + self._media_position: float | None = None + self._media_position_updated_at: datetime.datetime | None = None + self._media_image_url: str | None = None + self._media_channel: str | None = None + self._media_artist: str | None = None + self._media_album_name: str | None = None + self._media_title: str | None = None + self._queue_position: int | None = None + self._night_sound: bool | None = None + self._speech_enhance: bool | None = None + self._source_name: str | None = None + self._favorites: list[DidlFavorite] = [] + self._soco_snapshot: pysonos.snapshot.Snapshot | None = None + self._snapshot_group: list[SonosEntity] | None = None # Set these early since device_info() needs them - speaker_info = self.soco.get_speaker_info(True) - self._name = speaker_info["zone_name"] - self._model = speaker_info["model_name"] - self._sw_version = speaker_info["software_version"] - self._mac_address = speaker_info["mac_address"] + speaker_info: dict = self.soco.get_speaker_info(True) + self._name: str = speaker_info["zone_name"] + self._model: str = speaker_info["model_name"] + self._sw_version: str = speaker_info["software_version"] + self._mac_address: str = speaker_info["mac_address"] - async def async_added_to_hass(self): + async def async_added_to_hass(self) -> None: """Subscribe sonos events.""" await self.async_seen(self.soco) self.hass.data[DATA_SONOS].entities.append(self) for entity in self.hass.data[DATA_SONOS].entities: - await entity.async_update_groups_coro() + await entity.create_update_groups_coro() @property - def unique_id(self): + def unique_id(self) -> str: """Return a unique ID.""" return self._unique_id - def __hash__(self): + def __hash__(self) -> int: """Return a hash of self.""" return hash(self.unique_id) @property - def name(self): + def name(self) -> str: """Return the name of the entity.""" return self._name @property - def device_info(self): + def device_info(self) -> dict: """Return information about the device.""" return { "identifiers": {(SONOS_DOMAIN, self._unique_id)}, @@ -478,9 +473,9 @@ class SonosEntity(MediaPlayerEntity): "suggested_area": self._name, } - @property + @property # type: ignore[misc] @soco_coordinator - def state(self): + def state(self) -> str: """Return the state of the entity.""" if self._status in ( "PAUSED_PLAYBACK", @@ -496,21 +491,21 @@ class SonosEntity(MediaPlayerEntity): return STATE_IDLE @property - def is_coordinator(self): + def is_coordinator(self) -> bool: """Return true if player is a coordinator.""" return self._coordinator is None @property - def soco(self): + def soco(self) -> SoCo: """Return soco object.""" return self._player @property - def coordinator(self): + def coordinator(self) -> SoCo: """Return coordinator of this player.""" return self._coordinator - async def async_seen(self, player): + async def async_seen(self, player: SoCo) -> None: """Record that this player was seen right now.""" was_available = self.available _LOGGER.debug("Async seen: %s, was_available: %s", player, was_available) @@ -521,7 +516,7 @@ class SonosEntity(MediaPlayerEntity): self._seen_timer() self._seen_timer = self.hass.helpers.event.async_call_later( - 2.5 * DISCOVERY_INTERVAL, self.async_unseen + SEEN_EXPIRE_TIME, self.async_unseen ) if was_available: @@ -533,12 +528,13 @@ class SonosEntity(MediaPlayerEntity): done = await self._async_attach_player() if not done: + assert self._seen_timer is not None self._seen_timer() await self.async_unseen() self.async_write_ha_state() - async def async_unseen(self, now=None): + async def async_unseen(self, now: datetime.datetime | None = None) -> None: """Make this player unavailable when it was not seen recently.""" self._seen_timer = None @@ -558,12 +554,12 @@ class SonosEntity(MediaPlayerEntity): """Return True if entity is available.""" return self._seen_timer is not None - def _clear_media_position(self): + def _clear_media_position(self) -> None: """Clear the media_position.""" self._media_position = None self._media_position_updated_at = None - def _set_favorites(self): + def _set_favorites(self) -> None: """Set available favorites.""" self._favorites = [] for fav in self.soco.music_library.get_sonos_favorites(): @@ -575,13 +571,13 @@ class SonosEntity(MediaPlayerEntity): # Skip unknown types _LOGGER.error("Unhandled favorite '%s': %s", fav.title, ex) - def _attach_player(self): + def _attach_player(self) -> None: """Get basic information and add event subscriptions.""" self._play_mode = self.soco.play_mode self.update_volume() self._set_favorites() - async def _async_attach_player(self): + async def _async_attach_player(self) -> bool: """Get basic information and add event subscriptions.""" try: await self.hass.async_add_executor_job(self._attach_player) @@ -603,18 +599,20 @@ class SonosEntity(MediaPlayerEntity): _LOGGER.warning("Could not connect %s: %s", self.entity_id, ex) return False - async def _subscribe(self, target, sub_callback): + async def _subscribe( + self, target: SubscriptionBase, sub_callback: Callable + ) -> None: """Create a sonos subscription.""" subscription = await target.subscribe(auto_renew=True) subscription.callback = sub_callback self._subscriptions.append(subscription) @property - def should_poll(self): + def should_poll(self) -> bool: """Return that we should not be polled (we handle that internally).""" return False - def update(self, now=None): + def update(self, now: datetime.datetime | None = None) -> None: """Retrieve latest state.""" try: self.update_groups() @@ -625,11 +623,11 @@ class SonosEntity(MediaPlayerEntity): pass @callback - def async_update_media(self, event=None): + def async_update_media(self, event: Event | None = None) -> None: """Update information about currently playing media.""" - self.hass.async_add_job(self.update_media, event) + self.hass.async_add_executor_job(self.update_media, event) - def update_media(self, event=None): + def update_media(self, event: Event | None = None) -> None: """Update information about currently playing media.""" variables = event and event.variables @@ -679,7 +677,7 @@ class SonosEntity(MediaPlayerEntity): self._media_title = track_info.get("title") if music_source == MUSIC_SRC_RADIO: - self.update_media_radio(variables, track_info) + self.update_media_radio(variables) else: self.update_media_music(update_position, track_info) @@ -691,14 +689,14 @@ class SonosEntity(MediaPlayerEntity): if coordinator and coordinator.unique_id == self.unique_id: entity.schedule_update_ha_state() - def update_media_linein(self, source): + def update_media_linein(self, source: str) -> None: """Update state when playing from line-in/tv.""" self._clear_media_position() self._media_title = source self._source_name = source - def update_media_radio(self, variables, track_info): + def update_media_radio(self, variables: dict) -> None: """Update state when streaming radio.""" self._clear_media_position() @@ -720,7 +718,8 @@ class SonosEntity(MediaPlayerEntity): ) and ( self.state != STATE_PLAYING or self.soco.music_source_from_uri(self._media_title) == MUSIC_SRC_RADIO - or self._media_title in self._uri + and self._uri is not None + and self._media_title in self._uri # type: ignore[operator] ): self._media_title = uri_meta_data.title except (TypeError, KeyError, AttributeError): @@ -735,7 +734,7 @@ class SonosEntity(MediaPlayerEntity): if fav.reference.get_uri() == media_info["uri"]: self._source_name = fav.title - def update_media_music(self, update_media_position, track_info): + def update_media_music(self, update_media_position: bool, track_info: dict) -> None: """Update state when playing music tracks.""" self._media_duration = _timespan_secs(track_info.get("duration")) current_position = _timespan_secs(track_info.get("position")) @@ -747,8 +746,9 @@ class SonosEntity(MediaPlayerEntity): # position jumped? if current_position is not None and self._media_position is not None: if self.state == STATE_PLAYING: - time_diff = utcnow() - self._media_position_updated_at - time_diff = time_diff.total_seconds() + assert self._media_position_updated_at is not None + time_delta = utcnow() - self._media_position_updated_at + time_diff = time_delta.total_seconds() else: time_diff = 0 @@ -765,12 +765,12 @@ class SonosEntity(MediaPlayerEntity): self._media_image_url = track_info.get("album_art") - playlist_position = int(track_info.get("playlist_position")) + playlist_position = int(track_info.get("playlist_position")) # type: ignore if playlist_position > 0: self._queue_position = playlist_position - 1 @callback - def async_update_volume(self, event): + def async_update_volume(self, event: Event) -> None: """Update information about currently volume settings.""" variables = event.variables @@ -788,30 +788,30 @@ class SonosEntity(MediaPlayerEntity): self.async_write_ha_state() - def update_volume(self): + def update_volume(self) -> None: """Update information about currently volume settings.""" self._player_volume = self.soco.volume self._player_muted = self.soco.mute self._night_sound = self.soco.night_mode self._speech_enhance = self.soco.dialog_mode - def update_groups(self, event=None): + def update_groups(self, event: Event | None = None) -> None: """Handle callback for topology change event.""" - coro = self.async_update_groups_coro(event) + coro = self.create_update_groups_coro(event) if coro: - self.hass.add_job(coro) + self.hass.add_job(coro) # type: ignore @callback - def async_update_groups(self, event=None): + def async_update_groups(self, event: Event | None = None) -> None: """Handle callback for topology change event.""" - coro = self.async_update_groups_coro(event) + coro = self.create_update_groups_coro(event) if coro: - self.hass.async_add_job(coro) + self.hass.async_add_job(coro) # type: ignore - def async_update_groups_coro(self, event=None): + def create_update_groups_coro(self, event: Event | None = None) -> Coroutine | None: """Handle callback for topology change event.""" - def _get_soco_group(): + def _get_soco_group() -> list[str]: """Ask SoCo cache for existing topology.""" coordinator_uid = self.unique_id slave_uids = [] @@ -827,16 +827,17 @@ class SonosEntity(MediaPlayerEntity): return [coordinator_uid] + slave_uids - async def _async_extract_group(event): + async def _async_extract_group(event: Event) -> list[str]: """Extract group layout from a topology event.""" group = event and event.zone_player_uui_ds_in_group if group: + assert isinstance(group, str) return group.split(",") return await self.hass.async_add_executor_job(_get_soco_group) @callback - def _async_regroup(group): + def _async_regroup(group: list[str]) -> None: """Rebuild internal group layout.""" sonos_group = [] for uid in group: @@ -856,7 +857,7 @@ class SonosEntity(MediaPlayerEntity): slave._sonos_group = sonos_group slave.async_schedule_update_ha_state() - async def _async_handle_group_event(event): + async def _async_handle_group_event(event: Event) -> None: """Get async lock and handle event.""" if event and self._poll_timer: # Cancel poll timer since we do receive events @@ -872,136 +873,136 @@ class SonosEntity(MediaPlayerEntity): self.hass.data[DATA_SONOS].topology_condition.notify_all() if event and not hasattr(event, "zone_player_uui_ds_in_group"): - return + return None return _async_handle_group_event(event) - def async_update_content(self, event=None): + @callback + def async_update_content(self, event: Event | None = None) -> None: """Update information about available content.""" if event and "favorites_update_id" in event.variables: self.hass.async_add_job(self._set_favorites) self.async_write_ha_state() @property - def volume_level(self): + def volume_level(self) -> int | None: """Volume level of the media player (0..1).""" - if self._player_volume is None: - return None - return self._player_volume / 100 + return self._player_volume and int(self._player_volume / 100) @property - def is_volume_muted(self): + def is_volume_muted(self) -> bool | None: """Return true if volume is muted.""" return self._player_muted - @property + @property # type: ignore[misc] @soco_coordinator - def shuffle(self): + def shuffle(self) -> str | None: """Shuffling state.""" - return PLAY_MODES[self._play_mode][0] + shuffle: str = PLAY_MODES[self._play_mode][0] + return shuffle - @property + @property # type: ignore[misc] @soco_coordinator - def repeat(self): + def repeat(self) -> str | None: """Return current repeat mode.""" sonos_repeat = PLAY_MODES[self._play_mode][1] return SONOS_TO_REPEAT[sonos_repeat] - @property + @property # type: ignore[misc] @soco_coordinator - def media_content_id(self): + def media_content_id(self) -> str | None: """Content id of current playing media.""" return self._uri @property - def media_content_type(self): + def media_content_type(self) -> str: """Content type of current playing media.""" return MEDIA_TYPE_MUSIC - @property + @property # type: ignore[misc] @soco_coordinator - def media_duration(self): + def media_duration(self) -> float | None: """Duration of current playing media in seconds.""" return self._media_duration - @property + @property # type: ignore[misc] @soco_coordinator - def media_position(self): + def media_position(self) -> float | None: """Position of current playing media in seconds.""" return self._media_position - @property + @property # type: ignore[misc] @soco_coordinator - def media_position_updated_at(self): + def media_position_updated_at(self) -> datetime.datetime | None: """When was the position of the current playing media valid.""" return self._media_position_updated_at - @property + @property # type: ignore[misc] @soco_coordinator - def media_image_url(self): + def media_image_url(self) -> str | None: """Image url of current playing media.""" return self._media_image_url or None - @property + @property # type: ignore[misc] @soco_coordinator - def media_channel(self): + def media_channel(self) -> str | None: """Channel currently playing.""" return self._media_channel or None - @property + @property # type: ignore[misc] @soco_coordinator - def media_artist(self): + def media_artist(self) -> str | None: """Artist of current playing media, music track only.""" return self._media_artist or None - @property + @property # type: ignore[misc] @soco_coordinator - def media_album_name(self): + def media_album_name(self) -> str | None: """Album name of current playing media, music track only.""" return self._media_album_name or None - @property + @property # type: ignore[misc] @soco_coordinator - def media_title(self): + def media_title(self) -> str | None: """Title of current playing media.""" return self._media_title or None - @property + @property # type: ignore[misc] @soco_coordinator - def queue_position(self): + def queue_position(self) -> int | None: """If playing local queue return the position in the queue else None.""" return self._queue_position - @property + @property # type: ignore[misc] @soco_coordinator - def source(self): + def source(self) -> str | None: """Name of the current input source.""" return self._source_name or None - @property + @property # type: ignore[misc] @soco_coordinator - def supported_features(self): + def supported_features(self) -> int: """Flag media player features that are supported.""" return SUPPORT_SONOS @soco_error() - def volume_up(self): + def volume_up(self) -> None: """Volume up media player.""" self._player.volume += self._volume_increment @soco_error() - def volume_down(self): + def volume_down(self) -> None: """Volume down media player.""" self._player.volume -= self._volume_increment @soco_error() - def set_volume_level(self, volume): + def set_volume_level(self, volume: str) -> None: """Set volume level, range 0..1.""" self.soco.volume = str(int(volume * 100)) @soco_error(UPNP_ERRORS_TO_IGNORE) @soco_coordinator - def set_shuffle(self, shuffle): + def set_shuffle(self, shuffle: str) -> None: """Enable/Disable shuffle mode.""" sonos_shuffle = shuffle sonos_repeat = PLAY_MODES[self._play_mode][1] @@ -1009,20 +1010,20 @@ class SonosEntity(MediaPlayerEntity): @soco_error(UPNP_ERRORS_TO_IGNORE) @soco_coordinator - def set_repeat(self, repeat): + def set_repeat(self, repeat: str) -> None: """Set repeat mode.""" sonos_shuffle = PLAY_MODES[self._play_mode][0] sonos_repeat = REPEAT_TO_SONOS[repeat] self.soco.play_mode = PLAY_MODE_BY_MEANING[(sonos_shuffle, sonos_repeat)] @soco_error() - def mute_volume(self, mute): + def mute_volume(self, mute: bool) -> None: """Mute (true) or unmute (false) media player.""" self.soco.mute = mute @soco_error() @soco_coordinator - def select_source(self, source): + def select_source(self, source: str) -> None: """Select input source.""" if source == SOURCE_LINEIN: self.soco.switch_to_line_in() @@ -1043,9 +1044,9 @@ class SonosEntity(MediaPlayerEntity): self.soco.add_to_queue(src.reference) self.soco.play_from_queue(0) - @property + @property # type: ignore[misc] @soco_coordinator - def source_list(self): + def source_list(self) -> list[str]: """List of available input sources.""" sources = [fav.title for fav in self._favorites] @@ -1061,49 +1062,49 @@ class SonosEntity(MediaPlayerEntity): @soco_error(UPNP_ERRORS_TO_IGNORE) @soco_coordinator - def media_play(self): + def media_play(self) -> None: """Send play command.""" self.soco.play() @soco_error(UPNP_ERRORS_TO_IGNORE) @soco_coordinator - def media_stop(self): + def media_stop(self) -> None: """Send stop command.""" self.soco.stop() @soco_error(UPNP_ERRORS_TO_IGNORE) @soco_coordinator - def media_pause(self): + def media_pause(self) -> None: """Send pause command.""" self.soco.pause() @soco_error(UPNP_ERRORS_TO_IGNORE) @soco_coordinator - def media_next_track(self): + def media_next_track(self) -> None: """Send next track command.""" self.soco.next() @soco_error(UPNP_ERRORS_TO_IGNORE) @soco_coordinator - def media_previous_track(self): + def media_previous_track(self) -> None: """Send next track command.""" self.soco.previous() @soco_error(UPNP_ERRORS_TO_IGNORE) @soco_coordinator - def media_seek(self, position): + def media_seek(self, position: str) -> None: """Send seek command.""" self.soco.seek(str(datetime.timedelta(seconds=int(position)))) @soco_error() @soco_coordinator - def clear_playlist(self): + def clear_playlist(self) -> None: """Clear players playlist.""" self.soco.clear_queue() @soco_error() @soco_coordinator - def play_media(self, media_type, media_id, **kwargs): + def play_media(self, media_type: str, media_id: str, **kwargs: Any) -> None: """ Send the play_media command to the media player. @@ -1116,7 +1117,7 @@ class SonosEntity(MediaPlayerEntity): """ if media_id and media_id.startswith(PLEX_URI_SCHEME): media_id = media_id[len(PLEX_URI_SCHEME) :] - play_on_sonos(self.hass, media_type, media_id, self.name) + play_on_sonos(self.hass, media_type, media_id, self.name) # type: ignore[no-untyped-call] elif media_type in (MEDIA_TYPE_MUSIC, MEDIA_TYPE_TRACK): if kwargs.get(ATTR_MEDIA_ENQUEUE): try: @@ -1140,7 +1141,7 @@ class SonosEntity(MediaPlayerEntity): self.soco.play_uri(media_id) elif media_type == MEDIA_TYPE_PLAYLIST: if media_id.startswith("S:"): - item = get_media(self._media_library, media_id, media_type) + item = get_media(self._media_library, media_id, media_type) # type: ignore[no-untyped-call] self.soco.play_uri(item.get_uri()) return try: @@ -1152,7 +1153,7 @@ class SonosEntity(MediaPlayerEntity): except StopIteration: _LOGGER.error('Could not find a Sonos playlist named "%s"', media_id) elif media_type in PLAYABLE_MEDIA_TYPES: - item = get_media(self._media_library, media_id, media_type) + item = get_media(self._media_library, media_id, media_type) # type: ignore[no-untyped-call] if not item: _LOGGER.error('Could not find "%s" in the library', media_id) @@ -1163,7 +1164,7 @@ class SonosEntity(MediaPlayerEntity): _LOGGER.error('Sonos does not support a media type of "%s"', media_type) @soco_error() - def join(self, slaves): + def join(self, slaves: list[SonosEntity]) -> list[SonosEntity]: """Form a group with other players.""" if self._coordinator: self.unjoin() @@ -1182,23 +1183,27 @@ class SonosEntity(MediaPlayerEntity): return group @staticmethod - async def join_multi(hass, master, entities): + async def join_multi( + hass: HomeAssistant, master: SonosEntity, entities: list[SonosEntity] + ) -> None: """Form a group with other players.""" async with hass.data[DATA_SONOS].topology_condition: - group = await hass.async_add_executor_job(master.join, entities) + group: list[SonosEntity] = await hass.async_add_executor_job( + master.join, entities + ) await SonosEntity.wait_for_groups(hass, [group]) @soco_error() - def unjoin(self): + def unjoin(self) -> None: """Unjoin the player from a group.""" self.soco.unjoin() self._coordinator = None @staticmethod - async def unjoin_multi(hass, entities): + async def unjoin_multi(hass: HomeAssistant, entities: list[SonosEntity]) -> None: """Unjoin several players from their group.""" - def _unjoin_all(entities): + def _unjoin_all(entities: list[SonosEntity]) -> None: """Sync helper.""" # Unjoin slaves first to prevent inheritance of queues coordinators = [e for e in entities if e.is_coordinator] @@ -1212,7 +1217,7 @@ class SonosEntity(MediaPlayerEntity): await SonosEntity.wait_for_groups(hass, [[e] for e in entities]) @soco_error() - def snapshot(self, with_group): + def snapshot(self, with_group: bool) -> None: """Snapshot the state of a player.""" self._soco_snapshot = pysonos.snapshot.Snapshot(self.soco) self._soco_snapshot.snapshot() @@ -1222,30 +1227,33 @@ class SonosEntity(MediaPlayerEntity): self._snapshot_group = None @staticmethod - async def snapshot_multi(hass, entities, with_group): + async def snapshot_multi( + hass: HomeAssistant, entities: list[SonosEntity], with_group: bool + ) -> None: """Snapshot all the entities and optionally their groups.""" # pylint: disable=protected-access - def _snapshot_all(entities): + def _snapshot_all(entities: list[SonosEntity]) -> None: """Sync helper.""" for entity in entities: entity.snapshot(with_group) # Find all affected players - entities = set(entities) + entities_set = set(entities) if with_group: - for entity in list(entities): - entities.update(entity._sonos_group) + for entity in list(entities_set): + entities_set.update(entity._sonos_group) async with hass.data[DATA_SONOS].topology_condition: - await hass.async_add_executor_job(_snapshot_all, entities) + await hass.async_add_executor_job(_snapshot_all, entities_set) @soco_error() - def restore(self): + def restore(self) -> None: """Restore a snapshotted state to a player.""" try: + assert self._soco_snapshot is not None self._soco_snapshot.restore() - except (TypeError, AttributeError, SoCoException) as ex: + except (TypeError, AssertionError, AttributeError, SoCoException) as ex: # Can happen if restoring a coordinator onto a current slave _LOGGER.warning("Error on restore %s: %s", self.entity_id, ex) @@ -1253,11 +1261,15 @@ class SonosEntity(MediaPlayerEntity): self._snapshot_group = None @staticmethod - async def restore_multi(hass, entities, with_group): + async def restore_multi( + hass: HomeAssistant, entities: list[SonosEntity], with_group: bool + ) -> None: """Restore snapshots for all the entities.""" # pylint: disable=protected-access - def _restore_groups(entities, with_group): + def _restore_groups( + entities: list[SonosEntity], with_group: bool + ) -> list[list[SonosEntity]]: """Pause all current coordinators and restore groups.""" for entity in (e for e in entities if e.is_coordinator): if entity.state == STATE_PLAYING: @@ -1273,13 +1285,14 @@ class SonosEntity(MediaPlayerEntity): # Bring back the original group topology for entity in (e for e in entities if e._snapshot_group): + assert entity._snapshot_group is not None if entity._snapshot_group[0] == entity: entity.join(entity._snapshot_group) groups.append(entity._snapshot_group.copy()) return groups - def _restore_players(entities): + def _restore_players(entities: list[SonosEntity]) -> None: """Restore state of all players.""" for entity in (e for e in entities if not e.is_coordinator): entity.restore() @@ -1288,26 +1301,29 @@ class SonosEntity(MediaPlayerEntity): entity.restore() # Find all affected players - entities = {e for e in entities if e._soco_snapshot} + entities_set = {e for e in entities if e._soco_snapshot} if with_group: - for entity in [e for e in entities if e._snapshot_group]: - entities.update(entity._snapshot_group) + for entity in [e for e in entities_set if e._snapshot_group]: + assert entity._snapshot_group is not None + entities_set.update(entity._snapshot_group) async with hass.data[DATA_SONOS].topology_condition: groups = await hass.async_add_executor_job( - _restore_groups, entities, with_group + _restore_groups, entities_set, with_group ) await SonosEntity.wait_for_groups(hass, groups) - await hass.async_add_executor_job(_restore_players, entities) + await hass.async_add_executor_job(_restore_players, entities_set) @staticmethod - async def wait_for_groups(hass, groups): + async def wait_for_groups( + hass: HomeAssistant, groups: list[list[SonosEntity]] + ) -> None: """Wait until all groups are present, or timeout.""" # pylint: disable=protected-access - def _test_groups(groups): + def _test_groups(groups: list[list[SonosEntity]]) -> bool: """Return whether all groups exist now.""" for group in groups: coordinator = group[0] @@ -1335,21 +1351,26 @@ class SonosEntity(MediaPlayerEntity): @soco_error() @soco_coordinator - def set_sleep_timer(self, sleep_time): + def set_sleep_timer(self, sleep_time: int) -> None: """Set the timer on the player.""" self.soco.set_sleep_timer(sleep_time) @soco_error() @soco_coordinator - def clear_sleep_timer(self): + def clear_sleep_timer(self) -> None: """Clear the timer on the player.""" self.soco.set_sleep_timer(None) @soco_error() @soco_coordinator def set_alarm( - self, alarm_id, time=None, volume=None, enabled=None, include_linked_zones=None - ): + self, + alarm_id: int, + time: datetime.datetime | None = None, + volume: float | None = None, + enabled: bool | None = None, + include_linked_zones: bool | None = None, + ) -> None: """Set the alarm clock on the player.""" alarm = None for one_alarm in alarms.get_alarms(self.soco): @@ -1370,7 +1391,12 @@ class SonosEntity(MediaPlayerEntity): alarm.save() @soco_error() - def set_option(self, night_sound=None, speech_enhance=None, status_light=None): + def set_option( + self, + night_sound: bool | None = None, + speech_enhance: bool | None = None, + status_light: bool | None = None, + ) -> None: """Modify playback options.""" if night_sound is not None and self._night_sound is not None: self.soco.night_mode = night_sound @@ -1382,20 +1408,22 @@ class SonosEntity(MediaPlayerEntity): self.soco.status_light = status_light @soco_error() - def play_queue(self, queue_position=0): + def play_queue(self, queue_position: int = 0) -> None: """Start playing the queue.""" self.soco.play_from_queue(queue_position) @soco_error() @soco_coordinator - def remove_from_queue(self, queue_position=0): + def remove_from_queue(self, queue_position: int = 0) -> None: """Remove item from the queue.""" self.soco.remove_from_queue(queue_position) @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> dict[str, Any]: """Return entity specific state attributes.""" - attributes = {ATTR_SONOS_GROUP: [e.entity_id for e in self._sonos_group]} + attributes: dict[str, Any] = { + ATTR_SONOS_GROUP: [e.entity_id for e in self._sonos_group] + } if self._night_sound is not None: attributes[ATTR_NIGHT_SOUND] = self._night_sound @@ -1409,8 +1437,11 @@ class SonosEntity(MediaPlayerEntity): return attributes async def async_get_browse_image( - self, media_content_type, media_content_id, media_image_id=None - ): + self, + media_content_type: str | None, + media_content_id: str | None, + media_image_id: str | None = None, + ) -> tuple[None | str, None | str]: """Fetch media browser image to serve via proxy.""" if ( media_content_type in [MEDIA_TYPE_ALBUM, MEDIA_TYPE_ARTIST] @@ -1424,25 +1455,29 @@ class SonosEntity(MediaPlayerEntity): ) image_url = getattr(item, "album_art_uri", None) if image_url: - result = await self._async_fetch_image(image_url) - return result + result = await self._async_fetch_image(image_url) # type: ignore[no-untyped-call] + return result # type: ignore return (None, None) - async def async_browse_media(self, media_content_type=None, media_content_id=None): + async def async_browse_media( + self, media_content_type: str | None = None, media_content_id: str | None = None + ) -> Any: """Implement the websocket media browsing helper.""" is_internal = is_internal_request(self.hass) def _get_thumbnail_url( - media_content_type, media_content_id, media_image_id=None - ): + media_content_type: str, + media_content_id: str, + media_image_id: str | None = None, + ) -> str | None: if is_internal: - item = get_media( + item = get_media( # type: ignore[no-untyped-call] self._media_library, media_content_id, media_content_type, ) - return getattr(item, "album_art_uri", None) + return getattr(item, "album_art_uri", None) # type: ignore[no-any-return] return self.get_browse_image_url( media_content_type, diff --git a/setup.cfg b/setup.cfg index 65b598f4f6f..d8569ad2188 100644 --- a/setup.cfg +++ b/setup.cfg @@ -43,7 +43,7 @@ warn_redundant_casts = true warn_unused_configs = true -[mypy-homeassistant.block_async_io,homeassistant.bootstrap,homeassistant.components,homeassistant.config_entries,homeassistant.config,homeassistant.const,homeassistant.core,homeassistant.data_entry_flow,homeassistant.exceptions,homeassistant.__init__,homeassistant.loader,homeassistant.__main__,homeassistant.requirements,homeassistant.runner,homeassistant.setup,homeassistant.util,homeassistant.auth.*,homeassistant.components.automation.*,homeassistant.components.binary_sensor.*,homeassistant.components.bond.*,homeassistant.components.calendar.*,homeassistant.components.cover.*,homeassistant.components.device_automation.*,homeassistant.components.frontend.*,homeassistant.components.geo_location.*,homeassistant.components.group.*,homeassistant.components.history.*,homeassistant.components.http.*,homeassistant.components.huawei_lte.*,homeassistant.components.hyperion.*,homeassistant.components.image_processing.*,homeassistant.components.integration.*,homeassistant.components.knx.*,homeassistant.components.light.*,homeassistant.components.lock.*,homeassistant.components.mailbox.*,homeassistant.components.media_player.*,homeassistant.components.notify.*,homeassistant.components.number.*,homeassistant.components.persistent_notification.*,homeassistant.components.proximity.*,homeassistant.components.recorder.purge,homeassistant.components.recorder.repack,homeassistant.components.remote.*,homeassistant.components.scene.*,homeassistant.components.sensor.*,homeassistant.components.slack.*,homeassistant.components.sun.*,homeassistant.components.switch.*,homeassistant.components.systemmonitor.*,homeassistant.components.tts.*,homeassistant.components.vacuum.*,homeassistant.components.water_heater.*,homeassistant.components.weather.*,homeassistant.components.websocket_api.*,homeassistant.components.zeroconf.*,homeassistant.components.zone.*,homeassistant.components.zwave_js.*,homeassistant.helpers.*,homeassistant.scripts.*,homeassistant.util.*,tests.components.hyperion.*] +[mypy-homeassistant.block_async_io,homeassistant.bootstrap,homeassistant.components,homeassistant.config_entries,homeassistant.config,homeassistant.const,homeassistant.core,homeassistant.data_entry_flow,homeassistant.exceptions,homeassistant.__init__,homeassistant.loader,homeassistant.__main__,homeassistant.requirements,homeassistant.runner,homeassistant.setup,homeassistant.util,homeassistant.auth.*,homeassistant.components.automation.*,homeassistant.components.binary_sensor.*,homeassistant.components.bond.*,homeassistant.components.calendar.*,homeassistant.components.cover.*,homeassistant.components.device_automation.*,homeassistant.components.frontend.*,homeassistant.components.geo_location.*,homeassistant.components.group.*,homeassistant.components.history.*,homeassistant.components.http.*,homeassistant.components.huawei_lte.*,homeassistant.components.hyperion.*,homeassistant.components.image_processing.*,homeassistant.components.integration.*,homeassistant.components.knx.*,homeassistant.components.light.*,homeassistant.components.lock.*,homeassistant.components.mailbox.*,homeassistant.components.media_player.*,homeassistant.components.notify.*,homeassistant.components.number.*,homeassistant.components.persistent_notification.*,homeassistant.components.proximity.*,homeassistant.components.recorder.purge,homeassistant.components.recorder.repack,homeassistant.components.remote.*,homeassistant.components.scene.*,homeassistant.components.sensor.*,homeassistant.components.slack.*,homeassistant.components.sonos.media_player,homeassistant.components.sun.*,homeassistant.components.switch.*,homeassistant.components.systemmonitor.*,homeassistant.components.tts.*,homeassistant.components.vacuum.*,homeassistant.components.water_heater.*,homeassistant.components.weather.*,homeassistant.components.websocket_api.*,homeassistant.components.zeroconf.*,homeassistant.components.zone.*,homeassistant.components.zwave_js.*,homeassistant.helpers.*,homeassistant.scripts.*,homeassistant.util.*,tests.components.hyperion.*] strict = true ignore_errors = false warn_unreachable = true