"""Run the Music Assistant Server."""
+
from __future__ import annotations
import argparse
"""Music Assistant Client: Manage a Music Assistant server remotely."""
+
from .client import MusicAssistantClient # noqa: F401
"""Music Assistant Client: Manage a Music Assistant server remotely."""
+
from __future__ import annotations
import asyncio
""""Connect o a remote Music Assistant Server using the default Websocket API."""
+
from __future__ import annotations
import logging
"""Client-specific Exceptions for Music Assistant."""
+
from __future__ import annotations
"""Handle Music/library related endpoints for Music Assistant."""
+
from __future__ import annotations
import urllib.parse
"""Handle player related endpoints for Music Assistant."""
+
from __future__ import annotations
from collections.abc import Iterator
"""Helpers for date and time."""
+
from __future__ import annotations
import datetime
"""Helper and utility functions."""
+
from __future__ import annotations
import asyncio
"""Model and helpers for Config entries."""
+
from __future__ import annotations
import logging
"""All enums used by the Music Assistant models."""
+
from __future__ import annotations
from enum import StrEnum
"""Models and helpers for media items."""
+
from __future__ import annotations
from dataclasses import dataclass, field, fields
"""Model(s) for Player."""
+
from __future__ import annotations
import time
"""Model(s) for PlayerQueue."""
+
from __future__ import annotations
import time
"""Model a QueueItem."""
+
from __future__ import annotations
from dataclasses import dataclass
DB_TABLE_PROVIDER_MAPPINGS: Final[str] = "provider_mappings"
# all other
-MASS_LOGO_ONLINE: Final[
- str
-] = "https://github.com/home-assistant/brands/raw/master/custom_integrations/mass/icon%402x.png"
+MASS_LOGO_ONLINE: Final[str] = (
+ "https://github.com/home-assistant/brands/raw/master/custom_integrations/mass/icon%402x.png"
+)
ENCRYPT_SUFFIX = "_encrypted_"
SECURE_STRING_SUBSTITUTE = "this_value_is_encrypted"
CONFIGURABLE_CORE_CONTROLLERS = (
"""Provides a simple stateless caching system."""
+
from __future__ import annotations
import asyncio
"""Logic to handle storage of persistent (configuration) settings."""
+
from __future__ import annotations
import asyncio
"""Manage MediaItems of type Album."""
+
from __future__ import annotations
import asyncio
"""Manage MediaItems of type Artist."""
+
from __future__ import annotations
import asyncio
"""Base (ABC) MediaType specific controller."""
+
from __future__ import annotations
import logging
"""Manage MediaItems of type Playlist."""
+
from __future__ import annotations
import asyncio
"""Manage MediaItems of type Radio."""
+
from __future__ import annotations
import asyncio
"""Manage MediaItems of type Track."""
+
from __future__ import annotations
import asyncio
"""All logic for metadata retrieval."""
+
from __future__ import annotations
import asyncio
"""MusicController: Orchestrates all data from music providers and sync to internal database."""
+
from __future__ import annotations
import asyncio
"""Logic to play music from MusicProviders to supported players."""
+
from __future__ import annotations
import logging
"""Logic to play music from MusicProviders to supported players."""
+
from __future__ import annotations
import asyncio
purely to stream audio packets to players and some control endpoints such as
the upnp callbacks and json rpc api for slimproto clients.
"""
+
from __future__ import annotations
import asyncio
Unlike the streamserver (which is as simple and unprotected as possible),
this webserver allows for more fine grained configuration to better secure it.
"""
+
from __future__ import annotations
import asyncio
"""Helpers for dealing with API's to interact with Music Assistant."""
+
from __future__ import annotations
import inspect
"""Various helpers for audio manipulation."""
+
from __future__ import annotations
import asyncio
"""Helper(s) to deal with authentication for (music) providers."""
+
from __future__ import annotations
import asyncio
"""Several helper/utils to compare objects."""
+
from __future__ import annotations
import re
"""Database helpers and logic."""
+
from __future__ import annotations
from collections.abc import AsyncGenerator, Mapping
"""Helper(s) to create DIDL Lite metadata for Sonos/DLNA players."""
+
from __future__ import annotations
import datetime
"""Utilities for image manipulation and retrieval."""
+
from __future__ import annotations
import asyncio
All rights reserved.
"""
+
from __future__ import annotations
import asyncio
@overload
def catch_log_exception(
func: Callable[..., Coroutine[Any, Any, Any]], format_err: Callable[..., Any]
-) -> Callable[..., Coroutine[Any, Any, None]]:
- ...
+) -> Callable[..., Coroutine[Any, Any, None]]: ...
@overload
def catch_log_exception(
func: Callable[..., Any], format_err: Callable[..., Any]
-) -> Callable[..., None] | Callable[..., Coroutine[Any, Any, None]]:
- ...
+) -> Callable[..., None] | Callable[..., Coroutine[Any, Any, None]]: ...
def catch_log_exception(
"""Helpers for parsing playlists."""
+
from __future__ import annotations
import asyncio
The subprocess implementation in asyncio can (still) sometimes cause deadlocks,
even when properly handling reading/writes from different tasks.
"""
+
from __future__ import annotations
import asyncio
"""Helpers/utilities to parse ID3 tags from audio files with ffmpeg."""
+
from __future__ import annotations
import json
"""Various (server-only) tools and helpers."""
+
from __future__ import annotations
import asyncio
"""Base Webserver logic for an HTTPServer that can handle dynamic routes."""
+
from __future__ import annotations
import logging
"""Server specific/only models."""
+
from __future__ import annotations
from typing import TYPE_CHECKING, Protocol
"""Model/base for a Core controller within Music Assistant."""
+
from __future__ import annotations
import logging
"""Model/base for a Metadata Provider implementation."""
+
from __future__ import annotations
from typing import TYPE_CHECKING
"""Model/base for a Music Provider implementation."""
+
from __future__ import annotations
from collections.abc import AsyncGenerator
"""Model/base for a Metadata Provider implementation."""
+
from __future__ import annotations
from abc import abstractmethod
"""Model/base for a Plugin Provider implementation."""
+
from __future__ import annotations
from typing import TYPE_CHECKING
"""Model/base for a Provider implementation within Music Assistant."""
+
from __future__ import annotations
import logging
It uses the amazing work of Philippe44 who created a bridge from airplay to slimproto.
https://github.com/philippe44/LMS-Raop
"""
+
from __future__ import annotations
import asyncio
"""Chromecast Player provider for Music Assistant, utilizing the pychromecast library."""
+
from __future__ import annotations
import asyncio
stream_type = STREAM_TYPE_BUFFERED
metadata = {
"metadataType": 3,
- "albumName": queue_item.media_item.album.name
- if queue_item.media_item.album
- else "",
+ "albumName": (
+ queue_item.media_item.album.name if queue_item.media_item.album else ""
+ ),
"songName": queue_item.media_item.name,
- "artist": queue_item.media_item.artists[0].name
- if queue_item.media_item.artists
- else "",
+ "artist": (
+ queue_item.media_item.artists[0].name if queue_item.media_item.artists else ""
+ ),
"title": queue_item.media_item.name,
"images": [{"url": image_url}] if image_url else None,
}
"""Helpers to deal with Cast devices."""
+
from __future__ import annotations
import urllib.error
"""Deezer music provider support for MusicAssistant."""
+
import datetime
import hashlib
import uuid
Credits go out to RemixDev (https://gitlab.com/RemixDev) for figuring out, how to get the arl
cookie based on the api_token.
"""
+
import datetime
from http.cookies import BaseCookie, Morsel
All rights/credits reserved.
"""
+
from __future__ import annotations
import asyncio
"""Various helpers and utils for the DLNA Player Provider."""
+
from __future__ import annotations
from typing import TYPE_CHECKING
"""Fanart.tv Metadata provider for Music Assistant."""
+
from __future__ import annotations
from json import JSONDecodeError
"""Get data from api."""
url = f"http://webservice.fanart.tv/v3/{endpoint}"
kwargs["api_key"] = app_var(4)
- async with self.throttler, self.mass.http_session.get(
- url, params=kwargs, ssl=False
- ) as response:
+ async with (
+ self.throttler,
+ self.mass.http_session.get(url, params=kwargs, ssl=False) as response,
+ ):
try:
result = await response.json()
except (
"""Filesystem musicprovider support for MusicAssistant."""
+
from __future__ import annotations
import asyncio
"""Filesystem musicprovider support for MusicAssistant."""
+
from __future__ import annotations
import asyncio
"""Some helpers for Filesystem based Musicproviders."""
+
from __future__ import annotations
import os
"""SMB filesystem provider for Music Assistant."""
+
from __future__ import annotations
import asyncio
At this time only used for retrieval of ID's but to be expanded to fetch metadata too.
"""
+
from __future__ import annotations
import re
"User-Agent": f"Music Assistant/{self.mass.version} ( https://github.com/music-assistant )" # noqa: E501
}
kwargs["fmt"] = "json" # type: ignore[assignment]
- async with self.throttler, self.mass.http_session.get(
- url, headers=headers, params=kwargs, ssl=False
- ) as response:
+ async with (
+ self.throttler,
+ self.mass.http_session.get(url, headers=headers, params=kwargs, ssl=False) as response,
+ ):
try:
result = await response.json()
except (
"""Open Subsonic music provider support for MusicAssistant."""
+
from __future__ import annotations
from music_assistant.common.models.config_entries import (
"""The provider class for Open Subsonic."""
+
from __future__ import annotations
import asyncio
"""Plex musicprovider support for MusicAssistant."""
+
from __future__ import annotations
import asyncio
provider_instance=self.instance_id,
available=available,
audio_format=AudioFormat(
- content_type=ContentType.try_parse(content)
- if content
- else ContentType.UNKNOWN,
+ content_type=(
+ ContentType.try_parse(content) if content else ContentType.UNKNOWN
+ ),
),
url=plex_track.getWebURL(),
)
"""Several helpers/utils for the Plex Music Provider."""
+
from __future__ import annotations
import asyncio
"""Qobuz musicprovider support for MusicAssistant."""
+
from __future__ import annotations
import datetime
kwargs["request_sig"] = request_sig
kwargs["app_id"] = app_var(0)
kwargs["user_auth_token"] = await self._auth_token()
- async with self._throttler, self.mass.http_session.get(
- url, headers=headers, params=kwargs, ssl=False
- ) as response:
+ async with (
+ self._throttler,
+ self.mass.http_session.get(url, headers=headers, params=kwargs, ssl=False) as response,
+ ):
try:
result = await response.json()
# check for error in json
"""RadioBrowser musicprovider support for MusicAssistant."""
+
from __future__ import annotations
from collections.abc import AsyncGenerator
"""Base/builtin provider with support for players using slimproto."""
+
from __future__ import annotations
import asyncio
await client.play_url(
url=url,
mime_type=f"audio/{url.split('.')[-1].split('?')[0]}",
- metadata={"item_id": queue_item.queue_item_id, "title": queue_item.name}
- if queue_item
- else {"item_id": client.player_id, "title": "Music Assistant"},
+ metadata=(
+ {"item_id": queue_item.queue_item_id, "title": queue_item.name}
+ if queue_item
+ else {"item_id": client.player_id, "title": "Music Assistant"}
+ ),
send_flush=send_flush,
transition=SlimTransition.CROSSFADE if crossfade else SlimTransition.NONE,
transition_duration=transition_duration,
Goal is player compatibility, not API compatibility.
Users that need more, should just stay with a full blown LMS server.
"""
+
from __future__ import annotations
import asyncio
"timestamp": time.strftime("%a, %d %b %Y %H:%M:%S %Z", time.gmtime()),
"advice": {
# update interval for streaming mode
- "interval": 5000
- if streaming
- else 0
+ "interval": 5000 if streaming else 0
},
}
)
"favorites_title": item.name,
"favorites_url": item.uri,
"favorites_type": item.media_type.value,
- "icon": self.mass.metadata.get_image_url(item.image, 256)
- if item.image
- else "",
+ "icon": (
+ self.mass.metadata.get_image_url(item.image, 256) if item.image else ""
+ ),
},
"textkey": item.name[0].upper(),
"commonParams": {
"""Models used for the JSON-RPC API."""
+
from __future__ import annotations
from typing import TYPE_CHECKING, Any, TypedDict
"artist": artist,
"album": album,
"remote": 1,
- "artwork_url": mass.metadata.get_image_url(queue_item.image, 512)
- if queue_item.image
- else "",
+ "artwork_url": (
+ mass.metadata.get_image_url(queue_item.image, 512) if queue_item.image else ""
+ ),
"coverid": "-187651250107376",
"duration": queue_item.duration,
"bitrate": bitrate,
"""Snapcast Player provider for Music Assistant."""
+
from __future__ import annotations
import asyncio
"""Sample Player provider for Music Assistant."""
+
from __future__ import annotations
import asyncio
"""Soundcloud support for MusicAssistant."""
+
from __future__ import annotations
import asyncio
This file is based on soundcloudpy from Naím Rodríguez https://github.com/naim-prog
Original package https://github.com/naim-prog/soundcloud-py
"""
+
from __future__ import annotations
from collections.abc import AsyncGenerator
"""Spotify musicprovider support for MusicAssistant."""
+
from __future__ import annotations
import asyncio
"""The AudioDB Metadata provider for Music Assistant."""
+
from __future__ import annotations
from json import JSONDecodeError
async def _get_data(self, endpoint, **kwargs) -> dict | None:
"""Get data from api."""
url = f"https://theaudiodb.com/api/v1/json/{app_var(3)}/{endpoint}"
- async with self.throttler, self.mass.http_session.get(
- url, params=kwargs, ssl=False
- ) as response:
+ async with (
+ self.throttler,
+ self.mass.http_session.get(url, params=kwargs, ssl=False) as response,
+ ):
try:
result = await response.json()
except (
def inner() -> None:
match media_type:
case MediaType.ARTIST:
- TidalFavorites(session, user_id).add_artist(item_id) if add else TidalFavorites(
- session, user_id
- ).remove_artist(item_id)
+ (
+ TidalFavorites(session, user_id).add_artist(item_id)
+ if add
+ else TidalFavorites(session, user_id).remove_artist(item_id)
+ )
case MediaType.ALBUM:
- TidalFavorites(session, user_id).add_album(item_id) if add else TidalFavorites(
- session, user_id
- ).remove_album(item_id)
+ (
+ TidalFavorites(session, user_id).add_album(item_id)
+ if add
+ else TidalFavorites(session, user_id).remove_album(item_id)
+ )
case MediaType.TRACK:
- TidalFavorites(session, user_id).add_track(item_id) if add else TidalFavorites(
- session, user_id
- ).remove_track(item_id)
+ (
+ TidalFavorites(session, user_id).add_track(item_id)
+ if add
+ else TidalFavorites(session, user_id).remove_track(item_id)
+ )
case MediaType.PLAYLIST:
- TidalFavorites(session, user_id).add_playlist(item_id) if add else TidalFavorites(
- session, user_id
- ).remove_playlist(item_id)
+ (
+ TidalFavorites(session, user_id).add_playlist(item_id)
+ if add
+ else TidalFavorites(session, user_id).remove_playlist(item_id)
+ )
case MediaType.UNKNOWN:
return
"""Tune-In musicprovider support for MusicAssistant."""
+
from __future__ import annotations
from collections.abc import AsyncGenerator
kwargs["username"] = self.config.get_value(CONF_USERNAME)
kwargs["partnerId"] = "1"
kwargs["render"] = "json"
- async with self._throttler, self.mass.http_session.get(
- url, params=kwargs, ssl=False
- ) as response:
+ async with (
+ self._throttler,
+ self.mass.http_session.get(url, params=kwargs, ssl=False) as response,
+ ):
result = await response.json()
if not result or "error" in result:
self.logger.error(url)
This is more like a "virtual" player provider,
allowing the user to create player groups from all players known in the system.
"""
+
from __future__ import annotations
import asyncio
"""Basic provider allowing for external URL's to be streamed."""
+
from __future__ import annotations
import os
"""Youtube Music support for MusicAssistant."""
+
from __future__ import annotations
import asyncio
"""Main Music Assistant class."""
+
from __future__ import annotations
import asyncio
https://www.red-gate.com/simple-talk/development/python/memory-profiling-in-python-with-tracemalloc/
"""
+
import asyncio
import tracemalloc
projects / music-assistant-server.git / commitdiff