/
/
/
1"""
2Sendspin Source provider implementation.
3
4Exposes every connected Sendspin client with an active `source` role as a Music
5Assistant AudioSource. Decoded PCM from the client is pulled through an
6occupancy-controlled clock bridge so the capture clock never has to match MA's
7consumption clock. See README.md for the design rationale.
8"""
9
10from __future__ import annotations
11
12import asyncio
13import time
14from dataclasses import dataclass, field
15from typing import TYPE_CHECKING, cast
16
17from aiosendspin.audio import AsrcSourceBridge
18from aiosendspin.audio import AudioFormat as SendspinAudioFormat
19from aiosendspin.models.types import role_family
20from aiosendspin.server import (
21 ClientConnectedEvent,
22 ClientDisconnectedEvent,
23 ClientRemovedEvent,
24 SignalState,
25 SourceSignalChangedEvent,
26 SourceStreamEndedEvent,
27 SourceStreamStartedEvent,
28)
29from music_assistant_models.enums import (
30 ContentType,
31 MediaType,
32 QueueOption,
33 StreamType,
34)
35from music_assistant_models.errors import (
36 AudioError,
37 MediaNotFoundError,
38 PlayerCommandFailed,
39 PlayerUnavailableError,
40)
41from music_assistant_models.helpers import create_uri
42from music_assistant_models.media_items import AudioFormat, AudioSource, ProviderMapping
43from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
44
45from music_assistant.models.plugin import PluginProvider
46from music_assistant.providers.sendspin.constants import (
47 CONF_SOURCE_AUTOSTART_TARGET,
48 SOURCE_AUTOSTART_OFF,
49)
50
51from .constants import (
52 AUTOSTART_SIGNAL_ABSENT_HOLD_S,
53 AUTOSTART_SIGNAL_DEBOUNCE_S,
54 CHUNK_DURATION_MS,
55 COLD_START_TIMEOUT_S,
56 CONF_TARGET_LATENCY,
57 DEFAULT_TARGET_LATENCY_MS,
58 OUTPUT_BIT_DEPTH,
59 OUTPUT_CHANNELS,
60 OUTPUT_SAMPLE_RATE,
61 SOURCE_TIMEOUT_S,
62)
63
64if TYPE_CHECKING:
65 from collections.abc import AsyncGenerator, Callable
66
67 from aiosendspin.audio import SourceBridge
68 from aiosendspin.server import (
69 ClientEvent,
70 SendspinClient,
71 SendspinEvent,
72 SendspinServer,
73 SourceStream,
74 )
75 from aiosendspin.server.roles import SourceV1Role
76 from music_assistant_models.config_entries import ProviderConfig
77 from music_assistant_models.enums import ProviderFeature
78 from music_assistant_models.provider import ProviderManifest
79
80 from music_assistant.mass import MusicAssistant
81 from music_assistant.providers.sendspin.provider import SendspinProvider
82
83OUTPUT_FORMAT = AudioFormat(
84 content_type=ContentType.PCM_S16LE,
85 sample_rate=OUTPUT_SAMPLE_RATE,
86 bit_depth=OUTPUT_BIT_DEPTH,
87 channels=OUTPUT_CHANNELS,
88)
89BRIDGE_OUTPUT_FORMAT = SendspinAudioFormat(
90 sample_rate=OUTPUT_SAMPLE_RATE,
91 bit_depth=OUTPUT_BIT_DEPTH,
92 channels=OUTPUT_CHANNELS,
93)
94
95
96@dataclass
97class _SourceSession:
98 """State for one source's active exclusive stream."""
99
100 client_id: str
101 player_id: str
102 owner_player_id: str
103 stream_session_id: str
104 playback_session_id: str | None = None
105 # Retires the prior generator when the same player reclaims the session.
106 generation: int = 0
107 bridge: SourceBridge | None = None
108 ingest_task: asyncio.Task[None] | None = None
109 # Selection time counts toward the source timeout before PCM arrives.
110 last_pcm_monotonic: float = field(default_factory=time.monotonic)
111 pcm_received: asyncio.Event = field(default_factory=asyncio.Event)
112
113
114@dataclass
115class _SourceClientState:
116 """State retained for one source client."""
117
118 session: _SourceSession | None = None
119 unwatch: Callable[[], None] | None = None
120 signal: SignalState | None = None
121 autostart_queue_id: str | None = None
122 autostart_queue_session_id: str | None = None
123 suppressed_autostart_claim: tuple[str, str | None] | None = None
124 selection_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
125
126
127class SendspinSourceProvider(PluginProvider):
128 """Expose Sendspin source-role clients as MA AudioSources."""
129
130 def __init__(
131 self,
132 mass: MusicAssistant,
133 manifest: ProviderManifest,
134 config: ProviderConfig,
135 supported_features: set[ProviderFeature] | None = None,
136 ) -> None:
137 """Initialize the provider."""
138 super().__init__(mass, manifest, config, supported_features)
139 self._clients: dict[str, _SourceClientState] = {}
140 self._server_unsubscribe: Callable[[], None] | None = None
141 self._unloading = False
142
143 async def loaded_in_mass(self) -> None:
144 """Start watching every source client, including ones that are already idle."""
145 await super().loaded_in_mass()
146 self._unloading = False
147 if (sendspin := self._sendspin_provider) is None:
148 return
149 self._server_unsubscribe = sendspin.server_api.add_event_listener(self._on_server_event)
150 for client in sendspin.server_api.connected_clients:
151 self._watch_client(client)
152
153 async def unload(self, is_removed: bool = False) -> None:
154 """Handle unload/close of the provider."""
155 self._unloading = True
156 if self._server_unsubscribe is not None:
157 self._server_unsubscribe()
158 self._server_unsubscribe = None
159 for client_id, state in list(self._clients.items()):
160 self._cancel_pending_autostart(client_id)
161 self._cancel_pending_autostop(client_id)
162 if state.unwatch is not None:
163 state.unwatch()
164 state.unwatch = None
165 await self._teardown_session(client_id)
166 self._clients.clear()
167
168 async def get_audio_sources(self) -> list[AudioSource]:
169 """Return one AudioSource per connected client with an active source role."""
170 if (sendspin := self._sendspin_provider) is None:
171 return []
172 sources: list[AudioSource] = []
173 for client in sendspin.server_api.connected_clients:
174 if self._get_source_role(client) is None:
175 continue
176 info = client.info_or_none
177 name = info.name if info else client.client_id
178 sources.append(
179 AudioSource(
180 item_id=client.client_id,
181 provider=self.instance_id,
182 name=name,
183 provider_mappings={
184 ProviderMapping(
185 item_id=client.client_id,
186 provider_domain=self.domain,
187 provider_instance=self.instance_id,
188 audio_format=OUTPUT_FORMAT,
189 )
190 },
191 can_play_pause=False,
192 can_seek=False,
193 can_next_previous=False,
194 exclusive=True,
195 allow_external_trigger=True,
196 can_initiate=True,
197 )
198 )
199 return sources
200
201 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
202 """
203 Return StreamDetails for streaming the given source to a queue.
204
205 Side-effect-free: streaming is requested from the client in
206 on_source_selected, which the streams controller fires before this
207 method on the actual stream request (never on queue preload).
208 """
209 client = self._get_client(item_id)
210 if client is None or self._get_source_role(client) is None:
211 raise MediaNotFoundError(f"Unknown or unavailable Sendspin source: {item_id}")
212 info = client.info_or_none
213 return StreamDetails(
214 provider=self.instance_id,
215 item_id=item_id,
216 audio_format=OUTPUT_FORMAT,
217 media_type=media_type,
218 stream_type=StreamType.CUSTOM,
219 stream_metadata=StreamMetadata(title=info.name if info else item_id),
220 )
221
222 async def on_source_selected(
223 self, source_id: str, player_id: str, owner_player_id: str, stream_session_id: str
224 ) -> None:
225 """Claim the source and ask the client to start streaming."""
226 client = self._get_client(source_id)
227 role = self._get_source_role(client) if client else None
228 if client is None or role is None:
229 raise MediaNotFoundError(f"Sendspin source is not connected: {source_id}")
230 state = self._clients.setdefault(source_id, _SourceClientState())
231 # Capture the source's session before waiting so delayed requests keep their
232 # identity: a live source plays on the player, so the queue has no session for it.
233 queue_session_id = self._player_session_id(owner_player_id)
234 try:
235 # Serialize handoffs because stopping the previous player can suspend.
236 async with state.selection_lock:
237 # The client or its role may have changed while waiting for the lock.
238 client = self._get_client(source_id)
239 role = self._get_source_role(client) if client else None
240 if client is None or role is None:
241 raise MediaNotFoundError(f"Sendspin source is not connected: {source_id}")
242 # Reject only the delayed request from an autostart superseded by the user.
243 if state.suppressed_autostart_claim == (owner_player_id, queue_session_id):
244 raise RuntimeError("Superseded autostart stream request")
245 autostart_queue_id = state.autostart_queue_id
246 if autostart_queue_id == owner_player_id:
247 state.autostart_queue_id = None
248 state.autostart_queue_session_id = None
249 self._cancel_pending_autostart(source_id, cancel_running=False)
250 else:
251 if autostart_queue_id is not None:
252 autostart_session_id = (
253 state.autostart_queue_session_id
254 or self._player_session_id(autostart_queue_id)
255 )
256 state.autostart_queue_id = None
257 state.autostart_queue_session_id = None
258 state.suppressed_autostart_claim = (
259 autostart_queue_id,
260 autostart_session_id,
261 )
262 self._cancel_pending_autostart(source_id)
263 if autostart_queue_id is not None:
264 try:
265 # deselect rather than stopping the queue: the source plays on
266 # the player, so stopping the queue would leave the source
267 # published on it while resetting the queue we are preserving
268 await self.mass.players.deselect_source(
269 autostart_queue_id,
270 provider_instance_id=self.instance_id,
271 source_id=source_id,
272 playback_session_id=autostart_session_id,
273 )
274 except (KeyError, PlayerCommandFailed, PlayerUnavailableError) as err:
275 self.logger.debug(
276 "Failed to release autostart player %s: %s",
277 autostart_queue_id,
278 err,
279 )
280 # Keep the running bridge when renderers reopen the URL for the same queue.
281 if (live := state.session) is not None and (
282 live.player_id,
283 live.owner_player_id,
284 ) == (player_id, owner_player_id):
285 live.stream_session_id = stream_session_id
286 live.playback_session_id = queue_session_id
287 live.generation += 1
288 return
289 await self._teardown_session(source_id, superseded_by_player_id=player_id)
290 if self._unloading or self._clients.get(source_id) is not state:
291 raise PlayerUnavailableError(f"Sendspin source is unloading: {source_id}")
292 # Teardown can suspend long enough for the connection role to be recreated.
293 client = self._get_client(source_id)
294 role = self._get_source_role(client) if client else None
295 if client is None or role is None:
296 raise MediaNotFoundError(f"Sendspin source is not connected: {source_id}")
297 state.session = _SourceSession(
298 client_id=source_id,
299 player_id=player_id,
300 owner_player_id=owner_player_id,
301 stream_session_id=stream_session_id,
302 playback_session_id=queue_session_id,
303 )
304 role.request_start()
305 finally:
306 self._drop_empty_client_state(source_id, state)
307
308 async def on_source_unselected(
309 self, source_id: str, owner_player_id: str, stream_session_id: str
310 ) -> None:
311 """Release the source when MA tears down its stream."""
312 session = self._get_session(source_id)
313 # Reject stale callbacks from superseded same-queue requests.
314 if session is None or session.stream_session_id != stream_session_id:
315 return
316 await self._teardown_session(source_id)
317 if (state := self._clients.get(source_id)) is not None:
318 state.suppressed_autostart_claim = None
319 self._drop_empty_client_state(source_id, state)
320
321 async def get_audio_stream(
322 self, streamdetails: StreamDetails, seek_position: int = 0
323 ) -> AsyncGenerator[bytes]:
324 """
325 Yield fixed-format PCM pulled from the source's clock bridge.
326
327 The pull cadence of this loop is the master clock: the bridge converts
328 the client's drifting capture stream to it and pads silence on underrun,
329 so the stream keeps playing through gaps (an unplugged line-in is silent,
330 not stopped) until SOURCE_TIMEOUT_S passes without source audio. Pulling at
331 the output rate here makes the controller's own realtime pacer a no-op.
332 """
333 session = self._get_session(streamdetails.item_id)
334 if session is None:
335 raise AudioError(f"Sendspin source is not selected: {streamdetails.item_id}")
336 generation = session.generation
337 await self._await_first_audio(session)
338 if self._get_session(session.client_id) is not session or session.generation != generation:
339 return
340 frames_per_chunk = OUTPUT_SAMPLE_RATE * CHUNK_DURATION_MS // 1000
341 period = CHUNK_DURATION_MS / 1000
342 loop = self.mass.loop
343 next_deadline = loop.time()
344 while True:
345 if (
346 self._get_session(session.client_id) is not session
347 or session.generation != generation
348 ):
349 break
350 if time.monotonic() - session.last_pcm_monotonic > SOURCE_TIMEOUT_S:
351 self.logger.info(
352 "No audio from Sendspin source %s for %.0fs, ending stream",
353 session.client_id,
354 SOURCE_TIMEOUT_S,
355 )
356 break
357 if (bridge := session.bridge) is None:
358 break
359 yield bridge.read(frames_per_chunk)
360 next_deadline += period
361 delay = next_deadline - loop.time()
362 if delay > 0:
363 await asyncio.sleep(delay)
364 elif -delay * 1_000_000 > bridge.occupancy_us:
365 # Consumer stalled beyond the buffered audio. Catch-up reads past this
366 # point would fabricate silence into the timeline, so re-anchor instead.
367 next_deadline = loop.time()
368
369 @property
370 def _sendspin_provider(self) -> SendspinProvider | None:
371 return cast("SendspinProvider | None", self.mass.get_provider("sendspin"))
372
373 def _get_client(self, client_id: str) -> SendspinClient | None:
374 if (sendspin := self._sendspin_provider) is None:
375 return None
376 return sendspin.server_api.get_client(client_id)
377
378 @staticmethod
379 def _get_source_role(client: SendspinClient) -> SourceV1Role | None:
380 roles = client.roles_by_family("source")
381 return cast("SourceV1Role", roles[0]) if roles else None
382
383 async def _await_first_audio(self, session: _SourceSession) -> None:
384 """
385 Block until the source actually streams, so a failed acquisition raises.
386
387 The silence-hold only makes sense once audio has flowed: a client that never
388 answers the start command is a broken source, not a quiet one.
389 """
390 try:
391 async with asyncio.timeout(COLD_START_TIMEOUT_S):
392 await session.pcm_received.wait()
393 except TimeoutError:
394 if self._get_session(session.client_id) is not session:
395 return
396 client = self._get_client(session.client_id)
397 info = client.info_or_none if client else None
398 raise AudioError(
399 f"Sendspin source {session.client_id} did not start streaming",
400 translation_key="no_audio",
401 translation_owner=self.translation_owner,
402 translation_args=[info.name if info else session.client_id],
403 ) from None
404
405 def _on_server_event(self, server: SendspinServer, event: SendspinEvent) -> None:
406 if self._unloading:
407 return
408 match event:
409 case ClientConnectedEvent(client_id):
410 # Roles attach after this event, so defer until the next loop turn.
411 self.mass.create_task(self._on_client_connected(client_id), eager_start=False)
412 case ClientRemovedEvent(client_id) | ClientDisconnectedEvent(client_id):
413 self._cancel_pending_autostart(client_id)
414 self._cancel_pending_autostop(client_id)
415 if (state := self._clients.get(client_id)) is not None:
416 state.signal = None
417 state.autostart_queue_id = None
418 state.autostart_queue_session_id = None
419 if state.session is None:
420 state.suppressed_autostart_claim = None
421 if state.unwatch is not None:
422 state.unwatch()
423 state.unwatch = None
424 self._drop_empty_client_state(client_id, state)
425
426 async def _on_client_connected(self, client_id: str) -> None:
427 """Re-arm watching and streaming for a client that just (re)connected."""
428 if self._unloading:
429 return
430 client = self._get_client(client_id)
431 if client is None:
432 return
433 self._watch_client(client)
434 # A reconnect clears the client's start request, so ask again.
435 if self._get_session(client_id) is None:
436 return
437 role = self._get_source_role(client)
438 if role is None or role.stream_active:
439 return
440 self.logger.debug("Re-requesting stream start from %s", client_id)
441 role.request_start()
442
443 def _watch_client(self, client: SendspinClient) -> None:
444 """Subscribe to a source client's events, for signal presence while idle."""
445 # Watch negotiated source roles because pairing can activate the role later
446 # without reconnecting or emitting another event.
447 if "source" not in {role_family(role_id) for role_id in client.negotiated_role_ids}:
448 return
449 state = self._clients.setdefault(client.client_id, _SourceClientState())
450 if state.unwatch is not None:
451 return
452 state.unwatch = client.add_event_listener(self._on_client_event)
453
454 def _on_client_event(self, client: SendspinClient, event: ClientEvent) -> None:
455 client_id = client.client_id
456 if isinstance(event, SourceSignalChangedEvent):
457 self._on_signal_reported(client_id, event.signal)
458 return
459 session = self._get_session(client_id)
460 if session is None:
461 return
462 if isinstance(event, SourceStreamStartedEvent):
463 self.mass.create_task(self._attach_stream(session, event))
464 elif isinstance(event, SourceStreamEndedEvent):
465 self.logger.debug("Sendspin source %s ended its stream", client_id)
466
467 def _on_signal_reported(self, client_id: str, signal: SignalState) -> None:
468 """
469 Drive line-in autostart/autostop from a reported signal presence.
470
471 Only transitions act. The first report for a client is recorded silently so a
472 server restart or a reconnect with the needle already down starts nothing.
473 """
474 state = self._clients.setdefault(client_id, _SourceClientState())
475 previous = state.signal
476 if previous == signal:
477 return
478 state.signal = signal
479 if previous is None:
480 return
481 self.logger.debug("Sendspin source %s signal %s", client_id, signal.value)
482 self._cancel_pending_autostart(client_id)
483 self._cancel_pending_autostop(client_id)
484 if signal == SignalState.PRESENT:
485 self.mass.call_later(
486 AUTOSTART_SIGNAL_DEBOUNCE_S,
487 self._autostart,
488 client_id,
489 task_id=self._autostart_timer_id(client_id),
490 )
491 else:
492 self.mass.call_later(
493 AUTOSTART_SIGNAL_ABSENT_HOLD_S,
494 self._autostop,
495 client_id,
496 task_id=self._autostop_timer_id(client_id),
497 )
498
499 def _cancel_pending_autostart(self, client_id: str, *, cancel_running: bool = True) -> None:
500 self._cancel_scheduled_action(
501 self._autostart_timer_id(client_id), cancel_running=cancel_running
502 )
503
504 def _cancel_pending_autostop(self, client_id: str) -> None:
505 self._cancel_scheduled_action(self._autostop_timer_id(client_id))
506
507 async def _autostart(self, client_id: str) -> None:
508 """Start playing a source whose signal has stayed present."""
509 state = self._clients.get(client_id)
510 if state is None or state.signal != SignalState.PRESENT or state.session is not None:
511 return
512 if (target := await self._resolve_autostart_target(client_id)) is None:
513 return
514 state = self._clients.get(client_id)
515 if state is None or state.signal != SignalState.PRESENT or state.session is not None:
516 return
517 queue_id, uri = target
518 self.logger.info("Line-in signal on %s, starting playback on %s", client_id, queue_id)
519 state.suppressed_autostart_claim = None
520 state.autostart_queue_id = queue_id
521 state.autostart_queue_session_id = None
522 completed = False
523 try:
524 await self.mass.player_queues.play_media(queue_id, uri, option=QueueOption.PLAY)
525 completed = True
526 if self._clients.get(client_id) is state and state.autostart_queue_id == queue_id:
527 state.autostart_queue_session_id = self._player_session_id(queue_id)
528 finally:
529 if (
530 not completed
531 and self._clients.get(client_id) is state
532 and state.autostart_queue_id == queue_id
533 ):
534 state.autostart_queue_id = None
535 state.autostart_queue_session_id = None
536 self._drop_empty_client_state(client_id, state)
537
538 async def _autostop(self, client_id: str) -> None:
539 """Stop a source whose signal has stayed absent, e.g. a record that ended."""
540 state = self._clients.get(client_id)
541 if state is None or state.signal != SignalState.ABSENT or state.session is None:
542 return
543 session = state.session
544 self.logger.info("Line-in signal gone on %s, stopping playback", client_id)
545 try:
546 # Queue stop also cancels pending preload and enqueue timers.
547 await self.mass.players.deselect_source(
548 session.owner_player_id,
549 provider_instance_id=self.instance_id,
550 source_id=client_id,
551 playback_session_id=session.playback_session_id,
552 )
553 except (KeyError, PlayerCommandFailed, PlayerUnavailableError) as err:
554 self.logger.debug("Failed to stop player %s: %s", session.owner_player_id, err)
555
556 def _player_session_id(self, player_id: str) -> str | None:
557 """
558 Return the id of whatever playback session the given player is running.
559
560 A live source's session belongs to the player; anything else is the queue's.
561 Used to tell a superseded autostart request apart from the one that replaced it.
562
563 :param player_id: The player to read the session of.
564 """
565 if (session := self.mass.players.get_audio_source_session(player_id)) is not None:
566 return session.playback_session_id
567 return self.mass.player_queues.queue_data(player_id).session_id
568
569 async def _resolve_autostart_target(self, client_id: str) -> tuple[str, str] | None:
570 """Return the queue id and source uri to autostart, if the source is configured."""
571 # Config entry defaults are not persisted until the user saves the page.
572 target_player_id = await self.mass.config.get_player_config_value(
573 client_id, CONF_SOURCE_AUTOSTART_TARGET, default=SOURCE_AUTOSTART_OFF
574 )
575 if not target_player_id or target_player_id == SOURCE_AUTOSTART_OFF:
576 return None
577 player = self.mass.players.get_player(target_player_id)
578 if player is None:
579 self.logger.warning(
580 "Autostart target %s for Sendspin source %s no longer exists",
581 target_player_id,
582 client_id,
583 )
584 return None
585 # Keep a grouped target in its active group. A target already playing a live
586 # source resolves to no queue, which is not a reason to refuse - it is the
587 # player we start on either way.
588 queue = self.mass.players.get_active_queue(player)
589 # mirror the controller's owner resolution: a sync child starts on its leader and
590 # a group member on its group, never on itself - selecting on the child ungroups it
591 target_id = (
592 queue.queue_id
593 if queue
594 else (player.state.synced_to or player.state.active_group or player.player_id)
595 )
596 return target_id, create_uri(MediaType.AUDIO_SOURCE, self.instance_id, client_id)
597
598 async def _attach_stream(
599 self, session: _SourceSession, event: SourceStreamStartedEvent
600 ) -> None:
601 """Route a (re)started source stream into a fresh bridge."""
602 if self._get_session(session.client_id) is not session:
603 return
604 if session.ingest_task is not None:
605 session.ingest_task.cancel()
606 # Provider options are unresolved when the instance first loads.
607 target_latency = (
608 cast("int | None", self.config.get_value(CONF_TARGET_LATENCY))
609 or DEFAULT_TARGET_LATENCY_MS
610 )
611 session.bridge = self._create_bridge(event.audio_format, target_latency)
612 session.last_pcm_monotonic = time.monotonic()
613 session.ingest_task = self.mass.create_task(self._ingest(session, event.handle))
614
615 def _create_bridge(
616 self, input_format: SendspinAudioFormat, target_latency_ms: int
617 ) -> SourceBridge:
618 return AsrcSourceBridge(
619 input_format=input_format,
620 output_format=BRIDGE_OUTPUT_FORMAT,
621 target_latency_ms=target_latency_ms,
622 )
623
624 async def _ingest(self, session: _SourceSession, handle: SourceStream) -> None:
625 """Feed decoded source chunks into the session's bridge until the stream ends."""
626 bridge = session.bridge
627 if bridge is None:
628 return
629 async for pcm, timestamp_us in handle:
630 if self._get_session(session.client_id) is not session or session.bridge is not bridge:
631 break
632 try:
633 bridge.feed(pcm, timestamp_us)
634 except ValueError as err:
635 self.logger.warning("Dropping malformed chunk from %s: %s", session.client_id, err)
636 continue
637 session.last_pcm_monotonic = time.monotonic()
638 session.pcm_received.set()
639 else:
640 if self._get_session(session.client_id) is session and session.bridge is bridge:
641 bridge.flush()
642
643 async def _teardown_session(
644 self, source_id: str, superseded_by_player_id: str | None = None
645 ) -> None:
646 state = self._clients.get(source_id)
647 if state is None or state.session is None:
648 return
649 session, state.session = state.session, None
650 if session.ingest_task is not None:
651 session.ingest_task.cancel()
652 # Stop even when superseding: the replacement session only gets a bridge from a
653 # fresh client_stream/start, which the client sends after a stop/start cycle.
654 if (client := self._get_client(session.client_id)) is not None and (
655 role := self._get_source_role(client)
656 ) is not None:
657 role.request_stop()
658 if superseded_by_player_id is not None and superseded_by_player_id != session.player_id:
659 # Ending the generator leaves the handed-off player draining its buffer over
660 # the new one, so stop it. A same-player re-claim keeps playing.
661 try:
662 await self.mass.players.cmd_stop(session.player_id)
663 except (PlayerCommandFailed, PlayerUnavailableError) as err:
664 self.logger.debug("Failed to stop player %s: %s", session.player_id, err)
665 self._drop_empty_client_state(source_id, state)
666
667 def _get_session(self, client_id: str) -> _SourceSession | None:
668 state = self._clients.get(client_id)
669 return state.session if state is not None else None
670
671 def _drop_empty_client_state(self, client_id: str, state: _SourceClientState) -> None:
672 if (
673 self._clients.get(client_id) is state
674 and state.session is None
675 and state.unwatch is None
676 and state.signal is None
677 and state.autostart_queue_id is None
678 and state.autostart_queue_session_id is None
679 and state.suppressed_autostart_claim is None
680 and not state.selection_lock.locked()
681 ):
682 self._clients.pop(client_id)
683
684 def _cancel_scheduled_action(self, task_id: str, *, cancel_running: bool = True) -> None:
685 self.mass.cancel_timer(task_id)
686 task = self.mass.get_task(task_id)
687 if cancel_running and task is not None and task is not asyncio.current_task():
688 self.mass.cancel_task(task_id)
689
690 def _autostart_timer_id(self, client_id: str) -> str:
691 return f"{self.instance_id}_autostart_{client_id}"
692
693 def _autostop_timer_id(self, client_id: str) -> str:
694 return f"{self.instance_id}_autostop_{client_id}"
695