Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make Sonos typing more complete (#68072)

pull/77968/head
Robert Hillis 2022-09-05 14:12:37 -04:00 committed by Paulus Schoutsen
parent 0d042d496d
commit 2fa517b81b
14 changed files with 168 additions and 193 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
homeassistant/components/sonos/__init__.py
View File

@ -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(

2
homeassistant/components/sonos/binary_sensor.py
View File

@ -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

15
homeassistant/components/sonos/diagnostics.py
View File

@ -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):

3
homeassistant/components/sonos/household_coordinator.py
View File

@ -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."""

5
homeassistant/components/sonos/media.py
View File

@ -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

59
homeassistant/components/sonos/media_browser.py
View File

@ -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)

76
homeassistant/components/sonos/media_player.py
View File

@ -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()

11
homeassistant/components/sonos/number.py
View File

@ -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))

2
homeassistant/components/sonos/sensor.py
View File

@ -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):

60
homeassistant/components/sonos/speaker.py
View File

@ -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:

2
homeassistant/components/sonos/statistics.py
View File

@ -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

37
homeassistant/components/sonos/switch.py
View File

@ -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),

36
mypy.ini
View File

@ -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

15
script/hassfest/mypy_config.py
View File

@ -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] = {