music-assistant-server
7.9 KB•PY
parsers.py
7.9 KB • 269 lines • python
1"""Parsers for Emby API responses."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
6
7from music_assistant_models.enums import ContentType, ImageType
8from music_assistant_models.media_items import (
9 Album,
10 Artist,
11 AudioFormat,
12 ItemMapping,
13 MediaItemImage,
14 Playlist,
15 ProviderMapping,
16 Track,
17)
18from music_assistant_models.unique_list import UniqueList
19
20from music_assistant.providers.emby.const import (
21 AUDIO_STREAM_BIT_DEPTH,
22 AUDIO_STREAM_CHANNELS,
23 AUDIO_STREAM_CODEC,
24 AUDIO_STREAM_SAMPLE_RATE,
25 ITEM_KEY_ALBUM_ID,
26 ITEM_KEY_ALBUM_NAME,
27 ITEM_KEY_ARTIST_ITEMS,
28 ITEM_KEY_CONTAINER,
29 ITEM_KEY_ID,
30 ITEM_KEY_IMAGE_TAGS,
31 ITEM_KEY_MEDIA_STREAMS,
32 ITEM_KEY_NAME,
33 ITEM_KEY_RUNTIME_TICKS,
34 ITEM_KEY_TYPE,
35)
36
37if TYPE_CHECKING:
38 from music_assistant.providers.emby import EmbyProvider
39
40
41def parse_track(
42 instance_id: str,
43 provider: EmbyProvider,
44 item: dict[str, Any],
45) -> Track:
46 """Parse an Emby Audio item into a Track."""
47 track_id = str(item.get(ITEM_KEY_ID))
48 name = str(item.get(ITEM_KEY_NAME))
49
50 # Extract artist info
51 artists = UniqueList[Artist | ItemMapping]()
52 if artist_items := item.get(ITEM_KEY_ARTIST_ITEMS):
53 for artist_item in artist_items:
54 artist_name = str(artist_item.get(ITEM_KEY_NAME))
55 artist_id = str(artist_item.get(ITEM_KEY_ID))
56
57 artists.append(
58 Artist(
59 item_id=artist_id,
60 name=artist_name,
61 provider=instance_id,
62 provider_mappings={
63 ProviderMapping(
64 item_id=artist_id,
65 provider_domain=provider.domain,
66 provider_instance=instance_id,
67 )
68 },
69 )
70 )
71
72 album_id = str(item.get(ITEM_KEY_ALBUM_ID))
73 album_name = str(item.get(ITEM_KEY_ALBUM_NAME))
74
75 album = Album(
76 item_id=album_id,
77 name=album_name,
78 provider=instance_id,
79 provider_mappings={
80 ProviderMapping(
81 item_id=album_id,
82 provider_domain=provider.domain,
83 provider_instance=instance_id,
84 )
85 },
86 )
87
88 duration = int(item.get(ITEM_KEY_RUNTIME_TICKS, 0) / 10000000) # Convert ticks to seconds
89 media_streams = item.get(ITEM_KEY_MEDIA_STREAMS, [{}])
90 audio_stream = next((dict(s) for s in media_streams if s.get(ITEM_KEY_TYPE) == "Audio"), {})
91
92 track = Track(
93 item_id=track_id,
94 name=name,
95 album=album,
96 artists=artists,
97 duration=duration,
98 provider=instance_id,
99 provider_mappings={
100 ProviderMapping(
101 item_id=track_id,
102 provider_domain=provider.domain,
103 provider_instance=instance_id,
104 audio_format=AudioFormat(
105 content_type=ContentType.try_parse(str(item.get(ITEM_KEY_CONTAINER))),
106 codec_type=ContentType.try_parse(str(audio_stream.get(AUDIO_STREAM_CODEC))),
107 sample_rate=int(audio_stream.get(AUDIO_STREAM_SAMPLE_RATE, 44100)),
108 bit_depth=int(audio_stream.get(AUDIO_STREAM_BIT_DEPTH, 16)),
109 channels=int(audio_stream.get(AUDIO_STREAM_CHANNELS, 2)),
110 ),
111 )
112 },
113 )
114
115 # Extract images
116 if "Primary" in item.get(ITEM_KEY_IMAGE_TAGS, {}):
117 image_url = f"{provider._base_url}Items/{track_id}/Images/Primary"
118 if track.metadata.images is None:
119 track.metadata.images = UniqueList[MediaItemImage]()
120 track.metadata.images.append(
121 MediaItemImage(
122 type=ImageType.THUMB,
123 path=image_url,
124 provider=instance_id,
125 remotely_accessible=True,
126 )
127 )
128
129 return track
130
131
132def parse_artist(
133 instance_id: str,
134 provider: EmbyProvider,
135 item: dict[str, Any],
136) -> Artist:
137 """Parse an Emby MusicArtist item into an Artist."""
138 artist_id = str(item.get(ITEM_KEY_ID))
139 name = str(item.get(ITEM_KEY_NAME))
140
141 artist = Artist(
142 item_id=artist_id,
143 name=name,
144 provider=instance_id,
145 provider_mappings={
146 ProviderMapping(
147 item_id=artist_id,
148 provider_domain=provider.domain,
149 provider_instance=instance_id,
150 )
151 },
152 )
153
154 # Extract images
155 if "Primary" in item.get(ITEM_KEY_IMAGE_TAGS, {}):
156 image_url = f"{provider._base_url}Items/{artist_id}/Images/Primary"
157 if artist.metadata.images is None:
158 artist.metadata.images = UniqueList[MediaItemImage]()
159 artist.metadata.images.append(
160 MediaItemImage(
161 type=ImageType.THUMB,
162 path=image_url,
163 provider=instance_id,
164 remotely_accessible=True,
165 )
166 )
167
168 return artist
169
170
171def parse_album(
172 instance_id: str,
173 provider: EmbyProvider,
174 item: dict[str, Any],
175) -> Album:
176 """Parse an Emby MusicAlbum item into an Album."""
177 album_id = str(item.get(ITEM_KEY_ID))
178 name = str(item.get(ITEM_KEY_NAME))
179
180 # Extract artist info
181 artists = UniqueList[Artist | ItemMapping]()
182 if artist_items := item.get(ITEM_KEY_ARTIST_ITEMS):
183 for artist_item in artist_items:
184 artist_id = str(artist_item.get(ITEM_KEY_ID))
185 artist_name = str(artist_item.get(ITEM_KEY_NAME))
186
187 artists.append(
188 Artist(
189 item_id=artist_id,
190 name=artist_name,
191 provider=instance_id,
192 provider_mappings={
193 ProviderMapping(
194 item_id=artist_id,
195 provider_domain=provider.domain,
196 provider_instance=instance_id,
197 )
198 },
199 )
200 )
201
202 album = Album(
203 item_id=album_id,
204 name=name,
205 artists=artists,
206 provider=instance_id,
207 provider_mappings={
208 ProviderMapping(
209 item_id=album_id,
210 provider_domain=provider.domain,
211 provider_instance=instance_id,
212 )
213 },
214 )
215
216 # Extract images
217 if image_id := item.get("PrimaryImageItemId"):
218 image_url = f"{provider._base_url}Items/{image_id}/Images/Primary"
219 if album.metadata.images is None:
220 album.metadata.images = UniqueList[MediaItemImage]()
221 album.metadata.images.append(
222 MediaItemImage(
223 type=ImageType.THUMB,
224 path=image_url,
225 provider=instance_id,
226 remotely_accessible=True,
227 )
228 )
229
230 return album
231
232
233def parse_playlist(
234 instance_id: str,
235 provider: EmbyProvider,
236 item: dict[str, Any],
237) -> Playlist:
238 """Parse an Emby Playlist item into a Playlist."""
239 playlist_id = str(item.get(ITEM_KEY_ID))
240 name = str(item.get(ITEM_KEY_NAME))
241
242 playlist = Playlist(
243 item_id=playlist_id,
244 name=name,
245 provider=instance_id,
246 provider_mappings={
247 ProviderMapping(
248 item_id=playlist_id,
249 provider_domain=provider.domain,
250 provider_instance=instance_id,
251 )
252 },
253 )
254 # Extract images
255 if "Primary" in item.get(ITEM_KEY_IMAGE_TAGS, {}):
256 image_url = f"{provider._base_url}Items/{playlist_id}/Images/Primary"
257 if playlist.metadata.images is None:
258 playlist.metadata.images = UniqueList[MediaItemImage]()
259 playlist.metadata.images.append(
260 MediaItemImage(
261 type=ImageType.THUMB,
262 path=image_url,
263 provider=instance_id,
264 remotely_accessible=True,
265 )
266 )
267
268 return playlist
269