Rockbox open source high quality audio player as a Music Player Daemon
mpris
rockbox
mpd
libadwaita
audio
rust
zig
deno
1"""Playback control: status, transport, and one-shot play helpers."""
2
3from __future__ import annotations
4
5from typing import Any
6
7from ..transport import HttpTransport
8from ..types import PlaybackStatus, Track
9from ._fragments import TRACK_FIELDS
10
11
12class PlaybackApi:
13 def __init__(self, http: HttpTransport) -> None:
14 self._http = http
15
16 # --- status ---------------------------------------------------------
17
18 async def raw_status(self) -> int:
19 """Raw numeric playback status from the firmware."""
20 data = await self._http.execute("query PlaybackStatus { status }")
21 return int(data["status"])
22
23 async def status(self) -> PlaybackStatus:
24 """Typed playback status."""
25 return PlaybackStatus(await self.raw_status())
26
27 async def current_track(self) -> Track | None:
28 data = await self._http.execute(
29 f"{TRACK_FIELDS} query CurrentTrack {{ currentTrack {{ ...TrackFields }} }}"
30 )
31 raw = data.get("currentTrack")
32 return Track.model_validate(raw) if raw is not None else None
33
34 async def next_track(self) -> Track | None:
35 data = await self._http.execute(
36 f"{TRACK_FIELDS} query NextTrack {{ nextTrack {{ ...TrackFields }} }}"
37 )
38 raw = data.get("nextTrack")
39 return Track.model_validate(raw) if raw is not None else None
40
41 async def file_position(self) -> int:
42 data = await self._http.execute("query FilePosition { getFilePosition }")
43 return int(data["getFilePosition"])
44
45 # --- transport ------------------------------------------------------
46
47 async def play(self, elapsed: int = 0, offset: int = 0) -> None:
48 await self._http.execute(
49 "mutation Play($elapsed: Long!, $offset: Long!) "
50 "{ play(elapsed: $elapsed, offset: $offset) }",
51 {"elapsed": elapsed, "offset": offset},
52 )
53
54 async def pause(self) -> None:
55 await self._http.execute("mutation Pause { pause }")
56
57 async def resume(self) -> None:
58 await self._http.execute("mutation Resume { resume }")
59
60 async def next(self) -> None:
61 await self._http.execute("mutation Next { next }")
62
63 async def previous(self) -> None:
64 await self._http.execute("mutation Previous { previous }")
65
66 async def seek(self, position_ms: int) -> None:
67 """Seek to an absolute position in milliseconds."""
68 await self._http.execute(
69 "mutation Seek($newTime: Int!) { fastForwardRewind(newTime: $newTime) }",
70 {"newTime": position_ms},
71 )
72
73 async def stop(self) -> None:
74 await self._http.execute("mutation Stop { hardStop }")
75
76 async def flush_and_reload(self) -> None:
77 """Reload and flush the current track queue."""
78 await self._http.execute("mutation FlushReload { flushAndReloadTracks }")
79
80 # --- one-shot play helpers -----------------------------------------
81
82 async def play_track(self, path: str) -> None:
83 await self._http.execute(
84 "mutation PlayTrack($path: String!) { playTrack(path: $path) }",
85 {"path": path},
86 )
87
88 async def play_album(
89 self,
90 album_id: str,
91 *,
92 shuffle: bool | None = None,
93 position: int | None = None,
94 ) -> None:
95 await self._http.execute(
96 "mutation PlayAlbum($albumId: String!, $shuffle: Boolean, $position: Int) "
97 "{ playAlbum(albumId: $albumId, shuffle: $shuffle, position: $position) }",
98 {"albumId": album_id, "shuffle": shuffle, "position": position},
99 )
100
101 async def play_artist(
102 self,
103 artist_id: str,
104 *,
105 shuffle: bool | None = None,
106 position: int | None = None,
107 ) -> None:
108 await self._http.execute(
109 "mutation PlayArtist($artistId: String!, $shuffle: Boolean, $position: Int) "
110 "{ playArtistTracks(artistId: $artistId, shuffle: $shuffle, position: $position) }",
111 {"artistId": artist_id, "shuffle": shuffle, "position": position},
112 )
113
114 async def play_playlist(
115 self,
116 playlist_id: str,
117 *,
118 shuffle: bool | None = None,
119 position: int | None = None,
120 ) -> None:
121 await self._http.execute(
122 "mutation PlayPlaylist($playlistId: String!, $shuffle: Boolean, $position: Int) "
123 "{ playPlaylist(playlistId: $playlistId, shuffle: $shuffle, position: $position) }",
124 {"playlistId": playlist_id, "shuffle": shuffle, "position": position},
125 )
126
127 async def play_directory(
128 self,
129 path: str,
130 *,
131 recurse: bool | None = None,
132 shuffle: bool | None = None,
133 position: int | None = None,
134 ) -> None:
135 await self._http.execute(
136 "mutation PlayDirectory("
137 "$path: String!, $recurse: Boolean, $shuffle: Boolean, $position: Int"
138 ") { playDirectory(path: $path, recurse: $recurse, "
139 "shuffle: $shuffle, position: $position) }",
140 {"path": path, "recurse": recurse, "shuffle": shuffle, "position": position},
141 )
142
143 async def play_liked_tracks(
144 self,
145 *,
146 shuffle: bool | None = None,
147 position: int | None = None,
148 ) -> None:
149 await self._http.execute(
150 "mutation PlayLikedTracks($shuffle: Boolean, $position: Int) "
151 "{ playLikedTracks(shuffle: $shuffle, position: $position) }",
152 {"shuffle": shuffle, "position": position},
153 )
154
155 async def play_all_tracks(
156 self,
157 *,
158 shuffle: bool | None = None,
159 position: int | None = None,
160 ) -> None:
161 await self._http.execute(
162 "mutation PlayAllTracks($shuffle: Boolean, $position: Int) "
163 "{ playAllTracks(shuffle: $shuffle, position: $position) }",
164 {"shuffle": shuffle, "position": position},
165 )
166
167 # ``Any`` to keep mypy quiet about the generic dict shape returned by GraphQL
168 _: Any = None