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