from __future__ import annotations
import logging
+from collections.abc import Iterable # noqa: TCH003
from dataclasses import dataclass
from types import NoneType
-from typing import TYPE_CHECKING, Any
+from typing import Any
from mashumaro import DataClassDictMixin
+from music_assistant.common.models.enums import ProviderType # noqa: TCH001
from music_assistant.constants import (
CONF_AUTO_PLAY,
CONF_CROSSFADE,
from .enums import ConfigEntryType
-if TYPE_CHECKING:
- from collections.abc import Iterable
-
- from music_assistant.common.models.enums import ProviderType
-
LOGGER = logging.getLogger(__name__)
ENCRYPT_CALLBACK: callable[[str], str] | None = None
import time
from dataclasses import dataclass, field
-from typing import TYPE_CHECKING
from mashumaro import DataClassDictMixin
-from .enums import PlayerState, RepeatMode
-
-if TYPE_CHECKING:
- from music_assistant.common.models.media_items import MediaItemType
+from music_assistant.common.models.media_items import MediaItemType # noqa: TCH001
- from .queue_item import QueueItem
+from .enums import PlayerState, RepeatMode
+from .queue_item import QueueItem # noqa: TCH001
@dataclass
import asyncio
import random
import time
-from typing import TYPE_CHECKING, Any
+from collections.abc import AsyncGenerator # noqa: TCH003
+from typing import Any
from music_assistant.common.helpers.datetime import utc_timestamp
from music_assistant.common.helpers.json import serialize_to_json
ProviderUnavailableError,
UnsupportedFeaturedException,
)
-from music_assistant.common.models.media_items import ItemMapping, Playlist, PlaylistTrack, Track
+from music_assistant.common.models.media_items import (
+ ItemMapping,
+ Playlist,
+ PlaylistTrack,
+ Track,
+)
from music_assistant.constants import DB_TABLE_PLAYLISTS
from music_assistant.server.helpers.compare import compare_strings
from .base import MediaControllerBase
-if TYPE_CHECKING:
- from collections.abc import AsyncGenerator
-
class PlaylistController(MediaControllerBase[Playlist]):
"""Controller managing MediaItems of type Playlist."""
return library_item
async def tracks(
- self, item_id: str, provider_instance_id_or_domain: str, force_refresh: bool = False
+ self,
+ item_id: str,
+ provider_instance_id_or_domain: str,
+ force_refresh: bool = False,
) -> AsyncGenerator[PlaylistTrack, None]:
"""Return playlist tracks for the given provider playlist id."""
playlist = await self.get(
async for track in self._get_provider_playlist_tracks(
prov.item_id,
prov.provider_instance,
- cache_checksum=str(time.time()) if force_refresh else playlist.metadata.checksum,
+ cache_checksum=(str(time.time()) if force_refresh else playlist.metadata.checksum),
):
yield track
import os
import shutil
import statistics
+from collections.abc import AsyncGenerator # noqa: TCH003
from contextlib import suppress
from itertools import zip_longest
from typing import TYPE_CHECKING
ProviderType,
)
from music_assistant.common.models.errors import MediaNotFoundError, MusicAssistantError
-from music_assistant.common.models.media_items import BrowseFolder, MediaItemType, SearchResults
+from music_assistant.common.models.media_items import (
+ BrowseFolder,
+ MediaItemType,
+ SearchResults,
+)
from music_assistant.common.models.provider import SyncTask
from music_assistant.constants import (
DB_SCHEMA_VERSION,
from .media.tracks import TracksController
if TYPE_CHECKING:
- from collections.abc import AsyncGenerator
-
from music_assistant.common.models.config_entries import CoreConfig
from music_assistant.server.models.music_provider import MusicProvider
for item in result:
if item.available:
return await self.get_item(
- item.media_type, item.item_id, item.provider, lazy=False, add_to_library=True
+ item.media_type,
+ item.item_id,
+ item.provider,
+ lazy=False,
+ add_to_library=True,
)
return None
"""List integrated loudness for a track in db."""
await self.database.insert(
DB_TABLE_TRACK_LOUDNESS,
- {"item_id": item_id, "provider": provider_instance_id_or_domain, "loudness": loudness},
+ {
+ "item_id": item_id,
+ "provider": provider_instance_id_or_domain,
+ "loudness": loudness,
+ },
allow_replace=True,
)
self.in_progress_syncs.remove(sync_spec)
if task_err := task.exception():
self.logger.warning(
- "Sync task for %s completed with errors", provider.name, exc_info=task_err
+ "Sync task for %s completed with errors",
+ provider.name,
+ exc_info=task_err,
)
else:
self.logger.info("Sync task for %s completed", provider.name)
import logging
import random
import time
+from collections.abc import AsyncGenerator # noqa: TCH003
from contextlib import suppress
from typing import TYPE_CHECKING, Any
from music_assistant.server.models.core_controller import CoreController
if TYPE_CHECKING:
- from collections.abc import AsyncGenerator, Iterator
+ from collections.abc import Iterator
from music_assistant.common.models.media_items import Album, Artist, Track
from music_assistant.common.models.player import Player
if option is None:
option = QueueOption(
await self.mass.config.get_core_config_value(
- self.domain, f"default_enqueue_action_{media_item.media_type.value}"
+ self.domain,
+ f"default_enqueue_action_{media_item.media_type.value}",
)
)
if option == QueueOption.REPLACE_NEXT and queue.state not in (
queue_items = [QueueItem.from_dict(x) for x in prev_items]
except Exception as err:
self.logger.warning(
- "Failed to restore the queue(items) for %s - %s", player.display_name, str(err)
+ "Failed to restore the queue(items) for %s - %s",
+ player.display_name,
+ str(err),
)
if queue is None:
queue = PlayerQueue(
async def _get_artist_tracks(self, artist: Artist) -> list[Track]:
"""Return tracks for given artist, based on user preference."""
artist_items_conf = self.mass.config.get_raw_core_config_value(
- self.domain, CONF_DEFAULT_ENQUEUE_SELECT_ARTIST, ENQUEUE_SELECT_ARTIST_DEFAULT_VALUE
+ self.domain,
+ CONF_DEFAULT_ENQUEUE_SELECT_ARTIST,
+ ENQUEUE_SELECT_ARTIST_DEFAULT_VALUE,
)
if artist_items_conf == "library_tracks":
# make sure we have an in-library artist
async def _get_album_tracks(self, album: Album) -> list[Track]:
"""Return tracks for given album, based on user preference."""
album_items_conf = self.mass.config.get_raw_core_config_value(
- self.domain, CONF_DEFAULT_ENQUEUE_SELECT_ALBUM, ENQUEUE_SELECT_ALBUM_DEFAULT_VALUE
+ self.domain,
+ CONF_DEFAULT_ENQUEUE_SELECT_ALBUM,
+ ENQUEUE_SELECT_ALBUM_DEFAULT_VALUE,
)
if album_items_conf == "library_tracks":
# make sure we have an in-library album
import logging
import time
import urllib.parse
+from collections.abc import AsyncGenerator # noqa: TCH003
from contextlib import suppress
from typing import TYPE_CHECKING
from music_assistant.server.models.core_controller import CoreController
if TYPE_CHECKING:
- from collections.abc import AsyncGenerator
-
from music_assistant.common.models.config_entries import CoreConfig
from music_assistant.common.models.player import Player
from music_assistant.common.models.player_queue import PlayerQueue
"""Install package with pip, raise when install failed."""
cmd = f"python3 -m pip install --find-links {HA_WHEELS} {package}"
proc = await asyncio.create_subprocess_shell(
- cmd, stderr=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.DEVNULL
+ cmd, stderr=asyncio.subprocess.STDOUT, stdout=asyncio.subprocess.PIPE
)
- _, stderr = await proc.communicate()
+ stdout, _ = await proc.communicate()
if proc.returncode != 0:
- msg = f"Failed to install package {package}\n{stderr.decode()}"
+ msg = f"Failed to install package {package}\n{stdout.decode()}"
raise RuntimeError(msg)
import asyncio
import re
+from collections.abc import AsyncGenerator # noqa: TCH003
from operator import itemgetter
from time import time
-from typing import TYPE_CHECKING, AsyncGenerator # noqa: UP035
+from typing import TYPE_CHECKING
from urllib.parse import unquote
import pytube
from music_assistant.server.helpers.util import (
get_package_version,
get_provider_module,
+ install_package,
is_hass_supervisor,
)
+from .models import ProviderInstanceType # noqa: TCH001
+
if TYPE_CHECKING:
from types import TracebackType
from music_assistant.common.models.config_entries import ProviderConfig
from music_assistant.server.models.core_controller import CoreController
- from .models import ProviderInstanceType
EventCallBackType = Callable[[MassEvent], None]
EventSubscriptionType = tuple[
async def __load_provider_manifests(self) -> None:
"""Preload all available provider manifest files."""
- for dir_str in os.listdir(PROVIDERS_PATH):
- dir_path = os.path.join(PROVIDERS_PATH, dir_str)
- if not os.path.isdir(dir_path):
- continue
+
+ async def load_provider_manifest(provider_domain: str, provider_path: str) -> None:
+ """Preload all available provider manifest files."""
# get files in subdirectory
- for file_str in os.listdir(dir_path):
- file_path = os.path.join(dir_path, file_str)
+ for file_str in os.listdir(provider_path):
+ file_path = os.path.join(provider_path, file_str)
if not os.path.isfile(file_path):
continue
if file_str != "manifest.json":
provider_manifest = await ProviderManifest.parse(file_path)
# check for icon.svg file
if not provider_manifest.icon_svg:
- icon_path = os.path.join(dir_path, "icon.svg")
+ icon_path = os.path.join(provider_path, "icon.svg")
if os.path.isfile(icon_path):
provider_manifest.icon_svg = await get_icon_string(icon_path)
# check for dark_icon file
if not provider_manifest.icon_svg_dark:
- icon_path = os.path.join(dir_path, "icon_dark.svg")
+ icon_path = os.path.join(provider_path, "icon_dark.svg")
if os.path.isfile(icon_path):
provider_manifest.icon_svg_dark = await get_icon_string(icon_path)
+ # install requirements
+ for requirement in provider_manifest.requirements:
+ await install_package(requirement)
self._provider_manifests[provider_manifest.domain] = provider_manifest
- LOGGER.debug("Loaded manifest for provider %s", dir_str)
+ LOGGER.debug("Loaded manifest for provider %s", provider_manifest.name)
except Exception as exc: # pylint: disable=broad-except
LOGGER.exception(
"Error while loading manifest for provider %s",
- dir_str,
+ provider_domain,
exc_info=exc,
)
+ async with asyncio.TaskGroup() as tg:
+ for dir_str in os.listdir(PROVIDERS_PATH):
+ dir_path = os.path.join(PROVIDERS_PATH, dir_str)
+ if not os.path.isdir(dir_path):
+ continue
+ tg.create_task(load_provider_manifest(dir_str, dir_path))
+
async def _setup_discovery(self) -> None:
"""Make this Music Assistant instance discoverable on the network."""
zeroconf_type = "_mass._tcp.local."
-[build-system]
-requires = ["setuptools~=62.3", "wheel~=0.37.1"]
-build-backend = "setuptools.build_meta"
-
[project]
name = "music_assistant"
# The version is set by GH action on release
classifiers = [
"Environment :: Console",
"Programming Language :: Python :: 3.11",
+ "Programming Language :: Python :: 3.12",
]
dependencies = ["aiohttp", "orjson", "mashumaro"]
projects / music-assistant-server.git / commitdiff