Chore: A bit of cleanup and restructuring

Author: marcelveldt

music_assistant_models/player_queue.py

@@ -6,10 +6,10 @@
 from dataclasses import dataclass, field
 from typing import Any, Self
 
-from mashumaro import DataClassDictMixin
+from mashumaro import DataClassDictMixin, field_options, pass_through
 
 from .enums import PlayerState, RepeatMode
-from .media_items import MediaItemType
+from .media_items import MediaItemType, media_from_dict
 from .queue_item import QueueItem
 
 
@@ -31,55 +31,68 @@ class PlayerQueue(DataClassDictMixin):
     display_name: str
     available: bool
     items: int
-
     shuffle_enabled: bool = False
     repeat_mode: RepeatMode = RepeatMode.OFF
     dont_stop_the_music_enabled: bool = False
+
     # current_index: index that is active (e.g. being played) by the player
     current_index: int | None = None
     # index_in_buffer: index that has been preloaded/buffered by the player
     index_in_buffer: int | None = None
+
     elapsed_time: float = 0
     elapsed_time_last_updated: float = field(default_factory=time.time)
     state: PlayerState = PlayerState.IDLE
     current_item: QueueItem | None = None
     next_item: QueueItem | None = None
     radio_source: list[MediaItemType] = field(default_factory=list)
-    enqueued_media_items: list[MediaItemType] = field(default_factory=list)
+
     flow_mode: bool = False
     resume_pos: int = 0
-    flow_mode_stream_log: list[PlayLogEntry] = field(default_factory=list)
-    next_track_enqueued: str | None = None
+
+    #############################################################################
+    # the fields below will only be used server-side and not sent to the client #
+    #############################################################################
+
+    enqueued_media_items: list[MediaItemType] = field(
+        default_factory=list,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
+    flow_mode_stream_log: list[PlayLogEntry] = field(
+        default_factory=list,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
+    next_track_enqueued: str | None = field(
+        default=None,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
 
     @property
     def corrected_elapsed_time(self) -> float:
         """Return the corrected/realtime elapsed time."""
         return self.elapsed_time + (time.time() - self.elapsed_time_last_updated)
 
-    def __post_serialize__(self, d: dict[Any, Any]) -> dict[Any, Any]:
-        """Execute action(s) on serialization."""
-        d.pop("flow_mode_stream_log", None)
-        d.pop("next_track_enqueued", None)
-        return d
-
     def to_cache(self) -> dict[str, Any]:
         """Return the dict that is suitable for storing into the cache db."""
         d = self.to_dict()
+        d.pop("flow_mode", None)
         d.pop("current_item", None)
         d.pop("next_item", None)
         d.pop("index_in_buffer", None)
-        d.pop("flow_mode", None)
-        d.pop("next_track_enqueued", None)
-        d.pop("flow_mode_stream_log", None)
+        # enqueued_media_items needs to survive a restart
+        # otherwise 'dont stop the music' will not work
+        d["enqueued_media_items"] = [x.to_dict() for x in self.enqueued_media_items]
         return d
 
     @classmethod
     def from_cache(cls, d: dict[Any, Any]) -> Self:
         """Restore a PlayerQueue from a cache dict."""
-        d.pop("current_item", None)
-        d.pop("next_item", None)
-        d.pop("index_in_buffer", None)
-        d.pop("flow_mode", None)
-        d.pop("next_track_enqueued", None)
-        d.pop("flow_mode_stream_log", None)
+        if "enqueued_media_items" in d:
+            d["enqueued_media_items"] = [media_from_dict(x) for x in d["enqueued_media_items"]]
         return cls.from_dict(d)

music_assistant_models/queue_item.py

@@ -34,17 +34,6 @@ def __post_init__(self) -> None:
         if not self.name:
             self.name = self.uri
 
-    def __post_serialize__(self, d: dict[Any, Any]) -> dict[Any, Any]:
-        """Execute action(s) on serialization."""
-        # Exclude internal streamdetails fields from dict
-        if streamdetails := d.get("streamdetails"):
-            streamdetails.pop("data", None)
-            streamdetails.pop("direct", None)
-            streamdetails.pop("expires", None)
-            streamdetails.pop("path", None)
-            streamdetails.pop("decryption_key", None)
-        return d
-
     @property
     def uri(self) -> str:
         """Return uri for this QueueItem (for logging purposes)."""

music_assistant_models/streamdetails.py

@@ -5,7 +5,7 @@
 from dataclasses import dataclass, field
 from typing import Any
 
-from mashumaro import DataClassDictMixin
+from mashumaro import DataClassDictMixin, field_options, pass_through
 
 from .dsp import DSPDetails
 from .enums import MediaType, StreamType, VolumeNormalizationMode
@@ -31,72 +31,144 @@ class StreamDetails(DataClassDictMixin):
     # the streamdetails are only used to provide details about the content
     # that is going to be streamed.
 
-    # mandatory fields
+    #############################################################################
+    # mandatory fields                                                          #
+    #############################################################################
     provider: str
     item_id: str
     audio_format: AudioFormat
     media_type: MediaType = MediaType.TRACK
     stream_type: StreamType = StreamType.CUSTOM
 
-    # optional fields
-
-    # path: url or (local accessible) path to the stream (if not custom stream)
-    path: str | None = None
+    #############################################################################
+    # optional fields                                                           #
+    #############################################################################
 
     # duration of the item to stream, copied from media_item if omitted
     duration: int | None = None
+
     # total size in bytes of the item, calculated at eof when omitted
     size: int | None = None
+
+    # stream_metadata: radio/live streams can optionally set/use this field
+    # to set the metadata of any media during the stream
+    stream_metadata: LivestreamMetadata | None = None
+
+    #############################################################################
+    # the fields below will only be used server-side and not sent to the client #
+    #############################################################################
+
+    # path: url or (local accessible) path to the stream (if not custom stream)
+    # this field should be set by the provider when creating the streamdetails
+    # unless the stream is a custom stream
+    path: str | None = field(
+        default=None,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
     # data: provider specific data (not exposed externally)
     # this info is for example used to pass slong details to the get_audio_stream
-    data: Any = None
+    # this field may be set by the provider when creating the streamdetails
+    data: Any = field(
+        default=None,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
+    # extra_input_args: any additional input args to pass along to ffmpeg
+    # this field may be set by the provider when creating the streamdetails
+    extra_input_args: list[str] = field(
+        default_factory=list,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
+    # decryption_key: decryption key for encrypted streams (used with StreamType.ENCRYPTED_HTTP)
+    # this field may be set by the provider when creating the streamdetails
+    decryption_key: str | None = field(
+        default=None,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
     # allow_seek: bool to indicate that the content can/may be seeked
     # If set to False, seeking will be completely disabled.
     # NOTE: this is automatically disabled for duration-less streams (e/g. radio)
-    allow_seek: bool = True
+    allow_seek: bool = field(
+        default=False,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
+
     # can_seek: bool to indicate that the custom audio stream can be seeked
     # if set to False, and allow seek is set to True, the core logic will attempt
     # to seek in the incoming (bytes)stream, which is not a guarantee it will work.
     # If allow_seek is also set to False, seeking will be completely disabled.
-    can_seek: bool = True
-    # extra_input_args: any additional input args to pass along to ffmpeg
-    extra_input_args: list[str] = field(default_factory=list)
-    # decryption_key: decryption key for encrypted streams
-    decryption_key: str | None = None
-    # stream_metadata: radio/live streams can optionally set/use this field
-    # to set the metadata of any media during the stream
-    stream_metadata: LivestreamMetadata | None = None
-
-    # the fields below will be set/controlled by the streamcontroller
-    seek_position: int = 0
-    fade_in: bool = False
+    can_seek: bool = field(
+        default=False,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
+
+    #############################################################################
+    # the fields below will be set/controlled by the streamcontroller           #
+    #############################################################################
     loudness: float | None = None
     loudness_album: float | None = None
     prefer_album_loudness: bool = False
     volume_normalization_mode: VolumeNormalizationMode | None = None
     volume_normalization_gain_correct: float | None = None
-    queue_id: str | None = None
-    seconds_streamed: float | None = None
     target_loudness: float | None = None
     strip_silence_begin: bool = False
     strip_silence_end: bool = False
-    stream_error: bool | None = None
+
     # This contains the DSPDetails of all players in the group.
     # In case of single player playback, dict will contain only one entry.
     # The leader will have is_leader set to True.
     # (keep in mind that PlayerGroups have no (explicit) leader!)
     dsp: dict[str, DSPDetails] | None = None
 
+    # the fields below are managed by the queue/stream controller and may not be set by providers
+    fade_in: bool = field(
+        default=False,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
+    seek_position: int = field(
+        default=0,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
+    queue_id: str | None = field(
+        default=None,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
+    seconds_streamed: float | None = field(
+        default=None,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
+    stream_error: bool | None = field(
+        default=None,
+        compare=False,
+        metadata=field_options(serialize="omit", deserialize=pass_through),
+        repr=False,
+    )
+
     def __str__(self) -> str:
         """Return pretty printable string of object."""
         return self.uri
 
     def __post_serialize__(self, d: dict[Any, Any]) -> dict[Any, Any]:
         """Execute action(s) on serialization."""
-        d.pop("queue_id", None)
-        d.pop("seconds_streamed", None)
-        d.pop("seek_position", None)
-        d.pop("fade_in", None)
         # for backwards compatibility
         d["stream_title"] = self.stream_title
         return d