/
/
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.media_items import MediaItemType
19 from music_assistant_models.player_queue import PlayerQueue
20 from music_assistant_models.queue_item import QueueItem
21
22 from music_assistant.controllers.player_queues.autoplay import Autoplay
23 from music_assistant.controllers.player_queues.managed_pool import ManagedPool
24 from music_assistant.controllers.player_queues.media_resolver import MediaResolver
25 from music_assistant.controllers.player_queues.smart_shuffle import SmartShuffle
26 from music_assistant.controllers.player_queues.state import PlayerQueueData
27 from music_assistant.models.player import Player, PlayerMedia
28
29
30class _PlayerQueuesBase(CoreController):
31 """
32 Shared state and core-operation contract for the player-queues logic mixins.
33
34 Carries the per-queue records and stateful helper services the mixins operate on, plus the
35 signatures of the controller operations they call into.
36 """
37
38 domain: str = "player_queues"
39
40 # per-queue records + stateful helper services, populated in PlayerQueuesController.__init__
41 _queue_data: dict[str, PlayerQueueData]
42 _autoplay: Autoplay
43 _smart_shuffle: SmartShuffle
44 _managed_pool: ManagedPool
45 _media_resolver: MediaResolver
46
47 if TYPE_CHECKING:
48 # core operations implemented on PlayerQueuesController
49 def get(self, queue_id: str) -> PlayerQueue | None: ...
50 def get_item(
51 self, queue_id: str, item_id_or_index: int | str | None
52 ) -> QueueItem | None: ...
53 def index_by_id(self, queue_id: str, queue_item_id: str) -> int | None: ...
54 def get_next_item(self, queue_id: str, cur_index: int | str) -> QueueItem | None: ...
55 def clear(self, queue_id: str, skip_stop: bool = False) -> None: ...
56 def mark_ended(self, queue_id: str) -> None: ...
57 def signal_update(self, queue_id: str, items_changed: bool = False) -> None: ...
58 def on_player_update(
59 self, player: Player, changed_values: dict[str, tuple[Any, Any]]
60 ) -> None: ...
61 def is_smart_shuffle_active(self, queue: PlayerQueue) -> bool: ...
62 def store_sources(self, queue: PlayerQueue, items: list[MediaItemType]) -> None: ...
63 async def load(
64 self,
65 queue_id: str,
66 queue_items: list[QueueItem],
67 insert_at_index: int = 0,
68 keep_remaining: bool = True,
69 keep_played: bool = True,
70 shuffle: bool = False,
71 ) -> None: ...
72 async def play_index(
73 self,
74 queue_id: str,
75 index: int | str,
76 seek_position: int = 0,
77 fade_in: bool = False,
78 ) -> None: ...
79 async def load_next_queue_item(self, queue_id: str, current_item_id: str) -> QueueItem: ...
80 async def player_media_from_queue_item(self, queue_item: QueueItem) -> PlayerMedia: ...
81 def _set_transitioning(self, queue_id: str, value: bool) -> None: ...
82 # implemented on QueueLoaderMixin; declared here for the cross-mixin call in PlaybackTracker
83 def _apply_probed_duration(self, queue_item: QueueItem) -> None: ...
84 async def _fill_dynamic_tracks(self, queue_id: str) -> None: ...
85 async def _fill_autoplay_tracks(self, queue_id: str) -> None: ...
86