- id: mypy
additional_dependencies: [types-all]
exclude: ^examples/
+ - repo: https://github.com/pycqa/pydocstyle
+ rev: 6.1.1
+ hooks:
+ - id: pydocstyle
+ exclude: ^examples/|^.venv/|^.vscode/
- repo: local
hooks:
- id: pylint
types: [python]
args: ["-rn", "-sn", "--rcfile=pylintrc", "--fail-on=I"]
exclude: ^examples/|^.venv/|^.vscode/
-
providers = []
-# if args.spotify_username and args.spotify_password:
-# providers.append(SpotifyProvider(args.spotify_username, args.spotify_password))
-# if args.qobuz_username and args.qobuz_password:
-# providers.append(QobuzProvider(args.qobuz_username, args.qobuz_password))
-# if args.tunein_username:
-# providers.append(TuneInProvider(args.tunein_username))
+if args.spotify_username and args.spotify_password:
+ providers.append(SpotifyProvider(args.spotify_username, args.spotify_password))
+if args.qobuz_username and args.qobuz_password:
+ providers.append(QobuzProvider(args.qobuz_username, args.qobuz_password))
+if args.tunein_username:
+ providers.append(TuneInProvider(args.tunein_username))
if args.musicdir:
providers.append(FileSystemProvider(args.musicdir, args.playlistdir))
"""Logic to play music from MusicProviders to supported players."""
from __future__ import annotations
-from typing import Dict, Tuple
+from typing import Dict, Tuple, Union
from music_assistant.constants import EventType
from music_assistant.controllers.stream import StreamController
from music_assistant.models.player import Player, PlayerGroup
from music_assistant.models.player_queue import PlayerQueue
-PlayerType = Player | PlayerGroup
+PlayerType = Union[Player, PlayerGroup]
DB_TABLE = "queue_settings"
from __future__ import annotations
from contextlib import asynccontextmanager
-from typing import Any, Dict, List, Mapping
+from typing import Any, Dict, List, Mapping, Optional
from databases import Database as Db
from databases import DatabaseURL
self.logger = mass.logger.getChild("db")
@asynccontextmanager
- async def get_db(self, db: Db | None = None) -> Db:
+ async def get_db(self, db: Optional[Db] = None) -> Db:
"""Context manager helper to get the active db connection."""
if db is not None:
yield db
yield _db
async def get_rows(
- self, table: str, match: dict = None, order_by: str = None, db: Db | None = None
+ self,
+ table: str,
+ match: dict = None,
+ order_by: str = None,
+ db: Optional[Db] = None,
) -> List[Mapping]:
"""Get all rows for given table."""
async with self.get_db(db) as _db:
return await _db.fetch_all(sql_query, match)
async def get_rows_from_query(
- self, query: str, db: Db | None = None
+ self, query: str, db: Optional[Db] = None
) -> List[Mapping]:
"""Get all rows for given custom query."""
async with self.get_db(db) as _db:
return await _db.fetch_all(query)
async def search(
- self, table: str, search: str, column: str = "name", db: Db | None = None
+ self, table: str, search: str, column: str = "name", db: Optional[Db] = None
) -> List[Mapping]:
"""Search table by column."""
async with self.get_db(db) as _db:
return await _db.fetch_all(sql_query)
async def get_row(
- self, table: str, match: Dict[str, Any] = None, db: Db | None = None
+ self, table: str, match: Dict[str, Any] = None, db: Optional[Db] = None
) -> Mapping | None:
"""Get single row for given table where column matches keys/values."""
async with self.get_db(db) as _db:
return await _db.fetch_one(sql_query, match)
async def insert_or_replace(
- self, table: str, values: Dict[str, Any], db: Db | None = None
+ self, table: str, values: Dict[str, Any], db: Optional[Db] = None
) -> Mapping:
"""Insert or replace data in given table."""
async with self.get_db(db) as _db:
table: str,
match: Dict[str, Any],
values: Dict[str, Any],
- db: Db | None = None,
+ db: Optional[Db] = None,
) -> Mapping:
"""Update record."""
async with self.get_db(db) as _db:
return await self.get_row(table, match, db=_db)
async def delete(
- self, table: str, match: Dict[str, Any], db: Db | None = None
+ self, table: str, match: Dict[str, Any], db: Optional[Db] = None
) -> None:
"""Delete data in given table."""
async with self.get_db(db) as _db:
"""Typing helper."""
from __future__ import annotations
-from typing import TYPE_CHECKING, Any, List, Optional
+from typing import TYPE_CHECKING, List, Optional
# pylint: disable=invalid-name
if TYPE_CHECKING:
from music_assistant.mass import (
- MusicAssistant,
- EventDetails,
EventCallBackType,
EventSubscriptionType,
+ MusicAssistant,
)
from music_assistant.models.media_items import MediaType
- from music_assistant.models.player import (
- PlayerQueue,
- QueueItem,
- )
- from music_assistant.models.player import Player
+ from music_assistant.models.player import Player, PlayerQueue, QueueItem
else:
MusicAssistant = "MusicAssistant"
StreamDetails = "StreamDetails"
Player = "Player"
MediaType = "MediaType"
- EventDetails = Any | None
EventCallBackType = "EventCallBackType"
EventSubscriptionType = "EventSubscriptionType"
import asyncio
import logging
from time import time
-from typing import Any, Callable, Coroutine, Optional, Tuple
+from typing import Any, Callable, Coroutine, Optional, Tuple, Union
import aiohttp
from databases import DatabaseURL
from music_assistant.helpers.database import Database
from music_assistant.helpers.util import create_task
-EventDetails = Any | None
-EventCallBackType = Callable[[EventType, EventDetails], None]
+EventCallBackType = Callable[[EventType, Any], None]
EventSubscriptionType = Tuple[EventCallBackType, Optional[Tuple[EventType]]]
self,
db_url: DatabaseURL,
stream_port: int = 8095,
- session: aiohttp.ClientSession | None = None,
+ session: Optional[aiohttp.ClientSession] = None,
) -> None:
"""
Create an instance of MusicAssistant.
await self.http_session.connector.close()
self.http_session.detach()
- def signal_event(
- self, event_type: EventType, event_details: EventDetails = None
- ) -> None:
+ def signal_event(self, event_type: EventType, event_details: Any = None) -> None:
"""
Signal (systemwide) event.
def subscribe(
self,
cb_func: EventCallBackType,
- event_filter: EventType | Tuple[EventType] | None = None,
+ event_filter: Union[EventType, Tuple[EventType], None] = None,
) -> Callable:
"""
Add callback to event listeners.
return remove_listener
- def add_job(self, job: Coroutine, name: str | None = None) -> None:
+ def add_job(self, job: Coroutine, name: Optional[str] = None) -> None:
"""Add job to be (slowly) processed in the background (one by one)."""
if not name:
name = job.__qualname__ or job.__name__
from __future__ import annotations
from abc import ABCMeta, abstractmethod
-from typing import Generic, List, Tuple, TypeVar
+from typing import Generic, List, Optional, Tuple, TypeVar
from music_assistant.helpers.cache import cached
from music_assistant.helpers.typing import MusicAssistant
return (prov.provider, prov.item_id)
return None, None
- async def get_db_items(self, custom_query: str | None = None) -> List[ItemCls]:
+ async def get_db_items(self, custom_query: Optional[str] = None) -> List[ItemCls]:
"""Fetch all records from database."""
if custom_query is not None:
func = self.mass.database.get_rows_from_query(custom_query)
from dataclasses import dataclass, field
from enum import Enum, IntEnum
-from typing import Any, Dict, List, Mapping
+from typing import Any, Dict, List, Mapping, Optional, Union
from mashumaro import DataClassDictMixin
from music_assistant.helpers.json import json
item_id: str
provider: str
name: str
- sort_name: str | None = None
+ sort_name: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
provider_ids: List[MediaItemProviderId] = field(default_factory=list)
in_library: bool = False
media_type: MediaType = MediaType.ALBUM
version: str = ""
- year: int | None = None
- artist: ItemMapping | Artist | None = None
+ year: Optional[int] = None
+ artist: Union[ItemMapping, Artist, None] = None
album_type: AlbumType = AlbumType.UNKNOWN
- upc: str | None = None
+ upc: Optional[str] = None
def __hash__(self):
"""Return custom hash."""
duration: int = 0
version: str = ""
isrc: str = ""
- artists: List[ItemMapping | Artist] = field(default_factory=list)
+ artists: List[Union[ItemMapping, Artist]] = field(default_factory=list)
# album track only
- album: ItemMapping | Album | None = None
- disc_number: int | None = None
- track_number: int | None = None
+ album: Union[ItemMapping, Album, None] = None
+ disc_number: Optional[int] = None
+ track_number: Optional[int] = None
# playlist track only
- position: int | None = None
+ position: Optional[int] = None
def __hash__(self):
"""Return custom hash."""
return f"{provider_id}://{media_type.value}/{item_id}"
-MediaItemType = Artist | Album | Track | Radio | Playlist
+MediaItemType = Union[Artist, Album, Track, Radio, Playlist]
class StreamType(Enum):
details: Dict[str, Any] = field(default_factory=dict)
seconds_played: int = 0
gain_correct: float = 0
- loudness: float | None = None
- sample_rate: int | None = None
- bit_depth: int | None = None
+ loudness: Optional[float] = None
+ sample_rate: Optional[int] = None
+ bit_depth: Optional[int] = None
channels: int = 2
media_type: MediaType = MediaType.TRACK
queue_id: str = None
from asyncio import Task, TimerHandle
from dataclasses import dataclass
from enum import Enum
-from typing import TYPE_CHECKING, List, Tuple
+from typing import TYPE_CHECKING, List, Optional, Tuple, Union
from uuid import uuid4
from mashumaro import DataClassDictMixin
uri: str
name: str = ""
- duration: int | None = None
+ duration: Optional[int] = None
item_id: str = ""
sort_index: int = 0
- streamdetails: StreamDetails | None = None
+ streamdetails: Optional[StreamDetails] = None
is_media_item: bool = False
def __post_init__(self):
self._volume_normalization_enabled: bool = True
self._volume_normalization_target: int = -23
- self._current_index: int | None = None
+ self._current_index: Optional[int] = None
self._current_item_time: int = 0
- self._last_item: QueueItem | None = None
+ self._last_item: Optional[QueueItem] = None
self._start_index: int = 0
self._next_index: int = 0
self._last_state = PlayerState.IDLE
self._update_task: Task = None
self._signal_next: bool = False
self._last_player_update: int = 0
- self._stream_url: str | None = None
+ self._stream_url: Optional[str] = None
async def setup(self) -> None:
"""Handle async setup of instance."""
return self._items
@property
- def current_index(self) -> int | None:
+ def current_index(self) -> Optional[int]:
"""
Return the current index of the queue.
return self._items[self._current_index]
@property
- def next_index(self) -> int | None:
+ def next_index(self) -> Optional[int]:
"""
Return the next index for this PlayerQueue.
return None
return next((x for x in self.items if x.item_id == queue_item_id), None)
- def index_by_id(self, queue_item_id: str) -> int | None:
+ def index_by_id(self, queue_item_id: str) -> Optional[int]:
"""Get index by queue_item_id."""
for index, item in enumerate(self.items):
if item.item_id == queue_item_id:
"resume queue requested for %s but queue is empty", self.queue_id
)
- async def play_index(self, index: int | str) -> None:
+ async def play_index(self, index: Union[int, str]) -> None:
"""Play item at index (or item_id) X in queue."""
if not isinstance(index, int):
index = self.index_by_id(index)
"""Model for a Music Providers."""
-
from __future__ import annotations
from abc import abstractmethod
from logging import Logger
-from typing import List
+from typing import List, Optional
from music_assistant.helpers.typing import MusicAssistant
from music_assistant.models.media_items import (
return self._attr_supported_mediatypes
async def search(
- self, search_query: str, media_types=List[MediaType] | None, limit: int = 5
+ self, search_query: str, media_types=Optional[List[MediaType]], limit: int = 5
) -> List[MediaItemType]:
"""
Perform search on musicprovider.
Should be compatible with LMS
"""
- def __init__(self, music_dir: str, playlist_dir: str | None = None) -> None:
+ def __init__(self, music_dir: str, playlist_dir: Optional[str] = None) -> None:
"""
Initialize the Filesystem provider.
class TuneInProvider(MusicProvider):
"""Provider implementation for Tune In."""
- def __init__(self, username: str | None) -> None:
+ def __init__(self, username: Optional[str]) -> None:
"""Initialize the provider."""
self._attr_id = "tunein"
self._attr_name = "Tune-in Radio"
line_length = 88
[mypy]
-python_version = 3.7
+python_version = 3.9
ignore_errors = true
follow_imports = silent
ignore_missing_imports = true
projects / music-assistant-server.git / commitdiff