from __future__ import annotations
+import logging
from typing import TYPE_CHECKING
-from music_assistant_models.enums import ImageType
-from music_assistant_models.media_items import Artist, MediaItemImage, ProviderMapping
+from music_assistant_models.enums import (
+ ImageType,
+ MediaType,
+)
+from music_assistant_models.media_items import (
+ Album,
+ Artist,
+ ItemMapping,
+ MediaItemImage,
+ ProviderMapping,
+)
from music_assistant_models.unique_list import UniqueList
+from music_assistant.constants import UNKNOWN_ARTIST
+
if TYPE_CHECKING:
+ from libopensonic.media import Album as SonicAlbum
+ from libopensonic.media import AlbumInfo as SonicAlbumInfo
from libopensonic.media import Artist as SonicArtist
from libopensonic.media import ArtistInfo as SonicArtistInfo
+UNKNOWN_ARTIST_ID = "fake_artist_unknown"
+
def parse_artist(
instance_id: str, sonic_artist: SonicArtist, sonic_info: SonicArtistInfo = None
)
return artist
+
+
+def parse_album(
+ logger: logging.Logger,
+ instance_id: str,
+ sonic_album: SonicAlbum,
+ sonic_info: SonicAlbumInfo | None = None,
+) -> Album:
+ """Parse album and albumInfo into a Music Assistant Album."""
+ album_id = sonic_album.id
+ album = Album(
+ item_id=album_id,
+ provider="opensubsonic",
+ name=sonic_album.name,
+ favorite=bool(sonic_album.starred),
+ provider_mappings={
+ ProviderMapping(
+ item_id=album_id,
+ provider_domain="opensubsonic",
+ provider_instance=instance_id,
+ )
+ },
+ year=sonic_album.year,
+ )
+
+ album.metadata.images = UniqueList()
+ if sonic_album.cover_id:
+ album.metadata.images.append(
+ MediaItemImage(
+ type=ImageType.THUMB,
+ path=sonic_album.cover_id,
+ provider=instance_id,
+ remotely_accessible=False,
+ ),
+ )
+
+ if sonic_album.artist_id:
+ album.artists.append(
+ ItemMapping(
+ media_type=MediaType.ARTIST,
+ item_id=sonic_album.artist_id,
+ provider=instance_id,
+ name=sonic_album.artist if sonic_album.artist else UNKNOWN_ARTIST,
+ )
+ )
+ else:
+ logger.info(
+ "Unable to find an artist ID for album '%s' with ID '%s'.",
+ sonic_album.name,
+ sonic_album.id,
+ )
+ album.artists.append(
+ Artist(
+ item_id=UNKNOWN_ARTIST_ID,
+ name=UNKNOWN_ARTIST,
+ provider=instance_id,
+ provider_mappings={
+ ProviderMapping(
+ item_id=UNKNOWN_ARTIST_ID,
+ provider_domain="opensubsonic",
+ provider_instance=instance_id,
+ )
+ },
+ )
+ )
+
+ if sonic_info:
+ if sonic_info.small_url:
+ album.metadata.images.append(
+ MediaItemImage(
+ type=ImageType.THUMB,
+ path=sonic_info.small_url,
+ remotely_accessible=False,
+ provider=instance_id,
+ )
+ )
+ if sonic_info.notes:
+ album.metadata.description = sonic_info.notes
+
+ return album
)
from music_assistant.models.music_provider import MusicProvider
-from .parsers import parse_artist
+from .parsers import parse_album, parse_artist
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, Callable
from libopensonic.media import Album as SonicAlbum
- from libopensonic.media import AlbumInfo as SonicAlbumInfo
from libopensonic.media import Artist as SonicArtist
from libopensonic.media import Playlist as SonicPlaylist
from libopensonic.media import PodcastChannel as SonicPodcast
name=name,
)
- def _parse_album(self, sonic_album: SonicAlbum, sonic_info: SonicAlbumInfo = None) -> Album:
- album_id = sonic_album.id
- album = Album(
- item_id=album_id,
- provider=self.domain,
- name=sonic_album.name,
- favorite=bool(sonic_album.starred),
- provider_mappings={
- ProviderMapping(
- item_id=album_id,
- provider_domain=self.domain,
- provider_instance=self.instance_id,
- )
- },
- year=sonic_album.year,
- )
-
- album.metadata.images = UniqueList()
- if sonic_album.cover_id:
- album.metadata.images.append(
- MediaItemImage(
- type=ImageType.THUMB,
- path=sonic_album.cover_id,
- provider=self.instance_id,
- remotely_accessible=False,
- ),
- )
-
- if sonic_album.artist_id:
- album.artists.append(
- self._get_item_mapping(
- MediaType.ARTIST,
- sonic_album.artist_id,
- sonic_album.artist if sonic_album.artist else UNKNOWN_ARTIST,
- )
- )
- else:
- self.logger.info(
- "Unable to find an artist ID for album '%s' with ID '%s'.",
- sonic_album.name,
- sonic_album.id,
- )
- album.artists.append(
- Artist(
- item_id=UNKNOWN_ARTIST_ID,
- name=UNKNOWN_ARTIST,
- provider=self.instance_id,
- provider_mappings={
- ProviderMapping(
- item_id=UNKNOWN_ARTIST_ID,
- provider_domain=self.domain,
- provider_instance=self.instance_id,
- )
- },
- )
- )
-
- if sonic_info:
- if sonic_info.small_url:
- album.metadata.images.append(
- MediaItemImage(
- type=ImageType.THUMB,
- path=sonic_info.small_url,
- remotely_accessible=False,
- provider=self.instance_id,
- )
- )
- if sonic_info.notes:
- album.metadata.description = sonic_info.notes
-
- return album
-
def _parse_track(
self, sonic_song: SonicSong, album: Album | ItemMapping | None = None
) -> Track:
)
return SearchResults(
artists=[parse_artist(self.instance_id, entry) for entry in answer["artists"]],
- albums=[self._parse_album(entry) for entry in answer["albums"]],
+ albums=[
+ parse_album(self.logger, self.instance_id, entry) for entry in answer["albums"]
+ ],
tracks=[self._parse_track(entry) for entry in answer["songs"]],
)
)
while albums:
for album in albums:
- yield self._parse_album(album)
+ yield parse_album(self.logger, self.instance_id, album)
offset += size
albums = await self._run_async(
self._conn.getAlbumList2,
msg = f"Album {prov_album_id} not found"
raise MediaNotFoundError(msg) from e
- return self._parse_album(sonic_album, sonic_info)
+ return parse_album(self.logger, self.instance_id, sonic_album, sonic_info)
async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
"""Return a list of tracks on the specified Album."""
raise MediaNotFoundError(msg) from e
albums = []
for entry in sonic_artist.albums:
- albums.append(self._parse_album(entry))
+ albums.append(parse_album(self.logger, self.instance_id, entry))
return albums
async def get_playlist(self, prov_playlist_id: str) -> Playlist:
# serializer version: 1
+# name: test_parse_albums[spec.album]
+ dict({
+ 'album_type': 'unknown',
+ 'artists': list([
+ dict({
+ 'available': True,
+ 'external_ids': list([
+ ]),
+ 'image': None,
+ 'item_id': '91c3901ac465b9efc439e4be4270c2b6',
+ 'media_type': 'artist',
+ 'name': 'pornophonique',
+ 'provider': 'xx-instance-id-xx',
+ 'sort_name': 'pornophonique',
+ 'uri': 'xx-instance-id-xx://artist/91c3901ac465b9efc439e4be4270c2b6',
+ 'version': '',
+ }),
+ ]),
+ 'external_ids': list([
+ ]),
+ 'favorite': True,
+ 'item_id': 'ad0f112b6dcf83de5e9cae85d07f0d35',
+ 'media_type': 'album',
+ 'metadata': dict({
+ 'chapters': None,
+ 'copyright': None,
+ 'description': None,
+ 'explicit': None,
+ 'genres': None,
+ 'images': list([
+ dict({
+ 'path': 'al-ad0f112b6dcf83de5e9cae85d07f0d35_640a93a8',
+ 'provider': 'xx-instance-id-xx',
+ 'remotely_accessible': False,
+ 'type': 'thumb',
+ }),
+ ]),
+ 'label': None,
+ 'languages': None,
+ 'last_refresh': None,
+ 'links': None,
+ 'lyrics': None,
+ 'mood': None,
+ 'performers': None,
+ 'popularity': None,
+ 'preview': None,
+ 'release_date': None,
+ 'review': None,
+ 'style': None,
+ }),
+ 'name': '8-bit lagerfeuer',
+ 'position': None,
+ 'provider': 'opensubsonic',
+ 'provider_mappings': list([
+ dict({
+ 'audio_format': dict({
+ 'bit_depth': 16,
+ 'bit_rate': 0,
+ 'channels': 2,
+ 'content_type': '?',
+ 'output_format_str': '?',
+ 'sample_rate': 44100,
+ }),
+ 'available': True,
+ 'details': None,
+ 'item_id': 'ad0f112b6dcf83de5e9cae85d07f0d35',
+ 'provider_domain': 'opensubsonic',
+ 'provider_instance': 'xx-instance-id-xx',
+ 'url': None,
+ }),
+ ]),
+ 'sort_name': '8-bit lagerfeuer',
+ 'uri': 'opensubsonic://album/ad0f112b6dcf83de5e9cae85d07f0d35',
+ 'version': '',
+ 'year': 2007,
+ })
+# ---
+# name: test_parse_albums[spec.album].1
+ dict({
+ 'album_type': 'unknown',
+ 'artists': list([
+ dict({
+ 'available': True,
+ 'external_ids': list([
+ ]),
+ 'image': None,
+ 'item_id': '91c3901ac465b9efc439e4be4270c2b6',
+ 'media_type': 'artist',
+ 'name': 'pornophonique',
+ 'provider': 'xx-instance-id-xx',
+ 'sort_name': 'pornophonique',
+ 'uri': 'xx-instance-id-xx://artist/91c3901ac465b9efc439e4be4270c2b6',
+ 'version': '',
+ }),
+ ]),
+ 'external_ids': list([
+ ]),
+ 'favorite': True,
+ 'item_id': 'ad0f112b6dcf83de5e9cae85d07f0d35',
+ 'media_type': 'album',
+ 'metadata': dict({
+ 'chapters': None,
+ 'copyright': None,
+ 'description': 'Download the full release here (creative commons). These cripsy beats are ripe with thumping funk and techno influences, sample wizardry and daring shuffles. Composed with the help of unique sound plugins which were especially programmed to measure Comfort Fit’s needs and wishes, we think the chances aren’t bad that you’ll fall for the unique sound signature, bounce and elegance of this unusual Hip Hop production. Ltj bukem / Good looking Rec., UK: "Really love this music." Velanche / XLR8R, UK: "Awesome job he\'s done... overall production is dope." Kwesi / BBE Music, UK: "Wooooooowwwww... WHAT THE FUCK! THIS IS WHAT',
+ 'explicit': None,
+ 'genres': None,
+ 'images': list([
+ dict({
+ 'path': 'al-ad0f112b6dcf83de5e9cae85d07f0d35_640a93a8',
+ 'provider': 'xx-instance-id-xx',
+ 'remotely_accessible': False,
+ 'type': 'thumb',
+ }),
+ dict({
+ 'path': 'http://localhost:8989/play/art/0f8c3cbd6b0b22c3b5402141351ac812/album/21/thumb34.jpg',
+ 'provider': 'xx-instance-id-xx',
+ 'remotely_accessible': False,
+ 'type': 'thumb',
+ }),
+ ]),
+ 'label': None,
+ 'languages': None,
+ 'last_refresh': None,
+ 'links': None,
+ 'lyrics': None,
+ 'mood': None,
+ 'performers': None,
+ 'popularity': None,
+ 'preview': None,
+ 'release_date': None,
+ 'review': None,
+ 'style': None,
+ }),
+ 'name': '8-bit lagerfeuer',
+ 'position': None,
+ 'provider': 'opensubsonic',
+ 'provider_mappings': list([
+ dict({
+ 'audio_format': dict({
+ 'bit_depth': 16,
+ 'bit_rate': 0,
+ 'channels': 2,
+ 'content_type': '?',
+ 'output_format_str': '?',
+ 'sample_rate': 44100,
+ }),
+ 'available': True,
+ 'details': None,
+ 'item_id': 'ad0f112b6dcf83de5e9cae85d07f0d35',
+ 'provider_domain': 'opensubsonic',
+ 'provider_instance': 'xx-instance-id-xx',
+ 'url': None,
+ }),
+ ]),
+ 'sort_name': '8-bit lagerfeuer',
+ 'uri': 'opensubsonic://album/ad0f112b6dcf83de5e9cae85d07f0d35',
+ 'version': '',
+ 'year': 2007,
+ })
+# ---
# name: test_parse_artists[spec-artistid3.artist]
dict({
'external_ids': list([
--- /dev/null
+{
+ "id": "ad0f112b6dcf83de5e9cae85d07f0d35",
+ "name": "8-bit lagerfeuer",
+ "artist": "pornophonique",
+ "year": 2007,
+ "coverArt": "al-ad0f112b6dcf83de5e9cae85d07f0d35_640a93a8",
+ "starred": "2023-03-22T01:51:06Z",
+ "duration": 1954,
+ "playCount": 97,
+ "genre": "Hip-Hop",
+ "created": "2023-03-10T02:19:35.784818075Z",
+ "artistId": "91c3901ac465b9efc439e4be4270c2b6",
+ "songCount": 8,
+ "played": "2023-03-28T00:45:13Z",
+ "userRating": 4,
+ "recordLabels": [
+ {
+ "name": "Sony"
+ }
+ ],
+ "musicBrainzId": "189002e7-3285-4e2e-92a3-7f6c30d407a2",
+ "genres": [
+ {
+ "name": "Hip-Hop"
+ },
+ {
+ "name": "East coast"
+ }
+ ],
+ "artists": [
+ {
+ "id": "ar-1",
+ "name": "Artist 1"
+ },
+ {
+ "id": "ar-2",
+ "name": "Artist 2"
+ }
+ ],
+ "displayArtist": "Artist 1 feat. Artist 2",
+ "releaseTypes": [
+ "Album",
+ "Remixes"
+ ],
+ "moods": [
+ "slow",
+ "cool"
+ ],
+ "sortName": "lagerfeuer (8-bit)",
+ "originalReleaseDate": {
+ "year": 2001,
+ "month": 3,
+ "day": 10
+ },
+ "releaseDate": {
+ "year": 2001,
+ "month": 3,
+ "day": 10
+ },
+ "isCompilation": false,
+ "explicitStatus": "explicit",
+ "discTitles": [
+ {
+ "disc": 0,
+ "title": "Disc 0 title"
+ },
+ {
+ "disc": 2,
+ "title": "Disc 1 title"
+ }
+ ]
+}
--- /dev/null
+{
+ "notes": "Download the full release here (creative commons). These cripsy beats are ripe with thumping funk and techno influences, sample wizardry and daring shuffles. Composed with the help of unique sound plugins which were especially programmed to measure Comfort Fit’s needs and wishes, we think the chances aren’t bad that you’ll fall for the unique sound signature, bounce and elegance of this unusual Hip Hop production. Ltj bukem / Good looking Rec., UK: \"Really love this music.\" Velanche / XLR8R, UK: \"Awesome job he's done... overall production is dope.\" Kwesi / BBE Music, UK: \"Wooooooowwwww... WHAT THE FUCK! THIS IS WHAT",
+ "musicBrainzId": "6e1d48f7-717c-416e-af35-5d2454a13af2",
+ "smallImageUrl": "http://localhost:8989/play/art/0f8c3cbd6b0b22c3b5402141351ac812/album/21/thumb34.jpg",
+ "mediumImageUrl": "http://localhost:8989/play/art/41b16680dc1b3aaf5dfba24ddb6a1712/album/21/thumb64.jpg",
+ "largeImageUrl": "http://localhost:8989/play/art/e6fd8d4e0d35c4436e56991892bfb27b/album/21/thumb174.jpg"
+}
"""Test we can parse Jellyfin models into Music Assistant models."""
import json
+import logging
import pathlib
import aiofiles
import pytest
+from libopensonic.media.album import Album, AlbumInfo
from libopensonic.media.artist import Artist, ArtistInfo
from syrupy.assertion import SnapshotAssertion
-from music_assistant.providers.opensubsonic.parsers import parse_artist
+from music_assistant.providers.opensubsonic.parsers import parse_album, parse_artist
FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures"
ARTIST_FIXTURES = list(FIXTURES_DIR.glob("artists/*.artist.json"))
+ALBUM_FIXTURES = list(FIXTURES_DIR.glob("albums/*.album.json"))
+
+_LOGGER = logging.getLogger(__name__)
@pytest.mark.parametrize("example", ARTIST_FIXTURES, ids=lambda val: str(val.stem))
# sort external Ids to ensure they are always in the same order for snapshot testing
parsed["external_ids"].sort()
assert snapshot == parsed
+
+
+@pytest.mark.parametrize("example", ALBUM_FIXTURES, ids=lambda val: str(val.stem))
+async def test_parse_albums(example: pathlib.Path, snapshot: SnapshotAssertion) -> None:
+ """Test we can parse albums."""
+ async with aiofiles.open(example) as fp:
+ album = Album(json.loads(await fp.read()))
+
+ parsed = parse_album(_LOGGER, "xx-instance-id-xx", album).to_dict()
+ # sort external Ids to ensure they are always in the same order for snapshot testing
+ parsed["external_ids"].sort()
+ assert snapshot == parsed
+
+ # Find the corresponding info file
+ example_info = example.with_suffix("").with_suffix(".info.json")
+ async with aiofiles.open(example_info) as fp:
+ album_info = AlbumInfo(json.loads(await fp.read()))
+
+ parsed = parse_album(_LOGGER, "xx-instance-id-xx", album, album_info).to_dict()
+ # sort external Ids to ensure they are always in the same order for snapshot testing
+ parsed["external_ids"].sort()
+ assert snapshot == parsed
projects / music-assistant-server.git / commitdiff