ProviderUnavailableError,
UnsupportedFeaturedException,
)
+from music_assistant_models.player import PlayerOptionValueType # noqa: TC002
from music_assistant_models.player_control import PlayerControl # noqa: TC002
from music_assistant.constants import (
player.set_active_mass_source(media.source_id)
await player.play_media(media)
+ @api_command("players/cmd/select_sound_mode")
+ @handle_player_command
+ async def select_sound_mode(self, player_id: str, sound_mode: str) -> None:
+ """
+ Handle SELECT SOUND MODE command on given player.
+
+ - player_id: player_id of the player to handle the command
+ - sound_mode: The ID of the sound mode that needs to be activated/selected.
+ """
+ player = self.get(player_id, True)
+ assert player is not None # for type checking
+
+ if PlayerFeature.SELECT_SOUND_MODE not in player.supported_features:
+ raise UnsupportedFeaturedException(
+ f"Player {player.display_name} does not support sound mode selection"
+ )
+
+ prev_sound_mode = player.active_sound_mode
+ if sound_mode == prev_sound_mode:
+ return
+
+ # basic check if sound mode is valid for player
+ if not any(x for x in player.sound_mode_list if x.id == sound_mode):
+ raise PlayerCommandFailed(
+ f"{sound_mode} is an invalid sound_mode for player {player.display_name}"
+ )
+
+ # forward to player
+ await player.select_sound_mode(sound_mode)
+
+ @api_command("players/cmd/set_option")
+ @handle_player_command
+ async def set_option(
+ self, player_id: str, option_key: str, option_value: PlayerOptionValueType
+ ) -> None:
+ """
+ Handle SET_OPTION command on given player.
+
+ - player_id: player_id of the player to handle the command
+ - option_key: The key of the player option that needs to be activated/selected.
+ - option_value: The new value of the player option.
+ """
+ player = self.get(player_id, True)
+ assert player is not None # for type checking
+
+ if PlayerFeature.OPTIONS not in player.supported_features:
+ raise UnsupportedFeaturedException(
+ f"Player {player.display_name} does not support set_option"
+ )
+
+ prev_player_option = next((x for x in player.options if x.key == option_key), None)
+ if not prev_player_option:
+ return
+ if prev_player_option.value == option_value:
+ return
+
+ if prev_player_option.read_only:
+ raise UnsupportedFeaturedException(
+ f"Player {player.display_name} option {option_key} is read-only"
+ )
+
+ # forward to player
+ await player.set_option(option_key=option_key, option_value=option_value)
+
@api_command("players/cmd/select_source")
@handle_player_command
async def select_source(self, player_id: str, source: str | None) -> None:
# signal player update on the eventbus
self.mass.signal_event(EventType.PLAYER_UPDATED, object_id=player_id, data=player)
+ # signal a separate PlayerOptionsUpdated event
+ if options := changed_values.get("options"):
+ self.mass.signal_event(
+ EventType.PLAYER_OPTIONS_UPDATED, object_id=player_id, data=options
+ )
+
if skip_forward and not force_update:
return
PLAYER_CONTROL_NATIVE,
PLAYER_CONTROL_NONE,
)
-from music_assistant_models.enums import MediaType, PlaybackState, PlayerFeature, PlayerType
+from music_assistant_models.enums import (
+ MediaType,
+ PlaybackState,
+ PlayerFeature,
+ PlayerType,
+)
from music_assistant_models.errors import UnsupportedFeaturedException
-from music_assistant_models.player import DeviceInfo, PlayerMedia, PlayerSource
+from music_assistant_models.player import (
+ DeviceInfo,
+ PlayerMedia,
+ PlayerOption,
+ PlayerOptionValueType,
+ PlayerSoundMode,
+ PlayerSource,
+)
from music_assistant_models.player import Player as PlayerState
from music_assistant_models.unique_list import UniqueList
from propcache import under_cached_property as cached_property
_attr_device_info: DeviceInfo
_attr_can_group_with: set[str]
_attr_source_list: list[PlayerSource]
+ _attr_sound_mode_list: list[PlayerSoundMode]
+ _attr_options: list[PlayerOption]
_attr_available: bool = True
_attr_name: str | None = None
_attr_powered: bool | None = None
_attr_elapsed_time: float | None = None
_attr_elapsed_time_last_updated: float | None = None
_attr_active_source: str | None = None
+ _attr_active_sound_mode: str | None = None
_attr_current_media: PlayerMedia | None = None
_attr_needs_poll: bool = False
_attr_poll_interval: int = 30
self._attr_device_info = DeviceInfo()
self._attr_can_group_with = set()
self._attr_source_list = []
+ self._attr_sound_mode_list = []
+ self._attr_options = []
# do not override/overwrite these private attributes below!
self._cache: dict[str, Any] = {} # storage dict for cached properties
self._player_id = player_id
"select_source needs to be implemented when PlayerFeature.SELECT_SOURCE is set"
)
+ async def select_sound_mode(self, sound_mode: str) -> None:
+ """
+ Handle SELECT SOUND MODE command on the player.
+
+ Will only be called if the PlayerFeature.SELECT_SOUND_MODE is supported.
+
+ :param source: The sound_mode(id) to select, as defined in the sound_mode_list.
+ """
+ raise NotImplementedError(
+ "select_sound_mode needs to be implemented when PlayerFeature.SELECT_SOUND_MODE is set"
+ )
+
+ async def set_option(self, option_key: str, option_value: PlayerOptionValueType) -> None:
+ """
+ Handle SET_OPTION command on the player.
+
+ Will only be called if the PlayerFeature.OPTIONS is supported.
+
+ :param option_key: The option_key of the PlayerOption
+ :param option_value: The new value of the PlayerOption
+ """
+ raise NotImplementedError(
+ "set_option needs to be implemented when PlayerFeature.Option is set"
+ )
+
async def set_members(
self,
player_ids_to_add: list[str] | None = None,
return player.player_id
return None
+ @property
+ def active_sound_mode(self) -> str | None:
+ """Return active sound mode of this player."""
+ return self._attr_active_sound_mode
+
+ @cached_property
+ def sound_mode_list(self) -> UniqueList[PlayerSoundMode]:
+ """Return available PlayerSoundModes for Player."""
+ return UniqueList(self._attr_sound_mode_list)
+
+ @cached_property
+ def options(self) -> UniqueList[PlayerOption]:
+ """Return all PlayerOptions for Player."""
+ return UniqueList(self._attr_options)
+
def _on_player_media_updated(self) -> None: # noqa: B027
"""Handle callback when the current media of the player is updated."""
# optional callback for players that want to be informed when the final
synced_to=self.synced_to,
active_source=self.active_source,
source_list=self.source_list,
+ active_sound_mode=self.active_sound_mode,
+ sound_mode_list=self.sound_mode_list,
+ options=self.options,
active_group=self.active_group,
current_media=self.current_media,
name=self.display_name,
"""Constants for the MusicCast provider."""
+from aiomusiccast.capabilities import BinarySensor as MCBinarySensor
+from aiomusiccast.capabilities import BinarySetter as MCBinarySetter
+from aiomusiccast.capabilities import NumberSensor as MCNumberSensor
+from aiomusiccast.capabilities import NumberSetter as MCNumberSetter
+from aiomusiccast.capabilities import OptionSetter as MCOptionSetter
+from aiomusiccast.capabilities import TextSensor as MCTextSensor
+
from music_assistant.constants import (
CONF_ENTRY_HTTP_PROFILE_DEFAULT_2,
CONF_ENTRY_ICY_METADATA_HIDDEN_DISABLED,
# tuner can be controlled, will change the station
"tuner",
)
+
+# for most sound modes we can just split at '_' and capitalize
+# here are some exceptions:
+MC_SOUND_MODE_FRIENDLY_NAMES = {
+ "2ch_stereo": "2 Channel Stereo",
+ "all_ch_stereo": "All Channels Stereo",
+ "surr_decoder": "Surround Decoder",
+}
+
+# We translate aiomusiccast's capabilities to PlayerOptions
+MC_CAPABILITIES = (
+ MCBinarySensor
+ | MCBinarySetter
+ | MCNumberSensor
+ | MCNumberSetter
+ | MCTextSensor
+ | MCOptionSetter
+)
if self.device.group_reduce_by_source:
await entity._check_client_list()
+ @property
+ def sound_mode_id(self) -> str | None:
+ """ID of current sound mode."""
+ zone = self.device.data.zones.get(self.zone_name)
+ assert zone is not None # for type checking
+ assert isinstance(zone.sound_program, str | None) # for type checking
+ return zone.sound_program
+
+ @property
+ def sound_mode_list(self) -> list[str]:
+ """Return a list of available sound modes."""
+ zone = self.device.data.zones.get(self.zone_name)
+ assert zone is not None # for type checking
+ assert isinstance(zone.sound_program_list, list) # for type checking
+ return zone.sound_program_list
+
@property
def source_id(self) -> str:
"""ID of the current input source.
"""Select input source. Internal source name."""
await self.device.select_source(self.zone_name, source_id)
+ async def select_sound_mode(self, sound_mode_id: str) -> None:
+ """Select sound mode. Internal sound_mode name."""
+ await self.device.select_sound_mode(self.zone_name, sound_mode_id)
+
def is_part_of_group(self, group_server: "MusicCastZoneDevice") -> bool:
"""Return True if the given server is the server of self's group."""
return group_server != self and (
import asyncio
import time
from collections.abc import Callable, Coroutine
+from contextlib import suppress
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, cast
from aiohttp.client_exceptions import ClientError
+from aiomusiccast.capabilities import BinarySensor as MCBinarySensor
+from aiomusiccast.capabilities import BinarySetter as MCBinarySetter
+from aiomusiccast.capabilities import NumberSensor as MCNumberSensor
+from aiomusiccast.capabilities import NumberSetter as MCNumberSetter
+from aiomusiccast.capabilities import OptionSetter as MCOptionSetter
+from aiomusiccast.capabilities import TextSensor as MCTextSensor
from aiomusiccast.exceptions import MusicCastGroupException
from aiomusiccast.pyamaha import MusicCastConnectionException
from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
from music_assistant_models.enums import ConfigEntryType, PlaybackState, PlayerFeature
-from music_assistant_models.player import DeviceInfo, PlayerMedia, PlayerSource
+from music_assistant_models.player import (
+ DeviceInfo,
+ PlayerMedia,
+ PlayerOption,
+ PlayerOptionEntry,
+ PlayerOptionType,
+ PlayerOptionValueType,
+ PlayerSoundMode,
+ PlayerSource,
+)
+from music_assistant_models.unique_list import UniqueList
from propcache import under_cached_property as cached_property
from music_assistant.models.player import Player
CONF_PLAYER_HANDLE_SOURCE_DISABLED,
CONF_PLAYER_SWITCH_SOURCE_NON_NET,
CONF_PLAYER_TURN_OFF_ON_LEAVE,
+ MC_CAPABILITIES,
MC_CONTROL_SOURCE_IDS,
MC_NETUSB_SOURCE_IDS,
MC_PASSIVE_SOURCE_IDS,
MC_POLL_INTERVAL,
+ MC_SOUND_MODE_FRIENDLY_NAMES,
MC_SOURCE_MAIN_SYNC,
MC_SOURCE_MC_LINK,
PLAYER_CONFIG_ENTRIES,
PlayerFeature.NEXT_PREVIOUS,
PlayerFeature.ENQUEUE,
PlayerFeature.GAPLESS_PLAYBACK,
+ PlayerFeature.SELECT_SOUND_MODE,
+ PlayerFeature.OPTIONS,
}
self._attr_device_info = DeviceInfo(
)
)
+ # SOUND MODES
+ for source_id in self.zone_device.sound_mode_list:
+ friendly_name = MC_SOUND_MODE_FRIENDLY_NAMES.get(source_id) or " ".join(
+ [x.capitalize() for x in source_id.split("_")]
+ )
+ self._attr_sound_mode_list.append(
+ PlayerSoundMode(id=source_id, name=friendly_name, passive=False)
+ )
+
async def set_dynamic_attributes(self) -> None:
"""Update Player attributes."""
# ruff: noqa: PLR0915
else None
)
+ # SOUND MODE
+ self._attr_active_sound_mode = self.zone_device.sound_mode_id
+
# GROUPING
# A zone cannot be synced to another zone or main of the same device.
# Additionally, a zone can only be synced, if main is currently not using any netusb
self._get_player_id_from_zone_device(x) for x in self.zone_device.musiccast_group
]
+ # PLAYER OPTIONS
+ # see https://github.com/vigonotion/aiomusiccast/blob/main/aiomusiccast/capabilities.py
+ # capability can be any instance of OptionSetter, BinarySetter, NumberSetter, NumberSensor,
+ # BinarySensor, TextSensor
+ # the type hint of the lib's zone_data.capabilities is wrong (_not_ list[str])
+ self._attr_options = []
+ for capability in cast(
+ "list[MC_CAPABILITIES]",
+ zone_data.capabilities,
+ ):
+ if isinstance(capability, MCBinarySensor):
+ self._attr_options.append(
+ PlayerOption(
+ key=capability.id,
+ name=capability.name,
+ type=PlayerOptionType.BOOLEAN,
+ read_only=True,
+ value=capability.current,
+ )
+ )
+ elif isinstance(capability, MCBinarySetter):
+ self._attr_options.append(
+ PlayerOption(
+ key=capability.id,
+ name=capability.name,
+ type=PlayerOptionType.BOOLEAN,
+ value=capability.current,
+ read_only=False,
+ )
+ )
+ elif isinstance(capability, MCNumberSensor):
+ self._attr_options.append(
+ PlayerOption(
+ key=capability.id,
+ name=capability.name,
+ type=PlayerOptionType.INTEGER,
+ value=capability.current,
+ read_only=True,
+ )
+ )
+ elif isinstance(capability, MCNumberSetter):
+ self._attr_options.append(
+ PlayerOption(
+ key=capability.id,
+ name=capability.name,
+ type=PlayerOptionType.INTEGER,
+ value=capability.current,
+ read_only=False,
+ min_value=capability.value_range.minimum,
+ max_value=capability.value_range.maximum,
+ step=capability.value_range.step,
+ )
+ )
+ elif isinstance(capability, MCTextSensor):
+ self._attr_options.append(
+ PlayerOption(
+ key=capability.id,
+ name=capability.name,
+ type=PlayerOptionType.STRING,
+ value=capability.current,
+ read_only=True,
+ )
+ )
+ elif isinstance(capability, MCOptionSetter):
+ options = []
+ for option_key, option_name in capability.options.items():
+ options.append(
+ PlayerOptionEntry(
+ key=str(option_key), # aiomusiccast allows str and int.
+ name=option_name,
+ value=str(option_key),
+ type=PlayerOptionType.STRING,
+ )
+ )
+ self._attr_options.append(
+ PlayerOption(
+ key=capability.id,
+ name=capability.name,
+ type=PlayerOptionType.STRING,
+ value=str(capability.current),
+ read_only=False,
+ options=UniqueList(options),
+ )
+ )
+
self.update_state()
@cached_property
"""Select source command."""
await self._cmd_run(self.zone_device.select_source, source)
+ async def select_sound_mode(self, sound_mode: str) -> None:
+ """Select sound Mode Command."""
+ await self._cmd_run(self.zone_device.select_sound_mode, sound_mode)
+
+ async def set_option(self, option_key: str, option_value: PlayerOptionValueType) -> None:
+ """Set player option."""
+ if self.zone_device.zone_data is None:
+ return
+ for capability in cast(
+ "list[MC_CAPABILITIES]",
+ self.zone_device.zone_data.capabilities,
+ ):
+ if str(capability.id) != option_key:
+ continue
+ if not isinstance(capability, MCBinarySetter | MCNumberSetter | MCOptionSetter):
+ self.logger.error(f"Option {capability.name} is read only!")
+ return
+ if isinstance(capability, MCBinarySetter):
+ await capability.set(bool(option_value))
+ elif isinstance(capability, MCNumberSetter):
+ min_value = capability.value_range.minimum
+ max_value = capability.value_range.maximum
+ if not min_value <= int(option_value) <= max_value:
+ self.logger.error(
+ f"Option {capability.name} has numeric range of"
+ f"{min_value} <= value <= {max_value}"
+ )
+ return
+ await capability.set(int(option_value))
+ elif isinstance(capability, MCOptionSetter):
+ assert isinstance(option_value, str | int) # for type checking
+ _option_value = option_value # we may have an int in aiomusiccast as key
+ with suppress(ValueError):
+ _option_value = int(_option_value)
+ if _option_value not in capability.options:
+ self.logger.error(f"Option {_option_value} is not allowed for {option_key}")
+ return
+ await capability.set(_option_value)
+ break
+
async def ungroup(self) -> None:
"""Ungroup command."""
if self.zone_device.zone_name.startswith("zone"):
projects / music-assistant-server.git / commitdiff