- name: Run pylint on changed files
run: |
pylint -rn -sn --rcfile=pylintrc --fail-on=I $(git ls-files '*.py')
+ - name: Run unit tests with pytest
+ run: pytest tests/
"""All constants for Music Assistant."""
-from dataclasses import dataclass
-from enum import Enum
-from time import time
-from typing import Any, Coroutine, Optional
-
-
-class EventType(Enum):
- """Enum with possible Events."""
-
- PLAYER_ADDED = "player added"
- PLAYER_REMOVED = "player removed"
- PLAYER_UPDATED = "player updated"
- STREAM_STARTED = "streaming started"
- STREAM_ENDED = "streaming ended"
- MUSIC_SYNC_STATUS = "music sync status"
- QUEUE_ADDED = "queue_added"
- QUEUE_UPDATED = "queue updated"
- QUEUE_ITEMS_UPDATED = "queue items updated"
- QUEUE_TIME_UPDATED = "queue time updated"
- SHUTDOWN = "application shutdown"
- ARTIST_ADDED = "artist added"
- ALBUM_ADDED = "album added"
- TRACK_ADDED = "track added"
- PLAYLIST_ADDED = "playlist added"
- PLAYLIST_UPDATED = "playlist updated"
- RADIO_ADDED = "radio added"
- TASK_UPDATED = "task updated"
- PROVIDER_REGISTERED = "provider registered"
- BACKGROUND_JOB_UPDATED = "background_job_updated"
-
-
-@dataclass
-class MassEvent:
- """Representation of an Event emitted in/by Music Assistant."""
-
- type: EventType
- object_id: Optional[str] = None # player_id, queue_id or uri
- data: Optional[Any] = None # optional data (such as the object)
-
-
-class JobStatus(Enum):
- """Enum with Job status."""
-
- PENDING = "pending"
- RUNNING = "running"
- CANCELLED = "cancelled"
- FINISHED = "success"
- ERROR = "error"
-
-
-@dataclass
-class BackgroundJob:
- """Description of a background job/task."""
-
- id: str
- coro: Coroutine
- name: str
- timestamp: float = time()
- status: JobStatus = JobStatus.PENDING
-
- def to_dict(self):
- """Return serializable dict from object."""
- return {
- "id": self.id,
- "name": self.name,
- "timestamp": self.status.value,
- "status": self.status.value,
- }
-
# player attributes
ATTR_PLAYER_ID = "player_id"
from __future__ import annotations
from time import time
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
from music_assistant.helpers.images import create_thumbnail
-from music_assistant.helpers.typing import MusicAssistant
from music_assistant.models.media_items import Album, Artist, Playlist, Radio, Track
from .audiodb import TheAudioDb
from .fanarttv import FanartTv
from .musicbrainz import MusicBrainz
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
+
TABLE_THUMBS = "thumbnails"
from __future__ import annotations
from json.decoder import JSONDecodeError
-from typing import Any, Dict, Optional
+from typing import TYPE_CHECKING, Any, Dict, Optional
import aiohttp
from asyncio_throttle import Throttler
)
from music_assistant.helpers.cache import use_cache
from music_assistant.helpers.compare import compare_strings
-from music_assistant.helpers.typing import MusicAssistant
from music_assistant.models.media_items import (
Album,
AlbumType,
Track,
)
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
+
IMG_MAPPING = {
"strArtistThumb": ImageType.THUMB,
"strArtistLogo": ImageType.LOGO,
from __future__ import annotations
from json.decoder import JSONDecodeError
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
import aiohttp
from asyncio_throttle import Throttler
app_var,
)
from music_assistant.helpers.cache import use_cache
-from music_assistant.helpers.typing import MusicAssistant
from music_assistant.models.media_items import (
Album,
Artist,
MediaItemMetadata,
)
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
+
# TODO: add support for personal api keys ?
-# TODO: Add support for album artwork ?
+
IMG_MAPPING = {
"artistthumb": ImageType.THUMB,
import re
from json.decoder import JSONDecodeError
+from typing import TYPE_CHECKING
import aiohttp
from asyncio_throttle import Throttler
from music_assistant.helpers.cache import use_cache
from music_assistant.helpers.compare import compare_strings, get_compare_string
-from music_assistant.helpers.typing import MusicAssistant
+
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
LUCENE_SPECIAL = r'([+\-&|!(){}\[\]\^"~*?:\\\/])'
import asyncio
import statistics
-from typing import Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
from databases import Database as Db
-from music_assistant.constants import EventType, MassEvent
from music_assistant.controllers.music.albums import AlbumsController
from music_assistant.controllers.music.artists import ArtistsController
from music_assistant.controllers.music.playlists import PlaylistController
TABLE_TRACK_LOUDNESS,
)
from music_assistant.helpers.datetime import utc_timestamp
-from music_assistant.helpers.typing import MusicAssistant
+from music_assistant.helpers.uri import parse_uri
from music_assistant.helpers.util import run_periodic
+from music_assistant.models.enums import EventType, MediaType
from music_assistant.models.errors import (
AlreadyRegisteredError,
MusicAssistantError,
SetupFailedError,
)
+from music_assistant.models.event import MassEvent
from music_assistant.models.media_items import (
MediaItem,
MediaItemProviderId,
MediaItemType,
- MediaType,
)
from music_assistant.models.provider import MusicProvider
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
+
class MusicController:
"""Several helpers around the musicproviders."""
self, uri: str, force_refresh: bool = False, lazy: bool = True
) -> MediaItemType:
"""Fetch MediaItem by uri."""
- try:
- if "://" in uri:
- provider = uri.split("://")[0]
- item_id = uri.split("/")[-1]
- media_type = MediaType(uri.split("/")[-2])
- elif "spotify" in uri:
- # spotify new-style uri
- provider, media_type, item_id = uri.split(":")
- media_type = MediaType(media_type)
- except (TypeError, AttributeError, ValueError) as err:
- raise MusicAssistantError(
- f"Not a valid Music Assistant uri: {uri}"
- ) from err
+ media_type, provider, item_id = parse_uri(uri)
return await self.get_item(
item_id, provider, media_type, force_refresh=force_refresh, lazy=lazy
)
import asyncio
from typing import List
-from music_assistant.constants import EventType, MassEvent
from music_assistant.helpers.compare import compare_album, compare_strings
from music_assistant.helpers.database import TABLE_ALBUMS
from music_assistant.helpers.json import json_serializer
from music_assistant.helpers.util import create_sort_name
+from music_assistant.models.enums import EventType
+from music_assistant.models.event import MassEvent
from music_assistant.models.media_controller import MediaControllerBase
from music_assistant.models.media_items import (
Album,
import itertools
from typing import List
-from music_assistant.constants import EventType, MassEvent
from music_assistant.helpers.compare import (
compare_album,
compare_strings,
from music_assistant.helpers.database import TABLE_ARTISTS
from music_assistant.helpers.json import json_serializer
from music_assistant.helpers.util import create_sort_name
+from music_assistant.models.enums import EventType
+from music_assistant.models.event import MassEvent
from music_assistant.models.media_controller import MediaControllerBase
from music_assistant.models.media_items import (
Album,
from time import time
from typing import List
-from music_assistant.constants import EventType, MassEvent
from music_assistant.helpers.database import TABLE_PLAYLISTS
from music_assistant.helpers.json import json_serializer
from music_assistant.helpers.util import create_sort_name
+from music_assistant.models.enums import EventType, MediaType
from music_assistant.models.errors import InvalidDataError, MediaNotFoundError
+from music_assistant.models.event import MassEvent
from music_assistant.models.media_controller import MediaControllerBase
-from music_assistant.models.media_items import MediaType, Playlist, Track
+from music_assistant.models.media_items import Playlist, Track
class PlaylistController(MediaControllerBase[Playlist]):
from time import time
-from music_assistant.constants import EventType, MassEvent
from music_assistant.helpers.database import TABLE_RADIOS
from music_assistant.helpers.json import json_serializer
from music_assistant.helpers.util import create_sort_name
+from music_assistant.models.enums import EventType, MediaType
+from music_assistant.models.event import MassEvent
from music_assistant.models.media_controller import MediaControllerBase
-from music_assistant.models.media_items import MediaType, Radio
+from music_assistant.models.media_items import Radio
class RadioController(MediaControllerBase[Radio]):
import asyncio
from typing import List
-from music_assistant.constants import EventType, MassEvent
from music_assistant.helpers.compare import (
compare_artists,
compare_strings,
from music_assistant.helpers.database import TABLE_TRACKS
from music_assistant.helpers.json import json_serializer
from music_assistant.helpers.util import create_sort_name
+from music_assistant.models.enums import EventType, MediaType
+from music_assistant.models.event import MassEvent
from music_assistant.models.media_controller import MediaControllerBase
-from music_assistant.models.media_items import ItemMapping, MediaType, Track
+from music_assistant.models.media_items import ItemMapping, Track
class TracksController(MediaControllerBase[Track]):
"""Logic to play music from MusicProviders to supported players."""
from __future__ import annotations
-from typing import Dict, Tuple, Union
+from typing import TYPE_CHECKING, Dict, Tuple, Union
-from music_assistant.constants import EventType, MassEvent
-from music_assistant.helpers.typing import MusicAssistant
+from music_assistant.models.enums import EventType
from music_assistant.models.errors import AlreadyRegisteredError
+from music_assistant.models.event import MassEvent
from music_assistant.models.player import Player, PlayerGroup
from music_assistant.models.player_queue import PlayerQueue
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
+
PlayerType = Union[Player, PlayerGroup]
DB_TABLE = "queue_settings"
import asyncio
from asyncio import Task
from time import time
-from typing import AsyncGenerator, Dict, Optional, Set
+from typing import TYPE_CHECKING, AsyncGenerator, Dict, Optional, Set
from aiohttp import web
-from music_assistant.constants import EventType, MassEvent
from music_assistant.helpers.audio import (
check_audio_support,
crossfade_pcm_parts,
strip_silence,
)
from music_assistant.helpers.process import AsyncProcess
-from music_assistant.helpers.typing import MusicAssistant
from music_assistant.helpers.util import get_ip, select_stream_port
+from music_assistant.models.enums import (
+ ContentType,
+ CrossFadeMode,
+ EventType,
+ MediaType,
+)
from music_assistant.models.errors import MediaNotFoundError, MusicAssistantError
-from music_assistant.models.media_items import ContentType, MediaType
-from music_assistant.models.player_queue import CrossFadeMode, PlayerQueue, QueueItem
+from music_assistant.models.event import MassEvent
+from music_assistant.models.player_queue import PlayerQueue, QueueItem
+
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
class StreamController:
import logging
import struct
from io import BytesIO
-from typing import AsyncGenerator, List, Optional, Tuple
+from typing import TYPE_CHECKING, AsyncGenerator, List, Optional, Tuple
import aiofiles
-from music_assistant.constants import EventType, MassEvent
from music_assistant.helpers.process import AsyncProcess, check_output
-from music_assistant.helpers.typing import MusicAssistant, QueueItem
from music_assistant.helpers.util import create_tempfile
+from music_assistant.models.enums import EventType
from music_assistant.models.errors import AudioError, MediaNotFoundError
+from music_assistant.models.event import MassEvent
from music_assistant.models.media_items import (
ContentType,
MediaType,
StreamType,
)
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
+ from music_assistant.models.player_queue import QueueItem
+
LOGGER = logging.getLogger("audio")
# pylint:disable=consider-using-f-string
async def get_stream_details(
- mass: MusicAssistant, queue_item: QueueItem, queue_id: str = ""
+ mass: MusicAssistant, queue_item: "QueueItem", queue_id: str = ""
) -> StreamDetails:
"""
Get streamdetails for the given QueueItem.
import functools
import json
import time
+from typing import TYPE_CHECKING
from music_assistant.helpers.database import TABLE_CACHE
-from music_assistant.helpers.typing import MusicAssistant
+
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
class Cache:
from __future__ import annotations
from contextlib import asynccontextmanager
-from typing import Any, Dict, List, Mapping, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional
from databases import Database as Db
from databases import DatabaseURL
-from music_assistant.helpers.typing import MusicAssistant
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
# pylint: disable=invalid-name
from __future__ import annotations
from io import BytesIO
+from typing import TYPE_CHECKING
from PIL import Image
-from music_assistant.helpers.typing import MusicAssistant
-from music_assistant.models.media_items import (
- ImageType,
- ItemMapping,
- MediaItemType,
- MediaType,
-)
+from music_assistant.models.enums import ImageType, MediaType
+from music_assistant.models.media_items import ItemMapping, MediaItemType
+
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
async def create_thumbnail(mass: MusicAssistant, url, size: int = 150) -> bytes:
+++ /dev/null
-"""Typing helper."""
-from __future__ import annotations
-
-from typing import TYPE_CHECKING, List, Optional
-
-# pylint: disable=invalid-name
-if TYPE_CHECKING:
- from music_assistant.mass import (
- EventCallBackType,
- EventSubscriptionType,
- MusicAssistant,
- )
- from music_assistant.models.media_items import MediaType
- from music_assistant.models.player import Player
- from music_assistant.models.player_queue import PlayerQueue, QueueItem
-
-else:
- MusicAssistant = "MusicAssistant"
- QueueItem = "QueueItem"
- PlayerQueue = "PlayerQueue"
- StreamDetails = "StreamDetails"
- Player = "Player"
- MediaType = "MediaType"
- EventCallBackType = "EventCallBackType"
- EventSubscriptionType = "EventSubscriptionType"
-
-
-QueueItems = List[QueueItem]
-Players = List[Player]
-
-OptionalInt = Optional[int]
-OptionalStr = Optional[str]
--- /dev/null
+"""Helpers for creating/parsing URI's."""
+
+from typing import Tuple
+
+from music_assistant.models.enums import MediaType
+from music_assistant.models.errors import MusicAssistantError
+
+
+def parse_uri(uri: str) -> Tuple[MediaType, str, str]:
+ """
+ Try to parse URI to Mass identifiers.
+
+ Returns Tuple: MediaType, provider, item_id
+ """
+ try:
+ if uri.startswith("https://open."):
+ # public share URL (e.g. Spotify or Qobuz, not sure about others)
+ # https://open.spotify.com/playlist/5lH9NjOeJvctAO92ZrKQNB?si=04a63c8234ac413e
+ provider = uri.split(".")[1]
+ media_type_str = uri.split("/")[3]
+ media_type = MediaType(media_type_str)
+ item_id = uri.split("/")[4].split("?")[0]
+ elif "://" in uri:
+ # music assistant-style uri
+ # provider://media_type/item_id
+ provider = uri.split("://")[0]
+ media_type_str = uri.split("/")[2]
+ media_type = MediaType(media_type_str)
+ item_id = uri.split(f"{media_type_str}/")[1]
+ elif ":" in uri:
+ # spotify new-style uri
+ provider, media_type_str, item_id = uri.split(":")
+ media_type = MediaType(media_type_str)
+ else:
+ raise KeyError
+ except (TypeError, AttributeError, ValueError, KeyError) as err:
+ raise MusicAssistantError(f"Not a valid Music Assistant uri: {uri}") from err
+ return (media_type, provider, item_id)
+
+
+def create_uri(media_type: MediaType, provider: str, item_id: str) -> str:
+ """Create Music Assistant URI from MediaItem values."""
+ return f"{provider}://{media_type.value}/{item_id}"
import aiohttp
from databases import DatabaseURL
-from music_assistant.constants import BackgroundJob, EventType, JobStatus, MassEvent
from music_assistant.controllers.metadata import MetaDataController
from music_assistant.controllers.music import MusicController
from music_assistant.controllers.players import PlayerController
from music_assistant.controllers.stream import StreamController
from music_assistant.helpers.cache import Cache
from music_assistant.helpers.database import Database
+from music_assistant.models.background_job import BackgroundJob
+from music_assistant.models.enums import EventType, JobStatus
+from music_assistant.models.event import MassEvent
EventCallBackType = Callable[[MassEvent], None]
EventSubscriptionType = Tuple[
--- /dev/null
+"""Model for a Background Job."""
+from dataclasses import dataclass
+from time import time
+from typing import Coroutine
+
+from music_assistant.models.enums import JobStatus
+
+
+@dataclass
+class BackgroundJob:
+ """Description of a background job/task."""
+
+ id: str
+ coro: Coroutine
+ name: str
+ timestamp: float = time()
+ status: JobStatus = JobStatus.PENDING
+
+ def to_dict(self):
+ """Return serializable dict from object."""
+ return {
+ "id": self.id,
+ "name": self.name,
+ "timestamp": self.status.value,
+ "status": self.status.value,
+ }
--- /dev/null
+"""All enums used by the Music Assistant models."""
+
+from enum import Enum, IntEnum
+
+
+class MediaType(Enum):
+ """Enum for MediaType."""
+
+ ARTIST = "artist"
+ ALBUM = "album"
+ TRACK = "track"
+ PLAYLIST = "playlist"
+ RADIO = "radio"
+ UNKNOWN = "unknown"
+
+
+class MediaQuality(IntEnum):
+ """Enum for Media Quality."""
+
+ UNKNOWN = 0
+ LOSSY_MP3 = 1
+ LOSSY_OGG = 2
+ LOSSY_AAC = 3
+ FLAC_LOSSLESS = 10 # 44.1/48khz 16 bits
+ FLAC_LOSSLESS_HI_RES_1 = 20 # 44.1/48khz 24 bits HI-RES
+ FLAC_LOSSLESS_HI_RES_2 = 21 # 88.2/96khz 24 bits HI-RES
+ FLAC_LOSSLESS_HI_RES_3 = 22 # 176/192khz 24 bits HI-RES
+ FLAC_LOSSLESS_HI_RES_4 = 23 # above 192khz 24 bits HI-RES
+
+
+class LinkType(Enum):
+ """Enum wth link types."""
+
+ WEBSITE = "website"
+ FACEBOOK = "facebook"
+ TWITTER = "twitter"
+ LASTFM = "lastfm"
+ YOUTUBE = "youtube"
+ INSTAGRAM = "instagram"
+ SNAPCHAT = "snapchat"
+ TIKTOK = "tiktok"
+ DISCOGS = "discogs"
+ WIKIPEDIA = "wikipedia"
+ ALLMUSIC = "allmusic"
+
+
+class ImageType(Enum):
+ """Enum wth image types."""
+
+ THUMB = "thumb"
+ WIDE_THUMB = "wide_thumb"
+ FANART = "fanart"
+ LOGO = "logo"
+ CLEARART = "clearart"
+ BANNER = "banner"
+ CUTOUT = "cutout"
+ BACK = "back"
+ CDART = "cdart"
+ EMBEDDED_THUMB = "embedded_thumb"
+ OTHER = "other"
+
+
+class AlbumType(Enum):
+ """Enum for Album type."""
+
+ ALBUM = "album"
+ SINGLE = "single"
+ COMPILATION = "compilation"
+ UNKNOWN = "unknown"
+
+
+class StreamType(Enum):
+ """Enum with stream types."""
+
+ EXECUTABLE = "executable"
+ URL = "url"
+ FILE = "file"
+ CACHE = "cache"
+
+
+class ContentType(Enum):
+ """Enum with audio content types supported by ffmpeg."""
+
+ OGG = "ogg"
+ FLAC = "flac"
+ MP3 = "mp3"
+ AAC = "aac"
+ MPEG = "mpeg"
+ WAV = "wav"
+ PCM_S16LE = "s16le" # PCM signed 16-bit little-endian
+ PCM_S24LE = "s24le" # PCM signed 24-bit little-endian
+ PCM_S32LE = "s32le" # PCM signed 32-bit little-endian
+ PCM_F32LE = "f32le" # PCM 32-bit floating-point little-endian
+ PCM_F64LE = "f64le" # PCM 64-bit floating-point little-endian
+
+ @classmethod
+ def try_parse(
+ cls: "ContentType", string: str, fallback: str = "mp3"
+ ) -> "ContentType":
+ """Try to parse ContentType from (url)string."""
+ tempstr = string.lower()
+ if "." in tempstr:
+ tempstr = tempstr.split(".")[-1]
+ tempstr = tempstr.split("?")[0]
+ tempstr = tempstr.split("&")[0]
+ try:
+ return cls(tempstr)
+ except ValueError:
+ return cls(fallback)
+
+ def is_pcm(self):
+ """Return if contentype is PCM."""
+ return self.name.startswith("PCM")
+
+ def sox_supported(self):
+ """Return if ContentType is supported by SoX."""
+ return self not in [ContentType.AAC, ContentType.MPEG]
+
+ def sox_format(self):
+ """Convert the ContentType to SoX compatible format."""
+ if not self.sox_supported():
+ raise NotImplementedError
+ return self.value.replace("le", "")
+
+ @classmethod
+ def from_bit_depth(
+ cls, bit_depth: int, floating_point: bool = False
+ ) -> "ContentType":
+ """Return (PCM) Contenttype from PCM bit depth."""
+ if floating_point and bit_depth > 32:
+ return cls.PCM_F64LE
+ if floating_point:
+ return cls.PCM_F32LE
+ if bit_depth == 16:
+ return cls.PCM_S16LE
+ if bit_depth == 24:
+ return cls.PCM_S24LE
+ return cls.PCM_S32LE
+
+
+class QueueOption(Enum):
+ """Enum representation of the queue (play) options."""
+
+ PLAY = "play"
+ REPLACE = "replace"
+ NEXT = "next"
+ ADD = "add"
+
+
+class CrossFadeMode(Enum):
+ """Enum with crossfade modes."""
+
+ DISABLED = "disabled" # no crossfading at all
+ STRICT = "strict" # do not crossfade tracks of same album
+ SMART = "smart" # crossfade if possible (do not crossfade different sample rates)
+ ALWAYS = "always" # all tracks - resample to fixed sample rate
+
+
+class RepeatMode(Enum):
+ """Enum with repeat modes."""
+
+ OFF = "off" # no repeat at all
+ ONE = "one" # repeat one/single track
+ ALL = "all" # repeat entire queue
+
+
+class PlayerState(Enum):
+ """Enum for the (playback)state of a player."""
+
+ IDLE = "idle"
+ PAUSED = "paused"
+ PLAYING = "playing"
+ OFF = "off"
+
+
+class EventType(Enum):
+ """Enum with possible Events."""
+
+ PLAYER_ADDED = "player added"
+ PLAYER_REMOVED = "player removed"
+ PLAYER_UPDATED = "player updated"
+ STREAM_STARTED = "streaming started"
+ STREAM_ENDED = "streaming ended"
+ MUSIC_SYNC_STATUS = "music sync status"
+ QUEUE_ADDED = "queue_added"
+ QUEUE_UPDATED = "queue updated"
+ QUEUE_ITEMS_UPDATED = "queue items updated"
+ QUEUE_TIME_UPDATED = "queue time updated"
+ SHUTDOWN = "application shutdown"
+ ARTIST_ADDED = "artist added"
+ ALBUM_ADDED = "album added"
+ TRACK_ADDED = "track added"
+ PLAYLIST_ADDED = "playlist added"
+ PLAYLIST_UPDATED = "playlist updated"
+ RADIO_ADDED = "radio added"
+ TASK_UPDATED = "task updated"
+ PROVIDER_REGISTERED = "provider registered"
+ BACKGROUND_JOB_UPDATED = "background_job_updated"
+
+
+class JobStatus(Enum):
+ """Enum with Job status."""
+
+ PENDING = "pending"
+ RUNNING = "running"
+ CANCELLED = "cancelled"
+ FINISHED = "success"
+ ERROR = "error"
--- /dev/null
+"""Model for Music Assistant Event."""
+
+from dataclasses import dataclass
+from typing import Any, Optional
+
+from music_assistant.models.enums import EventType
+
+
+@dataclass
+class MassEvent:
+ """Representation of an Event emitted in/by Music Assistant."""
+
+ type: EventType
+ object_id: Optional[str] = None # player_id, queue_id or uri
+ data: Optional[Any] = None # optional data (such as the object)
from abc import ABCMeta, abstractmethod
from time import time
-from typing import Generic, List, Optional, Tuple, TypeVar
+from typing import TYPE_CHECKING, Generic, List, Optional, Tuple, TypeVar
from databases import Database as Db
-from music_assistant.helpers.typing import MusicAssistant
from music_assistant.models.errors import MediaNotFoundError, ProviderUnavailableError
-from .media_items import MediaItemType, MediaType
+from .enums import MediaType
+from .media_items import MediaItemType
+
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
ItemCls = TypeVar("ItemCls", bound="MediaControllerBase")
from __future__ import annotations
from dataclasses import dataclass, field, fields
-from enum import Enum, IntEnum
from typing import Any, Dict, List, Mapping, Optional, Set, Union
from mashumaro import DataClassDictMixin
from music_assistant.helpers.json import json
+from music_assistant.helpers.uri import create_uri
from music_assistant.helpers.util import create_sort_name
+from music_assistant.models.enums import (
+ AlbumType,
+ ContentType,
+ ImageType,
+ LinkType,
+ MediaQuality,
+ MediaType,
+ StreamType,
+)
MetadataTypes = Union[int, bool, str, List[str]]
JSON_KEYS = ("artists", "artist", "album", "metadata", "provider_ids")
-class MediaType(Enum):
- """Enum for MediaType."""
-
- ARTIST = "artist"
- ALBUM = "album"
- TRACK = "track"
- PLAYLIST = "playlist"
- RADIO = "radio"
- UNKNOWN = "unknown"
-
-
-class MediaQuality(IntEnum):
- """Enum for Media Quality."""
-
- UNKNOWN = 0
- LOSSY_MP3 = 1
- LOSSY_OGG = 2
- LOSSY_AAC = 3
- FLAC_LOSSLESS = 10 # 44.1/48khz 16 bits
- FLAC_LOSSLESS_HI_RES_1 = 20 # 44.1/48khz 24 bits HI-RES
- FLAC_LOSSLESS_HI_RES_2 = 21 # 88.2/96khz 24 bits HI-RES
- FLAC_LOSSLESS_HI_RES_3 = 22 # 176/192khz 24 bits HI-RES
- FLAC_LOSSLESS_HI_RES_4 = 23 # above 192khz 24 bits HI-RES
-
-
@dataclass(frozen=True)
class MediaItemProviderId(DataClassDictMixin):
"""Model for a MediaItem's provider id."""
return hash((self.provider, self.item_id, self.quality))
-class LinkType(Enum):
- """Enum wth link types."""
-
- WEBSITE = "website"
- FACEBOOK = "facebook"
- TWITTER = "twitter"
- LASTFM = "lastfm"
- YOUTUBE = "youtube"
- INSTAGRAM = "instagram"
- SNAPCHAT = "snapchat"
- TIKTOK = "tiktok"
- DISCOGS = "discogs"
- WIKIPEDIA = "wikipedia"
- ALLMUSIC = "allmusic"
-
-
@dataclass(frozen=True)
class MediaItemLink(DataClassDictMixin):
"""Model for a link."""
return hash((self.type.value))
-class ImageType(Enum):
- """Enum wth image types."""
-
- THUMB = "thumb"
- WIDE_THUMB = "wide_thumb"
- FANART = "fanart"
- LOGO = "logo"
- CLEARART = "clearart"
- BANNER = "banner"
- CUTOUT = "cutout"
- BACK = "back"
- CDART = "cdart"
- EMBEDDED_THUMB = "embedded_thumb"
- OTHER = "other"
-
-
@dataclass(frozen=True)
class MediaItemImage(DataClassDictMixin):
"""Model for a image."""
musicbrainz_id: Optional[str] = None
-class AlbumType(Enum):
- """Enum for Album type."""
-
- ALBUM = "album"
- SINGLE = "single"
- COMPILATION = "compilation"
- UNKNOWN = "unknown"
-
-
@dataclass
class Album(MediaItem):
"""Model for an album."""
return val
-def create_uri(media_type: MediaType, provider_id: str, item_id: str):
- """Create uri for mediaitem."""
- return f"{provider_id}://{media_type.value}/{item_id}"
-
-
MediaItemType = Union[Artist, Album, Track, Radio, Playlist]
-class StreamType(Enum):
- """Enum with stream types."""
-
- EXECUTABLE = "executable"
- URL = "url"
- FILE = "file"
- CACHE = "cache"
-
-
-class ContentType(Enum):
- """Enum with audio content types supported by ffmpeg."""
-
- OGG = "ogg"
- FLAC = "flac"
- MP3 = "mp3"
- AAC = "aac"
- MPEG = "mpeg"
- WAV = "wav"
- PCM_S16LE = "s16le" # PCM signed 16-bit little-endian
- PCM_S24LE = "s24le" # PCM signed 24-bit little-endian
- PCM_S32LE = "s32le" # PCM signed 32-bit little-endian
- PCM_F32LE = "f32le" # PCM 32-bit floating-point little-endian
- PCM_F64LE = "f64le" # PCM 64-bit floating-point little-endian
-
- @classmethod
- def try_parse(
- cls: "ContentType", string: str, fallback: str = "mp3"
- ) -> "ContentType":
- """Try to parse ContentType from (url)string."""
- tempstr = string.lower()
- if "." in tempstr:
- tempstr = tempstr.split(".")[-1]
- tempstr = tempstr.split("?")[0]
- tempstr = tempstr.split("&")[0]
- try:
- return cls(tempstr)
- except ValueError:
- return cls(fallback)
-
- def is_pcm(self):
- """Return if contentype is PCM."""
- return self.name.startswith("PCM")
-
- def sox_supported(self):
- """Return if ContentType is supported by SoX."""
- return self not in [ContentType.AAC, ContentType.MPEG]
-
- def sox_format(self):
- """Convert the ContentType to SoX compatible format."""
- if not self.sox_supported():
- raise NotImplementedError
- return self.value.replace("le", "")
-
- @classmethod
- def from_bit_depth(
- cls, bit_depth: int, floating_point: bool = False
- ) -> "ContentType":
- """Return (PCM) Contenttype from PCM bit depth."""
- if floating_point and bit_depth > 32:
- return cls.PCM_F64LE
- if floating_point:
- return cls.PCM_F32LE
- if bit_depth == 16:
- return cls.PCM_S16LE
- if bit_depth == 24:
- return cls.PCM_S24LE
- return cls.PCM_S32LE
-
-
@dataclass
class StreamDetails(DataClassDictMixin):
"""Model for streamdetails."""
import asyncio
from abc import ABC
from dataclasses import dataclass
-from enum import Enum
from typing import TYPE_CHECKING, Any, Dict, List, Tuple
from mashumaro import DataClassDictMixin
-from music_assistant.constants import EventType, MassEvent
-from music_assistant.helpers.typing import MusicAssistant
from music_assistant.helpers.util import get_changed_keys
+from music_assistant.models.enums import EventType, PlayerState
+from music_assistant.models.event import MassEvent
from music_assistant.models.media_items import ContentType
if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
+
from .player_queue import PlayerQueue
)
-class PlayerState(Enum):
- """Enum for the (playback)state of a player."""
-
- IDLE = "idle"
- PAUSED = "paused"
- PLAYING = "playing"
- OFF = "off"
-
-
@dataclass(frozen=True)
class DeviceInfo(DataClassDictMixin):
"""Model for a player's deviceinfo."""
import random
from asyncio import Task, TimerHandle
from dataclasses import dataclass
-from enum import Enum
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
from uuid import uuid4
from mashumaro import DataClassDictMixin
-from music_assistant.constants import EventType, MassEvent
from music_assistant.helpers.audio import get_stream_details
-from music_assistant.helpers.typing import MusicAssistant
-from music_assistant.models.errors import MediaNotFoundError, QueueEmpty
-from music_assistant.models.media_items import (
+from music_assistant.models.enums import (
ContentType,
+ CrossFadeMode,
+ EventType,
MediaType,
- Radio,
- StreamDetails,
- Track,
+ QueueOption,
+ RepeatMode,
)
+from music_assistant.models.errors import MediaNotFoundError, QueueEmpty
+from music_assistant.models.event import MassEvent
+from music_assistant.models.media_items import Radio, StreamDetails, Track
from .player import Player, PlayerGroup, PlayerState
-
-class QueueOption(Enum):
- """Enum representation of the queue (play) options."""
-
- PLAY = "play"
- REPLACE = "replace"
- NEXT = "next"
- ADD = "add"
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
@dataclass
)
-class CrossFadeMode(Enum):
- """Enum with crossfade modes."""
-
- DISABLED = "disabled" # no crossfading at all
- STRICT = "strict" # do not crossfade tracks of same album
- SMART = "smart" # crossfade if possible (do not crossfade different sample rates)
- ALWAYS = "always" # all tracks - resample to fixed sample rate
-
-
-class RepeatMode(Enum):
- """Enum with repeat modes."""
-
- OFF = "off" # no repeat at all
- ONE = "one" # repeat one/single track
- ALL = "all" # repeat entire queue
-
-
class QueueSettings:
"""Representation of (user adjustable) PlayerQueue settings/preferences."""
from abc import abstractmethod
from logging import Logger
-from typing import List, Optional
+from typing import TYPE_CHECKING, List, Optional
-from music_assistant.helpers.typing import MusicAssistant
+from music_assistant.models.enums import MediaType
from music_assistant.models.media_items import (
Album,
Artist,
MediaItemType,
- MediaType,
Playlist,
Radio,
Track,
)
from music_assistant.models.player_queue import StreamDetails
+if TYPE_CHECKING:
+ from music_assistant.mass import MusicAssistant
+
class MusicProvider:
"""Model for a Music Provider."""
import aiohttp
from asyncio_throttle import Throttler
-from music_assistant.constants import EventType, MassEvent
from music_assistant.helpers.app_vars import ( # pylint: disable=no-name-in-module
app_var,
)
from music_assistant.helpers.cache import use_cache
from music_assistant.helpers.util import parse_title_and_version, try_parse_int
+from music_assistant.models.enums import EventType
from music_assistant.models.errors import LoginFailed
+from music_assistant.models.event import MassEvent
from music_assistant.models.media_items import (
Album,
AlbumType,
--- /dev/null
+"""Tests for utility/helper functions."""
+
+from pytest import raises
+
+from music_assistant.helpers import uri, util
+from music_assistant.models import media_items
+from music_assistant.models.errors import MusicAssistantError
+
+
+def test_version_extract():
+ """Test the extraction of version from title."""
+
+ test_str = "Bam Bam (feat. Ed Sheeran) - Karaoke Version"
+ title, version = util.parse_title_and_version(test_str)
+ assert title == "Bam Bam"
+ assert version == "Karaoke Version"
+
+
+def test_uri_parsing():
+ """Test parsing of URI."""
+ # test regular uri
+ test_uri = "spotify://track/123456789"
+ media_type, provider, item_id = uri.parse_uri(test_uri)
+ assert media_type == media_items.MediaType.TRACK
+ assert provider == "spotify"
+ assert item_id == "123456789"
+ # test spotify uri
+ test_uri = "spotify:track:123456789"
+ media_type, provider, item_id = uri.parse_uri(test_uri)
+ assert media_type == media_items.MediaType.TRACK
+ assert provider == "spotify"
+ assert item_id == "123456789"
+ # test public play/open url
+ test_uri = (
+ "https://open.spotify.com/playlist/5lH9NjOeJvctAO92ZrKQNB?si=04a63c8234ac413e"
+ )
+ media_type, provider, item_id = uri.parse_uri(test_uri)
+ assert media_type == media_items.MediaType.PLAYLIST
+ assert provider == "spotify"
+ assert item_id == "5lH9NjOeJvctAO92ZrKQNB"
+ # test filename with slashes as item_id
+ test_uri = "filesystem://track/Artist/Album/Track.flac"
+ media_type, provider, item_id = uri.parse_uri(test_uri)
+ assert media_type == media_items.MediaType.TRACK
+ assert provider == "filesystem"
+ assert item_id == "Artist/Album/Track.flac"
+ # test invalid uri
+ with raises(MusicAssistantError):
+ uri.parse_uri("invalid://blah")
+++ /dev/null
-"""Tests for utility functions."""
-
-from music_assistant.helpers import util
-
-
-def test_version_extract():
- """Test the extraction of version from title."""
-
- test_str = "Bam Bam (feat. Ed Sheeran) - Karaoke Version"
- title, version = util.parse_title_and_version(test_str)
- assert title == "Bam Bam"
- assert version == "Karaoke Version"
projects / music-assistant-server.git / commitdiff