music-assistant-server

1"""Group Player implementation."""
2
3from __future__ import annotations
4
5import asyncio
6from copy import deepcopy
7from time import time
8from typing import TYPE_CHECKING, cast
9
10from aiohttp import HttpVersion11, web
11from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
12from music_assistant_models.constants import PLAYER_CONTROL_FAKE, PLAYER_CONTROL_NONE
13from music_assistant_models.enums import (
14    ConfigEntryType,
15    ContentType,
16    MediaType,
17    PlaybackState,
18    PlayerFeature,
19    PlayerType,
20)
21from music_assistant_models.errors import UnsupportedFeaturedException
22from music_assistant_models.media_items import AudioFormat
23from propcache import under_cached_property as cached_property
24
25from music_assistant.constants import (
26    CONF_DYNAMIC_GROUP_MEMBERS,
27    CONF_ENTRY_HTTP_PROFILE_DEFAULT_1,
28    CONF_GROUP_MEMBERS,
29    CONF_HTTP_PROFILE,
30    CONF_POWER_CONTROL,
31    DEFAULT_STREAM_HEADERS,
32    DLNA_CONTENT_FEATURES_REALTIME,
33)
34from music_assistant.controllers.streams.audio_processing import get_media_session_id
35from music_assistant.helpers.audio import get_mime_type
36from music_assistant.helpers.util import TaskManager
37from music_assistant.models.player import DeviceInfo, Player, PlayerMedia
38
39from .constants import (
40    CONF_ENTRY_UGP_OUTPUT_FORMAT,
41    CONF_UGP_OUTPUT_FORMAT,
42    CONFIG_ENTRY_UGP_NOTE,
43    EXTRA_FEATURES_FROM_MEMBERS,
44    IDLE_GRACE_SECONDS,
45    UGP_OUTPUT_MP3,
46    resolve_ugp_output_format,
47)
48from .ugp_stream import UGPStream
49
50if TYPE_CHECKING:
51    from .provider import UniversalGroupProvider
52
53# The features the group carries on its own. Everything else is resolved per read in
54# the supported_features property: POWER when the user assigns 'Fake power control',
55# SET_MEMBERS for dynamic groups, and EXTRA_FEATURES_FROM_MEMBERS from the members.
56# PlayerFeature.POWER is intentionally not a base feature: the lifecycle (form on
57# play, dissolve on stop, debounced idle deform) governs whether the group captures
58# its members.
59BASE_FEATURES = {
60    PlayerFeature.PLAY_MEDIA,
61    PlayerFeature.MULTI_DEVICE_DSP,
62}
63
64
65class UniversalGroupPlayer(Player):
66    """Universal Group Player implementation."""
67
68    _attr_type: PlayerType = PlayerType.GROUP
69
70    def __init__(
71        self,
72        provider: UniversalGroupProvider,
73        player_id: str,
74    ) -> None:
75        """Initialize UniversalGroupPlayer instance."""
76        super().__init__(provider, player_id)
77        self.stream: UGPStream | None = None
78        self._attr_name = self.config.name or f"Universal Group {player_id}"
79        self._attr_available = True
80        # See SyncGroupPlayer: groups have no opinion on power by default; the
81        # session lifecycle is what governs activity. Fake power control is the
82        # opt-in mechanism for explicit on/off semantics.
83        self._attr_powered = None
84        self._attr_device_info = DeviceInfo(model="Universal Group", manufacturer=provider.name)
85        self._attr_needs_poll = True
86        self._attr_poll_interval = 30
87        # task that releases members after the idle grace window expires
88        self._idle_grace_task: asyncio.Task[None] | None = None
89        # register dynamic routes for the ugp stream (FLAC + MP3 cover the configured
90        # output formats; the actual codec served is decided by the UGP's own config,
91        # not by the request URL)
92        self._on_unload_callbacks.append(
93            self.mass.streams.register_dynamic_route(
94                f"/ugp/{self.player_id}.flac", self._serve_ugp_stream
95            )
96        )
97        self._on_unload_callbacks.append(
98            self.mass.streams.register_dynamic_route(
99                f"/ugp/{self.player_id}.mp3", self._serve_ugp_stream
100            )
101        )
102        self._set_attributes()
103
104    @property
105    def supported_features(self) -> set[PlayerFeature]:
106        """Return the supported features of the player."""
107        features = {*BASE_FEATURES}
108        # The raw config value is read here to avoid recursion via the power_control
109        # property (which itself may inspect supported features).
110        raw_power_conf = self.mass.config.get_raw_player_config_value(
111            self.player_id, CONF_POWER_CONTROL
112        )
113        if raw_power_conf == PLAYER_CONTROL_FAKE:
114            features.add(PlayerFeature.POWER)
115        if self.is_dynamic:
116            features.add(PlayerFeature.SET_MEMBERS)
117        # derive the fanned-out features from all (configured) members, so volume and
118        # mute are advertised whether or not the group currently has a live session.
119        for member_id in self._attr_group_members:
120            member_player = self.mass.players.get_player(member_id)
121            if member_player and member_player.state.available:
122                for feature in EXTRA_FEATURES_FROM_MEMBERS:
123                    if feature in member_player.state.supported_features:
124                        features.add(feature)
125        return features
126
127    @property
128    def requires_flow_mode(self) -> bool:
129        """Return if the player requires flow mode."""
130        return True
131
132    @property
133    def synced_to(self) -> str | None:
134        """Return the id of the player this player is synced to (sync leader)."""
135        # groups can't be synced
136        return None
137
138    @property
139    def is_active_session(self) -> bool:
140        """
141        Return whether this group currently has captured members.
142
143        The session is considered active while the multicast stream is live or
144        while the idle grace timer is still pending. ``__final_active_group``
145        reads this to decide whether the configured members should be marked
146        as ``active_group`` for this group.
147        """
148        if self.stream is not None and not self.stream.done:
149            return True
150        return self._idle_grace_task is not None
151
152    @property
153    def can_group_with(self) -> set[str]:
154        """Return the id's of players this player can group with."""
155        if not self.is_dynamic:
156            # in case of static members,
157            # we can only group with the players defined in the config, so we return those directly
158            return set(self._attr_static_group_members)
159        # allow grouping with all providers, except the ugp provider itself
160        return {
161            x.instance_id
162            for x in self.mass.players.providers
163            if x.instance_id != self.provider.instance_id
164        }
165
166    @cached_property
167    def supported_sample_rates(self) -> list[tuple[int, int]] | None:
168        """Return the (sample_rate, bit_depth) pair the UGP serves to its members."""
169        # UGP delivers the same encoded stream to every member, so its only natively
170        # supported rate is whatever the configured output format produces. Returning a
171        # single-rate list keeps the upstream MA flow stream pinned and prevents
172        # smart/bit-perfect modes from triggering needless restarts.
173        output_format, _ = resolve_ugp_output_format(
174            cast("str", self.config.get_value(CONF_UGP_OUTPUT_FORMAT, UGP_OUTPUT_MP3))
175        )
176        return [(output_format.sample_rate, output_format.bit_depth)]
177
178    async def on_config_updated(self) -> None:
179        """Handle logic when the PlayerConfig is first loaded or updated."""
180        static_members = cast("list[str]", self.config.get_value(CONF_GROUP_MEMBERS, []))
181        self._attr_static_group_members = static_members.copy()
182        if not self.is_active_session:
183            # only realign members to the configured static set when the group
184            # is dormant — otherwise we would lose any dynamic adds mid-session.
185            self._attr_group_members = static_members.copy()
186
187    @cached_property
188    def is_dynamic(self) -> bool:
189        """Return if the player is a dynamic group player."""
190        return bool(self.config.get_value(CONF_DYNAMIC_GROUP_MEMBERS, False))
191
192    async def get_config_entries(self) -> list[ConfigEntry]:
193        """Return all (provider/player specific) Config Entries for the given player (if any)."""
194        return [
195            # add universal group specific entries
196            CONFIG_ENTRY_UGP_NOTE,
197            ConfigEntry(
198                key=CONF_GROUP_MEMBERS,
199                type=ConfigEntryType.STRING,
200                multi_value=True,
201                default_value=[],
202                required=False,  # needed for dynamic members (which allows empty members list)
203                options=[
204                    ConfigValueOption(x.player_id, title=x.display_name)
205                    for x in self.mass.players.all_players(True, False)
206                    if x.type != PlayerType.GROUP
207                ],
208            ),
209            ConfigEntry(
210                key=CONF_DYNAMIC_GROUP_MEMBERS,
211                type=ConfigEntryType.BOOLEAN,
212                default_value=False,
213                required=False,
214            ),
215            CONF_ENTRY_UGP_OUTPUT_FORMAT,
216            CONF_ENTRY_HTTP_PROFILE_DEFAULT_1,
217        ]
218
219    async def stop(self) -> None:
220        """
221        Handle STOP command.
222
223        An explicit stop releases the captured members immediately so they
224        return to individual control. The idle grace timer is only used for
225        natural end-of-queue transitions (see :meth:`_set_attributes`). Users
226        who want the group to stay 'active' across stops can assign Fake
227        power control and use that to pin the group.
228        """
229        # an explicit stop overrides any pending idle-grace release
230        self._cancel_idle_grace_timer()
231        async with TaskManager(self.mass) as tg:
232            for member in self.mass.players.iter_group_members(self, active_only=True):
233                # Use internal handler to get protocol selection and avoid redirect
234                tg.create_task(self.mass.players._handle_cmd_stop(member.player_id))
235        # abort the stream session — this drops is_active_session to False so
236        # the (former) members will see active_group=None on their next state
237        # update and accept direct playback commands again.
238        if self.stream and not self.stream.done:
239            await self.stream.stop()
240            self.stream = None
241        # snap group_members back to the configured static set so we don't
242        # keep stale dynamic adds around once the session has ended.
243        if self._attr_powered is not True:
244            self._attr_group_members = self._attr_static_group_members.copy()
245        self._set_attributes()
246
247    async def power(self, powered: bool) -> None:
248        """
249        Handle POWER command to group player.
250
251        Only called when the user has assigned a power control (native or fake)
252        to the group. Powering ON prepares the members so the group is
253        considered 'active' immediately (matching the legacy behaviour for
254        users who opt in). Powering OFF stops any playback and releases the
255        captured members.
256
257        :param powered: True to power on (capture members), False to power off (release).
258        """
259        # any pending idle-grace release is moot — we're on an explicit transition
260        self._cancel_idle_grace_timer()
261
262        # always stop at power off
263        if not powered and self._attr_playback_state in (
264            PlaybackState.PLAYING,
265            PlaybackState.PAUSED,
266        ):
267            await self.stop()
268
269        prev_power = self._attr_powered
270        self._attr_powered = powered
271
272        if powered:
273            await self._capture_members()
274        elif prev_power:
275            # handle TURN_OFF of the group player by turning off all members
276            for member in self.mass.players.iter_group_members(
277                self, only_powered=True, active_only=True
278            ):
279                if member.powered and member.power_control != PLAYER_CONTROL_NONE:
280                    await self.mass.players.cmd_power(member.player_id, False)
281
282        if not powered:
283            # reset the original group members when powered off
284            self._attr_group_members = self._attr_static_group_members.copy()
285        self.update_state()
286
287    async def play_media(self, media: PlayerMedia) -> None:
288        """Handle PLAY MEDIA on given player."""
289        # form on play: cancel any pending idle-grace release, then capture the
290        # configured members and free them of any conflicting prior allegiance.
291        self._cancel_idle_grace_timer()
292        await self._capture_members()
293
294        if self.stream and not self.stream.done:
295            # stop any existing stream first
296            await self.stream.stop()
297
298        # resolve the static output format the UGP serves to all members
299        output_format, fmt_str = resolve_ugp_output_format(
300            cast("str", self.config.get_value(CONF_UGP_OUTPUT_FORMAT, UGP_OUTPUT_MP3))
301        )
302        # internal PCM pivot for the multiplexer: F32 at the configured output rate
303        # so the per-member encoder doesn't have to resample
304        pivot_format = AudioFormat(
305            content_type=ContentType.PCM_F32LE,
306            sample_rate=output_format.sample_rate,
307            bit_depth=32,
308            channels=2,
309        )
310        audio_source = self.mass.streams.get_stream(media, pivot_format, self.player_id)
311        self.stream = UGPStream(
312            audio_source=audio_source,
313            audio_format=pivot_format,
314            base_pcm_format=pivot_format,
315            queue_id=media.source_id,
316            session_id=get_media_session_id(media),
317        )
318        base_url = f"{self.mass.streams.base_url}/ugp/{self.player_id}.{fmt_str}"
319
320        # set the state optimistically
321        self._attr_current_media = deepcopy(media)
322        self._attr_elapsed_time = 0
323        self._attr_elapsed_time_last_updated = time() - 1
324        self._attr_playback_state = PlaybackState.PLAYING
325        self.update_state()
326
327        # forward to downstream play_media commands
328        async with TaskManager(self.mass) as tg:
329            for member in self.mass.players.iter_group_members(self, only_powered=True):
330                # Use internal handler to get protocol selection and avoid redirect
331                tg.create_task(
332                    self.mass.players._handle_play_media(
333                        member.player_id,
334                        PlayerMedia(
335                            uri=f"{base_url}?player_id={member.player_id}",
336                            media_type=MediaType.FLOW_STREAM,
337                            title=self.display_name,
338                            source_id=self.player_id,
339                            custom_data={
340                                "ugp_player_id": self.player_id,
341                                "session_id": self.stream.session_id,
342                            },
343                        ),
344                    )
345                )
346
347    async def set_members(
348        self,
349        player_ids_to_add: list[str] | None = None,
350        player_ids_to_remove: list[str] | None = None,
351    ) -> None:
352        """Handle SET_MEMBERS command on the player."""
353        if not self.is_dynamic:
354            raise UnsupportedFeaturedException(
355                f"Group {self.display_name} does not allow dynamically adding/removing members!",
356                translation_key="group_not_dynamic",
357                translation_owner=self.translation_owner,
358                translation_args=[self.display_name],
359            )
360        # handle additions
361        for player_id in player_ids_to_add or []:
362            if player_id in self._attr_group_members:
363                continue
364            if player_id == self.player_id:
365                raise UnsupportedFeaturedException(
366                    f"Cannot add {self.display_name} to itself as a member!",
367                    translation_key="cannot_add_group_to_itself",
368                    translation_owner=self.translation_owner,
369                    translation_args=[self.display_name],
370                )
371            child_player = self.mass.players.get_player(player_id, True)
372            assert child_player  # for type checking
373            if child_player.synced_to:
374                # This is player is part of a syncgroup - ungroup it first
375                await child_player.ungroup()
376            self._attr_group_members.append(player_id)
377            # let the newly added member join the stream if it's still live —
378            # the `self.powered` gate that used to guard this is gone with the
379            # session-lifecycle refactor (groups now have `_attr_powered=None`
380            # unless the user assigned Fake control).
381            if self.stream and not self.stream.done:
382                _, fmt_str = resolve_ugp_output_format(
383                    cast("str", self.config.get_value(CONF_UGP_OUTPUT_FORMAT, UGP_OUTPUT_MP3))
384                )
385                base_url = f"{self.mass.streams.base_url}/ugp/{self.player_id}.{fmt_str}"
386                # Use internal handler to get protocol selection and avoid redirect
387                await self.mass.players._handle_play_media(
388                    player_id,
389                    PlayerMedia(
390                        uri=f"{base_url}?player_id={player_id}",
391                        media_type=MediaType.FLOW_STREAM,
392                        title=self.display_name,
393                        source_id=self.player_id,
394                        custom_data={
395                            "ugp_player_id": self.player_id,
396                            "session_id": self.stream.session_id,
397                        },
398                    ),
399                )
400        # handle removals
401        for player_id in player_ids_to_remove or []:
402            if player_id not in self._attr_group_members:
403                continue
404            if player_id == self.player_id:
405                raise UnsupportedFeaturedException(
406                    f"Cannot remove {self.display_name} from itself as a member!",
407                    translation_key=(
408                        "provider.universal_group.errors.cannot_remove_group_from_itself"
409                    ),
410                    translation_args=[self.display_name],
411                )
412            self._attr_group_members.remove(player_id)
413            child_player = self.mass.players.get_player(player_id, True)
414            assert child_player is not None  # for type checking
415            if child_player.playback_state in (
416                PlaybackState.PLAYING,
417                PlaybackState.PAUSED,
418            ):
419                # if the child player is playing the group stream, stop it
420                # Use internal handler to get protocol selection and avoid redirect
421                await self.mass.players._handle_cmd_stop(player_id)
422        self.update_state()
423
424    async def poll(self) -> None:
425        """Poll player for state updates."""
426        self._set_attributes()
427
428    async def on_unload(self) -> None:
429        """Handle logic when the player is unloaded from the Player controller."""
430        self._cancel_idle_grace_timer()
431        await super().on_unload()
432        if self.is_active_session or self._attr_powered is True:
433            # tear down any in-flight session before unloading
434            await self.stop()
435            self._attr_powered = False
436
437    async def _capture_members(self) -> None:
438        """
439        Resolve collisions and prepare the configured members for grouping.
440
441        Rebuilds the effective member list from the configured static set,
442        powers on each member that has a power control, releases members
443        that are currently captured by another group / sync session, and
444        leaves the group ready for playback. Idempotent: safe to call on an
445        already-prepared group.
446        """
447        # rebuild the effective member list from the configured static set
448        self._attr_group_members = []
449        for static_group_member in self._attr_static_group_members:
450            if (
451                (member_player := self.mass.players.get_player(static_group_member))
452                and member_player.available
453                and member_player.enabled
454            ):
455                self._attr_group_members.append(static_group_member)
456        # ensure each member is free of any prior group/sync allegiance and ready to play
457        for member in self.mass.players.iter_group_members(
458            self, only_powered=False, active_only=False
459        ):
460            if (
461                member.playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED)
462                and member.active_source != self.active_source
463            ):
464                # Use internal handler to get protocol selection and avoid redirect
465                await self.mass.players._handle_cmd_stop(member.player_id)
466            if (
467                member.state.active_group is not None
468                and member.state.active_group != self.player_id
469            ):
470                # collision: child is currently captured by a different group
471                if other_group := self.mass.players.get_player(member.state.active_group):
472                    if (
473                        other_group.supports_feature(PlayerFeature.SET_MEMBERS)
474                        and member.player_id not in other_group.static_group_members
475                    ):
476                        async with self.mass.players.wait_for_player_update(
477                            member.player_id, timeout=5
478                        ):
479                            await other_group.set_members(player_ids_to_remove=[member.player_id])
480                    # the other group can't release this member dynamically — stop
481                    # it entirely so the member is freed. Route power-off through
482                    # the controller so a FAKE-power group also gets its extra_data
483                    # updated; calling other_group.power() directly would only set
484                    # _attr_powered and leave the cached fake state out of sync.
485                    elif other_group.state.power_control != PLAYER_CONTROL_NONE:
486                        async with self.mass.players.wait_for_player_update(
487                            member.player_id, timeout=5
488                        ):
489                            await self.mass.players._handle_cmd_power(other_group.player_id, False)
490                    else:
491                        async with self.mass.players.wait_for_player_update(
492                            member.player_id, timeout=5
493                        ):
494                            await other_group.stop()
495            if member.synced_to:
496                # member is part of a syncgroup — release it first
497                await member.ungroup()
498            if not member.powered and member.power_control != PLAYER_CONTROL_NONE:
499                await self.mass.players.cmd_power(member.player_id, True)
500
501    def _set_attributes(self) -> None:
502        """Set attributes of the group player."""
503        prev_state = self._attr_playback_state
504        # grab current media and state from one of the active players
505        # use state properties (not raw attributes) to account for protocol player propagation
506        for child_player in self.mass.players.iter_group_members(self, active_only=True):
507            self._attr_playback_state = child_player.state.playback_state
508            # a position is only meaningful together with the timestamp it was taken at,
509            # so the pair is adopted as a whole or not at all. Position 0 is a valid
510            # position: members that anchor the group stream once report a fixed 0 and
511            # let the timestamp carry both the progression and their own buffer delay.
512            if (
513                child_player.state.elapsed_time is not None
514                and child_player.state.elapsed_time_last_updated is not None
515            ):
516                self._attr_elapsed_time = child_player.state.elapsed_time
517                self._attr_elapsed_time_last_updated = child_player.state.elapsed_time_last_updated
518            break
519        else:
520            self._attr_playback_state = PlaybackState.IDLE
521        # idle grace handling: schedule a debounced release when playback
522        # naturally transitions to IDLE (e.g. queue ended). Skipped if the
523        # user has pinned the group with Fake power control.
524        if (
525            self._attr_playback_state == PlaybackState.IDLE
526            and prev_state in (PlaybackState.PLAYING, PlaybackState.PAUSED)
527            and self._attr_powered is not True
528            and self.stream is not None
529            and not self.stream.done
530        ):
531            self._schedule_idle_grace_timer()
532        elif self._attr_playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
533            self._cancel_idle_grace_timer()
534        self.update_state()
535
536    def _schedule_idle_grace_timer(self) -> None:
537        """Schedule a debounced session release after the stream becomes idle."""
538        self._cancel_idle_grace_timer()
539        self.logger.debug(
540            "Scheduling idle-grace release for universal group %s in %ss",
541            self.display_name,
542            IDLE_GRACE_SECONDS,
543        )
544        self._idle_grace_task = self.mass.create_task(self._idle_grace_runner())
545
546    def _cancel_idle_grace_timer(self) -> None:
547        """Cancel any pending idle-grace release task."""
548        if self._idle_grace_task is not None:
549            if not self._idle_grace_task.done():
550                self._idle_grace_task.cancel()
551            self._idle_grace_task = None
552
553    async def _idle_grace_runner(self) -> None:
554        """Wait the grace window, then release members if still idle."""
555        try:
556            await asyncio.sleep(IDLE_GRACE_SECONDS)
557        except asyncio.CancelledError:
558            return
559        # re-check state at fire time — a new play may have arrived, the user
560        # may have powered the group on, or another path may have torn down
561        # the session already.
562        self._idle_grace_task = None
563        if self._attr_powered is True:
564            return
565        if self._attr_playback_state != PlaybackState.IDLE:
566            return
567        self.logger.info(
568            "Idle-grace expired for universal group %s, releasing members",
569            self.display_name,
570        )
571        if self.stream and not self.stream.done:
572            await self.stream.stop()
573            self.stream = None
574        # snap group_members back to the configured static set; this drops
575        # is_active_session to False so children see active_group=None.
576        self._attr_group_members = self._attr_static_group_members.copy()
577        self.update_state()
578
579    async def _serve_ugp_stream(self, request: web.Request) -> web.StreamResponse:
580        """Serve the UGP (multi-client) flow stream audio to a player."""
581        ugp_player_id = request.path.rsplit(".")[0].rsplit("/")[-1]
582        # child_player_id is optional and only used for per-member DSP — never to
583        # decide the output codec/rate. The output format is dictated by the UGP
584        # player's own CONF_UGP_OUTPUT_FORMAT so every member receives an identical
585        # encoded stream.
586        child_player_id = request.query.get("player_id")
587
588        if not (ugp_player := self.mass.players.get_player(ugp_player_id)):
589            raise web.HTTPNotFound(reason=f"Unknown UGP player: {ugp_player_id}")
590        if not self.stream or self.stream.done:
591            raise web.HTTPNotFound(body=f"There is no active UGP stream for {ugp_player_id}!")
592
593        output_format, output_format_str = resolve_ugp_output_format(
594            cast("str", self.config.get_value(CONF_UGP_OUTPUT_FORMAT, UGP_OUTPUT_MP3))
595        )
596        headers = {
597            **DEFAULT_STREAM_HEADERS,
598            "contentFeatures.dlna.org": DLNA_CONTENT_FEATURES_REALTIME,
599            "Content-Type": get_mime_type(output_format_str),
600        }
601        resp = web.StreamResponse(status=200, reason="OK", headers=headers)
602        http_profile = self.get_config_value(CONF_HTTP_PROFILE, "chunked")
603        # prefer the configuration of the player that actually renders the audio
604        # (the member's active protocol player when it outputs via a protocol);
605        # child player_id may be stale/invalid, then fall back to the group profile
606        if child_player_id and (child_player := self.mass.players.get_player(child_player_id)):
607            http_profile = child_player.get_output_config_value(CONF_HTTP_PROFILE, http_profile)
608        if http_profile == "chunked" and request.version < HttpVersion11:
609            # chunked encoding is not allowed on HTTP/1.0; fall back to
610            # connection-close streaming to avoid raising in resp.prepare()
611            self.logger.debug(
612                "Disabling chunked encoding for UGP stream to HTTP/1.0 client %s",
613                child_player_id or request.remote,
614            )
615            http_profile = "no_content_length"
616        if http_profile == "forced_content_length":
617            # some clients (notably older Chromecast firmware) refuse to play unless
618            # they see a Content-Length header up front
619            resp.content_length = 4294967296
620        elif http_profile == "chunked":
621            resp.enable_chunked_encoding()
622        await resp.prepare(request)
623
624        # return early if this is not a GET request
625        if request.method != "GET":
626            return resp
627
628        self.logger.debug(
629            "Start serving UGP flow audio stream for UGP-player %s to %s",
630            ugp_player.display_name,
631            child_player_id or request.remote,
632        )
633
634        # Generate filter params for the player specific DSP settings
635        output_plan = None
636        if child_player_id:
637            output_plan = self.mass.streams.audio.get_player_output_plan(
638                child_player_id,
639                self.stream.input_format,
640                output_format,
641                queue_id=self.stream.queue_id,
642                session_id=self.stream.session_id,
643            )
644
645        async for chunk in self.stream.get_stream(
646            output_format,
647            filter_params=output_plan.filter_params if output_plan else None,
648        ):
649            try:
650                await resp.write(chunk)
651            except ConnectionError, ConnectionResetError:
652                break
653
654        return resp
655