# Keep internal reference so we don't need to check None on each call
self._heos = self._device.heos
+ self._attr_type = PlayerType.PLAYER
+ self._attr_supported_features = PLAYER_FEATURES
+ self._attr_can_group_with = {self.provider.instance_id}
+
async def setup(self) -> None:
"""Set up the player."""
- self.set_static_attributes()
+ self.set_device_info()
self.set_dynamic_attributes()
await self.mass.players.register_or_update(self)
- if self.enabled:
- self._on_unload_callbacks.append(
- self._device.add_on_player_event(self._player_event_received)
- )
+ self._on_unload_callbacks.append(
+ self._device.add_on_player_event(self._player_event_received)
+ )
- await self.build_group_list()
- await self.build_source_list()
+ await self.build_group_list()
+ await self.build_source_list()
- def set_static_attributes(self) -> None:
- """Set all player static attributes."""
+ def set_device_info(self) -> None:
+ """Set all device info attributes."""
# Extract manufacturer and model from device model string, if available
model_parts = self._device.model.split(maxsplit=1)
manufacturer = model_parts[0] if len(model_parts) == 2 else "HEOS"
model = model_parts[1] if len(model_parts) == 2 else self._device.model
- self._attr_type = PlayerType.PLAYER
- self._attr_supported_features = PLAYER_FEATURES
_device_info = DeviceInfo(
model=model,
software_version=self._device.version,
)
_device_info.ip_address = self._device.ip_address
self._attr_device_info = _device_info
- self._attr_can_group_with = {self.provider.instance_id}
self._attr_available = self._device.available
self._attr_name = self._device.name
from __future__ import annotations
import logging
+from typing import TYPE_CHECKING, cast
from music_assistant_models.errors import SetupFailedError
from music_assistant_models.player import PlayerSource
from pyheos import Heos, HeosError, HeosOptions, MediaItem, PlayerUpdateResult, const
+from zeroconf import ServiceStateChange
from music_assistant.constants import CONF_ENABLED, CONF_IP_ADDRESS, VERBOSE_LOG_LEVEL
+from music_assistant.helpers.util import get_primary_ip_address_from_zeroconf
from music_assistant.models.player_provider import PlayerProvider
from music_assistant.providers.heos.constants import HEOS_PASSIVE_SOURCES
from .player import HeosPlayer
+if TYPE_CHECKING:
+ from zeroconf.asyncio import AsyncServiceInfo
+
class HeosPlayerProvider(PlayerProvider):
"""Player provided for Denon HEOS."""
- _heos: Heos
+ _heos: Heos | None = None
_music_source_list: list[PlayerSource] = []
_input_source_list: list[MediaItem] = []
- _discovery_running: bool = False
+ _player_discovery_running: bool = False
+ _controller_discovery_running: bool = False
async def handle_async_init(self) -> None:
"""Handle async initialization of the provider."""
else:
logging.getLogger("pyheos").setLevel(self.logger.level + 10)
- self._heos = Heos(
- HeosOptions(
- str(self.config.get_value(CONF_IP_ADDRESS)),
- auto_reconnect=True,
- )
- )
+ if ip_address := self.config.get_value(CONF_IP_ADDRESS):
+ # Manual IP path
+ ip_address = cast("str", ip_address)
+ await self._setup_controller(ip_address)
+
+ async def _setup_controller(self, controller_ip: str, connect_preferred: bool = False) -> None:
+ """Set up the HEOS controller."""
+ self.logger.debug("Attempting HEOS controller setup on IP %s", controller_ip)
+ self._heos = Heos(HeosOptions(controller_ip, auto_reconnect=True, auto_failover=True))
try:
await self._heos.connect()
- self._heos.add_on_controller_event(self._handle_controller_event)
+ self.logger.debug("HEOS controller connected, checking preferred setup")
+ system_info = await self._heos.get_system_info()
+ preferred_ips: list[str] | None = [
+ host.ip_address for host in system_info.preferred_hosts if host.ip_address
+ ]
+
+ if preferred_ips and controller_ip not in preferred_ips:
+ if connect_preferred:
+ await self._heos.disconnect()
+ # Set up controller with preferred host instead
+ return await self._setup_controller(preferred_ips[0], connect_preferred=False)
+
+ # Just log a warning, it still works but might be less reliable
+ self.logger.warning(f"Configured IP {controller_ip} is not a preferred HEOS host")
except HeosError as e:
self.logger.error(f"Failed to connect to HEOS controller: {e}")
raise SetupFailedError("Failed to connect to HEOS controller") from e
# Initialize library values
try:
- # Populate source lists
+ self._heos.add_on_controller_event(self._handle_controller_event)
await self._populate_sources()
- # NOTE: players are discovered via discovery method (called automatically by core)
+
+ # Explicitly discover players now, in case we are set up from discovery
+ await self.discover_players()
except HeosError as e:
self.logger.error(f"Unexpected error setting up HEOS controller: {e}")
raise SetupFailedError("Unexpected error setting up HEOS controller") from e
if result is None:
return
- for removed_player_id in result.removed_player_ids:
- await self.mass.players.unregister(str(removed_player_id))
-
- for new_player_id in result.added_player_ids:
- try:
- device = await self._heos.get_player_info(new_player_id)
- heos_player = HeosPlayer(self, device)
-
- await heos_player.setup()
- except HeosError as e:
- self.logger.error(
- "Error adding new HEOS player with id %s: %s", new_player_id, e
- )
- continue
+ await self.discover_players()
async def _populate_sources(self) -> None:
"""Build source list based on data from controller."""
+ if not self._heos:
+ return
self._input_source_list = list(await self._heos.get_input_sources())
music_sources = await self._heos.get_music_sources()
async def unload(self, is_removed: bool = False) -> None:
"""Handle unload/close of the provider."""
- self._heos.dispatcher.disconnect_all() # Remove all event connections
- await self._heos.disconnect()
+ if self._heos:
+ self._heos.dispatcher.disconnect_all() # Remove all event connections
+ await self._heos.disconnect()
for player in self.players:
self.logger.debug("Unloading player %s", player.name)
async def discover_players(self) -> None:
"""Discover players for this provider."""
- if self._discovery_running:
- return # discovery already running
+ if self._player_discovery_running or not self._heos:
+ return # discovery already running or not set up
+
try:
- self._discovery_running = True
+ self._player_discovery_running = True
self.logger.debug("Discovering HEOS players")
devices = await self._heos.get_players()
- already_registered = {p.player_id for p in self.players}
for device in devices.values():
player_id = str(device.player_id)
- if player_id in already_registered:
- continue # already registered
- # ignore disabled players in discovery
+ if player := cast("HeosPlayer", self.mass.players.get(player_id)):
+ self.logger.debug(
+ "Updating existing HEOS player: %s (%s)", device.name, player_id
+ )
+ # Update properties such as name or availability
+ player.set_device_info()
+ player.update_state()
+ continue
+
player_enabled = self.mass.config.get_raw_player_config_value(
player_id, CONF_ENABLED, default=True
)
if not player_enabled:
+ self.logger.debug("Skipping disabled player: %s (%s)", device.name, player_id)
continue
self.logger.info("Discovered new HEOS player: %s (%s)", device.name, player_id)
heos_player = HeosPlayer(self, device)
await heos_player.setup()
finally:
- self._discovery_running = False
- # reschedule discovery
- task_id = f"discover_players_{self.instance_id}"
- self.mass.call_later(600, self.discover_players, task_id=task_id)
+ self._player_discovery_running = False
+
+ async def on_mdns_service_state_change(
+ self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
+ ) -> None:
+ """Discovery via mdns."""
+ if state_change == ServiceStateChange.Removed:
+ return
+
+ if not info:
+ return
+
+ if self._heos or self._controller_discovery_running:
+ self.logger.debug("Ignoring mDNS configuration because we're already set up")
+ # We're already set up or in the process of setting up
+ return
+
+ device_ip = get_primary_ip_address_from_zeroconf(info)
+ if not device_ip:
+ self.logger.debug("Ignoring incomplete mdns discovery for HEOS player: %s", name)
+ return
+
+ self.logger.debug("Discovered HEOS device %s on %s", name, device_ip)
+
+ self._controller_discovery_running = True
+ try:
+ await self._setup_controller(device_ip, True)
+ except SetupFailedError:
+ self.logger.error("Failed to set up HEOS controller at %s discovered via mDNS")
+ finally:
+ self._controller_discovery_running = False
projects / music-assistant-server.git / commitdiff