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