jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 19:27:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make Sonos typing more complete (#68072)

Browse Source
This commit is contained in:
Robert Hillis 2022-09-05 14:12:37 -04:00 committed by GitHub
parent 6fbc0a8103
commit 73ba7a989b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
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 from functools import partial
import logging import logging
import socket import socket
from typing import TYPE_CHECKING, Any, Optional, cast
from urllib.parse import urlparse from urllib.parse import urlparse
from soco import events_asyncio 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.components.media_player import DOMAIN as MP_DOMAIN
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOSTS, EVENT_HOMEASSISTANT_STOP 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 import config_validation as cv, device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_send, dispatcher_send from homeassistant.helpers.dispatcher import async_dispatcher_send, dispatcher_send
from homeassistant.helpers.event import async_track_time_interval, call_later from homeassistant.helpers.event import async_track_time_interval, call_later
@ -93,7 +94,7 @@ class SonosData:
self.favorites: dict[str, SonosFavorites] = {} self.favorites: dict[str, SonosFavorites] = {}
self.alarms: dict[str, SonosAlarms] = {} self.alarms: dict[str, SonosAlarms] = {}
self.topology_condition = asyncio.Condition() self.topology_condition = asyncio.Condition()
self.hosts_heartbeat = None self.hosts_heartbeat: CALLBACK_TYPE | None = None
self.discovery_known: set[str] = set() self.discovery_known: set[str] = set()
self.boot_counts: dict[str, int] = {} self.boot_counts: dict[str, int] = {}
self.mdns_names: dict[str, str] = {} self.mdns_names: dict[str, str] = {}
@ -168,10 +169,10 @@ class SonosDiscoveryManager:
self.data = data self.data = data
self.hosts = set(hosts) self.hosts = set(hosts)
self.discovery_lock = asyncio.Lock() self.discovery_lock = asyncio.Lock()
self._known_invisible = set() self._known_invisible: set[SoCo] = set()
self._manual_config_required = bool(hosts) self._manual_config_required = bool(hosts)
async def async_shutdown(self): async def async_shutdown(self) -> None:
"""Stop all running tasks.""" """Stop all running tasks."""
await self._async_stop_event_listener() await self._async_stop_event_listener()
self._stop_manual_heartbeat() self._stop_manual_heartbeat()
@ -236,6 +237,8 @@ class SonosDiscoveryManager:
(SonosAlarms, self.data.alarms), (SonosAlarms, self.data.alarms),
(SonosFavorites, self.data.favorites), (SonosFavorites, self.data.favorites),
): ):
if TYPE_CHECKING:
coord_dict = cast(dict[str, Any], coord_dict)
if soco.household_id not in coord_dict: if soco.household_id not in coord_dict:
new_coordinator = coordinator(self.hass, soco.household_id) new_coordinator = coordinator(self.hass, soco.household_id)
new_coordinator.setup(soco) new_coordinator.setup(soco)
@ -298,7 +301,7 @@ class SonosDiscoveryManager:
) )
async def _async_handle_discovery_message( 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: ) -> None:
"""Handle discovered player creation and activity.""" """Handle discovered player creation and activity."""
async with self.discovery_lock: async with self.discovery_lock:
@ -338,22 +341,27 @@ class SonosDiscoveryManager:
async_dispatcher_send(self.hass, f"{SONOS_VANISHED}-{uid}", reason) async_dispatcher_send(self.hass, f"{SONOS_VANISHED}-{uid}", reason)
return return
discovered_ip = urlparse(info.ssdp_location).hostname
boot_seqnum = info.ssdp_headers.get("X-RINCON-BOOTSEQ")
self.async_discovered_player( self.async_discovered_player(
"SSDP", "SSDP",
info, info,
discovered_ip, cast(str, urlparse(info.ssdp_location).hostname),
uid, uid,
boot_seqnum, info.ssdp_headers.get("X-RINCON-BOOTSEQ"),
info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME), cast(str, info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME)),
None, None,
) )
@callback @callback
def async_discovered_player( 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.""" """Handle discovery via ssdp or zeroconf."""
if self._manual_config_required: if self._manual_config_required:
_LOGGER.warning( _LOGGER.warning(
@ -376,10 +384,12 @@ class SonosDiscoveryManager:
_LOGGER.debug("New %s discovery uid=%s: %s", source, uid, info) _LOGGER.debug("New %s discovery uid=%s: %s", source, uid, info)
self.data.discovery_known.add(uid) self.data.discovery_known.add(uid)
asyncio.create_task( 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.""" """Set up platforms and discovery."""
await self.hass.config_entries.async_forward_entry_setups(self.entry, PLATFORMS) await self.hass.config_entries.async_forward_entry_setups(self.entry, PLATFORMS)
self.entry.async_on_unload( 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 self.speaker.mic_enabled = self.soco.mic_enabled
@property @property
def is_on(self) -> bool: def is_on(self) -> bool | None:
"""Return the state of the binary sensor.""" """Return the state of the binary sensor."""
return self.speaker.mic_enabled 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 hass: HomeAssistant, config_entry: ConfigEntry
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Return diagnostics for a config entry.""" """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"): for section in ("discovered", "discovery_known"):
payload[section] = {} 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): if isinstance(data, set):
payload[section] = data payload[section] = data
continue continue
@ -60,7 +60,6 @@ async def async_get_config_entry_diagnostics(
payload[section][key] = await async_generate_speaker_info(hass, value) payload[section][key] = await async_generate_speaker_info(hass, value)
else: else:
payload[section][key] = value payload[section][key] = value
return payload return payload
@ -85,12 +84,12 @@ async def async_generate_media_info(
hass: HomeAssistant, speaker: SonosSpeaker hass: HomeAssistant, speaker: SonosSpeaker
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Generate a diagnostic payload for current media metadata.""" """Generate a diagnostic payload for current media metadata."""
payload = {} payload: dict[str, Any] = {}
for attrib in MEDIA_DIAGNOSTIC_ATTRIBUTES: for attrib in MEDIA_DIAGNOSTIC_ATTRIBUTES:
payload[attrib] = getattr(speaker.media, attrib) payload[attrib] = getattr(speaker.media, attrib)
def poll_current_track_info(): def poll_current_track_info() -> dict[str, Any] | str:
try: try:
return speaker.soco.avTransport.GetPositionInfo( return speaker.soco.avTransport.GetPositionInfo(
[("InstanceID", 0), ("Channel", "Master")], [("InstanceID", 0), ("Channel", "Master")],
@ -110,9 +109,11 @@ async def async_generate_speaker_info(
hass: HomeAssistant, speaker: SonosSpeaker hass: HomeAssistant, speaker: SonosSpeaker
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Generate the diagnostic payload for a specific speaker.""" """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)): if isinstance(item, (int, float, str)):
return item return item
if isinstance(item, dict): if isinstance(item, dict):

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

@ -20,13 +20,14 @@ _LOGGER = logging.getLogger(__name__)
class SonosHouseholdCoordinator: class SonosHouseholdCoordinator:
"""Base class for Sonos household-level storage.""" """Base class for Sonos household-level storage."""
cache_update_lock: asyncio.Lock
def __init__(self, hass: HomeAssistant, household_id: str) -> None: def __init__(self, hass: HomeAssistant, household_id: str) -> None:
"""Initialize the data.""" """Initialize the data."""
self.hass = hass self.hass = hass
self.household_id = household_id self.household_id = household_id
self.async_poll: Callable[[], Coroutine[None, None, None]] | None = None self.async_poll: Callable[[], Coroutine[None, None, None]] | None = None
self.last_processed_event_id: int | None = None self.last_processed_event_id: int | None = None
self.cache_update_lock: asyncio.Lock | None = None
def setup(self, soco: SoCo) -> None: def setup(self, soco: SoCo) -> None:
"""Set up the SonosAlarm instance.""" """Set up the SonosAlarm instance."""

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

@ -2,7 +2,6 @@
from __future__ import annotations from __future__ import annotations
import datetime import datetime
import logging
from typing import Any from typing import Any
from soco.core import ( from soco.core import (
@ -43,8 +42,6 @@ UNAVAILABLE_VALUES = {"", "NOT_IMPLEMENTED", None}
DURATION_SECONDS = "duration_in_s" DURATION_SECONDS = "duration_in_s"
POSITION_SECONDS = "position_in_s" POSITION_SECONDS = "position_in_s"
_LOGGER = logging.getLogger(__name__)
def _timespan_secs(timespan: str | None) -> None | float: def _timespan_secs(timespan: str | None) -> None | float:
"""Parse a time-span into number of seconds.""" """Parse a time-span into number of seconds."""
@ -106,7 +103,7 @@ class SonosMedia:
@soco_error() @soco_error()
def poll_track_info(self) -> dict[str, Any]: def poll_track_info(self) -> dict[str, Any]:
"""Poll the speaker for current track info, add converted position values, and return.""" """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[DURATION_SECONDS] = _timespan_secs(track_info.get("duration"))
track_info[POSITION_SECONDS] = _timespan_secs(track_info.get("position")) track_info[POSITION_SECONDS] = _timespan_secs(track_info.get("position"))
return track_info return track_info

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

@ -5,8 +5,13 @@ from collections.abc import Callable
from contextlib import suppress from contextlib import suppress
from functools import partial from functools import partial
import logging import logging
from typing import cast
from urllib.parse import quote_plus, unquote 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 import media_source, plex, spotify
from homeassistant.components.media_player import BrowseMedia from homeassistant.components.media_player import BrowseMedia
from homeassistant.components.media_player.const import ( from homeassistant.components.media_player.const import (
@ -50,12 +55,12 @@ def get_thumbnail_url_full(
) -> str | None: ) -> str | None:
"""Get thumbnail URL.""" """Get thumbnail URL."""
if is_internal: if is_internal:
item = get_media( # type: ignore[no-untyped-call] item = get_media(
media.library, media.library,
media_content_id, media_content_id,
media_content_type, 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( return get_browse_image_url(
media_content_type, 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.""" """Filter media sources."""
return item.media_content_type.startswith("audio/") return item.media_content_type.startswith("audio/")
async def async_browse_media( async def async_browse_media(
hass, hass: HomeAssistant,
speaker: SonosSpeaker, speaker: SonosSpeaker,
media: SonosMedia, media: SonosMedia,
get_browse_image_url: GetBrowseImageUrlType, get_browse_image_url: GetBrowseImageUrlType,
media_content_id: str | None, media_content_id: str | None,
media_content_type: str | None, media_content_type: str | None,
): ) -> BrowseMedia:
"""Browse media.""" """Browse media."""
if media_content_id is None: if media_content_id is None:
@ -86,6 +91,7 @@ async def async_browse_media(
media, media,
get_browse_image_url, get_browse_image_url,
) )
assert media_content_type is not None
if media_source.is_media_source_id(media_content_id): if media_source.is_media_source_id(media_content_id):
return await media_source.async_browse_media( return await media_source.async_browse_media(
@ -150,7 +156,9 @@ async def async_browse_media(
return response 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.""" """Create response payload for the provided media query."""
if payload["search_type"] == MEDIA_TYPE_ALBUM and payload["idstring"].startswith( if payload["search_type"] == MEDIA_TYPE_ALBUM and payload["idstring"].startswith(
("A:GENRE", "A:COMPOSER") ("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", "Unknown media type received when building item response: %s",
payload["search_type"], payload["search_type"],
) )
return return None
media = media_library.browse_by_idstring( media = media_library.browse_by_idstring(
search_type, search_type,
@ -176,7 +184,7 @@ def build_item_response(media_library, payload, get_thumbnail_url=None):
) )
if media is None: if media is None:
return return None
thumbnail = None thumbnail = None
title = 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. Create response payload for a single media item.
@ -256,9 +264,9 @@ async def root_payload(
speaker: SonosSpeaker, speaker: SonosSpeaker,
media: SonosMedia, media: SonosMedia,
get_browse_image_url: GetBrowseImageUrlType, get_browse_image_url: GetBrowseImageUrlType,
): ) -> BrowseMedia:
"""Return root payload for Sonos.""" """Return root payload for Sonos."""
children = [] children: list[BrowseMedia] = []
if speaker.favorites: if speaker.favorites:
children.append( children.append(
@ -303,6 +311,7 @@ async def root_payload(
if "spotify" in hass.config.components: if "spotify" in hass.config.components:
result = await spotify.async_browse_media(hass, None, None) result = await spotify.async_browse_media(hass, None, None)
if result.children:
children.extend(result.children) children.extend(result.children)
try: try:
@ -310,7 +319,7 @@ async def root_payload(
hass, None, content_filter=media_source_filter hass, None, content_filter=media_source_filter
) )
# If domain is None, it's overview of available sources # 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) children.extend(item.children)
else: else:
children.append(item) 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. 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. 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. """Create response payload to describe all items of a type of favorite.
Used by async_browse_media. 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.""" """Extract media type of item."""
if item.item_class == "object.item.audioItem.musicTrack": if item.item_class == "object.item.audioItem.musicTrack":
return SONOS_TRACKS 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) 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. Test if playable.
@ -459,7 +470,7 @@ def can_play(item):
return SONOS_TO_MEDIA_TYPES.get(item) in PLAYABLE_MEDIA_TYPES return SONOS_TO_MEDIA_TYPES.get(item) in PLAYABLE_MEDIA_TYPES
def can_expand(item): def can_expand(item: DidlObject) -> bool:
""" """
Test if expandable. Test if expandable.
@ -474,14 +485,16 @@ def can_expand(item):
return SONOS_TYPES_MAPPING.get(item.item_id) in EXPANDABLE_MEDIA_TYPES 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.""" """Extract content id or uri."""
if item.item_class == "object.item.audioItem.musicTrack": if item.item_class == "object.item.audioItem.musicTrack":
return item.get_uri() return cast(str, item.get_uri())
return item.item_id 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.""" """Fetch media/album."""
search_type = MEDIA_TYPES_TO_SONOS.get(search_type, search_type) 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: if service_call.service == SERVICE_SNAPSHOT:
await SonosSpeaker.snapshot_multi( 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: elif service_call.service == SERVICE_RESTORE:
await SonosSpeaker.restore_multi( 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( config_entry.async_on_unload(
@ -153,7 +153,7 @@ async def async_setup_entry(
SONOS_DOMAIN, SERVICE_RESTORE, async_service_handle, join_unjoin_schema 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, SERVICE_SET_TIMER,
{ {
vol.Required(ATTR_SLEEP_TIME): vol.All( vol.Required(ATTR_SLEEP_TIME): vol.All(
@ -163,9 +163,9 @@ async def async_setup_entry(
"set_sleep_timer", "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, SERVICE_UPDATE_ALARM,
{ {
vol.Required(ATTR_ALARM_ID): cv.positive_int, vol.Required(ATTR_ALARM_ID): cv.positive_int,
@ -177,13 +177,13 @@ async def async_setup_entry(
"set_alarm", "set_alarm",
) )
platform.async_register_entity_service( # type: ignore platform.async_register_entity_service(
SERVICE_PLAY_QUEUE, SERVICE_PLAY_QUEUE,
{vol.Optional(ATTR_QUEUE_POSITION): cv.positive_int}, {vol.Optional(ATTR_QUEUE_POSITION): cv.positive_int},
"play_queue", "play_queue",
) )
platform.async_register_entity_service( # type: ignore platform.async_register_entity_service(
SERVICE_REMOVE_FROM_QUEUE, SERVICE_REMOVE_FROM_QUEUE,
{vol.Optional(ATTR_QUEUE_POSITION): cv.positive_int}, {vol.Optional(ATTR_QUEUE_POSITION): cv.positive_int},
"remove_from_queue", "remove_from_queue",
@ -239,8 +239,8 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
"""Return if the media_player is available.""" """Return if the media_player is available."""
return ( return (
self.speaker.available self.speaker.available
and self.speaker.sonos_group_entities and bool(self.speaker.sonos_group_entities)
and self.media.playback_status and self.media.playback_status is not None
) )
@property @property
@ -257,7 +257,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
"""Return a hash of self.""" """Return a hash of self."""
return hash(self.unique_id) return hash(self.unique_id)
@property # type: ignore[misc] @property
def state(self) -> str: def state(self) -> str:
"""Return the state of the entity.""" """Return the state of the entity."""
if self.media.playback_status in ( if self.media.playback_status in (
@ -300,13 +300,12 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
"""Return true if volume is muted.""" """Return true if volume is muted."""
return self.speaker.muted return self.speaker.muted
@property # type: ignore[misc] @property
def shuffle(self) -> str | None: def shuffle(self) -> bool | None:
"""Shuffling state.""" """Shuffling state."""
shuffle: str = PLAY_MODES[self.media.play_mode][0] return PLAY_MODES[self.media.play_mode][0]
return shuffle
@property # type: ignore[misc] @property
def repeat(self) -> str | None: def repeat(self) -> str | None:
"""Return current repeat mode.""" """Return current repeat mode."""
sonos_repeat = PLAY_MODES[self.media.play_mode][1] 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 the SonosMedia object from the coordinator speaker."""
return self.coordinator.media return self.coordinator.media
@property # type: ignore[misc] @property
def media_content_id(self) -> str | None: def media_content_id(self) -> str | None:
"""Content id of current playing media.""" """Content id of current playing media."""
return self.media.uri return self.media.uri
@property # type: ignore[misc] @property
def media_duration(self) -> float | None: def media_duration(self) -> int | None:
"""Duration of current playing media in seconds.""" """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] @property
def media_position(self) -> float | None: def media_position(self) -> int | None:
"""Position of current playing media in seconds.""" """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: def media_position_updated_at(self) -> datetime.datetime | None:
"""When was the position of the current playing media valid.""" """When was the position of the current playing media valid."""
return self.media.position_updated_at return self.media.position_updated_at
@property # type: ignore[misc] @property
def media_image_url(self) -> str | None: def media_image_url(self) -> str | None:
"""Image url of current playing media.""" """Image url of current playing media."""
return self.media.image_url or None return self.media.image_url or None
@property # type: ignore[misc] @property
def media_channel(self) -> str | None: def media_channel(self) -> str | None:
"""Channel currently playing.""" """Channel currently playing."""
return self.media.channel or None return self.media.channel or None
@ -352,22 +351,22 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
"""Title of playlist currently playing.""" """Title of playlist currently playing."""
return self.media.playlist_name return self.media.playlist_name
@property # type: ignore[misc] @property
def media_artist(self) -> str | None: def media_artist(self) -> str | None:
"""Artist of current playing media, music track only.""" """Artist of current playing media, music track only."""
return self.media.artist or None return self.media.artist or None
@property # type: ignore[misc] @property
def media_album_name(self) -> str | None: def media_album_name(self) -> str | None:
"""Album name of current playing media, music track only.""" """Album name of current playing media, music track only."""
return self.media.album_name or None return self.media.album_name or None
@property # type: ignore[misc] @property
def media_title(self) -> str | None: def media_title(self) -> str | None:
"""Title of current playing media.""" """Title of current playing media."""
return self.media.title or None return self.media.title or None
@property # type: ignore[misc] @property
def source(self) -> str | None: def source(self) -> str | None:
"""Name of the current input source.""" """Name of the current input source."""
return self.media.source_name or None return self.media.source_name or None
@ -383,12 +382,12 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
self.soco.volume -= VOLUME_INCREMENT self.soco.volume -= VOLUME_INCREMENT
@soco_error() @soco_error()
def set_volume_level(self, volume: str) -> None: def set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1.""" """Set volume level, range 0..1."""
self.soco.volume = str(int(volume * 100)) self.soco.volume = str(int(volume * 100))
@soco_error(UPNP_ERRORS_TO_IGNORE) @soco_error(UPNP_ERRORS_TO_IGNORE)
def set_shuffle(self, shuffle: str) -> None: def set_shuffle(self, shuffle: bool) -> None:
"""Enable/Disable shuffle mode.""" """Enable/Disable shuffle mode."""
sonos_shuffle = shuffle sonos_shuffle = shuffle
sonos_repeat = PLAY_MODES[self.media.play_mode][1] sonos_repeat = PLAY_MODES[self.media.play_mode][1]
@ -486,7 +485,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
self.coordinator.soco.previous() self.coordinator.soco.previous()
@soco_error(UPNP_ERRORS_TO_IGNORE) @soco_error(UPNP_ERRORS_TO_IGNORE)
def media_seek(self, position: str) -> None: def media_seek(self, position: float) -> None:
"""Send seek command.""" """Send seek command."""
self.coordinator.soco.seek(str(datetime.timedelta(seconds=int(position)))) 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) soco.play_uri(media_id, force_radio=is_radio)
elif media_type == MEDIA_TYPE_PLAYLIST: elif media_type == MEDIA_TYPE_PLAYLIST:
if media_id.startswith("S:"): 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()) soco.play_uri(item.get_uri())
return return
try: try:
@ -619,7 +618,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
soco.add_to_queue(playlist) soco.add_to_queue(playlist)
soco.play_from_queue(0) soco.play_from_queue(0)
elif media_type in PLAYABLE_MEDIA_TYPES: 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: if not item:
_LOGGER.error('Could not find "%s" in the library', media_id) _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, include_linked_zones: bool | None = None,
) -> None: ) -> None:
"""Set the alarm clock on the player.""" """Set the alarm clock on the player."""
alarm = None alarm: alarms.Alarm | None = None
for one_alarm in alarms.get_alarms(self.coordinator.soco): for one_alarm in alarms.get_alarms(self.coordinator.soco):
if one_alarm.alarm_id == str(alarm_id): if one_alarm.alarm_id == str(alarm_id):
alarm = one_alarm alarm = one_alarm
@ -710,8 +709,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
MEDIA_TYPES_TO_SONOS[media_content_type], MEDIA_TYPES_TO_SONOS[media_content_type],
) )
if image_url := getattr(item, "album_art_uri", None): if image_url := getattr(item, "album_art_uri", None):
result = await self._async_fetch_image(image_url) # type: ignore[no-untyped-call] return await self._async_fetch_image(image_url)
return result # type: ignore
return (None, None) return (None, None)
@ -728,7 +726,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
media_content_type, 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.""" """Join `group_members` as a player group with the current player."""
speakers = [] speakers = []
for entity_id in group_members: for entity_id in group_members:
@ -739,7 +737,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
await SonosSpeaker.join_multi(self.hass, self.speaker, speakers) 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. """Remove this player from any group.
Coalesces all calls within UNJOIN_SERVICE_TIMEOUT to allow use of SonosSpeaker.unjoin_multi() 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 from __future__ import annotations
import logging import logging
from typing import cast
from homeassistant.components.number import NumberEntity from homeassistant.components.number import NumberEntity
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
@ -24,6 +25,8 @@ LEVEL_TYPES = {
"music_surround_level": (-15, 15), "music_surround_level": (-15, 15),
} }
SocoFeatures = list[tuple[str, tuple[int, int]]]
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -34,8 +37,8 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up the Sonos number platform from a config entry.""" """Set up the Sonos number platform from a config entry."""
def available_soco_attributes(speaker: SonosSpeaker) -> list[str]: def available_soco_attributes(speaker: SonosSpeaker) -> SocoFeatures:
features = [] features: SocoFeatures = []
for level_type, valid_range in LEVEL_TYPES.items(): for level_type, valid_range in LEVEL_TYPES.items():
if (state := getattr(speaker.soco, level_type, None)) is not None: if (state := getattr(speaker.soco, level_type, None)) is not None:
setattr(speaker, level_type, state) setattr(speaker, level_type, state)
@ -67,7 +70,7 @@ class SonosLevelEntity(SonosEntity, NumberEntity):
_attr_entity_category = EntityCategory.CONFIG _attr_entity_category = EntityCategory.CONFIG
def __init__( def __init__(
self, speaker: SonosSpeaker, level_type: str, valid_range: tuple[int] self, speaker: SonosSpeaker, level_type: str, valid_range: tuple[int, int]
) -> None: ) -> None:
"""Initialize the level entity.""" """Initialize the level entity."""
super().__init__(speaker) super().__init__(speaker)
@ -94,4 +97,4 @@ class SonosLevelEntity(SonosEntity, NumberEntity):
@property @property
def native_value(self) -> float: def native_value(self) -> float:
"""Return the current value.""" """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 @property
def available(self) -> bool: def available(self) -> bool:
"""Return whether this device is available.""" """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): class SonosAudioInputFormatSensorEntity(SonosPollingEntity, SensorEntity):

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

@ -8,7 +8,7 @@ import datetime
from functools import partial from functools import partial
import logging import logging
import time import time
from typing import Any from typing import Any, cast
import async_timeout import async_timeout
import defusedxml.ElementTree as ET import defusedxml.ElementTree as ET
@ -97,17 +97,17 @@ class SonosSpeaker:
self.media = SonosMedia(hass, soco) self.media = SonosMedia(hass, soco)
self._plex_plugin: PlexPlugin | None = None self._plex_plugin: PlexPlugin | None = None
self._share_link_plugin: ShareLinkPlugin | None = None self._share_link_plugin: ShareLinkPlugin | None = None
self.available = True self.available: bool = True
# Device information # Device information
self.hardware_version = speaker_info["hardware_version"] self.hardware_version: str = speaker_info["hardware_version"]
self.software_version = speaker_info["software_version"] self.software_version: str = speaker_info["software_version"]
self.mac_address = speaker_info["mac_address"] self.mac_address: str = speaker_info["mac_address"]
self.model_name = speaker_info["model_name"] self.model_name: str = speaker_info["model_name"]
self.model_number = speaker_info["model_number"] self.model_number: str = speaker_info["model_number"]
self.uid = speaker_info["uid"] self.uid: str = speaker_info["uid"]
self.version = speaker_info["display_version"] self.version: str = speaker_info["display_version"]
self.zone_name = speaker_info["zone_name"] self.zone_name: str = speaker_info["zone_name"]
# Subscriptions and events # Subscriptions and events
self.subscriptions_failed: bool = False self.subscriptions_failed: bool = False
@ -160,12 +160,12 @@ class SonosSpeaker:
self.sonos_group: list[SonosSpeaker] = [self] self.sonos_group: list[SonosSpeaker] = [self]
self.sonos_group_entities: list[str] = [] self.sonos_group_entities: list[str] = []
self.soco_snapshot: Snapshot | None = None 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() self._group_members_missing: set[str] = set()
async def async_setup_dispatchers(self, entry: ConfigEntry) -> None: async def async_setup_dispatchers(self, entry: ConfigEntry) -> None:
"""Connect dispatchers in async context during setup.""" """Connect dispatchers in async context during setup."""
dispatch_pairs = ( dispatch_pairs: tuple[tuple[str, Callable[..., Any]], ...] = (
(SONOS_CHECK_ACTIVITY, self.async_check_activity), (SONOS_CHECK_ACTIVITY, self.async_check_activity),
(SONOS_SPEAKER_ADDED, self.update_group_for_uid), (SONOS_SPEAKER_ADDED, self.update_group_for_uid),
(f"{SONOS_REBOOTED}-{self.soco.uid}", self.async_rebooted), (f"{SONOS_REBOOTED}-{self.soco.uid}", self.async_rebooted),
@ -283,18 +283,17 @@ class SonosSpeaker:
return self._share_link_plugin return self._share_link_plugin
@property @property
def subscription_address(self) -> str | None: def subscription_address(self) -> str:
"""Return the current subscription callback address if any.""" """Return the current subscription callback address."""
if self._subscriptions: assert len(self._subscriptions) > 0
addr, port = self._subscriptions[0].event_listener.address addr, port = self._subscriptions[0].event_listener.address
return ":".join([addr, str(port)]) return ":".join([addr, str(port)])
return None
# #
# Subscription handling and event dispatchers # Subscription handling and event dispatchers
# #
def log_subscription_result( def log_subscription_result(
self, result: Any, event: str, level: str = logging.DEBUG self, result: Any, event: str, level: int = logging.DEBUG
) -> None: ) -> None:
"""Log a message if a subscription action (create/renew/stop) results in an exception.""" """Log a message if a subscription action (create/renew/stop) results in an exception."""
if not isinstance(result, Exception): if not isinstance(result, Exception):
@ -304,7 +303,7 @@ class SonosSpeaker:
message = "Request timed out" message = "Request timed out"
exc_info = None exc_info = None
else: else:
message = result message = str(result)
exc_info = result if not str(result) else None exc_info = result if not str(result) else None
_LOGGER.log( _LOGGER.log(
@ -554,7 +553,7 @@ class SonosSpeaker:
) )
@callback @callback
def speaker_activity(self, source): def speaker_activity(self, source: str) -> None:
"""Track the last activity on this speaker, set availability and resubscribe.""" """Track the last activity on this speaker, set availability and resubscribe."""
if self._resub_cooldown_expires_at: if self._resub_cooldown_expires_at:
if time.monotonic() < 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: async def async_offline(self) -> None:
"""Handle removal of speaker when unavailable.""" """Handle removal of speaker when unavailable."""
assert self._subscription_lock is not None
async with self._subscription_lock: async with self._subscription_lock:
await self._async_offline() await self._async_offline()
@ -826,8 +826,8 @@ class SonosSpeaker:
if speaker: if speaker:
self._group_members_missing.discard(uid) self._group_members_missing.discard(uid)
sonos_group.append(speaker) sonos_group.append(speaker)
entity_id = entity_registry.async_get_entity_id( entity_id = cast(
MP_DOMAIN, DOMAIN, uid str, entity_registry.async_get_entity_id(MP_DOMAIN, DOMAIN, uid)
) )
sonos_group_entities.append(entity_id) sonos_group_entities.append(entity_id)
else: else:
@ -850,7 +850,9 @@ class SonosSpeaker:
self.async_write_entity_states() self.async_write_entity_states()
for joined_uid in group[1:]: 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: if joined_speaker:
joined_speaker.coordinator = self joined_speaker.coordinator = self
joined_speaker.sonos_group = sonos_group joined_speaker.sonos_group = sonos_group
@ -936,7 +938,7 @@ class SonosSpeaker:
if with_group: if with_group:
self.snapshot_group = self.sonos_group.copy() self.snapshot_group = self.sonos_group.copy()
else: else:
self.snapshot_group = None self.snapshot_group = []
@staticmethod @staticmethod
async def snapshot_multi( async def snapshot_multi(
@ -969,7 +971,7 @@ class SonosSpeaker:
_LOGGER.warning("Error on restore %s: %s", self.zone_name, ex) _LOGGER.warning("Error on restore %s: %s", self.zone_name, ex)
self.soco_snapshot = None self.soco_snapshot = None
self.snapshot_group = None self.snapshot_group = []
@staticmethod @staticmethod
async def restore_multi( async def restore_multi(
@ -996,7 +998,7 @@ class SonosSpeaker:
exc_info=exc, exc_info=exc,
) )
groups = [] groups: list[list[SonosSpeaker]] = []
if not with_group: if not with_group:
return groups return groups
@ -1022,7 +1024,7 @@ class SonosSpeaker:
# Bring back the original group topology # Bring back the original group topology
for speaker in (s for s in speakers if s.snapshot_group): 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[0] == speaker:
if speaker.snapshot_group not in (speaker.sonos_group, [speaker]): if speaker.snapshot_group not in (speaker.sonos_group, [speaker]):
speaker.join(speaker.snapshot_group) speaker.join(speaker.snapshot_group)
@ -1047,7 +1049,7 @@ class SonosSpeaker:
if with_group: if with_group:
for speaker in [s for s in speakers_set if s.snapshot_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) speakers_set.update(speaker.snapshot_group)
async with hass.data[DATA_SONOS].topology_condition: 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: def __init__(self, zone_name: str, kind: str) -> None:
"""Initialize SonosStatistics.""" """Initialize SonosStatistics."""
self._stats = {} self._stats: dict[str, dict[str, int | float]] = {}
self._stat_type = kind self._stat_type = kind
self.zone_name = zone_name self.zone_name = zone_name

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

@ -3,8 +3,9 @@ from __future__ import annotations
import datetime import datetime
import logging import logging
from typing import Any from typing import Any, cast
from soco.alarms import Alarm
from soco.exceptions import SoCoSlaveException, SoCoUPnPException from soco.exceptions import SoCoSlaveException, SoCoUPnPException
from homeassistant.components.switch import ENTITY_ID_FORMAT, SwitchEntity from homeassistant.components.switch import ENTITY_ID_FORMAT, SwitchEntity
@ -183,14 +184,14 @@ class SonosSwitchEntity(SonosPollingEntity, SwitchEntity):
def is_on(self) -> bool: def is_on(self) -> bool:
"""Return True if entity is on.""" """Return True if entity is on."""
if self.needs_coordinator and not self.speaker.is_coordinator: if self.needs_coordinator and not self.speaker.is_coordinator:
return getattr(self.speaker.coordinator, self.feature_type) return cast(bool, getattr(self.speaker.coordinator, self.feature_type))
return getattr(self.speaker, 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.""" """Turn the entity on."""
self.send_command(True) self.send_command(True)
def turn_off(self, **kwargs) -> None: def turn_off(self, **kwargs: Any) -> None:
"""Turn the entity off.""" """Turn the entity off."""
self.send_command(False) self.send_command(False)
@ -233,7 +234,7 @@ class SonosAlarmEntity(SonosEntity, SwitchEntity):
) )
@property @property
def alarm(self): def alarm(self) -> Alarm:
"""Return the alarm instance.""" """Return the alarm instance."""
return self.hass.data[DATA_SONOS].alarms[self.household_id].get(self.alarm_id) 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() await self.hass.data[DATA_SONOS].alarms[self.household_id].async_poll()
@callback @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.""" """Check if alarm exists and remove alarm entity if not available."""
if self.alarm: if self.alarm:
return True return True
@ -279,7 +280,7 @@ class SonosAlarmEntity(SonosEntity, SwitchEntity):
self.async_write_ha_state() self.async_write_ha_state()
@callback @callback
def _async_update_device(self): def _async_update_device(self) -> None:
"""Update the device, since this alarm moved to a different player.""" """Update the device, since this alarm moved to a different player."""
device_registry = dr.async_get(self.hass) device_registry = dr.async_get(self.hass)
entity_registry = er.async_get(self.hass) entity_registry = er.async_get(self.hass)
@ -288,22 +289,20 @@ class SonosAlarmEntity(SonosEntity, SwitchEntity):
if entity is None: if entity is None:
raise RuntimeError("Alarm has been deleted by accident.") raise RuntimeError("Alarm has been deleted by accident.")
entry_id = entity.config_entry_id
new_device = device_registry.async_get_or_create( 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)}, identifiers={(SONOS_DOMAIN, self.soco.uid)},
connections={(dr.CONNECTION_NETWORK_MAC, self.speaker.mac_address)}, 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) _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 @property
def _is_today(self): def _is_today(self) -> bool:
"""Return whether this alarm is scheduled for today."""
recurrence = self.alarm.recurrence recurrence = self.alarm.recurrence
timestr = int(datetime.datetime.today().strftime("%w")) timestr = int(datetime.datetime.today().strftime("%w"))
return ( return (
@ -321,12 +320,12 @@ class SonosAlarmEntity(SonosEntity, SwitchEntity):
return (self.alarm is not None) and self.speaker.available return (self.alarm is not None) and self.speaker.available
@property @property
def is_on(self): def is_on(self) -> bool:
"""Return state of Sonos alarm switch.""" """Return state of Sonos alarm switch."""
return self.alarm.enabled return self.alarm.enabled
@property @property
def extra_state_attributes(self): def extra_state_attributes(self) -> dict[str, Any]:
"""Return attributes of Sonos alarm switch.""" """Return attributes of Sonos alarm switch."""
return { return {
ATTR_ID: str(self.alarm_id), ATTR_ID: str(self.alarm_id),

36
mypy.ini
View File

@ -2589,39 +2589,3 @@ disallow_untyped_decorators = false
disallow_untyped_defs = false disallow_untyped_defs = false
warn_return_any = false warn_return_any = false
warn_unreachable = 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 # If you are an author of component listed here, please fix these errors and
# remove your component from this list to enable type checks. # remove your component from this list to enable type checks.
# Do your best to not add anything new here. # Do your best to not add anything new here.
IGNORED_MODULES: Final[list[str]] = [ 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",
]
# Component modules which should set no_implicit_reexport = true. # Component modules which should set no_implicit_reexport = true.
NO_IMPLICIT_REEXPORT_MODULES: set[str] = { NO_IMPLICIT_REEXPORT_MODULES: set[str] = {