music-assistant-server
6.2 KB•PY
plugin.py
6.2 KB • 186 lines • python
1"""Model/base for a Plugin Provider implementation."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator, Awaitable, Callable
6from dataclasses import dataclass, field
7from typing import TYPE_CHECKING
8
9from mashumaro import field_options, pass_through
10from music_assistant_models.enums import ContentType, StreamType
11from music_assistant_models.media_items.audio_format import AudioFormat
12
13from music_assistant.models.player import PlayerSource
14
15from .provider import Provider
16
17if TYPE_CHECKING:
18 from music_assistant_models.streamdetails import StreamMetadata
19
20
21@dataclass
22class PluginSource(PlayerSource):
23 """
24 Model for a PluginSource, which is a player (audio)source provided by a plugin.
25
26 A PluginSource is for example a live audio stream such as a aux/microphone input.
27
28 This (intermediate) model is not exposed on the api,
29 but is used internally by the plugin provider.
30 """
31
32 # The PCM audio format provided by this source
33 # for realtime audio, we recommend using PCM 16bit 44.1kHz stereo
34 audio_format: AudioFormat = field(
35 default=AudioFormat(
36 content_type=ContentType.PCM_S16LE,
37 sample_rate=44100,
38 bit_depth=16,
39 channels=2,
40 ),
41 compare=False,
42 metadata=field_options(serialize="omit", deserialize=pass_through),
43 repr=False,
44 )
45
46 # metadata of the current playing media (if known)
47 metadata: StreamMetadata | None = field(
48 default=None,
49 compare=False,
50 metadata=field_options(serialize="omit", deserialize=pass_through),
51 repr=False,
52 )
53
54 # The type of stream that is provided by this source
55 stream_type: StreamType | None = field(
56 default=StreamType.CUSTOM,
57 compare=False,
58 metadata=field_options(serialize="omit", deserialize=pass_through),
59 repr=False,
60 )
61
62 # The path to the source/audio (if streamtype is not custom)
63 path: str | None = field(
64 default=None,
65 compare=False,
66 metadata=field_options(serialize="omit", deserialize=pass_through),
67 repr=False,
68 )
69 # in_use_by specifies the player id that is currently using this plugin (if any)
70 in_use_by: str | None = field(
71 default=None,
72 compare=False,
73 metadata=field_options(serialize="omit", deserialize=pass_through),
74 repr=False,
75 )
76
77 # Optional callbacks for playback control
78 # These callbacks will be called by the player controller when control commands are issued
79 # and the source reports the corresponding capability (can_play_pause, can_seek, etc.)
80
81 # Callback for play command: async def callback() -> None
82 on_play: Callable[[], Awaitable[None]] | None = field(
83 default=None,
84 compare=False,
85 metadata=field_options(serialize="omit", deserialize=pass_through),
86 repr=False,
87 )
88
89 # Callback for pause command: async def callback() -> None
90 on_pause: Callable[[], Awaitable[None]] | None = field(
91 default=None,
92 compare=False,
93 metadata=field_options(serialize="omit", deserialize=pass_through),
94 repr=False,
95 )
96
97 # Callback for next track command: async def callback() -> None
98 on_next: Callable[[], Awaitable[None]] | None = field(
99 default=None,
100 compare=False,
101 metadata=field_options(serialize="omit", deserialize=pass_through),
102 repr=False,
103 )
104
105 # Callback for previous track command: async def callback() -> None
106 on_previous: Callable[[], Awaitable[None]] | None = field(
107 default=None,
108 compare=False,
109 metadata=field_options(serialize="omit", deserialize=pass_through),
110 repr=False,
111 )
112
113 # Callback for seek command: async def callback(position: int) -> None
114 on_seek: Callable[[int], Awaitable[None]] | None = field(
115 default=None,
116 compare=False,
117 metadata=field_options(serialize="omit", deserialize=pass_through),
118 repr=False,
119 )
120
121 # Callback for volume change command: async def callback(volume: int) -> None
122 on_volume: Callable[[int], Awaitable[None]] | None = field(
123 default=None,
124 compare=False,
125 metadata=field_options(serialize="omit", deserialize=pass_through),
126 repr=False,
127 )
128
129 # Callback for when this source is selected: async def callback() -> None
130 on_select: Callable[[], Awaitable[None]] | None = field(
131 default=None,
132 compare=False,
133 metadata=field_options(serialize="omit", deserialize=pass_through),
134 repr=False,
135 )
136
137 def as_player_source(self) -> PlayerSource:
138 """Return a basic PlayerSource representation without unpicklable callbacks."""
139 return PlayerSource(
140 id=self.id,
141 name=self.name,
142 passive=self.passive,
143 can_play_pause=self.can_play_pause,
144 can_seek=self.can_seek,
145 can_next_previous=self.can_next_previous,
146 )
147
148
149class PluginProvider(Provider):
150 """
151 Base representation of a Plugin for Music Assistant.
152
153 Plugin Provider implementations should inherit from this base model.
154 """
155
156 def get_source(self) -> PluginSource:
157 """
158 Get (audio)source details for this plugin.
159
160 # Will only be called if ProviderFeature.AUDIO_SOURCE is declared
161 """
162 raise NotImplementedError
163
164 async def get_audio_stream(self, player_id: str) -> AsyncGenerator[bytes, None]:
165 """
166 Return the (custom) audio stream for the audio source provided by this plugin.
167
168 Will only be called if this plugin is a PluginSource, meaning that
169 the ProviderFeature.AUDIO_SOURCE is declared AND if the streamtype is StreamType.CUSTOM.
170
171 The player_id is the id of the player that is requesting the stream.
172
173 Must return audio data as bytes generator (in the format specified by the audio_format).
174 """
175 yield b""
176 raise NotImplementedError
177
178 async def resolve_image(self, path: str) -> str | bytes:
179 """
180 Resolve an image from an image path.
181
182 This either returns (a generator to get) raw bytes of the image or
183 a string with an http(s) URL or local path that is accessible from the server.
184 """
185 return path
186