diff --git a/homeassistant/components/sonos/__init__.py b/homeassistant/components/sonos/__init__.py index d94b49e52f2..f0dd8e668fa 100644 --- a/homeassistant/components/sonos/__init__.py +++ b/homeassistant/components/sonos/__init__.py @@ -8,6 +8,7 @@ import datetime from functools import partial import logging import socket +from typing import TYPE_CHECKING, Any, Optional, cast from urllib.parse import urlparse from soco import events_asyncio @@ -21,7 +22,7 @@ from homeassistant.components import ssdp from homeassistant.components.media_player import DOMAIN as MP_DOMAIN from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_HOSTS, EVENT_HOMEASSISTANT_STOP -from homeassistant.core import Event, HomeAssistant, callback +from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant, callback from homeassistant.helpers import config_validation as cv, device_registry as dr from homeassistant.helpers.dispatcher import async_dispatcher_send, dispatcher_send from homeassistant.helpers.event import async_track_time_interval, call_later @@ -93,7 +94,7 @@ class SonosData: self.favorites: dict[str, SonosFavorites] = {} self.alarms: dict[str, SonosAlarms] = {} self.topology_condition = asyncio.Condition() - self.hosts_heartbeat = None + self.hosts_heartbeat: CALLBACK_TYPE | None = None self.discovery_known: set[str] = set() self.boot_counts: dict[str, int] = {} self.mdns_names: dict[str, str] = {} @@ -168,10 +169,10 @@ class SonosDiscoveryManager: self.data = data self.hosts = set(hosts) self.discovery_lock = asyncio.Lock() - self._known_invisible = set() + self._known_invisible: set[SoCo] = set() self._manual_config_required = bool(hosts) - async def async_shutdown(self): + async def async_shutdown(self) -> None: """Stop all running tasks.""" await self._async_stop_event_listener() self._stop_manual_heartbeat() @@ -236,6 +237,8 @@ class SonosDiscoveryManager: (SonosAlarms, self.data.alarms), (SonosFavorites, self.data.favorites), ): + if TYPE_CHECKING: + coord_dict = cast(dict[str, Any], coord_dict) if soco.household_id not in coord_dict: new_coordinator = coordinator(self.hass, soco.household_id) new_coordinator.setup(soco) @@ -298,7 +301,7 @@ class SonosDiscoveryManager: ) async def _async_handle_discovery_message( - self, uid: str, discovered_ip: str, boot_seqnum: int + self, uid: str, discovered_ip: str, boot_seqnum: int | None ) -> None: """Handle discovered player creation and activity.""" async with self.discovery_lock: @@ -338,22 +341,27 @@ class SonosDiscoveryManager: async_dispatcher_send(self.hass, f"{SONOS_VANISHED}-{uid}", reason) return - discovered_ip = urlparse(info.ssdp_location).hostname - boot_seqnum = info.ssdp_headers.get("X-RINCON-BOOTSEQ") self.async_discovered_player( "SSDP", info, - discovered_ip, + cast(str, urlparse(info.ssdp_location).hostname), uid, - boot_seqnum, - info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME), + info.ssdp_headers.get("X-RINCON-BOOTSEQ"), + cast(str, info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME)), None, ) @callback def async_discovered_player( - self, source, info, discovered_ip, uid, boot_seqnum, model, mdns_name - ): + self, + source: str, + info: ssdp.SsdpServiceInfo, + discovered_ip: str, + uid: str, + boot_seqnum: str | int | None, + model: str, + mdns_name: str | None, + ) -> None: """Handle discovery via ssdp or zeroconf.""" if self._manual_config_required: _LOGGER.warning( @@ -376,10 +384,12 @@ class SonosDiscoveryManager: _LOGGER.debug("New %s discovery uid=%s: %s", source, uid, info) self.data.discovery_known.add(uid) asyncio.create_task( - self._async_handle_discovery_message(uid, discovered_ip, boot_seqnum) + self._async_handle_discovery_message( + uid, discovered_ip, cast(Optional[int], boot_seqnum) + ) ) - async def setup_platforms_and_discovery(self): + async def setup_platforms_and_discovery(self) -> None: """Set up platforms and discovery.""" await self.hass.config_entries.async_forward_entry_setups(self.entry, PLATFORMS) self.entry.async_on_unload( diff --git a/homeassistant/components/sonos/binary_sensor.py b/homeassistant/components/sonos/binary_sensor.py index e890c1c64a8..3f736f83922 100644 --- a/homeassistant/components/sonos/binary_sensor.py +++ b/homeassistant/components/sonos/binary_sensor.py @@ -109,6 +109,6 @@ class SonosMicrophoneSensorEntity(SonosEntity, BinarySensorEntity): self.speaker.mic_enabled = self.soco.mic_enabled @property - def is_on(self) -> bool: + def is_on(self) -> bool | None: """Return the state of the binary sensor.""" return self.speaker.mic_enabled diff --git a/homeassistant/components/sonos/diagnostics.py b/homeassistant/components/sonos/diagnostics.py index 463884e1ea8..fda96b86215 100644 --- a/homeassistant/components/sonos/diagnostics.py +++ b/homeassistant/components/sonos/diagnostics.py @@ -47,11 +47,11 @@ async def async_get_config_entry_diagnostics( hass: HomeAssistant, config_entry: ConfigEntry ) -> dict[str, Any]: """Return diagnostics for a config entry.""" - payload = {"current_timestamp": time.monotonic()} + payload: dict[str, Any] = {"current_timestamp": time.monotonic()} for section in ("discovered", "discovery_known"): payload[section] = {} - data = getattr(hass.data[DATA_SONOS], section) + data: set[Any] | dict[str, Any] = getattr(hass.data[DATA_SONOS], section) if isinstance(data, set): payload[section] = data continue @@ -60,7 +60,6 @@ async def async_get_config_entry_diagnostics( payload[section][key] = await async_generate_speaker_info(hass, value) else: payload[section][key] = value - return payload @@ -85,12 +84,12 @@ async def async_generate_media_info( hass: HomeAssistant, speaker: SonosSpeaker ) -> dict[str, Any]: """Generate a diagnostic payload for current media metadata.""" - payload = {} + payload: dict[str, Any] = {} for attrib in MEDIA_DIAGNOSTIC_ATTRIBUTES: payload[attrib] = getattr(speaker.media, attrib) - def poll_current_track_info(): + def poll_current_track_info() -> dict[str, Any] | str: try: return speaker.soco.avTransport.GetPositionInfo( [("InstanceID", 0), ("Channel", "Master")], @@ -110,9 +109,11 @@ async def async_generate_speaker_info( hass: HomeAssistant, speaker: SonosSpeaker ) -> dict[str, Any]: """Generate the diagnostic payload for a specific speaker.""" - payload = {} + payload: dict[str, Any] = {} - def get_contents(item): + def get_contents( + item: int | float | str | dict[str, Any] + ) -> int | float | str | dict[str, Any]: if isinstance(item, (int, float, str)): return item if isinstance(item, dict): diff --git a/homeassistant/components/sonos/household_coordinator.py b/homeassistant/components/sonos/household_coordinator.py index 51d7e9cec8c..29b9a005552 100644 --- a/homeassistant/components/sonos/household_coordinator.py +++ b/homeassistant/components/sonos/household_coordinator.py @@ -20,13 +20,14 @@ _LOGGER = logging.getLogger(__name__) class SonosHouseholdCoordinator: """Base class for Sonos household-level storage.""" + cache_update_lock: asyncio.Lock + def __init__(self, hass: HomeAssistant, household_id: str) -> None: """Initialize the data.""" self.hass = hass self.household_id = household_id self.async_poll: Callable[[], Coroutine[None, None, None]] | None = None self.last_processed_event_id: int | None = None - self.cache_update_lock: asyncio.Lock | None = None def setup(self, soco: SoCo) -> None: """Set up the SonosAlarm instance.""" diff --git a/homeassistant/components/sonos/media.py b/homeassistant/components/sonos/media.py index 9608356ba64..24233b1316f 100644 --- a/homeassistant/components/sonos/media.py +++ b/homeassistant/components/sonos/media.py @@ -2,7 +2,6 @@ from __future__ import annotations import datetime -import logging from typing import Any from soco.core import ( @@ -43,8 +42,6 @@ UNAVAILABLE_VALUES = {"", "NOT_IMPLEMENTED", None} DURATION_SECONDS = "duration_in_s" POSITION_SECONDS = "position_in_s" -_LOGGER = logging.getLogger(__name__) - def _timespan_secs(timespan: str | None) -> None | float: """Parse a time-span into number of seconds.""" @@ -106,7 +103,7 @@ class SonosMedia: @soco_error() def poll_track_info(self) -> dict[str, Any]: """Poll the speaker for current track info, add converted position values, and return.""" - track_info = self.soco.get_current_track_info() + track_info: dict[str, Any] = self.soco.get_current_track_info() track_info[DURATION_SECONDS] = _timespan_secs(track_info.get("duration")) track_info[POSITION_SECONDS] = _timespan_secs(track_info.get("position")) return track_info diff --git a/homeassistant/components/sonos/media_browser.py b/homeassistant/components/sonos/media_browser.py index b2d881e8bf2..95ff08cb87b 100644 --- a/homeassistant/components/sonos/media_browser.py +++ b/homeassistant/components/sonos/media_browser.py @@ -5,8 +5,13 @@ from collections.abc import Callable from contextlib import suppress from functools import partial import logging +from typing import cast from urllib.parse import quote_plus, unquote +from soco.data_structures import DidlFavorite, DidlObject +from soco.ms_data_structures import MusicServiceItem +from soco.music_library import MusicLibrary + from homeassistant.components import media_source, plex, spotify from homeassistant.components.media_player import BrowseMedia from homeassistant.components.media_player.const import ( @@ -50,12 +55,12 @@ def get_thumbnail_url_full( ) -> str | None: """Get thumbnail URL.""" if is_internal: - item = get_media( # type: ignore[no-untyped-call] + item = get_media( media.library, media_content_id, media_content_type, ) - return getattr(item, "album_art_uri", None) # type: ignore[no-any-return] + return getattr(item, "album_art_uri", None) return get_browse_image_url( media_content_type, @@ -64,19 +69,19 @@ def get_thumbnail_url_full( ) -def media_source_filter(item: BrowseMedia): +def media_source_filter(item: BrowseMedia) -> bool: """Filter media sources.""" return item.media_content_type.startswith("audio/") async def async_browse_media( - hass, + hass: HomeAssistant, speaker: SonosSpeaker, media: SonosMedia, get_browse_image_url: GetBrowseImageUrlType, media_content_id: str | None, media_content_type: str | None, -): +) -> BrowseMedia: """Browse media.""" if media_content_id is None: @@ -86,6 +91,7 @@ async def async_browse_media( media, get_browse_image_url, ) + assert media_content_type is not None if media_source.is_media_source_id(media_content_id): return await media_source.async_browse_media( @@ -150,7 +156,9 @@ async def async_browse_media( return response -def build_item_response(media_library, payload, get_thumbnail_url=None): +def build_item_response( + media_library: MusicLibrary, payload: dict[str, str], get_thumbnail_url=None +) -> BrowseMedia | None: """Create response payload for the provided media query.""" if payload["search_type"] == MEDIA_TYPE_ALBUM and payload["idstring"].startswith( ("A:GENRE", "A:COMPOSER") @@ -166,7 +174,7 @@ def build_item_response(media_library, payload, get_thumbnail_url=None): "Unknown media type received when building item response: %s", payload["search_type"], ) - return + return None media = media_library.browse_by_idstring( search_type, @@ -176,7 +184,7 @@ def build_item_response(media_library, payload, get_thumbnail_url=None): ) if media is None: - return + return None thumbnail = None title = None @@ -222,7 +230,7 @@ def build_item_response(media_library, payload, get_thumbnail_url=None): ) -def item_payload(item, get_thumbnail_url=None): +def item_payload(item: DidlObject, get_thumbnail_url=None) -> BrowseMedia: """ Create response payload for a single media item. @@ -256,9 +264,9 @@ async def root_payload( speaker: SonosSpeaker, media: SonosMedia, get_browse_image_url: GetBrowseImageUrlType, -): +) -> BrowseMedia: """Return root payload for Sonos.""" - children = [] + children: list[BrowseMedia] = [] if speaker.favorites: children.append( @@ -303,14 +311,15 @@ async def root_payload( if "spotify" in hass.config.components: result = await spotify.async_browse_media(hass, None, None) - children.extend(result.children) + if result.children: + children.extend(result.children) try: item = await media_source.async_browse_media( hass, None, content_filter=media_source_filter ) # If domain is None, it's overview of available sources - if item.domain is None: + if item.domain is None and item.children is not None: children.extend(item.children) else: children.append(item) @@ -338,7 +347,7 @@ async def root_payload( ) -def library_payload(media_library, get_thumbnail_url=None): +def library_payload(media_library: MusicLibrary, get_thumbnail_url=None) -> BrowseMedia: """ Create response payload to describe contents of a specific library. @@ -360,7 +369,7 @@ def library_payload(media_library, get_thumbnail_url=None): ) -def favorites_payload(favorites): +def favorites_payload(favorites: list[DidlFavorite]) -> BrowseMedia: """ Create response payload to describe contents of a specific library. @@ -398,7 +407,9 @@ def favorites_payload(favorites): ) -def favorites_folder_payload(favorites, media_content_id): +def favorites_folder_payload( + favorites: list[DidlFavorite], media_content_id: str +) -> BrowseMedia: """Create response payload to describe all items of a type of favorite. Used by async_browse_media. @@ -432,7 +443,7 @@ def favorites_folder_payload(favorites, media_content_id): ) -def get_media_type(item): +def get_media_type(item: DidlObject) -> str: """Extract media type of item.""" if item.item_class == "object.item.audioItem.musicTrack": return SONOS_TRACKS @@ -450,7 +461,7 @@ def get_media_type(item): return SONOS_TYPES_MAPPING.get(item.item_id.split("/")[0], item.item_class) -def can_play(item): +def can_play(item: DidlObject) -> bool: """ Test if playable. @@ -459,7 +470,7 @@ def can_play(item): return SONOS_TO_MEDIA_TYPES.get(item) in PLAYABLE_MEDIA_TYPES -def can_expand(item): +def can_expand(item: DidlObject) -> bool: """ Test if expandable. @@ -474,14 +485,16 @@ def can_expand(item): return SONOS_TYPES_MAPPING.get(item.item_id) in EXPANDABLE_MEDIA_TYPES -def get_content_id(item): +def get_content_id(item: DidlObject) -> str: """Extract content id or uri.""" if item.item_class == "object.item.audioItem.musicTrack": - return item.get_uri() - return item.item_id + return cast(str, item.get_uri()) + return cast(str, item.item_id) -def get_media(media_library, item_id, search_type): +def get_media( + media_library: MusicLibrary, item_id: str, search_type: str +) -> MusicServiceItem: """Fetch media/album.""" search_type = MEDIA_TYPES_TO_SONOS.get(search_type, search_type) diff --git a/homeassistant/components/sonos/media_player.py b/homeassistant/components/sonos/media_player.py index 1f57cafbf09..14e0693f55a 100644 --- a/homeassistant/components/sonos/media_player.py +++ b/homeassistant/components/sonos/media_player.py @@ -130,11 +130,11 @@ async def async_setup_entry( if service_call.service == SERVICE_SNAPSHOT: await SonosSpeaker.snapshot_multi( - hass, speakers, service_call.data[ATTR_WITH_GROUP] # type: ignore[arg-type] + hass, speakers, service_call.data[ATTR_WITH_GROUP] ) elif service_call.service == SERVICE_RESTORE: await SonosSpeaker.restore_multi( - hass, speakers, service_call.data[ATTR_WITH_GROUP] # type: ignore[arg-type] + hass, speakers, service_call.data[ATTR_WITH_GROUP] ) config_entry.async_on_unload( @@ -153,7 +153,7 @@ async def async_setup_entry( SONOS_DOMAIN, SERVICE_RESTORE, async_service_handle, join_unjoin_schema ) - platform.async_register_entity_service( # type: ignore + platform.async_register_entity_service( SERVICE_SET_TIMER, { vol.Required(ATTR_SLEEP_TIME): vol.All( @@ -163,9 +163,9 @@ async def async_setup_entry( "set_sleep_timer", ) - platform.async_register_entity_service(SERVICE_CLEAR_TIMER, {}, "clear_sleep_timer") # type: ignore + platform.async_register_entity_service(SERVICE_CLEAR_TIMER, {}, "clear_sleep_timer") - platform.async_register_entity_service( # type: ignore + platform.async_register_entity_service( SERVICE_UPDATE_ALARM, { vol.Required(ATTR_ALARM_ID): cv.positive_int, @@ -177,13 +177,13 @@ async def async_setup_entry( "set_alarm", ) - platform.async_register_entity_service( # type: ignore + platform.async_register_entity_service( SERVICE_PLAY_QUEUE, {vol.Optional(ATTR_QUEUE_POSITION): cv.positive_int}, "play_queue", ) - platform.async_register_entity_service( # type: ignore + platform.async_register_entity_service( SERVICE_REMOVE_FROM_QUEUE, {vol.Optional(ATTR_QUEUE_POSITION): cv.positive_int}, "remove_from_queue", @@ -239,8 +239,8 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): """Return if the media_player is available.""" return ( self.speaker.available - and self.speaker.sonos_group_entities - and self.media.playback_status + and bool(self.speaker.sonos_group_entities) + and self.media.playback_status is not None ) @property @@ -257,7 +257,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): """Return a hash of self.""" return hash(self.unique_id) - @property # type: ignore[misc] + @property def state(self) -> str: """Return the state of the entity.""" if self.media.playback_status in ( @@ -300,13 +300,12 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): """Return true if volume is muted.""" return self.speaker.muted - @property # type: ignore[misc] - def shuffle(self) -> str | None: + @property + def shuffle(self) -> bool | None: """Shuffling state.""" - shuffle: str = PLAY_MODES[self.media.play_mode][0] - return shuffle + return PLAY_MODES[self.media.play_mode][0] - @property # type: ignore[misc] + @property def repeat(self) -> str | None: """Return current repeat mode.""" sonos_repeat = PLAY_MODES[self.media.play_mode][1] @@ -317,32 +316,32 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): """Return the SonosMedia object from the coordinator speaker.""" return self.coordinator.media - @property # type: ignore[misc] + @property def media_content_id(self) -> str | None: """Content id of current playing media.""" return self.media.uri - @property # type: ignore[misc] - def media_duration(self) -> float | None: + @property + def media_duration(self) -> int | None: """Duration of current playing media in seconds.""" - return self.media.duration + return int(self.media.duration) if self.media.duration else None - @property # type: ignore[misc] - def media_position(self) -> float | None: + @property + def media_position(self) -> int | None: """Position of current playing media in seconds.""" - return self.media.position + return int(self.media.position) if self.media.position else None - @property # type: ignore[misc] + @property 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 # type: ignore[misc] + @property def media_image_url(self) -> str | None: """Image url of current playing media.""" return self.media.image_url or None - @property # type: ignore[misc] + @property def media_channel(self) -> str | None: """Channel currently playing.""" return self.media.channel or None @@ -352,22 +351,22 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): """Title of playlist currently playing.""" return self.media.playlist_name - @property # type: ignore[misc] + @property def media_artist(self) -> str | None: """Artist of current playing media, music track only.""" return self.media.artist or None - @property # type: ignore[misc] + @property def media_album_name(self) -> str | None: """Album name of current playing media, music track only.""" return self.media.album_name or None - @property # type: ignore[misc] + @property def media_title(self) -> str | None: """Title of current playing media.""" return self.media.title or None - @property # type: ignore[misc] + @property def source(self) -> str | None: """Name of the current input source.""" return self.media.source_name or None @@ -383,12 +382,12 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): self.soco.volume -= VOLUME_INCREMENT @soco_error() - def set_volume_level(self, volume: str) -> None: + def set_volume_level(self, volume: float) -> None: """Set volume level, range 0..1.""" self.soco.volume = str(int(volume * 100)) @soco_error(UPNP_ERRORS_TO_IGNORE) - def set_shuffle(self, shuffle: str) -> None: + def set_shuffle(self, shuffle: bool) -> None: """Enable/Disable shuffle mode.""" sonos_shuffle = shuffle sonos_repeat = PLAY_MODES[self.media.play_mode][1] @@ -486,7 +485,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): self.coordinator.soco.previous() @soco_error(UPNP_ERRORS_TO_IGNORE) - def media_seek(self, position: str) -> None: + def media_seek(self, position: float) -> None: """Send seek command.""" self.coordinator.soco.seek(str(datetime.timedelta(seconds=int(position)))) @@ -606,7 +605,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): soco.play_uri(media_id, force_radio=is_radio) elif media_type == MEDIA_TYPE_PLAYLIST: if media_id.startswith("S:"): - item = media_browser.get_media(self.media.library, media_id, media_type) # type: ignore[no-untyped-call] + item = media_browser.get_media(self.media.library, media_id, media_type) soco.play_uri(item.get_uri()) return try: @@ -619,7 +618,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): soco.add_to_queue(playlist) soco.play_from_queue(0) elif media_type in PLAYABLE_MEDIA_TYPES: - item = media_browser.get_media(self.media.library, media_id, media_type) # type: ignore[no-untyped-call] + item = media_browser.get_media(self.media.library, media_id, media_type) if not item: _LOGGER.error('Could not find "%s" in the library', media_id) @@ -649,7 +648,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): include_linked_zones: bool | None = None, ) -> None: """Set the alarm clock on the player.""" - alarm = None + alarm: alarms.Alarm | None = None for one_alarm in alarms.get_alarms(self.coordinator.soco): if one_alarm.alarm_id == str(alarm_id): alarm = one_alarm @@ -710,8 +709,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): MEDIA_TYPES_TO_SONOS[media_content_type], ) if image_url := getattr(item, "album_art_uri", None): - result = await self._async_fetch_image(image_url) # type: ignore[no-untyped-call] - return result # type: ignore + return await self._async_fetch_image(image_url) return (None, None) @@ -728,7 +726,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): media_content_type, ) - async def async_join_players(self, group_members): + async def async_join_players(self, group_members: list[str]) -> None: """Join `group_members` as a player group with the current player.""" speakers = [] for entity_id in group_members: @@ -739,7 +737,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity): await SonosSpeaker.join_multi(self.hass, self.speaker, speakers) - async def async_unjoin_player(self): + async def async_unjoin_player(self) -> None: """Remove this player from any group. Coalesces all calls within UNJOIN_SERVICE_TIMEOUT to allow use of SonosSpeaker.unjoin_multi() diff --git a/homeassistant/components/sonos/number.py b/homeassistant/components/sonos/number.py index ccbcbc3c339..7a6edb0d293 100644 --- a/homeassistant/components/sonos/number.py +++ b/homeassistant/components/sonos/number.py @@ -2,6 +2,7 @@ from __future__ import annotations import logging +from typing import cast from homeassistant.components.number import NumberEntity from homeassistant.config_entries import ConfigEntry @@ -24,6 +25,8 @@ LEVEL_TYPES = { "music_surround_level": (-15, 15), } +SocoFeatures = list[tuple[str, tuple[int, int]]] + _LOGGER = logging.getLogger(__name__) @@ -34,8 +37,8 @@ async def async_setup_entry( ) -> None: """Set up the Sonos number platform from a config entry.""" - def available_soco_attributes(speaker: SonosSpeaker) -> list[str]: - features = [] + def available_soco_attributes(speaker: SonosSpeaker) -> SocoFeatures: + features: SocoFeatures = [] for level_type, valid_range in LEVEL_TYPES.items(): if (state := getattr(speaker.soco, level_type, None)) is not None: setattr(speaker, level_type, state) @@ -67,7 +70,7 @@ class SonosLevelEntity(SonosEntity, NumberEntity): _attr_entity_category = EntityCategory.CONFIG def __init__( - self, speaker: SonosSpeaker, level_type: str, valid_range: tuple[int] + self, speaker: SonosSpeaker, level_type: str, valid_range: tuple[int, int] ) -> None: """Initialize the level entity.""" super().__init__(speaker) @@ -94,4 +97,4 @@ class SonosLevelEntity(SonosEntity, NumberEntity): @property def native_value(self) -> float: """Return the current value.""" - return getattr(self.speaker, self.level_type) + return cast(float, getattr(self.speaker, self.level_type)) diff --git a/homeassistant/components/sonos/sensor.py b/homeassistant/components/sonos/sensor.py index 8477e523a40..d1705fb030d 100644 --- a/homeassistant/components/sonos/sensor.py +++ b/homeassistant/components/sonos/sensor.py @@ -100,7 +100,7 @@ class SonosBatteryEntity(SonosEntity, SensorEntity): @property def available(self) -> bool: """Return whether this device is available.""" - return self.speaker.available and self.speaker.power_source + return self.speaker.available and self.speaker.power_source is not None class SonosAudioInputFormatSensorEntity(SonosPollingEntity, SensorEntity): diff --git a/homeassistant/components/sonos/speaker.py b/homeassistant/components/sonos/speaker.py index 0c5bec06dfb..516a431295a 100644 --- a/homeassistant/components/sonos/speaker.py +++ b/homeassistant/components/sonos/speaker.py @@ -8,7 +8,7 @@ import datetime from functools import partial import logging import time -from typing import Any +from typing import Any, cast import async_timeout import defusedxml.ElementTree as ET @@ -97,17 +97,17 @@ class SonosSpeaker: self.media = SonosMedia(hass, soco) self._plex_plugin: PlexPlugin | None = None self._share_link_plugin: ShareLinkPlugin | None = None - self.available = True + self.available: bool = True # Device information - self.hardware_version = speaker_info["hardware_version"] - self.software_version = speaker_info["software_version"] - self.mac_address = speaker_info["mac_address"] - self.model_name = speaker_info["model_name"] - self.model_number = speaker_info["model_number"] - self.uid = speaker_info["uid"] - self.version = speaker_info["display_version"] - self.zone_name = speaker_info["zone_name"] + self.hardware_version: str = speaker_info["hardware_version"] + self.software_version: str = speaker_info["software_version"] + self.mac_address: str = speaker_info["mac_address"] + self.model_name: str = speaker_info["model_name"] + self.model_number: str = speaker_info["model_number"] + self.uid: str = speaker_info["uid"] + self.version: str = speaker_info["display_version"] + self.zone_name: str = speaker_info["zone_name"] # Subscriptions and events self.subscriptions_failed: bool = False @@ -160,12 +160,12 @@ class SonosSpeaker: self.sonos_group: list[SonosSpeaker] = [self] self.sonos_group_entities: list[str] = [] self.soco_snapshot: Snapshot | None = None - self.snapshot_group: list[SonosSpeaker] | None = None + self.snapshot_group: list[SonosSpeaker] = [] self._group_members_missing: set[str] = set() async def async_setup_dispatchers(self, entry: ConfigEntry) -> None: """Connect dispatchers in async context during setup.""" - dispatch_pairs = ( + dispatch_pairs: tuple[tuple[str, Callable[..., Any]], ...] = ( (SONOS_CHECK_ACTIVITY, self.async_check_activity), (SONOS_SPEAKER_ADDED, self.update_group_for_uid), (f"{SONOS_REBOOTED}-{self.soco.uid}", self.async_rebooted), @@ -283,18 +283,17 @@ class SonosSpeaker: return self._share_link_plugin @property - def subscription_address(self) -> str | None: - """Return the current subscription callback address if any.""" - if self._subscriptions: - addr, port = self._subscriptions[0].event_listener.address - return ":".join([addr, str(port)]) - return None + def subscription_address(self) -> str: + """Return the current subscription callback address.""" + assert len(self._subscriptions) > 0 + addr, port = self._subscriptions[0].event_listener.address + return ":".join([addr, str(port)]) # # Subscription handling and event dispatchers # def log_subscription_result( - self, result: Any, event: str, level: str = logging.DEBUG + self, result: Any, event: str, level: int = logging.DEBUG ) -> None: """Log a message if a subscription action (create/renew/stop) results in an exception.""" if not isinstance(result, Exception): @@ -304,7 +303,7 @@ class SonosSpeaker: message = "Request timed out" exc_info = None else: - message = result + message = str(result) exc_info = result if not str(result) else None _LOGGER.log( @@ -554,7 +553,7 @@ class SonosSpeaker: ) @callback - def speaker_activity(self, source): + def speaker_activity(self, source: str) -> None: """Track the last activity on this speaker, set availability and resubscribe.""" if self._resub_cooldown_expires_at: if time.monotonic() < self._resub_cooldown_expires_at: @@ -593,6 +592,7 @@ class SonosSpeaker: async def async_offline(self) -> None: """Handle removal of speaker when unavailable.""" + assert self._subscription_lock is not None async with self._subscription_lock: await self._async_offline() @@ -826,8 +826,8 @@ class SonosSpeaker: if speaker: self._group_members_missing.discard(uid) sonos_group.append(speaker) - entity_id = entity_registry.async_get_entity_id( - MP_DOMAIN, DOMAIN, uid + entity_id = cast( + str, entity_registry.async_get_entity_id(MP_DOMAIN, DOMAIN, uid) ) sonos_group_entities.append(entity_id) else: @@ -850,7 +850,9 @@ class SonosSpeaker: self.async_write_entity_states() for joined_uid in group[1:]: - joined_speaker = self.hass.data[DATA_SONOS].discovered.get(joined_uid) + joined_speaker: SonosSpeaker = self.hass.data[ + DATA_SONOS + ].discovered.get(joined_uid) if joined_speaker: joined_speaker.coordinator = self joined_speaker.sonos_group = sonos_group @@ -936,7 +938,7 @@ class SonosSpeaker: if with_group: self.snapshot_group = self.sonos_group.copy() else: - self.snapshot_group = None + self.snapshot_group = [] @staticmethod async def snapshot_multi( @@ -969,7 +971,7 @@ class SonosSpeaker: _LOGGER.warning("Error on restore %s: %s", self.zone_name, ex) self.soco_snapshot = None - self.snapshot_group = None + self.snapshot_group = [] @staticmethod async def restore_multi( @@ -996,7 +998,7 @@ class SonosSpeaker: exc_info=exc, ) - groups = [] + groups: list[list[SonosSpeaker]] = [] if not with_group: return groups @@ -1022,7 +1024,7 @@ class SonosSpeaker: # Bring back the original group topology for speaker in (s for s in speakers if s.snapshot_group): - assert speaker.snapshot_group is not None + assert len(speaker.snapshot_group) if speaker.snapshot_group[0] == speaker: if speaker.snapshot_group not in (speaker.sonos_group, [speaker]): speaker.join(speaker.snapshot_group) @@ -1047,7 +1049,7 @@ class SonosSpeaker: if with_group: for speaker in [s for s in speakers_set if s.snapshot_group]: - assert speaker.snapshot_group is not None + assert len(speaker.snapshot_group) speakers_set.update(speaker.snapshot_group) async with hass.data[DATA_SONOS].topology_condition: diff --git a/homeassistant/components/sonos/statistics.py b/homeassistant/components/sonos/statistics.py index a850e5a8caf..b761469aea5 100644 --- a/homeassistant/components/sonos/statistics.py +++ b/homeassistant/components/sonos/statistics.py @@ -14,7 +14,7 @@ class SonosStatistics: def __init__(self, zone_name: str, kind: str) -> None: """Initialize SonosStatistics.""" - self._stats = {} + self._stats: dict[str, dict[str, int | float]] = {} self._stat_type = kind self.zone_name = zone_name diff --git a/homeassistant/components/sonos/switch.py b/homeassistant/components/sonos/switch.py index a348b40cb0f..acf33ea34aa 100644 --- a/homeassistant/components/sonos/switch.py +++ b/homeassistant/components/sonos/switch.py @@ -3,8 +3,9 @@ from __future__ import annotations import datetime import logging -from typing import Any +from typing import Any, cast +from soco.alarms import Alarm from soco.exceptions import SoCoSlaveException, SoCoUPnPException from homeassistant.components.switch import ENTITY_ID_FORMAT, SwitchEntity @@ -183,14 +184,14 @@ class SonosSwitchEntity(SonosPollingEntity, SwitchEntity): def is_on(self) -> bool: """Return True if entity is on.""" if self.needs_coordinator and not self.speaker.is_coordinator: - return getattr(self.speaker.coordinator, self.feature_type) - return getattr(self.speaker, self.feature_type) + return cast(bool, getattr(self.speaker.coordinator, self.feature_type)) + return cast(bool, getattr(self.speaker, self.feature_type)) - def turn_on(self, **kwargs) -> None: + def turn_on(self, **kwargs: Any) -> None: """Turn the entity on.""" self.send_command(True) - def turn_off(self, **kwargs) -> None: + def turn_off(self, **kwargs: Any) -> None: """Turn the entity off.""" self.send_command(False) @@ -233,7 +234,7 @@ class SonosAlarmEntity(SonosEntity, SwitchEntity): ) @property - def alarm(self): + def alarm(self) -> Alarm: """Return the alarm instance.""" return self.hass.data[DATA_SONOS].alarms[self.household_id].get(self.alarm_id) @@ -247,7 +248,7 @@ class SonosAlarmEntity(SonosEntity, SwitchEntity): await self.hass.data[DATA_SONOS].alarms[self.household_id].async_poll() @callback - def async_check_if_available(self): + def async_check_if_available(self) -> bool: """Check if alarm exists and remove alarm entity if not available.""" if self.alarm: return True @@ -279,7 +280,7 @@ class SonosAlarmEntity(SonosEntity, SwitchEntity): self.async_write_ha_state() @callback - def _async_update_device(self): + def _async_update_device(self) -> None: """Update the device, since this alarm moved to a different player.""" device_registry = dr.async_get(self.hass) entity_registry = er.async_get(self.hass) @@ -288,22 +289,20 @@ class SonosAlarmEntity(SonosEntity, SwitchEntity): if entity is None: raise RuntimeError("Alarm has been deleted by accident.") - entry_id = entity.config_entry_id - new_device = device_registry.async_get_or_create( - config_entry_id=entry_id, + config_entry_id=cast(str, entity.config_entry_id), identifiers={(SONOS_DOMAIN, self.soco.uid)}, connections={(dr.CONNECTION_NETWORK_MAC, self.speaker.mac_address)}, ) - if not entity_registry.async_get(self.entity_id).device_id == new_device.id: + if ( + device := entity_registry.async_get(self.entity_id) + ) and device.device_id != new_device.id: _LOGGER.debug("%s is moving to %s", self.entity_id, new_device.name) - # pylint: disable=protected-access - entity_registry._async_update_entity( - self.entity_id, device_id=new_device.id - ) + entity_registry.async_update_entity(self.entity_id, device_id=new_device.id) @property - def _is_today(self): + def _is_today(self) -> bool: + """Return whether this alarm is scheduled for today.""" recurrence = self.alarm.recurrence timestr = int(datetime.datetime.today().strftime("%w")) return ( @@ -321,12 +320,12 @@ class SonosAlarmEntity(SonosEntity, SwitchEntity): return (self.alarm is not None) and self.speaker.available @property - def is_on(self): + def is_on(self) -> bool: """Return state of Sonos alarm switch.""" return self.alarm.enabled @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> dict[str, Any]: """Return attributes of Sonos alarm switch.""" return { ATTR_ID: str(self.alarm_id), diff --git a/mypy.ini b/mypy.ini index d6665cb40c8..957da7254eb 100644 --- a/mypy.ini +++ b/mypy.ini @@ -2589,39 +2589,3 @@ disallow_untyped_decorators = false disallow_untyped_defs = false warn_return_any = false warn_unreachable = false - -[mypy-homeassistant.components.sonos] -ignore_errors = true - -[mypy-homeassistant.components.sonos.alarms] -ignore_errors = true - -[mypy-homeassistant.components.sonos.binary_sensor] -ignore_errors = true - -[mypy-homeassistant.components.sonos.diagnostics] -ignore_errors = true - -[mypy-homeassistant.components.sonos.entity] -ignore_errors = true - -[mypy-homeassistant.components.sonos.favorites] -ignore_errors = true - -[mypy-homeassistant.components.sonos.media_browser] -ignore_errors = true - -[mypy-homeassistant.components.sonos.media_player] -ignore_errors = true - -[mypy-homeassistant.components.sonos.number] -ignore_errors = true - -[mypy-homeassistant.components.sonos.sensor] -ignore_errors = true - -[mypy-homeassistant.components.sonos.speaker] -ignore_errors = true - -[mypy-homeassistant.components.sonos.statistics] -ignore_errors = true diff --git a/script/hassfest/mypy_config.py b/script/hassfest/mypy_config.py index b6c31751e12..0c598df9cd1 100644 --- a/script/hassfest/mypy_config.py +++ b/script/hassfest/mypy_config.py @@ -15,20 +15,7 @@ from .model import Config, Integration # If you are an author of component listed here, please fix these errors and # remove your component from this list to enable type checks. # Do your best to not add anything new here. -IGNORED_MODULES: Final[list[str]] = [ - "homeassistant.components.sonos", - "homeassistant.components.sonos.alarms", - "homeassistant.components.sonos.binary_sensor", - "homeassistant.components.sonos.diagnostics", - "homeassistant.components.sonos.entity", - "homeassistant.components.sonos.favorites", - "homeassistant.components.sonos.media_browser", - "homeassistant.components.sonos.media_player", - "homeassistant.components.sonos.number", - "homeassistant.components.sonos.sensor", - "homeassistant.components.sonos.speaker", - "homeassistant.components.sonos.statistics", -] +IGNORED_MODULES: Final[list[str]] = [] # Component modules which should set no_implicit_reexport = true. NO_IMPLICIT_REEXPORT_MODULES: set[str] = {