from __future__ import annotations
+import asyncio
from collections.abc import AsyncGenerator, Sequence
from typing import TYPE_CHECKING
from music_assistant_models.streamdetails import StreamDetails
from music_assistant.models.music_provider import MusicProvider
-from music_assistant.providers.audiobookshelf.abs_client import ABSClient
+from music_assistant.providers.audiobookshelf.abs_client import ABSClient, LibraryWithItemIDs
from music_assistant.providers.audiobookshelf.abs_schema import (
ABSDeviceInfo,
- ABSLibrary,
ABSLibraryItemExpandedBook,
ABSLibraryItemExpandedPodcast,
+ ABSLibraryItemMinifiedBook,
+ ABSLibraryItemMinifiedPodcast,
ABSPlaybackSessionExpanded,
ABSPodcastEpisodeExpanded,
)
CONF_USERNAME = "username"
CONF_PASSWORD = "password"
CONF_VERIFY_SSL = "verify_ssl"
+# optionally hide podcasts with no episodes
+CONF_HIDE_EMPTY_PODCASTS = "hide_empty_podcasts"
async def setup(
category="advanced",
default_value=True,
),
+ ConfigEntry(
+ key=CONF_HIDE_EMPTY_PODCASTS,
+ type=ConfigEntryType.BOOLEAN,
+ label="Hide empty podcasts.",
+ required=False,
+ description="This will skip podcasts with no episodes associated.",
+ category="advanced",
+ default_value=False,
+ ),
)
except RuntimeError:
# login details were not correct
raise LoginFailed(f"Login to abs instance at {base_url} failed.")
- await self._client.sync()
# this will be provided when creating sessions or receive already opened sessions
self.device_info = ABSDeviceInfo(
await self._client.sync()
await super().sync_library(media_types=media_types)
- def _parse_podcast(self, abs_podcast: ABSLibraryItemExpandedPodcast) -> Podcast:
+ def _parse_podcast(
+ self, abs_podcast: ABSLibraryItemExpandedPodcast | ABSLibraryItemMinifiedPodcast
+ ) -> Podcast:
"""Translate ABSPodcast to MassPodcast."""
title = abs_podcast.media.metadata.title
# Per API doc title may be None.
name=title,
publisher=abs_podcast.media.metadata.author,
provider=self.lookup_key,
- total_episodes=len(abs_podcast.media.episodes),
provider_mappings={
ProviderMapping(
item_id=abs_podcast.id_,
mass_podcast.metadata.genres = set(abs_podcast.media.metadata.genres)
mass_podcast.metadata.release_date = abs_podcast.media.metadata.release_date
+ if isinstance(abs_podcast, ABSLibraryItemExpandedPodcast):
+ mass_podcast.total_episodes = len(abs_podcast.media.episodes)
+ elif isinstance(abs_podcast, ABSLibraryItemMinifiedPodcast):
+ mass_podcast.total_episodes = abs_podcast.media.num_episodes
+
return mass_podcast
async def _parse_podcast_episode(
async def get_library_podcasts(self) -> AsyncGenerator[Podcast, None]:
"""Retrieve library/subscribed podcasts from the provider."""
- async for abs_podcast in self._client.get_all_podcasts():
+ async for abs_podcast in self._client.get_all_podcasts_minified():
mass_podcast = self._parse_podcast(abs_podcast)
+ if (
+ bool(self.config.get_value(CONF_HIDE_EMPTY_PODCASTS))
+ and mass_podcast.total_episodes == 0
+ ):
+ continue
yield mass_podcast
async def get_podcast(self, prov_podcast_id: str) -> Podcast:
"""Get single podcast."""
- abs_podcast = await self._client.get_podcast(prov_podcast_id)
+ abs_podcast = await self._client.get_podcast_expanded(prov_podcast_id)
return self._parse_podcast(abs_podcast)
async def get_podcast_episodes(self, prov_podcast_id: str) -> list[PodcastEpisode]:
"""Get all podcast episodes of podcast."""
- abs_podcast = await self._client.get_podcast(prov_podcast_id)
+ abs_podcast = await self._client.get_podcast_expanded(prov_podcast_id)
episode_list = []
episode_cnt = 1
for abs_episode in abs_podcast.media.episodes:
async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
"""Get single podcast episode."""
prov_podcast_id, e_id = prov_episode_id.split(" ")
- abs_podcast = await self._client.get_podcast(prov_podcast_id)
+ abs_podcast = await self._client.get_podcast_expanded(prov_podcast_id)
episode_cnt = 1
for abs_episode in abs_podcast.media.episodes:
if abs_episode.id_ == e_id:
episode_cnt += 1
raise MediaNotFoundError("Episode not found")
- async def _parse_audiobook(self, abs_audiobook: ABSLibraryItemExpandedBook) -> Audiobook:
+ async def _parse_audiobook(
+ self, abs_audiobook: ABSLibraryItemExpandedBook | ABSLibraryItemMinifiedBook
+ ) -> Audiobook:
mass_audiobook = Audiobook(
item_id=abs_audiobook.id_,
provider=self.lookup_key,
)
},
publisher=abs_audiobook.media.metadata.publisher,
- authors=UniqueList([x.name for x in abs_audiobook.media.metadata.authors]),
- narrators=UniqueList(abs_audiobook.media.metadata.narrators),
)
mass_audiobook.metadata.description = abs_audiobook.media.metadata.description
if abs_audiobook.media.metadata.language is not None:
if abs_audiobook.media.metadata.genres is not None:
mass_audiobook.metadata.genres = set(abs_audiobook.media.metadata.genres)
- # chapters
- chapters = []
- for idx, chapter in enumerate(abs_audiobook.media.chapters):
- chapters.append(
- MediaItemChapter(
- position=idx + 1, # chapter starting at 1
- name=chapter.title,
- start=chapter.start,
- end=chapter.end,
- )
- )
- mass_audiobook.metadata.chapters = chapters
-
mass_audiobook.metadata.explicit = abs_audiobook.media.metadata.explicit
- progress, finished = await self._client.get_audiobook_progress_ms(abs_audiobook.id_)
- if progress is not None:
- mass_audiobook.resume_position_ms = progress
- mass_audiobook.fully_played = finished
# cover
base_url = f"{self.config.get_value(CONF_URL)}"
[MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=self.lookup_key)]
)
+ # expanded version
+ if isinstance(abs_audiobook, ABSLibraryItemExpandedBook):
+ authors = UniqueList([x.name for x in abs_audiobook.media.metadata.authors])
+ narrators = UniqueList(abs_audiobook.media.metadata.narrators)
+ mass_audiobook.authors = authors
+ mass_audiobook.narrators = narrators
+ chapters = []
+ for idx, chapter in enumerate(abs_audiobook.media.chapters):
+ chapters.append(
+ MediaItemChapter(
+ position=idx + 1, # chapter starting at 1
+ name=chapter.title,
+ start=chapter.start,
+ end=chapter.end,
+ )
+ )
+ mass_audiobook.metadata.chapters = chapters
+
+ progress, finished = await self._client.get_audiobook_progress_ms(abs_audiobook.id_)
+ if progress is not None:
+ mass_audiobook.resume_position_ms = progress
+ mass_audiobook.fully_played = finished
+ elif isinstance(abs_audiobook, ABSLibraryItemMinifiedBook):
+ mass_audiobook.authors = UniqueList([abs_audiobook.media.metadata.author_name])
+ mass_audiobook.narrators = UniqueList([abs_audiobook.media.metadata.narrator_name])
+
return mass_audiobook
async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook, None]:
"""Get Audiobook libraries."""
- async for abs_audiobook in self._client.get_all_audiobooks():
+ async for abs_audiobook in self._client.get_all_audiobooks_minified():
mass_audiobook = await self._parse_audiobook(abs_audiobook)
yield mass_audiobook
async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
"""Get a single audiobook."""
- abs_audiobook = await self._client.get_audiobook(prov_audiobook_id)
+ abs_audiobook = await self._client.get_audiobook_expanded(prov_audiobook_id)
return await self._parse_audiobook(abs_audiobook)
async def get_streamdetails_from_playback_session(
if media_type == MediaType.PODCAST_EPISODE:
return await self._get_stream_details_podcast_episode(item_id)
elif media_type == MediaType.AUDIOBOOK:
- abs_audiobook = await self._client.get_audiobook(item_id)
+ abs_audiobook = await self._client.get_audiobook_expanded(item_id)
tracks = abs_audiobook.media.tracks
if len(tracks) == 0:
raise MediaNotFoundError("Stream not found")
session = await self._client.get_playback_session_audiobook(
device_info=self.device_info, audiobook_id=item_id
)
+ # small delay, allow abs to launch ffmpeg process
+ await asyncio.sleep(1)
return await self.get_streamdetails_from_playback_session(session)
return await self._get_stream_details_audiobook(abs_audiobook)
raise MediaNotFoundError("Stream unknown")
abs_podcast_id, abs_episode_id = podcast_id.split(" ")
abs_episode = None
- abs_podcast = await self._client.get_podcast(abs_podcast_id)
+ abs_podcast = await self._client.get_podcast_expanded(abs_podcast_id)
for abs_episode in abs_podcast.media.episodes:
if abs_episode.id_ == abs_episode_id:
break
)
async def _browse_root(
- self, library_list: list[ABSLibrary], item_path: str
+ self, library_list: list[LibraryWithItemIDs], item_path: str
) -> Sequence[MediaItemType | ItemMapping]:
"""Browse root folder in browse view.
async def _browse_lib(
self,
library_id: str,
- library_list: list[ABSLibrary],
+ library_list: list[LibraryWithItemIDs],
media_type: MediaType,
) -> Sequence[MediaItemType | ItemMapping]:
"""Browse lib folder in browse view.
if library is None:
raise MediaNotFoundError("Lib missing.")
- def get_item_mapping(
- item: ABSLibraryItemExpandedBook | ABSLibraryItemExpandedPodcast,
- ) -> ItemMapping:
- title = item.media.metadata.title
- if title is None:
- title = "UNKNOWN"
- token = self._client.token
- url = f"{self.config.get_value(CONF_URL)}/api/items/{item.id_}/cover?token={token}"
- image = MediaItemImage(type=ImageType.THUMB, path=url, provider=self.lookup_key)
- return ItemMapping(
- media_type=media_type,
- item_id=item.id_,
- provider=self.lookup_key,
- name=title,
- image=image,
- )
-
items: list[MediaItemType | ItemMapping] = []
- if media_type == MediaType.PODCAST:
- async for podcast in self._client.get_all_podcasts_by_library(library):
- items.append(get_item_mapping(podcast))
- elif media_type == MediaType.AUDIOBOOK:
- async for audiobook in self._client.get_all_audiobooks_by_library(library):
- items.append(get_item_mapping(audiobook))
+ if media_type in [MediaType.PODCAST, MediaType.AUDIOBOOK]:
+ for item_id in library.item_ids:
+ mass_item = await self.mass.music.get_library_item_by_prov_id(
+ media_type=media_type,
+ item_id=item_id,
+ provider_instance_id_or_domain=self.instance_id,
+ )
+ if mass_item is not None:
+ items.append(mass_item)
else:
raise RuntimeError(f"Media type must not be {media_type}")
return items
import logging
from collections.abc import AsyncGenerator
+from dataclasses import dataclass, field
from enum import Enum
from typing import Any
from aiohttp import ClientSession
+from mashumaro.exceptions import InvalidFieldValue, MissingField
from music_assistant_models.media_items import UniqueList
from music_assistant.providers.audiobookshelf.abs_schema import (
ABSLibrariesItemsMinifiedBookResponse,
ABSLibrariesItemsMinifiedPodcastResponse,
ABSLibrariesResponse,
- ABSLibrary,
ABSLibraryItemExpandedBook,
ABSLibraryItemExpandedPodcast,
ABSLibraryItemMinifiedBook,
ABSLibraryItemMinifiedPodcast,
ABSLoginResponse,
ABSMediaProgress,
- ABSPlaybackSession,
ABSPlaybackSessionExpanded,
ABSPlayRequest,
- ABSSessionsResponse,
ABSSessionUpdate,
ABSUser,
)
LIMIT_ITEMS_PER_PAGE = 10
+@dataclass
+class LibraryWithItemIDs:
+ """Helper class to store ABSLibrary, and the ids of the items associated."""
+
+ id_: str
+ name: str = ""
+ item_ids: UniqueList[str] = field(default_factory=UniqueList[str])
+
+
class ABSStatus(Enum):
"""ABS Status Enum."""
def __init__(self) -> None:
"""Client authorization."""
- self.podcast_libraries: list[ABSLibrary] = []
- self.audiobook_libraries: list[ABSLibrary] = []
+ self.podcast_libraries: list[LibraryWithItemIDs] = []
+ self.audiobook_libraries: list[LibraryWithItemIDs] = []
self.user: ABSUser
self.check_ssl: bool
# I would like to receive opened sessions via the API, however, it appears
async def sync(self) -> None:
"""Update available book and podcast libraries."""
data = await self._get("libraries")
- libraries = ABSLibrariesResponse.from_json(data)
+ try:
+ libraries = ABSLibrariesResponse.from_json(data)
+ except (MissingField, InvalidFieldValue) as exc:
+ self.logger.error(exc)
+ return
ids = [x.id_ for x in self.audiobook_libraries]
ids.extend([x.id_ for x in self.podcast_libraries])
for library in libraries.libraries:
media_type = library.media_type
if library.id_ not in ids:
+ _library = LibraryWithItemIDs(library.id_, library.name)
if media_type == "book":
- self.audiobook_libraries.append(library)
+ self.audiobook_libraries.append(_library)
elif media_type == "podcast":
- self.podcast_libraries.append(library)
+ self.podcast_libraries.append(_library)
self.user = await self.get_authenticated_user()
- async def get_all_podcasts(self) -> AsyncGenerator[ABSLibraryItemExpandedPodcast]:
+ async def get_all_podcasts_minified(self) -> AsyncGenerator[ABSLibraryItemMinifiedPodcast]:
"""Get all available podcasts."""
for library in self.podcast_libraries:
- async for podcast in self.get_all_podcasts_by_library(library):
+ async for podcast in self.get_all_podcasts_by_library_minified(library):
yield podcast
- async def _get_lib_items(self, lib: ABSLibrary) -> AsyncGenerator[bytes]:
+ async def _get_lib_items(self, lib: LibraryWithItemIDs) -> AsyncGenerator[bytes]:
"""Get library items with pagination.
Note:
page_cnt += 1
yield data
- async def get_all_podcasts_by_library(
- self, lib: ABSLibrary
- ) -> AsyncGenerator[ABSLibraryItemExpandedPodcast]:
+ async def get_all_podcasts_by_library_minified(
+ self, lib: LibraryWithItemIDs
+ ) -> AsyncGenerator[ABSLibraryItemMinifiedPodcast]:
"""Get all podcasts in a library."""
async for podcast_data in self._get_lib_items(lib):
- podcast_list = ABSLibrariesItemsMinifiedPodcastResponse.from_json(podcast_data).results
+ try:
+ podcast_list = ABSLibrariesItemsMinifiedPodcastResponse.from_json(
+ podcast_data
+ ).results
+ except (MissingField, InvalidFieldValue) as exc:
+ self.logger.error(exc)
+ return
if not podcast_list: # [] if page exceeds
return
- async def _get_id(
- plist: list[ABSLibraryItemMinifiedPodcast] = podcast_list,
- ) -> AsyncGenerator[str]:
- for entry in plist:
- yield entry.id_
-
- async for id_ in _get_id():
- podcast = await self.get_podcast(id_)
+ for podcast in podcast_list:
+ # store ids of library items for later use
+ lib.item_ids.append(podcast.id_)
yield podcast
- async def get_podcast(self, id_: str) -> ABSLibraryItemExpandedPodcast:
+ async def get_podcast_expanded(self, id_: str) -> ABSLibraryItemExpandedPodcast:
"""Get a single Podcast by ID."""
# this endpoint gives more podcast extra data
data = await self._get(f"items/{id_}?expanded=1")
- return ABSLibraryItemExpandedPodcast.from_json(data)
+ try:
+ abs_podcast = ABSLibraryItemExpandedPodcast.from_json(data)
+ except (MissingField, InvalidFieldValue) as exc:
+ self.logger.error(exc)
+ raise RuntimeError from exc
+ return abs_podcast
async def _get_progress_ms(
self,
if not data:
# entry doesn't exist, so it wasn't played yet
return 0, False
- abs_media_progress = ABSMediaProgress.from_json(data)
+ try:
+ abs_media_progress = ABSMediaProgress.from_json(data)
+ except (MissingField, InvalidFieldValue) as exc:
+ self.logger.error(exc)
+ return None, False
return (
int(abs_media_progress.current_time * 1000),
endpoint = f"me/progress/{audiobook_id}"
await self._update_progress(endpoint, progress_s, duration_s, is_finished)
- async def get_all_audiobooks(self) -> AsyncGenerator[ABSLibraryItemExpandedBook]:
+ async def get_all_audiobooks_minified(self) -> AsyncGenerator[ABSLibraryItemMinifiedBook]:
"""Get all audiobooks."""
for library in self.audiobook_libraries:
- async for book in self.get_all_audiobooks_by_library(library):
+ async for book in self.get_all_audiobooks_by_library_minified(library):
yield book
- async def get_all_audiobooks_by_library(
- self, lib: ABSLibrary
- ) -> AsyncGenerator[ABSLibraryItemExpandedBook]:
+ async def get_all_audiobooks_by_library_minified(
+ self, lib: LibraryWithItemIDs
+ ) -> AsyncGenerator[ABSLibraryItemMinifiedBook]:
"""Get all Audiobooks in a library."""
async for audiobook_data in self._get_lib_items(lib):
- audiobook_list = ABSLibrariesItemsMinifiedBookResponse.from_json(audiobook_data).results
+ try:
+ audiobook_list = ABSLibrariesItemsMinifiedBookResponse.from_json(
+ audiobook_data
+ ).results
+ except (MissingField, InvalidFieldValue) as exc:
+ self.logger.error(exc)
+ return
if not audiobook_list: # [] if page exceeds
return
- async def _get_id(
- alist: list[ABSLibraryItemMinifiedBook] = audiobook_list,
- ) -> AsyncGenerator[str]:
- for entry in alist:
- yield entry.id_
-
- async for id_ in _get_id():
- audiobook = await self.get_audiobook(id_)
+ for audiobook in audiobook_list:
+ # store ids of library items for later use
+ lib.item_ids.append(audiobook.id_)
yield audiobook
- async def get_audiobook(self, id_: str) -> ABSLibraryItemExpandedBook:
+ async def get_audiobook_expanded(self, id_: str) -> ABSLibraryItemExpandedBook:
"""Get a single Audiobook by ID."""
# this endpoint gives more audiobook extra data
audiobook = await self._get(f"items/{id_}?expanded=1")
- return ABSLibraryItemExpandedBook.from_json(audiobook)
+ try:
+ abs_book = ABSLibraryItemExpandedBook.from_json(audiobook)
+ except (MissingField, InvalidFieldValue) as exc:
+ self.logger.error(exc)
+ raise RuntimeError from exc
+ return abs_book
async def get_playback_session_podcast(
self, device_info: ABSDeviceInfo, podcast_id: str, episode_id: str
device_info.device_id += f"/{audiobook_id}"
return await self._get_playback_session(endpoint, device_info=device_info)
- async def get_open_playback_session(self, session_id: str) -> ABSPlaybackSessionExpanded | None:
- """Return open playback session."""
- data = await self._get(f"session/{session_id}")
- if data:
- return ABSPlaybackSessionExpanded.from_json(data)
- else:
- return None
-
async def _get_playback_session(
self, endpoint: str, device_info: ABSDeviceInfo
) -> ABSPlaybackSessionExpanded:
supported_mime_types=[],
)
data = await self._post(endpoint, data=play_request.to_dict())
- session = ABSPlaybackSessionExpanded.from_json(data)
+ try:
+ session = ABSPlaybackSessionExpanded.from_json(data)
+ except (MissingField, InvalidFieldValue) as exc:
+ self.logger.error(exc)
+ raise RuntimeError from exc
+
self.logger.debug(
f"Got playback session {session.id_} "
f"for {session.media_type} named {session.display_title}"
"""Sync an open playback session."""
await self._post(f"session/{playback_session_id}/sync", data=update.to_dict())
- async def get_all_closed_playback_sessions(self) -> AsyncGenerator[ABSPlaybackSession]:
- """Get library items with pagination.
-
- This returns only sessions, which are already closed.
- """
- page_cnt = 0
- while True:
- data = await self._get(
- "me/listening-sessions",
- params={"itemsPerPage": LIMIT_ITEMS_PER_PAGE, "page": page_cnt},
- )
- page_cnt += 1
-
- sessions = ABSSessionsResponse.from_json(data).sessions
- self.logger.debug([session.device_info for session in sessions])
- if sessions:
- for session in sessions:
- yield session
- else:
- return
+ # async def get_all_closed_playback_sessions(self) -> AsyncGenerator[ABSPlaybackSession]:
+ # """Get library items with pagination.
+ #
+ # This returns only sessions, which are already closed.
+ # """
+ # page_cnt = 0
+ # while True:
+ # data = await self._get(
+ # "me/listening-sessions",
+ # params={"itemsPerPage": LIMIT_ITEMS_PER_PAGE, "page": page_cnt},
+ # )
+ # page_cnt += 1
+ #
+ # sessions = ABSSessionsResponse.from_json(data).sessions
+ # # self.logger.debug([session.device_info for session in sessions])
+ # if sessions:
+ # for session in sessions:
+ # yield session
+ # else:
+ # return
async def close_all_playback_sessions(self) -> None:
"""Cleanup all playback sessions opened by us."""
https://api.audiobookshelf.org/#audio-track
"""
- index: int | None
- start_offset: Annotated[float, Alias("startOffset")] = 0.0
- duration: float = 0.0
- title: str = ""
+ # index: int | None
+ # start_offset: Annotated[float, Alias("startOffset")] = 0.0
+ # duration: float = 0.0
+ # title: str = ""
content_url: Annotated[str, Alias("contentUrl")] = ""
mime_type: str = ""
# metadata: # not needed for mass application
https://api.audiobookshelf.org/#user-permissions
"""
- download: bool
- update: bool
- delete: bool
- upload: bool
- access_all_libraries: Annotated[bool, Alias("accessAllLibraries")]
- access_all_tags: Annotated[bool, Alias("accessAllTags")]
- access_explicit_content: Annotated[bool, Alias("accessExplicitContent")]
+ # download: bool
+ # update: bool
+ # delete: bool
+ # upload: bool
+ # access_all_libraries: Annotated[bool, Alias("accessAllLibraries")]
+ # access_all_tags: Annotated[bool, Alias("accessAllTags")]
+ # access_explicit_content: Annotated[bool, Alias("accessExplicitContent")]
@dataclass
# displayOrder: Integer
# icon: String
media_type: Annotated[str, Alias("mediaType")]
- provider: str
+ # provider: str
# settings
- created_at: Annotated[int, Alias("createdAt")]
- last_update: Annotated[int, Alias("lastUpdate")]
+ # created_at: Annotated[int, Alias("createdAt")]
+ # last_update: Annotated[int, Alias("lastUpdate")]
@dataclass
class ABSAuthor(ABSAuthorMinified):
"""ABSAuthor."""
- asin: str | None
+ # asin: str | None
description: str | None
- image_path: Annotated[str | None, Alias("imagePath")]
- added_at: Annotated[int, Alias("addedAt")] # ms epoch
- updated_at: Annotated[int, Alias("updatedAt")] # ms epoch
+ # image_path: Annotated[str | None, Alias("imagePath")]
+ # added_at: Annotated[int, Alias("addedAt")] # ms epoch
+ # updated_at: Annotated[int, Alias("updatedAt")] # ms epoch
@dataclass
"""ABSSeries."""
description: str | None
- added_at: Annotated[int, Alias("addedAt")] # ms epoch
- updated_at: Annotated[int, Alias("updatedAt")] # ms epoch
+ # added_at: Annotated[int, Alias("addedAt")] # ms epoch
+ # updated_at: Annotated[int, Alias("updatedAt")] # ms epoch
@dataclass
progress: float # percent 0->1
current_time: Annotated[float, Alias("currentTime")] # seconds
is_finished: Annotated[bool, Alias("isFinished")]
- hide_from_continue_listening: Annotated[bool, Alias("hideFromContinueListening")]
- last_update: Annotated[int, Alias("lastUpdate")] # ms epoch
- started_at: Annotated[int, Alias("startedAt")] # ms epoch
- finished_at: Annotated[int | None, Alias("finishedAt")] # ms epoch
+ # hide_from_continue_listening: Annotated[bool, Alias("hideFromContinueListening")]
+ # last_update: Annotated[int, Alias("lastUpdate")] # ms epoch
+ # started_at: Annotated[int, Alias("startedAt")] # ms epoch
+ # finished_at: Annotated[int | None, Alias("finishedAt")] # ms epoch
# two additional progress variants, 'with media' book and podcast, further down.
type_: Annotated[str, Alias("type")]
token: str
media_progress: Annotated[list[ABSMediaProgress], Alias("mediaProgress")]
- series_hide_from_continue_listening: Annotated[
- list[str], Alias("seriesHideFromContinueListening")
- ]
- bookmarks: list[ABSAudioBookmark]
- is_active: Annotated[bool, Alias("isActive")]
- is_locked: Annotated[bool, Alias("isLocked")]
- last_seen: Annotated[int | None, Alias("lastSeen")]
- created_at: Annotated[int, Alias("createdAt")]
- permissions: ABSUserPermissions
+ # series_hide_from_continue_listening: Annotated[
+ # list[str], Alias("seriesHideFromContinueListening")
+ # ]
+ # bookmarks: list[ABSAudioBookmark]
+ # is_active: Annotated[bool, Alias("isActive")]
+ # is_locked: Annotated[bool, Alias("isLocked")]
+ # last_seen: Annotated[int | None, Alias("lastSeen")]
+ # created_at: Annotated[int, Alias("createdAt")]
+ # permissions: ABSUserPermissions
libraries_accessible: Annotated[list[str], Alias("librariesAccessible")]
# this seems to be missing
"""ABSPlaybackSession."""
id_: Annotated[str, Alias("id")]
- user_id: Annotated[str, Alias("userId")]
- library_id: Annotated[str, Alias("libraryId")]
+ # user_id: Annotated[str, Alias("userId")]
+ # library_id: Annotated[str, Alias("libraryId")]
library_item_id: Annotated[str, Alias("libraryItemId")]
episode_id: Annotated[str | None, Alias("episodeId")]
media_type: Annotated[str, Alias("mediaType")]
# media_metadata: Annotated[ABSPodcastMetaData | ABSAudioBookMetaData, Alias("mediaMetadata")]
# chapters: list[ABSAudioBookChapter]
display_title: Annotated[str, Alias("displayTitle")]
- display_author: Annotated[str, Alias("displayAuthor")]
- cover_path: Annotated[str, Alias("coverPath")]
- duration: float
+ # display_author: Annotated[str, Alias("displayAuthor")]
+ # cover_path: Annotated[str, Alias("coverPath")]
+ # duration: float
# 0: direct play, 1: direct stream, 2: transcode, 3: local
- play_method: Annotated[ABSPlayMethod, Alias("playMethod")]
- media_player: Annotated[str, Alias("mediaPlayer")]
- device_info: Annotated[ABSDeviceInfo, Alias("deviceInfo")]
- server_version: Annotated[str, Alias("serverVersion")]
+ # play_method: Annotated[ABSPlayMethod, Alias("playMethod")]
+ # media_player: Annotated[str, Alias("mediaPlayer")]
+ # device_info: Annotated[ABSDeviceInfo, Alias("deviceInfo")]
+ # server_version: Annotated[str, Alias("serverVersion")]
# YYYY-MM-DD
- date: str
- day_of_week: Annotated[str, Alias("dayOfWeek")]
- time_listening: Annotated[float, Alias("timeListening")] # s
- start_time: Annotated[float, Alias("startTime")] # s
- current_time: Annotated[float, Alias("currentTime")] # s
- started_at: Annotated[int, Alias("startedAt")] # ms since Unix Epoch
- updated_at: Annotated[int, Alias("updatedAt")] # ms since Unix Epoch
+ # date: str
+ # day_of_week: Annotated[str, Alias("dayOfWeek")]
+ # time_listening: Annotated[float, Alias("timeListening")] # s
+ # start_time: Annotated[float, Alias("startTime")] # s
+ # current_time: Annotated[float, Alias("currentTime")] # s
+ # started_at: Annotated[int, Alias("startedAt")] # ms since Unix Epoch
+ # updated_at: Annotated[int, Alias("updatedAt")] # ms since Unix Epoch
@dataclass
description: str | None
release_date: Annotated[str | None, Alias("releaseDate")]
genres: list[str] | None
- feed_url: Annotated[str | None, Alias("feedUrl")]
- image_url: Annotated[str | None, Alias("imageUrl")]
- itunes_page_url: Annotated[str | None, Alias("itunesPageUrl")]
- itunes_id: Annotated[int | None, Alias("itunesId")]
- itunes_artist_id: Annotated[int | None, Alias("itunesArtistId")]
+ # feed_url: Annotated[str | None, Alias("feedUrl")]
+ # image_url: Annotated[str | None, Alias("imageUrl")]
+ # itunes_page_url: Annotated[str | None, Alias("itunesPageUrl")]
+ # itunes_id: Annotated[int | None, Alias("itunesId")]
+ # itunes_artist_id: Annotated[int | None, Alias("itunesArtistId")]
explicit: bool
language: str | None
- type_: Annotated[str | None, Alias("type")]
+ # type_: Annotated[str | None, Alias("type")]
@dataclass
class ABSPodcastMetadataMinified(ABSPodcastMetadata):
"""ABSPodcastMetadataMinified."""
- title_ignore_prefix: Annotated[str, Alias("titleIgnorePrefix")]
+ # title_ignore_prefix: Annotated[str, Alias("titleIgnorePrefix")]
ABSPodcastMetaDataExpanded = ABSPodcastMetadataMinified
published_at: Annotated[int | None, Alias("publishedAt")] # ms posix epoch
added_at: Annotated[int | None, Alias("addedAt")] # ms posix epoch
updated_at: Annotated[int | None, Alias("updatedAt")] # ms posix epoch
- season: str = ""
+ # season: str = ""
episode: str = ""
- episode_type: Annotated[str, Alias("episodeType")] = ""
+ # episode_type: Annotated[str, Alias("episodeType")] = ""
title: str = ""
subtitle: str = ""
description: str = ""
- enclosure: str = ""
+ # enclosure: str = ""
pub_date: Annotated[str, Alias("pubDate")] = ""
- guid: str = ""
+ # guid: str = ""
# chapters
# audio_file: # not needed for mass application
published_at: Annotated[int | None, Alias("publishedAt")] # ms posix epoch
added_at: Annotated[int | None, Alias("addedAt")] # ms posix epoch
- updated_at: Annotated[int | None, Alias("updatedAt")] # ms posix epoch
+ # updated_at: Annotated[int | None, Alias("updatedAt")] # ms posix epoch
audio_track: Annotated[ABSAudioTrack, Alias("audioTrack")]
- size: int # in bytes
- season: str = ""
+ # size: int # in bytes
+ # season: str = ""
episode: str = ""
- episode_type: Annotated[str, Alias("episodeType")] = ""
+ # episode_type: Annotated[str, Alias("episodeType")] = ""
title: str = ""
subtitle: str = ""
description: str = ""
- enclosure: str = ""
- pub_date: Annotated[str, Alias("pubDate")] = ""
- guid: str = ""
+ # enclosure: str = ""
+ # pub_date: Annotated[str, Alias("pubDate")] = ""
+ # guid: str = ""
# chapters
duration: float = 0.0
"""ABSPodcastMinified."""
metadata: ABSPodcastMetadataMinified
- size: int # bytes
+ # size: int # bytes
num_episodes: Annotated[int, Alias("numEpisodes")] = 0
published_date: Annotated[str | None, Alias("publishedDate")]
publisher: str | None
description: str | None
- isbn: str | None
- asin: str | None
+ # isbn: str | None
+ # asin: str | None
language: str | None
explicit: bool
"""ABSBookMetadataMinified."""
# these are normally there
- title_ignore_prefix: Annotated[str, Alias("titleIgnorePrefix")]
+ # title_ignore_prefix: Annotated[str, Alias("titleIgnorePrefix")]
author_name: Annotated[str, Alias("authorName")]
- author_name_lf: Annotated[str, Alias("authorNameLF")]
+ # author_name_lf: Annotated[str, Alias("authorNameLF")]
narrator_name: Annotated[str, Alias("narratorName")]
series_name: Annotated[str, Alias("seriesName")]
"""ABSBookBase."""
metadata: ABSBookMetadataMinified
- num_tracks: Annotated[int, Alias("numTracks")]
- num_audiofiles: Annotated[int, Alias("numAudioFiles")]
+ # num_tracks: Annotated[int, Alias("numTracks")]
+ # num_audiofiles: Annotated[int, Alias("numAudioFiles")]
num_chapters: Annotated[int, Alias("numChapters")]
duration: float # in s
- size: int # in bytes
+ # size: int # in bytes
# ebookFormat
"""_ABSLibraryItemBase."""
id_: Annotated[str, Alias("id")]
- ino: str
- library_id: Annotated[str, Alias("libraryId")]
- folder_id: Annotated[str, Alias("folderId")]
- path: str
- relative_path: Annotated[str, Alias("relPath")]
- is_file: Annotated[bool, Alias("isFile")]
- last_modified_ms: Annotated[int, Alias("mtimeMs")] # epoch
- last_changed_ms: Annotated[int, Alias("ctimeMs")] # epoch
- birthtime_ms: Annotated[int, Alias("birthtimeMs")] # epoch
- added_at: Annotated[int, Alias("addedAt")] # ms epoch
- updated_at: Annotated[int, Alias("updatedAt")] # ms epoch
- is_missing: Annotated[bool, Alias("isMissing")]
- is_invalid: Annotated[bool, Alias("isInvalid")]
+ # ino: str
+ # library_id: Annotated[str, Alias("libraryId")]
+ # folder_id: Annotated[str, Alias("folderId")]
+ # path: str
+ # relative_path: Annotated[str, Alias("relPath")]
+ # is_file: Annotated[bool, Alias("isFile")]
+ # last_modified_ms: Annotated[int, Alias("mtimeMs")] # epoch
+ # last_changed_ms: Annotated[int, Alias("ctimeMs")] # epoch
+ # birthtime_ms: Annotated[int, Alias("birthtimeMs")] # epoch
+ # added_at: Annotated[int, Alias("addedAt")] # ms epoch
+ # updated_at: Annotated[int, Alias("updatedAt")] # ms epoch
+ # is_missing: Annotated[bool, Alias("isMissing")]
+ # is_invalid: Annotated[bool, Alias("isInvalid")]
media_type: Annotated[str, Alias("mediaType")]
class _ABSLibraryItem(_ABSLibraryItemBase):
"""ABSLibraryItem."""
- last_scan: Annotated[int | None, Alias("lastScan")] # ms epoch
- scan_version: Annotated[str | None, Alias("scanVersion")]
+ # last_scan: Annotated[int | None, Alias("lastScan")] # ms epoch
+ # scan_version: Annotated[str | None, Alias("scanVersion")]
# libraryFiles
"""ABSSeriesBooks."""
added_at: Annotated[int, Alias("addedAt")] # ms epoch
- name_ignore_prefix: Annotated[str, Alias("nameIgnorePrefix")]
- name_ignore_prefix_sort: Annotated[str, Alias("nameIgnorePrefixSort")]
- type_: Annotated[str, Alias("type")]
+ # name_ignore_prefix: Annotated[str, Alias("nameIgnorePrefix")]
+ # name_ignore_prefix_sort: Annotated[str, Alias("nameIgnorePrefixSort")]
+ # type_: Annotated[str, Alias("type")]
books: list[ABSLibraryItemBookSeries]
total_duration: Annotated[float, Alias("totalDuration")] # s
projects / music-assistant-server.git / commitdiff