music-assistant-server

1"""
2Shared base for the Player Queues controller and its logic mixins.
3
4The controller's loading, playback-tracking and stream-feeding logic is split across mixins
5(see queue_loader/playback_tracker/stream_feeder). They operate on the controller's own per-queue
6records and call back into its core operations. This base declares that shared surface - the
7per-queue state, the stateful helper services and the core-operation signatures - so each mixin
8type-checks on its own while the real implementations live on PlayerQueuesController.
9"""
10
11from __future__ import annotations
12
13from typing import TYPE_CHECKING, Any
14
15from music_assistant.models.core_controller import CoreController
16
17if TYPE_CHECKING:
18    from music_assistant_models.enums import QueueOption
19    from music_assistant_models.media_items import MediaItemType
20    from music_assistant_models.player_queue import PlayerQueue
21    from music_assistant_models.queue_item import QueueItem
22
23    from music_assistant.controllers.player_queues.autoplay import Autoplay
24    from music_assistant.controllers.player_queues.managed_pool import ManagedPool
25    from music_assistant.controllers.player_queues.media_resolver import MediaResolver
26    from music_assistant.controllers.player_queues.smart_shuffle import SmartShuffle
27    from music_assistant.controllers.player_queues.state import PlayerQueueData
28    from music_assistant.models.player import Player, PlayerMedia
29
30
31class _PlayerQueuesBase(CoreController):
32    """
33    Shared state and core-operation contract for the player-queues logic mixins.
34
35    Carries the per-queue records and stateful helper services the mixins operate on, plus the
36    signatures of the controller operations they call into.
37    """
38
39    domain: str = "player_queues"
40
41    # per-queue records + stateful helper services, populated in PlayerQueuesController.__init__
42    _queue_data: dict[str, PlayerQueueData]
43    _autoplay: Autoplay
44    _smart_shuffle: SmartShuffle
45    _managed_pool: ManagedPool
46    _media_resolver: MediaResolver
47
48    if TYPE_CHECKING:
49        # core operations implemented on PlayerQueuesController
50        def get(self, queue_id: str) -> PlayerQueue | None: ...
51        def get_item(
52            self, queue_id: str, item_id_or_index: int | str | None
53        ) -> QueueItem | None: ...
54        def index_by_id(self, queue_id: str, queue_item_id: str) -> int | None: ...
55        def get_next_item(self, queue_id: str, cur_index: int | str) -> QueueItem | None: ...
56        def clear(self, queue_id: str, skip_stop: bool = False) -> None: ...
57        def mark_ended(self, queue_id: str) -> None: ...
58        def signal_update(self, queue_id: str, items_changed: bool = False) -> None: ...
59        def on_player_update(
60            self, player: Player, changed_values: dict[str, tuple[Any, Any]]
61        ) -> None: ...
62        def is_smart_shuffle_active(self, queue: PlayerQueue) -> bool: ...
63        def store_sources(self, queue: PlayerQueue, items: list[MediaItemType]) -> None: ...
64        async def load(
65            self,
66            queue_id: str,
67            queue_items: list[QueueItem],
68            insert_at_index: int = 0,
69            keep_remaining: bool = True,
70            keep_played: bool = True,
71            shuffle: bool = False,
72            pin_first: bool = False,
73        ) -> None: ...
74        async def play_index(
75            self,
76            queue_id: str,
77            index: int | str,
78            seek_position: int = 0,
79            fade_in: bool = False,
80        ) -> None: ...
81        async def load_next_queue_item(self, queue_id: str, current_item_id: str) -> QueueItem: ...
82        async def player_media_from_queue_item(self, queue_item: QueueItem) -> PlayerMedia: ...
83        def update_next_item_on_player(self, queue_id: str, force: bool = False) -> None: ...
84        def _set_transitioning(self, queue_id: str, value: bool) -> None: ...
85        def _clear(self, queue_id: str, skip_stop: bool = False) -> None: ...
86        async def _cleanup_queue_audio_data(
87            self, queue_id: str, session_id: str | None = None
88        ) -> None: ...
89        async def _apply_shuffle(
90            self, queue_id: str, option: QueueOption, shuffle: bool | None
91        ) -> None: ...
92        # implemented on QueueLoaderMixin; declared here for the cross-mixin call in PlaybackTracker
93        def _apply_probed_duration(self, queue_item: QueueItem) -> None: ...
94        async def _fill_dynamic_tracks(self, queue_id: str) -> None: ...
95        async def _fill_autoplay_tracks(self, queue_id: str) -> None: ...
96