/
/
/
1"""Bluesound Player Provider implementation."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, ClassVar, TypedDict, cast
6
7from zeroconf import ServiceStateChange
8
9from music_assistant.helpers.util import (
10 get_mac_address,
11 get_port_from_zeroconf,
12 get_primary_ip_address_from_zeroconf,
13)
14from music_assistant.models.player_provider import PlayerProvider
15
16from .const import MUSP_MDNS_TYPE
17from .player import BluesoundPlayer
18
19if TYPE_CHECKING:
20 from music_assistant_models.config_entries import ConfigEntry
21 from zeroconf.asyncio import AsyncServiceInfo
22
23
24class BluesoundDiscoveryInfo(TypedDict):
25 """Template for MDNS discovery info."""
26
27 _objectType: str
28 ip_address: str
29 port: str
30 mac: str
31 model: str
32
33
34class BluesoundPlayerProvider(PlayerProvider):
35 """Bluos compatible player provider, providing support for bluesound speakers."""
36
37 player_map: ClassVar[dict[tuple[str, int], str]] = {}
38
39 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
40 """Set up legacy BluOS devices."""
41 return ()
42
43 async def handle_async_init(self) -> None:
44 """Handle async initialization of the provider."""
45
46 async def on_mdns_service_state_change(
47 self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
48 ) -> None:
49 """Handle MDNS service state callback for BluOS."""
50 if state_change == ServiceStateChange.Removed:
51 # Wait for connection to fail, same as sonos.
52 return
53 name = name.split(".", 1)[0]
54 assert info is not None
55
56 ip_address = get_primary_ip_address_from_zeroconf(info)
57 port = get_port_from_zeroconf(info)
58
59 if not ip_address or not port:
60 self.logger.debug("Ignoring incomplete mdns discovery for Bluesound player: %s", name)
61 return
62
63 player_id: str | None
64 if info.type == MUSP_MDNS_TYPE:
65 # this is a multi-zone device, we need to fetch the mac address of the main device
66 mac_address = await get_mac_address(ip_address)
67 player_id = f"{mac_address}:{port}"
68 else:
69 mac_address = info.decoded_properties.get("mac")
70 player_id = mac_address
71
72 if not mac_address:
73 self.logger.debug(
74 "Ignoring mdns discovery for Bluesound player without MAC address: %s",
75 name,
76 )
77 return
78
79 # Handle update of existing player
80 assert player_id is not None # for type checker
81 if bluos_player := self.mass.players.get_player(player_id):
82 bluos_player = cast("BluesoundPlayer", bluos_player)
83 # Check if the IP address has changed
84 if ip_address and ip_address != bluos_player.ip_address:
85 self.logger.debug(
86 "IP address for player %s updated to %s", bluos_player.name, ip_address
87 )
88 else:
89 # IP address not changed
90 self.logger.debug("Player back online: %s", bluos_player.name)
91 bluos_player._attr_available = True
92 await bluos_player.update_attributes()
93 return
94
95 # New player discovered
96 self.logger.debug("Discovered player: %s", name)
97
98 discovery_info = BluesoundDiscoveryInfo(
99 _objectType=info.decoded_properties.get("_objectType") or "",
100 ip_address=ip_address,
101 port=str(port),
102 mac=mac_address,
103 model=info.decoded_properties.get("model") or "",
104 )
105
106 # Create BluOS player
107 bluos_player = BluesoundPlayer(self, player_id, discovery_info, name, ip_address, port)
108 self.player_map[(ip_address, port)] = player_id
109
110 # Register with Music Assistant
111 await bluos_player.setup()
112