# ty:ignore[unresolved-import] """Service actions for mass_queue.""" from __future__ import annotations from homeassistant.core import ( ServiceCall, SupportsResponse, callback, ) from .const import ( ATTR_CONFIG_ENTRY_ID, ATTR_PAGE, ATTR_PLAYER_ENTITY, ATTR_PLAYLIST_ID, ATTR_POSITIONS_TO_REMOVE, ATTR_QUEUE_ITEM_ID, ATTR_URI, DOMAIN, LOGGER, SERVICE_CLEAR_QUEUE_FROM_HERE, SERVICE_GET_ALBUM, SERVICE_GET_ALBUM_TRACKS, SERVICE_GET_ARTIST, SERVICE_GET_ARTIST_TRACKS, SERVICE_GET_GROUP_VOLUME, SERVICE_GET_PLAYLIST, SERVICE_GET_PLAYLIST_TRACKS, SERVICE_GET_PODCAST, SERVICE_GET_PODCAST_EPISODES, SERVICE_GET_QUEUE_ITEMS, SERVICE_GET_RECOMMENDATIONS, SERVICE_MOVE_QUEUE_ITEM_DOWN, SERVICE_MOVE_QUEUE_ITEM_NEXT, SERVICE_MOVE_QUEUE_ITEM_UP, SERVICE_PLAY_QUEUE_ITEM, SERVICE_REMOVE_PLAYLIST_TRACKS, SERVICE_REMOVE_QUEUE_ITEM, SERVICE_SEND_COMMAND, SERVICE_SET_GROUP_VOLUME, SERVICE_UNFAVORITE_CURRENT_ITEM, ) from .schemas import ( CLEAR_QUEUE_FROM_HERE_SERVICE_SCHEMA, GET_DATA_SERVICE_SCHEMA, GET_GROUP_VOLUME_SERVICE_SCHEMA, GET_PODCAST_EPISODES_SERVICE_SCHEMA, GET_RECOMMENDATIONS_SERVICE_SCHEMA, GET_TRACKS_SERVICE_SCHEMA, MOVE_QUEUE_ITEM_DOWN_SERVICE_SCHEMA, MOVE_QUEUE_ITEM_NEXT_SERVICE_SCHEMA, MOVE_QUEUE_ITEM_UP_SERVICE_SCHEMA, PLAY_QUEUE_ITEM_SERVICE_SCHEMA, QUEUE_ITEMS_SERVICE_SCHEMA, REMOVE_PLAYLIST_TRACKS_SERVICE_SCHEMA, REMOVE_QUEUE_ITEM_SERVICE_SCHEMA, SEND_COMMAND_SERVICE_SCHEMA, SET_GROUP_VOLUME_SERVICE_SCHEMA, UNFAVORITE_CURRENT_ITEM_SERVICE_SCHEMA, ) from .utils import get_entity_actions_controller, process_recommendations @callback def register_actions(hass) -> None: """Registers actions with Home Assistant.""" hass.services.async_register( DOMAIN, SERVICE_GET_QUEUE_ITEMS, get_queue_items, schema=QUEUE_ITEMS_SERVICE_SCHEMA, supports_response=SupportsResponse.ONLY, ) hass.services.async_register( DOMAIN, SERVICE_MOVE_QUEUE_ITEM_DOWN, move_queue_item_down, schema=MOVE_QUEUE_ITEM_DOWN_SERVICE_SCHEMA, supports_response=SupportsResponse.NONE, ) hass.services.async_register( DOMAIN, SERVICE_MOVE_QUEUE_ITEM_NEXT, move_queue_item_next, schema=MOVE_QUEUE_ITEM_NEXT_SERVICE_SCHEMA, supports_response=SupportsResponse.NONE, ) hass.services.async_register( DOMAIN, SERVICE_MOVE_QUEUE_ITEM_UP, move_queue_item_up, schema=MOVE_QUEUE_ITEM_UP_SERVICE_SCHEMA, supports_response=SupportsResponse.NONE, ) hass.services.async_register( DOMAIN, SERVICE_PLAY_QUEUE_ITEM, play_queue_item, schema=PLAY_QUEUE_ITEM_SERVICE_SCHEMA, supports_response=SupportsResponse.NONE, ) hass.services.async_register( DOMAIN, SERVICE_REMOVE_QUEUE_ITEM, remove_queue_item, schema=REMOVE_QUEUE_ITEM_SERVICE_SCHEMA, supports_response=SupportsResponse.NONE, ) hass.services.async_register( DOMAIN, SERVICE_SEND_COMMAND, send_command, schema=SEND_COMMAND_SERVICE_SCHEMA, supports_response=SupportsResponse.OPTIONAL, ) hass.services.async_register( DOMAIN, SERVICE_UNFAVORITE_CURRENT_ITEM, unfavorite_current_item, schema=UNFAVORITE_CURRENT_ITEM_SERVICE_SCHEMA, supports_response=SupportsResponse.NONE, ) hass.services.async_register( DOMAIN, SERVICE_GET_RECOMMENDATIONS, get_recommendations, schema=GET_RECOMMENDATIONS_SERVICE_SCHEMA, supports_response=SupportsResponse.ONLY, ) hass.services.async_register( DOMAIN, SERVICE_GET_GROUP_VOLUME, get_group_volume, schema=GET_GROUP_VOLUME_SERVICE_SCHEMA, supports_response=SupportsResponse.ONLY, ) hass.services.async_register( DOMAIN, SERVICE_SET_GROUP_VOLUME, set_group_volume, schema=SET_GROUP_VOLUME_SERVICE_SCHEMA, supports_response=SupportsResponse.NONE, ) hass.services.async_register( DOMAIN, SERVICE_CLEAR_QUEUE_FROM_HERE, clear_queue_from_here, schema=CLEAR_QUEUE_FROM_HERE_SERVICE_SCHEMA, supports_response=SupportsResponse.NONE, ) hass.services.async_register( DOMAIN, SERVICE_GET_PLAYLIST_TRACKS, get_playlist_tracks, schema=GET_TRACKS_SERVICE_SCHEMA, supports_response=SupportsResponse.ONLY, ) hass.services.async_register( DOMAIN, SERVICE_GET_ALBUM_TRACKS, get_album_tracks, schema=GET_TRACKS_SERVICE_SCHEMA, supports_response=SupportsResponse.ONLY, ) hass.services.async_register( DOMAIN, SERVICE_GET_ARTIST_TRACKS, get_artist_tracks, schema=GET_TRACKS_SERVICE_SCHEMA, supports_response=SupportsResponse.ONLY, ) hass.services.async_register( DOMAIN, SERVICE_GET_PODCAST_EPISODES, get_podcast_episodes, schema=GET_PODCAST_EPISODES_SERVICE_SCHEMA, supports_response=SupportsResponse.ONLY, ) hass.services.async_register( DOMAIN, SERVICE_GET_ALBUM, get_album, schema=GET_DATA_SERVICE_SCHEMA, supports_response=SupportsResponse.ONLY, ) hass.services.async_register( DOMAIN, SERVICE_GET_ARTIST, get_artist, schema=GET_DATA_SERVICE_SCHEMA, supports_response=SupportsResponse.ONLY, ) hass.services.async_register( DOMAIN, SERVICE_GET_PLAYLIST, get_playlist, schema=GET_DATA_SERVICE_SCHEMA, supports_response=SupportsResponse.ONLY, ) hass.services.async_register( DOMAIN, SERVICE_GET_PODCAST, get_podcast, schema=GET_DATA_SERVICE_SCHEMA, supports_response=SupportsResponse.ONLY, ) hass.services.async_register( DOMAIN, SERVICE_REMOVE_PLAYLIST_TRACKS, remove_playlist_tracks, schema=REMOVE_PLAYLIST_TRACKS_SERVICE_SCHEMA, supports_response=SupportsResponse.NONE, ) async def get_queue_items(call: ServiceCall): """Service wrapper to get queue items.""" entity_id = call.data[ATTR_PLAYER_ENTITY] hass = call.hass actions = get_entity_actions_controller(hass, entity_id) return await actions.get_queue_items(call) async def move_queue_item_down(call: ServiceCall): """Service wrapper to move queue item down.""" entity_id = call.data[ATTR_PLAYER_ENTITY] hass = call.hass actions = get_entity_actions_controller(hass, entity_id) return await actions.move_queue_item_down(call) async def move_queue_item_next(call: ServiceCall): """Service wrapper to move queue item next.""" entity_id = call.data[ATTR_PLAYER_ENTITY] hass = call.hass actions = get_entity_actions_controller(hass, entity_id) return await actions.move_queue_item_next(call) async def move_queue_item_up(call: ServiceCall): """Service wrapper to move queue item up.""" entity_id = call.data[ATTR_PLAYER_ENTITY] hass = call.hass actions = get_entity_actions_controller(hass, entity_id) return await actions.move_queue_item_up(call) async def play_queue_item(call: ServiceCall): """Service wrapper to play a queue item.""" entity_id = call.data[ATTR_PLAYER_ENTITY] hass = call.hass actions = get_entity_actions_controller(hass, entity_id) return await actions.play_queue_item(call) async def remove_queue_item(call: ServiceCall): """Service wrapper to remove a queue item.""" entity_id = call.data[ATTR_PLAYER_ENTITY] hass = call.hass actions = get_entity_actions_controller(hass, entity_id) return await actions.remove_queue_item(call) async def send_command(call: ServiceCall): """Service wrapper to send command to Music Assistant.""" entry_id = call.data[ATTR_CONFIG_ENTRY_ID] hass = call.hass entry = hass.config_entries.async_get_entry(entry_id) actions = entry.runtime_data.actions return await actions.send_command(call) async def unfavorite_current_item(call: ServiceCall): """Service wrapper to unfavorite currently playing item.""" entity_id = call.data[ATTR_PLAYER_ENTITY] hass = call.hass actions = get_entity_actions_controller(hass, entity_id) await actions.unfavorite_item(call) async def get_recommendations(call: ServiceCall): """Service wrapper to get recommendations from providers.""" entity_id = call.data[ATTR_PLAYER_ENTITY] hass = call.hass actions = get_entity_actions_controller(hass, entity_id) result = await actions.get_recommendations(call) return {"response": process_recommendations(result)} async def get_group_volume(call: ServiceCall): """Service wrapper to get grouped volume.""" entity_id = call.data[ATTR_PLAYER_ENTITY] hass = call.hass actions = get_entity_actions_controller(hass, entity_id) result = await actions.get_group_volume(call) return {"volume_level": result} async def set_group_volume(call: ServiceCall): """Service wrapper to set grouped volume.""" entity_id = call.data[ATTR_PLAYER_ENTITY] hass = call.hass actions = get_entity_actions_controller(hass, entity_id) await actions.set_group_volume(call) def filter_queue_after(queue, current_idx): """Returns all items after the current active track.""" if current_idx == len(queue): return [] return queue[current_idx + 1 :] async def clear_queue_from_here(call: ServiceCall): """Service wrapper to clear queue from point.""" entity_id = call.data[ATTR_PLAYER_ENTITY] hass = call.hass actions = get_entity_actions_controller(hass, entity_id) current_idx = await actions.get_queue_index(entity_id) LOGGER.debug(f"Current Index: {current_idx}") queue_id = actions.get_queue_id(entity_id) LOGGER.debug(f"Queue ID: {queue_id}") queue = actions._controller.queues.get(queue_id) LOGGER.debug(f"Queue length: {len(queue)}") client = actions._client if len(queue) == current_idx: return items = queue[current_idx + 1 :] LOGGER.debug(f"Filtered length: {len(items)}") LOGGER.debug(f"First item to remove {items[0]}") LOGGER.debug(f"Last item to remove {items[-1]}") for item in items: queue_item_id = item[ATTR_QUEUE_ITEM_ID] await client.player_queues.queue_command_delete(queue_id, queue_item_id) async def get_album_tracks(call: ServiceCall): """Gets all tracks in an album.""" config_entry = call.data[ATTR_CONFIG_ENTRY_ID] uri = call.data[ATTR_URI] page = call.data.get(ATTR_PAGE) hass = call.hass entry = hass.config_entries.async_get_entry(config_entry) actions = entry.runtime_data.actions return { "tracks": await actions.get_album_tracks(uri, page), } async def get_artist_tracks(call: ServiceCall): """Gets all tracks for an artist.""" config_entry = call.data[ATTR_CONFIG_ENTRY_ID] uri = call.data[ATTR_URI] hass = call.hass entry = hass.config_entries.async_get_entry(config_entry) actions = entry.runtime_data.actions return { "tracks": await actions.get_artist_tracks(uri), } async def get_playlist_tracks(call: ServiceCall): """Gets all tracks in a playlist.""" config_entry = call.data[ATTR_CONFIG_ENTRY_ID] uri = call.data[ATTR_URI] page = call.data.get(ATTR_PAGE) hass = call.hass entry = hass.config_entries.async_get_entry(config_entry) actions = entry.runtime_data.actions return { "tracks": await actions.get_playlist_tracks(uri, page), } async def get_podcast_episodes(call: ServiceCall): """Gets all episodes for a podcast.""" config_entry = call.data[ATTR_CONFIG_ENTRY_ID] uri = call.data[ATTR_URI] hass = call.hass entry = hass.config_entries.async_get_entry(config_entry) actions = entry.runtime_data.actions return { "episodes": await actions.get_podcast_episodes(uri), } async def get_album(call: ServiceCall): """Returns the details about an album from the server.""" config_entry = call.data[ATTR_CONFIG_ENTRY_ID] uri = call.data[ATTR_URI] hass = call.hass entry = hass.config_entries.async_get_entry(config_entry) actions = entry.runtime_data.actions return (await actions.get_album_details(uri)).to_dict() async def get_artist(call: ServiceCall): """Returns the details about an artist from the server.""" config_entry = call.data[ATTR_CONFIG_ENTRY_ID] uri = call.data[ATTR_URI] hass = call.hass entry = hass.config_entries.async_get_entry(config_entry) actions = entry.runtime_data.actions return (await actions.get_artist_details(uri)).to_dict() async def get_playlist(call: ServiceCall): """Returns the details about a playlist from the server.""" config_entry = call.data[ATTR_CONFIG_ENTRY_ID] uri = call.data[ATTR_URI] hass = call.hass entry = hass.config_entries.async_get_entry(config_entry) actions = entry.runtime_data.actions return (await actions.get_playlist_details(uri)).to_dict() async def get_podcast(call: ServiceCall): """Returns the details about a podcast from the server.""" config_entry = call.data[ATTR_CONFIG_ENTRY_ID] uri = call.data[ATTR_URI] hass = call.hass entry = hass.config_entries.async_get_entry(config_entry) actions = entry.runtime_data.actions return (await actions.get_podcast_details(uri)).to_dict() async def remove_playlist_tracks(call: ServiceCall): """Removes one or more items from a playlist.""" config_entry = call.data[ATTR_CONFIG_ENTRY_ID] playlist = call.data[ATTR_PLAYLIST_ID] positions = call.data[ATTR_POSITIONS_TO_REMOVE] if isinstance(positions, int): positions = [positions] positions = [int(position) for position in positions] hass = call.hass entry = hass.config_entries.async_get_entry(config_entry) actions = entry.runtime_data.actions await actions.remove_playlist_tracks(playlist, positions)