# ty:ignore[unresolved-import] """Utilities.""" from __future__ import annotations import base64 import urllib.parse from typing import TYPE_CHECKING from aiocache import cached from aiocache.serializers import PickleSerializer from homeassistant.config_entries import ConfigEntryState from homeassistant.core import async_get_hass, callback from homeassistant.exceptions import ServiceValidationError from homeassistant.helpers import aiohttp_client from homeassistant.helpers import device_registry as dr from homeassistant.helpers import entity_registry as er if TYPE_CHECKING: from homeassistant.core import HomeAssistant from music_assistant_client import MusicAssistantClient from . import MassQueueEntryData from .const import ATTR_QUEUE_ID, LOGGER @callback def _get_config_entry( hass: HomeAssistant, config_entry_id: str, ) -> MusicAssistantClient: """Get Music Assistant Client from config_entry_id.""" entry: MassQueueEntryData | None if not (entry := hass.config_entries.async_get_entry(config_entry_id)): exc = "Entry not found." raise ServiceValidationError(exc) if entry.state is not ConfigEntryState.LOADED: exc = "Entry not loaded" raise ServiceValidationError(exc) return entry def get_mass_queue_entry(hass, entity_id): """Gets the actions for the selected entity.""" mass_entry = get_mass_entry(hass, entity_id) unique_id = mass_entry.unique_id return find_mass_queue_entry_from_unique_id(hass, unique_id) def get_entity_actions_controller(hass, entity_id): """Gets the actions for the selected entity.""" mass_queue_entry = get_mass_queue_entry(hass, entity_id) return mass_queue_entry.runtime_data.actions def get_mass_client(hass, entity_id): """Gets the actions for the selected entity.""" mass_queue_entry = get_mass_queue_entry(hass, entity_id) return mass_queue_entry.runtime_data.mass def get_mass_entry(hass, entity_id): """Helper function to pull MA Config Entry.""" config_id = _get_mass_entity_config_entry_id(hass, entity_id) return _get_config_entry(hass, config_id) def _get_mass_entity_config_entry_id(hass, entity_id): """Helper to grab config entry ID from entity ID.""" registry = er.async_get(hass) return registry.async_get(entity_id).config_entry_id def find_mass_queue_entry_from_unique_id(hass: HomeAssistant, unique_id: str): """Finds the mass_queue entry for the given MA URL.""" entries = _get_mass_queue_entries(hass) for entry in entries: if entry.unique_id == unique_id: return entry msg = f"Cannot find entry for Music Assistant Queue Actions with unique ID {unique_id}. Are the integrations for Music Assistant and Music Assistant Queue Actions configured?" raise ServiceValidationError(msg) def _get_mass_queue_entries(hass): """Gets all entries for mass_queue domain.""" entries = hass.config_entries.async_entries() return [entry for entry in entries if entry.domain == "mass_queue"] def format_event_data_queue_item(queue_item): """Format event data results for usage by controller.""" if queue_item is None: return None if queue_item.get("queue_id") is None: return queue_item item_cp = queue_item.copy() if "streamdetails" in item_cp: item_cp.pop("streamdetails") if "media_item" in item_cp: item_cp.pop("media_item") return item_cp def format_queue_updated_event_data(event: dict): """Format queue updated results for usage by controller.""" event_data = event.copy() event_data["current_item"] = format_event_data_queue_item( event_data.get("current_item"), ) event_data["next_item"] = format_event_data_queue_item(event_data.get("next_item")) return event_data def get_queue_id_from_player_data(player_data): """Force as dict if not already.""" data = player_data.to_dict() if type(player_data) is not dict else player_data current_media = data.get("current_media", None) if current_media is None: return None return current_media.get("queue_id") def return_image_or_none(img_data: dict | None, remotely_accessible: bool): """Returns None if image is not present or not remotely accessible.""" if type(img_data) is dict: img = img_data.get("path") remote = img_data.get("remotely_accessible") if remote or not remotely_accessible: return img return None def search_image_list(images: list, remotely_accessible: bool): """Checks through a list of image data and attempts to find an image.""" result = None for item in images: image = return_image_or_none(item, remotely_accessible) if image is not None: result = image break return result def find_image_from_image(data: dict, remotely_accessible: bool): """Attempts to find the image via the image key.""" img_data = data.get("image") return return_image_or_none(img_data, remotely_accessible) def find_image_from_metadata(data: dict, remotely_accessible: bool): """Attempts to find the image via the metadata key.""" metadata = data.get("metadata", {}) img_data = metadata.get("images") if img_data is None: return None return search_image_list(img_data, remotely_accessible) def find_image_from_album(data: dict, remotely_accessible: bool): """Attempts to find the image via the album key.""" album = data.get("album") or {} metadata = album.get("metadata") or {} img_data = metadata.get("images") if img_data is None: return None return search_image_list(img_data, remotely_accessible) def find_image_from_artists(data: dict, remotely_accessible: bool): """Attempts to find the image via the artists key.""" artist = data.get("artist", {}) img_data = artist.get("image") or [] img_data += artist.get("metadata", {}) if len(img_data): return search_image_list(img_data, remotely_accessible) if isinstance(img_data, dict): return return_image_or_none(img_data, remotely_accessible) return None def find_image(data: dict, remotely_accessible: bool = True): """Returns None if image is not present or not remotely accessible.""" media_item = data.get("media_item", data) from_image = find_image_from_image(data, remotely_accessible) from_metadata = find_image_from_metadata(media_item, remotely_accessible) from_album = find_image_from_album(data, remotely_accessible) from_artists = find_image_from_artists(data, remotely_accessible) return from_image or from_metadata or from_album or from_artists def _get_recommendation_item_image_from_metadata(item: dict): try: images = item["metadata"]["images"] accessible = [image for image in images if image["remotely_accessible"]] if accessible: return accessible[0]["path"] except: # noqa: E722 S110 pass return "" def _get_recommendation_item_image_from_image(item: dict): try: image_data = item["image"] accessible = image_data["remotely_accessible"] if accessible: return image_data["path"] except: # noqa: E722 S110 pass return "" def _get_recommendation_item_image(item: dict): meta_img = _get_recommendation_item_image_from_metadata(item) img_img = _get_recommendation_item_image_from_image(item) if len(meta_img): return meta_img return img_img def process_recommendation_section_item(item: dict): """Process and reformat a single recommendation item.""" LOGGER.debug(f"Got section item: {item}") return { "item_id": item["item_id"], "name": item["name"], "sort_name": item["sort_name"], "uri": item["uri"], "media_type": item["media_type"], "image": _get_recommendation_item_image(item), } def process_recommendation_section_items(items: list): """Process and reformat items for a single recommendation section.""" return [process_recommendation_section_item(item) for item in items] def process_recommendation_section(section): """Process and reformat a single recommendation section.""" LOGGER.debug(f"Got section: {section}") section = section.to_dict() return { "item_id": section["item_id"], "provider": section["provider"], "sort_name": section["sort_name"], "name": section["name"], "uri": section["uri"], "icon": section["icon"], "image": section["image"], "items": process_recommendation_section_items(section["items"]), } def process_recommendations(recs: list): """Process and reformat items all recommendation sections.""" result = [] for rec in recs: processed = process_recommendation_section(rec) if len(processed["items"]): result.append(processed) return result def generate_image_url_from_image_data(image_data: dict, client): """Generates an image URL from `image_data`.""" img_path = image_data["path"] provider = image_data["provider"] base_url = "" if img_path.startswith("http") else client.server_url img = urllib.parse.quote_plus(urllib.parse.quote_plus(img_path)) return f"{base_url}/imageproxy?provider={provider}&size=256&format=png&path={img}" async def download_single_image_from_image_data( image_data: dict, entity_id, hass, session, ): """Downloads a single image from Music Assistant and returns the base64 encoded string.""" entry = get_mass_entry(hass, entity_id) client = entry.runtime_data.mass url = generate_image_url_from_image_data(image_data, client) try: req = await session.get(url) read = await req.content.read() return f"data:image;base64,{base64.b64encode(read).decode('utf-8')}" except: # noqa: E722 LOGGER.error(f"Unable to get image with data {image_data}") return None @cached(serializer=PickleSerializer()) async def download_and_encode_image(url: str): """Downloads and encodes a single image from the given URL.""" hass = async_get_hass() session = aiohttp_client.async_get_clientsession(hass) req = await session.get(url) read = await req.content.read() return f"data:image;base64,{base64.b64encode(read).decode('utf-8')}" async def get_user_info(hass: HomeAssistant, entity_id: str, username: str): """Returns the user information for the given username.""" client = get_mass_client(hass, entity_id) users = await client.auth.list_users() LOGGER.debug(f"Client: {client}") LOGGER.debug(f"Users: {users}") return [user.to_dict() for user in users if user.username == username][0] def get_entity_info(hass: HomeAssistant, entity_id: str): """Gets the server and client info for a given player.""" client = get_mass_client(hass, entity_id) state = hass.states.get(entity_id) device_registry = dr.async_get(hass) entity_registry = er.async_get(hass) dev_id = entity_registry.async_get(entity_id).device_id dev = device_registry.async_get(dev_id) identifiers = dev.identifiers player_id = [_id[1] for _id in identifiers if _id[0] == "music_assistant"][0] player = client.players.get(player_id) mass_entry_id = _get_mass_entity_config_entry_id(hass, entity_id) mass_queue_id = get_mass_queue_entry(hass, entity_id).entry_id queue_id = state.attributes.get(ATTR_QUEUE_ID) server_url = client.server_info.base_url ws_url = client.connection.ws_server_url config_url = dev.configuration_url manufacturer = dev.manufacturer model = dev.model available = player.available can_group_with = player.can_group_with ip_address = player.device_info.ip_address features = list(player.supported_features) name = player.name provider = player.provider synced_to = player.synced_to player_type = player.type return { "available": available, "can_group_with": can_group_with, "connection": { "configuration_url": config_url, "url": ip_address, }, "entries": { "music_assistant": mass_entry_id, "mass_queue": mass_queue_id, }, "features": features, "manufacturer": manufacturer, "model": model, "name": name, "player_id": player_id, "provider": provider, "queue_id": queue_id, "server": { "connection": { "url": server_url, "websocket": ws_url, }, }, "synced_to": synced_to, "type": player_type, } def parse_uri(uri): """Parse a URI and split to provider and item ID.""" provider = uri.split("://")[0] item_id = uri.split("/")[-1] return [provider, item_id]