/
/
/
1"""HEOS Player Provider implementation."""
2
3from __future__ import annotations
4
5import logging
6from typing import TYPE_CHECKING, cast
7
8from music_assistant_models.errors import SetupFailedError
9from music_assistant_models.player import PlayerSource
10from pyheos import Heos, HeosError, HeosOptions, MediaItem, PlayerUpdateResult, const
11from zeroconf import ServiceStateChange
12
13from music_assistant.constants import CONF_ENABLED, CONF_IP_ADDRESS, VERBOSE_LOG_LEVEL
14from music_assistant.helpers.util import get_primary_ip_address_from_zeroconf
15from music_assistant.models.player_provider import PlayerProvider
16from music_assistant.providers.heos.constants import HEOS_PASSIVE_SOURCES
17
18from .player import HeosPlayer
19
20if TYPE_CHECKING:
21 from zeroconf.asyncio import AsyncServiceInfo
22
23
24class HeosPlayerProvider(PlayerProvider):
25 """Player provided for Denon HEOS."""
26
27 _heos: Heos | None = None
28 _music_source_list: list[PlayerSource] = []
29 _input_source_list: list[MediaItem] = []
30 _player_discovery_running: bool = False
31 _controller_discovery_running: bool = False
32
33 async def handle_async_init(self) -> None:
34 """Handle async initialization of the provider."""
35 if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
36 logging.getLogger("pyheos").setLevel(logging.DEBUG)
37 else:
38 logging.getLogger("pyheos").setLevel(self.logger.level + 10)
39
40 if ip_address := self.config.get_value(CONF_IP_ADDRESS):
41 # Manual IP path
42 ip_address = cast("str", ip_address)
43 await self._setup_controller(ip_address)
44
45 async def _setup_controller(self, controller_ip: str, connect_preferred: bool = False) -> None:
46 """Set up the HEOS controller."""
47 self.logger.debug("Attempting HEOS controller setup on IP %s", controller_ip)
48 self._heos = Heos(HeosOptions(controller_ip, auto_reconnect=True, auto_failover=True))
49
50 try:
51 await self._heos.connect()
52
53 self.logger.debug("HEOS controller connected, checking preferred setup")
54 system_info = await self._heos.get_system_info()
55 preferred_ips: list[str] | None = [
56 host.ip_address for host in system_info.preferred_hosts if host.ip_address
57 ]
58
59 if preferred_ips and controller_ip not in preferred_ips:
60 if connect_preferred:
61 await self._heos.disconnect()
62 # Set up controller with preferred host instead
63 return await self._setup_controller(preferred_ips[0], connect_preferred=False)
64
65 # Just log a warning, it still works but might be less reliable
66 self.logger.warning(f"Configured IP {controller_ip} is not a preferred HEOS host")
67 except HeosError as e:
68 self.logger.error(f"Failed to connect to HEOS controller: {e}")
69 raise SetupFailedError("Failed to connect to HEOS controller") from e
70
71 # Initialize library values
72 try:
73 self._heos.add_on_controller_event(self._handle_controller_event)
74 await self._populate_sources()
75
76 # Explicitly discover players now, in case we are set up from discovery
77 await self.discover_players()
78 except HeosError as e:
79 self.logger.error(f"Unexpected error setting up HEOS controller: {e}")
80 raise SetupFailedError("Unexpected error setting up HEOS controller") from e
81
82 async def _handle_controller_event(
83 self, event: str, result: PlayerUpdateResult | None = None
84 ) -> None:
85 self.logger.debug("Controller event received: %s", event)
86
87 if event == const.EVENT_GROUPS_CHANGED:
88 for player in self.mass.players.all_players(provider_filter=self.instance_id):
89 assert isinstance(player, HeosPlayer) # for type checking
90 await player.build_group_list()
91
92 if event == const.EVENT_PLAYERS_CHANGED:
93 if result is None:
94 return
95
96 await self.discover_players()
97
98 async def _populate_sources(self) -> None:
99 """Build source list based on data from controller."""
100 if not self._heos:
101 return
102 self._input_source_list = list(await self._heos.get_input_sources())
103
104 music_sources = await self._heos.get_music_sources()
105 for source_id, source in music_sources.items():
106 self._music_source_list.append(
107 PlayerSource(
108 id=str(source_id),
109 name=source.name,
110 passive=source_id in HEOS_PASSIVE_SOURCES or not source.available,
111 can_play_pause=True, # All sources support play/pause
112 can_next_previous=source_id == 1024, # TODO: properly check
113 )
114 )
115
116 @property
117 def music_source_list(self) -> list[PlayerSource]:
118 """Get mapped music source list from controller info."""
119 return self._music_source_list
120
121 @property
122 def input_source_list(self) -> list[MediaItem]:
123 """Get input list from controller info. This represents all inputs across all players."""
124 return self._input_source_list
125
126 async def unload(self, is_removed: bool = False) -> None:
127 """Handle unload/close of the provider."""
128 if self._heos:
129 self._heos.dispatcher.disconnect_all() # Remove all event connections
130 await self._heos.disconnect()
131
132 for player in self.players:
133 self.logger.debug("Unloading player %s", player.name)
134 await self.mass.players.unregister(player.player_id)
135
136 async def discover_players(self) -> None:
137 """Discover players for this provider."""
138 if self._player_discovery_running or not self._heos:
139 return # discovery already running or not set up
140
141 try:
142 self._player_discovery_running = True
143 self.logger.debug("Discovering HEOS players")
144 devices = await self._heos.get_players()
145 for device in devices.values():
146 player_id = str(device.player_id)
147 if player := cast("HeosPlayer", self.mass.players.get_player(player_id)):
148 self.logger.debug(
149 "Updating existing HEOS player: %s (%s)", device.name, player_id
150 )
151 # Update properties such as name or availability
152 player.set_device_info()
153 player.update_state()
154 continue
155
156 player_enabled = self.mass.config.get_raw_player_config_value(
157 player_id, CONF_ENABLED, default=True
158 )
159 if not player_enabled:
160 self.logger.debug("Skipping disabled player: %s (%s)", device.name, player_id)
161 continue
162 self.logger.info("Discovered new HEOS player: %s (%s)", device.name, player_id)
163
164 heos_player = HeosPlayer(self, device)
165 await heos_player.setup()
166 finally:
167 self._player_discovery_running = False
168
169 async def on_mdns_service_state_change(
170 self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
171 ) -> None:
172 """Discovery via mdns."""
173 if state_change == ServiceStateChange.Removed:
174 return
175
176 if not info:
177 return
178
179 if self._heos or self._controller_discovery_running:
180 self.logger.debug("Ignoring mDNS configuration because we're already set up")
181 # We're already set up or in the process of setting up
182 return
183
184 device_ip = get_primary_ip_address_from_zeroconf(info)
185 if not device_ip:
186 self.logger.debug("Ignoring incomplete mdns discovery for HEOS player: %s", name)
187 return
188
189 self.logger.debug("Discovered HEOS device %s on %s", name, device_ip)
190
191 self._controller_discovery_running = True
192 try:
193 await self._setup_controller(device_ip, True)
194 except SetupFailedError:
195 self.logger.error("Failed to set up HEOS controller at %s discovered via mDNS")
196 finally:
197 self._controller_discovery_running = False
198