Use subscription callbacks to discover Sonos speakers

homeassistant/components/sonos/__init__.py

@@ -11,9 +11,10 @@
 from typing import TYPE_CHECKING, Any, Optional, cast
 from urllib.parse import urlparse
 
-from soco import events_asyncio
+from soco import events_asyncio, zonegroupstate
 import soco.config as soco_config
 from soco.core import SoCo
+from soco.events_base import Event as SonosEvent, SubscriptionBase
 from soco.exceptions import SoCoException
 import voluptuous as vol
 
@@ -24,8 +25,8 @@
 from homeassistant.const import CONF_HOSTS, EVENT_HOMEASSISTANT_STOP
 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
+from homeassistant.helpers.dispatcher import async_dispatcher_send
+from homeassistant.helpers.event import async_call_later, async_track_time_interval
 from homeassistant.helpers.typing import ConfigType
 
 from .alarms import SonosAlarms
@@ -40,6 +41,7 @@
     SONOS_REBOOTED,
     SONOS_SPEAKER_ACTIVITY,
     SONOS_VANISHED,
+    SUBSCRIPTION_TIMEOUT,
     UPNP_ST,
 )
 from .exception import SonosUpdateError
@@ -51,7 +53,7 @@
 CONF_ADVERTISE_ADDR = "advertise_addr"
 CONF_INTERFACE_ADDR = "interface_addr"
 DISCOVERY_IGNORED_MODELS = ["Sonos Boost"]
-
+ZGS_SUBSCRIPTION_TIMEOUT = 2
 
 CONFIG_SCHEMA = vol.Schema(
     {
@@ -122,6 +124,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up Sonos from a config entry."""
     soco_config.EVENTS_MODULE = events_asyncio
     soco_config.REQUEST_TIMEOUT = 9.5
+    zonegroupstate.EVENT_CACHE_TIMEOUT = SUBSCRIPTION_TIMEOUT
 
     if DATA_SONOS not in hass.data:
         hass.data[DATA_SONOS] = SonosData()
@@ -172,6 +175,7 @@ def __init__(
         self.data = data
         self.hosts = set(hosts)
         self.discovery_lock = asyncio.Lock()
+        self.creation_lock = asyncio.Lock()
         self._known_invisible: set[SoCo] = set()
         self._manual_config_required = bool(hosts)
 
@@ -184,21 +188,70 @@ def is_device_invisible(self, ip_address: str) -> bool:
         """Check if device at provided IP is known to be invisible."""
         return any(x for x in self._known_invisible if x.ip_address == ip_address)
 
-    def _create_visible_speakers(self, ip_address: str) -> None:
-        """Create all visible SonosSpeaker instances with the provided seed IP."""
-        try:
-            soco = SoCo(ip_address)
+    async def async_subscribe_to_zone_updates(self, ip_address: str) -> None:
+        """Test subscriptions and create SonosSpeakers based on results."""
+        soco = SoCo(ip_address)
+        # Cache now to avoid household ID lookup during first ZoneGroupState processing
+        await self.hass.async_add_executor_job(
+            getattr,
+            soco,
+            "household_id",
+        )
+        sub = await soco.zoneGroupTopology.subscribe()
+
+        @callback
+        def _async_add_visible_zones(subscription_succeeded: bool = False) -> None:
+            """Determine visible zones and create SonosSpeaker instances."""
+            zones_to_add = set()
+            subscription = None
+            if subscription_succeeded:
+                subscription = sub
+
             visible_zones = soco.visible_zones
             self._known_invisible = soco.all_zones - visible_zones
-        except (OSError, SoCoException) as ex:
+            for zone in visible_zones:
+                if zone.uid not in self.data.discovered:
+                    zones_to_add.add(zone)
+
+            if not zones_to_add:
+                return
+
+            self.hass.async_create_task(
+                self.async_add_speakers(zones_to_add, subscription, soco.uid)
+            )
+
+        async def async_subscription_failed(now: datetime.datetime) -> None:
+            """Fallback logic if the subscription callback never arrives."""
+            await sub.unsubscribe()
             _LOGGER.warning(
-                "Failed to request visible zones from %s: %s", ip_address, ex
+                "Subscription to %s failed, attempting to poll directly", ip_address
             )
-            return
+            try:
+                await self.hass.async_add_executor_job(soco.zone_group_state.poll, soco)
+            except (OSError, SoCoException) as ex:
+                _LOGGER.warning(
+                    "Fallback pollling to %s failed, setup cannot continue: %s",
+                    ip_address,
+                    ex,
+                )
+                return
+            _LOGGER.debug("Fallback ZoneGroupState poll to %s succeeded", ip_address)
+            _async_add_visible_zones()
+
+        cancel_failure_callback = async_call_later(
+            self.hass, ZGS_SUBSCRIPTION_TIMEOUT, async_subscription_failed
+        )
+
+        @callback
+        def _async_subscription_succeeded(event: SonosEvent) -> None:
+            """Create SonosSpeakers when subscription callbacks successfully arrive."""
+            _LOGGER.debug("Subscription to %s succeeded", ip_address)
+            cancel_failure_callback()
+            _async_add_visible_zones(subscription_succeeded=True)
 
-        for zone in visible_zones:
-            if zone.uid not in self.data.discovered:
-                self._add_speaker(zone)
+        sub.callback = _async_subscription_succeeded
+        # Hold lock to prevent concurrent subscription attempts
+        await asyncio.sleep(ZGS_SUBSCRIPTION_TIMEOUT * 2)
 
     async def _async_stop_event_listener(self, event: Event | None = None) -> None:
         for speaker in self.data.discovered.values():
@@ -227,14 +280,35 @@ def _stop_manual_heartbeat(self, event: Event | None = None) -> None:
             self.data.hosts_heartbeat()
             self.data.hosts_heartbeat = None
 
-    def _add_speaker(self, soco: SoCo) -> None:
+    async def async_add_speakers(
+        self,
+        socos: set[SoCo],
+        zgs_subscription: SubscriptionBase | None,
+        zgs_subscription_uid: str | None,
+    ) -> None:
+        """Create and set up new SonosSpeaker instances."""
+
+        def _add_speakers():
+            """Add all speakers in a single executor job."""
+            for soco in socos:
+                sub = None
+                if soco.uid == zgs_subscription_uid and zgs_subscription:
+                    sub = zgs_subscription
+                self._add_speaker(soco, sub)
+
+        async with self.creation_lock:
+            await self.hass.async_add_executor_job(_add_speakers)
+
+    def _add_speaker(
+        self, soco: SoCo, zone_group_state_sub: SubscriptionBase | None
+    ) -> None:
         """Create and set up a new SonosSpeaker instance."""
         try:
             speaker_info = soco.get_speaker_info(True, timeout=7)
             if soco.uid not in self.data.boot_counts:
                 self.data.boot_counts[soco.uid] = soco.boot_seqnum
             _LOGGER.debug("Adding new speaker: %s", speaker_info)
-            speaker = SonosSpeaker(self.hass, soco, speaker_info)
+            speaker = SonosSpeaker(self.hass, soco, speaker_info, zone_group_state_sub)
             self.data.discovered[soco.uid] = speaker
             for coordinator, coord_dict in (
                 (SonosAlarms, self.data.alarms),
@@ -250,13 +324,25 @@ def _add_speaker(self, soco: SoCo) -> None:
         except (OSError, SoCoException):
             _LOGGER.warning("Failed to add SonosSpeaker using %s", soco, exc_info=True)
 
-    def _poll_manual_hosts(self, now: datetime.datetime | None = None) -> None:
+    async def async_poll_manual_hosts(
+        self, now: datetime.datetime | None = None
+    ) -> None:
         """Add and maintain Sonos devices from a manual configuration."""
+
+        def get_sync_attributes(soco: SoCo) -> set[SoCo]:
+            """Ensure I/O attributes are cached and return visible zones."""
+            _ = soco.household_id
+            _ = soco.uid
+            return soco.visible_zones
+
         for host in self.hosts:
             ip_addr = socket.gethostbyname(host)
             soco = SoCo(ip_addr)
             try:
-                visible_zones = soco.visible_zones
+                visible_zones = await self.hass.async_add_executor_job(
+                    get_sync_attributes,
+                    soco,
+                )
             except OSError:
                 _LOGGER.warning("Could not get visible Sonos devices from %s", ip_addr)
             else:
@@ -267,7 +353,7 @@ def _poll_manual_hosts(self, now: datetime.datetime | None = None) -> None:
                 }:
                     _LOGGER.debug("Adding to manual hosts: %s", new_hosts)
                     self.hosts.update(new_hosts)
-                dispatcher_send(
+                async_dispatcher_send(
                     self.hass,
                     f"{SONOS_SPEAKER_ACTIVITY}-{soco.uid}",
                     "manual zone scan",
@@ -290,7 +376,9 @@ def _poll_manual_hosts(self, now: datetime.datetime | None = None) -> None:
                 None,
             )
             if not known_speaker:
-                self._create_visible_speakers(ip_addr)
+                await self._async_handle_discovery_message(
+                    soco.uid, ip_addr, "manual zone scan"
+                )
             elif not known_speaker.available:
                 try:
                     known_speaker.ping()
@@ -299,33 +387,32 @@ def _poll_manual_hosts(self, now: datetime.datetime | None = None) -> None:
                         "Manual poll to %s failed, keeping unavailable", ip_addr
                     )
 
-        self.data.hosts_heartbeat = call_later(
-            self.hass, DISCOVERY_INTERVAL.total_seconds(), self._poll_manual_hosts
+        self.data.hosts_heartbeat = async_call_later(
+            self.hass, DISCOVERY_INTERVAL.total_seconds(), self.async_poll_manual_hosts
         )
 
     async def _async_handle_discovery_message(
-        self, uid: str, discovered_ip: str, boot_seqnum: int | None
+        self,
+        uid: str,
+        discovered_ip: str,
+        source: str,
+        boot_seqnum: int | None = None,
     ) -> None:
         """Handle discovered player creation and activity."""
         async with self.discovery_lock:
             if not self.data.discovered:
                 # Initial discovery, attempt to add all visible zones
-                await self.hass.async_add_executor_job(
-                    self._create_visible_speakers,
-                    discovered_ip,
-                )
+                await self.async_subscribe_to_zone_updates(discovered_ip)
             elif uid not in self.data.discovered:
                 if self.is_device_invisible(discovered_ip):
                     return
-                await self.hass.async_add_executor_job(
-                    self._add_speaker, SoCo(discovered_ip)
-                )
+                await self.async_subscribe_to_zone_updates(discovered_ip)
             elif boot_seqnum and boot_seqnum > self.data.boot_counts[uid]:
                 self.data.boot_counts[uid] = boot_seqnum
                 async_dispatcher_send(self.hass, f"{SONOS_REBOOTED}-{uid}")
             else:
                 async_dispatcher_send(
-                    self.hass, f"{SONOS_SPEAKER_ACTIVITY}-{uid}", "discovery"
+                    self.hass, f"{SONOS_SPEAKER_ACTIVITY}-{uid}", source
                 )
 
     async def _async_ssdp_discovered_player(
@@ -389,7 +476,10 @@ def async_discovered_player(
             self.data.discovery_known.add(uid)
         asyncio.create_task(
             self._async_handle_discovery_message(
-                uid, discovered_ip, cast(Optional[int], boot_seqnum)
+                uid,
+                discovered_ip,
+                "discovery",
+                boot_seqnum=cast(Optional[int], boot_seqnum),
             )
         )
 
@@ -408,7 +498,7 @@ async def setup_platforms_and_discovery(self) -> None:
                     EVENT_HOMEASSISTANT_STOP, self._stop_manual_heartbeat
                 )
             )
-            await self.hass.async_add_executor_job(self._poll_manual_hosts)
+            await self.async_poll_manual_hosts()
 
         self.entry.async_on_unload(
             await ssdp.async_register_callback(

homeassistant/components/sonos/speaker.py

@@ -69,14 +69,14 @@
     "CHARGING": True,
     "NOT_CHARGING": False,
 }
-SUBSCRIPTION_SERVICES = [
+SUBSCRIPTION_SERVICES = {
     "alarmClock",
     "avTransport",
     "contentDirectory",
     "deviceProperties",
     "renderingControl",
     "zoneGroupTopology",
-]
+}
 SUPPORTED_VANISH_REASONS = ("sleeping", "switch to bluetooth", "upgrade")
 UNUSED_DEVICE_KEYS = ["SPID", "TargetRoomName"]
 
@@ -88,7 +88,11 @@ class SonosSpeaker:
     """Representation of a Sonos speaker."""
 
     def __init__(
-        self, hass: HomeAssistant, soco: SoCo, speaker_info: dict[str, Any]
+        self,
+        hass: HomeAssistant,
+        soco: SoCo,
+        speaker_info: dict[str, Any],
+        zone_group_state_sub: SubscriptionBase | None,
     ) -> None:
         """Initialize a SonosSpeaker."""
         self.hass = hass
@@ -112,6 +116,9 @@ def __init__(
         # Subscriptions and events
         self.subscriptions_failed: bool = False
         self._subscriptions: list[SubscriptionBase] = []
+        if zone_group_state_sub:
+            zone_group_state_sub.callback = self.async_dispatch_event
+            self._subscriptions.append(zone_group_state_sub)
         self._subscription_lock: asyncio.Lock | None = None
         self._event_dispatchers: dict[str, Callable] = {}
         self._last_activity: float = NEVER_TIME
@@ -289,6 +296,12 @@ def subscription_address(self) -> str:
         addr, port = self._subscriptions[0].event_listener.address
         return ":".join([addr, str(port)])
 
+    @property
+    def missing_subscriptions(self) -> set[str]:
+        """Return a list of missing service subscriptions."""
+        subscribed_services = {sub.service.service_type for sub in self._subscriptions}
+        return SUBSCRIPTION_SERVICES - subscribed_services
+
     #
     # Subscription handling and event dispatchers
     #
@@ -321,8 +334,6 @@ async def async_subscribe(self) -> None:
             self._subscription_lock = asyncio.Lock()
 
         async with self._subscription_lock:
-            if self._subscriptions:
-                return
             try:
                 await self._async_subscribe()
             except SonosSubscriptionsFailed:
@@ -331,12 +342,14 @@ async def async_subscribe(self) -> None:
 
     async def _async_subscribe(self) -> None:
         """Create event subscriptions."""
-        _LOGGER.debug("Creating subscriptions for %s", self.zone_name)
-
         subscriptions = [
             self._subscribe(getattr(self.soco, service), self.async_dispatch_event)
-            for service in SUBSCRIPTION_SERVICES
+            for service in self.missing_subscriptions
         ]
+        if not subscriptions:
+            return
+
+        _LOGGER.debug("Creating subscriptions for %s", self.zone_name)
         results = await asyncio.gather(*subscriptions, return_exceptions=True)
         for result in results:
             self.log_subscription_result(