Allow for advanced networking in the Sonos provider.

music_assistant/providers/sonos/__init__.py

@@ -10,12 +10,14 @@
 import logging
 from typing import TYPE_CHECKING
 
+from music_assistant_models.config_entries import ConfigEntry, ConfigEntryType
+
 from music_assistant.constants import VERBOSE_LOG_LEVEL
 
-from .provider import SonosPlayerProvider
+from .provider import CONF_IPS, SonosPlayerProvider
 
 if TYPE_CHECKING:
-    from music_assistant_models.config_entries import ConfigEntry, ConfigValueType, ProviderConfig
+    from music_assistant_models.config_entries import ConfigValueType, ProviderConfig
     from music_assistant_models.provider import ProviderManifest
 
     from music_assistant import MusicAssistant
@@ -49,4 +51,20 @@ async def get_config_entries(
     values: the (intermediate) raw values for config entries sent with the action.
     """
     # ruff: noqa: ARG001
-    return ()
+    return (
+        ConfigEntry(
+            key=CONF_IPS,
+            type=ConfigEntryType.STRING,
+            label="IP addresses (ADVANCED, NOT SUPPORTED)",
+            description="Additional fixed IP addresses for speakers. "
+            "Should be formatted as a comma separated list of IP addresses "
+            "(e.g. '10.0.0.42, 10.0.0.45').\n"
+            "Invalid addresses may result in the Sonos provider "
+            "becoming unresponsive and server crashes.\n"
+            "Bidirectional unicast communication to and between all IPs is required.\n"
+            "NOT SUPPORTED, USE ON YOU'RE OWN RISK",
+            category="advanced",
+            default_value=None,
+            required=False,
+        ),
+    )

music_assistant/providers/sonos/provider.py

@@ -40,6 +40,8 @@
 if TYPE_CHECKING:
     from zeroconf.asyncio import AsyncServiceInfo
 
+CONF_IPS = "ips"
+
 
 class SonosPlayerProvider(PlayerProvider):
     """Sonos Player provider."""
@@ -67,6 +69,34 @@ async def handle_async_init(self) -> None:
             "/sonos_queue/v2.3/timePlayed", self._handle_sonos_queue_time_played
         )
 
+    async def loaded_in_mass(self) -> None:
+        """Call after the provider has been loaded."""
+        await super().loaded_in_mass()
+
+        manual_ip_config: str | None
+        # Handle config option for manual IP's (comma separated list)
+        if (manual_ip_config := self.config.get_value(CONF_IPS)) is not None:
+            ips = manual_ip_config.split(",")
+            for raw_ip in ips:
+                # strip to ignore whitespace
+                # (e.g. '10.0.0.42, 10.0.0.43' -> ('10.0.0.42', ' 10.0.0.43'))
+                ip = raw_ip.strip()
+                if ip == "":
+                    continue
+                try:
+                    # get discovery info from SONOS speaker so we can provide an ID & other info
+                    discovery_info = await get_discovery_info(self.mass.http_session, ip)
+                except ClientError as err:
+                    self.logger.debug(
+                        "Ignoring %s (manual IP) as it is not reachable: %s", ip, str(err)
+                    )
+                    continue
+                player_id = discovery_info["device"]["id"]
+                self.sonos_players[player_id] = sonos_player = SonosPlayer(
+                    self, player_id, discovery_info=discovery_info, ip_address=ip
+                )
+                await sonos_player.setup()
+
     async def unload(self, is_removed: bool = False) -> None:
         """Handle close/cleanup of the provider."""
         # disconnect all players