/
/
/
1"""
2Sonos Player S1 provider for Music Assistant.
3
4Based on the SoCo library for Sonos which uses the legacy/V1 UPnP API.
5
6Note that large parts of this code are copied over from the Home Assistant
7integration for Sonos.
8"""
9
10from __future__ import annotations
11
12import asyncio
13from typing import TYPE_CHECKING, cast
14
15from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
16from music_assistant_models.enums import ConfigEntryType, ProviderFeature
17from soco.discovery import scan_network
18
19from music_assistant.constants import CONF_ENTRY_MANUAL_DISCOVERY_IPS
20
21from .constants import CONF_HOUSEHOLD_ID, CONF_NETWORK_SCAN
22from .provider import SonosPlayerProvider
23
24if TYPE_CHECKING:
25 from music_assistant_models.config_entries import ProviderConfig
26 from music_assistant_models.provider import ProviderManifest
27 from soco import SoCo
28
29 from music_assistant.mass import MusicAssistant
30 from music_assistant.models import ProviderInstanceType
31
32SUPPORTED_FEATURES = {
33 ProviderFeature.SYNC_PLAYERS,
34}
35
36
37async def setup(
38 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
39) -> ProviderInstanceType:
40 """Initialize provider(instance) with given configuration."""
41 return SonosPlayerProvider(mass, manifest, config, SUPPORTED_FEATURES)
42
43
44async def get_config_entries(
45 mass: MusicAssistant,
46 instance_id: str | None = None, # noqa: ARG001
47 action: str | None = None, # noqa: ARG001
48 values: dict[str, ConfigValueType] | None = None, # noqa: ARG001
49) -> tuple[ConfigEntry, ...]:
50 """
51 Return Config entries to setup this provider.
52
53 instance_id: id of an existing provider instance (None if new instance setup).
54 action: [optional] action key called from config entries UI.
55 values: the (intermediate) raw values for config entries sent with the action.
56 """
57 household_ids = await discover_household_ids(mass)
58 return (
59 CONF_ENTRY_MANUAL_DISCOVERY_IPS,
60 ConfigEntry(
61 key=CONF_NETWORK_SCAN,
62 type=ConfigEntryType.BOOLEAN,
63 label="Enable network scan for discovery",
64 default_value=False,
65 description="Enable network scan for discovery of players. \n"
66 "Can be used if (some of) your players are not automatically discovered.\n"
67 "Should normally not be needed",
68 ),
69 ConfigEntry(
70 key=CONF_HOUSEHOLD_ID,
71 type=ConfigEntryType.STRING,
72 label="Household ID",
73 default_value=household_ids[0] if household_ids else None,
74 description="Household ID for the Sonos (S1) system. Will be auto detected if empty.",
75 advanced=True,
76 required=False,
77 ),
78 )
79
80
81async def discover_household_ids(mass: MusicAssistant, prefer_s1: bool = True) -> list[str]:
82 """Discover the HouseHold ID of S1 speaker(s) the network."""
83 if cache := await mass.cache.get("sonos_household_ids"):
84 return cast("list[str]", cache)
85 household_ids: list[str] = []
86
87 def get_all_sonos_ips() -> set[SoCo]:
88 """Run full network discovery and return IP's of all devices found on the network."""
89 discovered_zones: set[SoCo] | None
90 if discovered_zones := scan_network(multi_household=True):
91 return {zone.ip_address for zone in discovered_zones}
92 return set()
93
94 all_sonos_ips = await asyncio.to_thread(get_all_sonos_ips)
95 for ip_address in all_sonos_ips:
96 async with mass.http_session.get(f"http://{ip_address}:1400/status/zp") as resp:
97 if resp.status == 200:
98 data = await resp.text()
99 if prefer_s1 and "<SWGen>2</SWGen>" in data:
100 continue
101 if "HouseholdControlID" in data:
102 household_id = data.split("<HouseholdControlID>")[1].split(
103 "</HouseholdControlID>"
104 )[0]
105 household_ids.append(household_id)
106 await mass.cache.set("sonos_household_ids", household_ids, 3600)
107 return household_ids
108