Bugfixes for Airplay and HLS streams (#1731)

music_assistant/server/controllers/config.py

@@ -394,6 +394,7 @@ async def save_player_config(
  object_id=config.player_id,
  data=config,
  )
+ self.mass.players.update(config.player_id, force_update=True)
  # return full player config (just in case)
  return await self.get_player_config(player_id)
 

music_assistant/server/controllers/player_queues.py

@@ -309,6 +309,7 @@ def set_repeat(self, queue_id: str, repeat_mode: RepeatMode) -> None:
  self.mass.create_task(self.resume(queue_id))
  else:
  task_id = f"enqueue_next_{queue_id}"
+ self.logger.info("Repeat mode detected, enqueue next item")
  self.mass.call_later(2, self._enqueue_next, queue, queue.current_index, task_id=task_id)
 
  @api_command("player_queues/play_media")

music_assistant/server/controllers/players.py

@@ -275,7 +275,7 @@ async def cmd_seek(self, player_id: str, position: int) -> None:
  if PlayerFeature.SEEK not in player.supported_features:
  msg = f"Player {player.display_name} does not support seeking"
  raise UnsupportedFeaturedException(msg)
- player_prov = self.mass.players.get_player_provider(player.player_id)
+ player_prov = self.get_player_provider(player.player_id)
  await player_prov.cmd_seek(player.player_id, position)
 
  @api_command("players/cmd/next")
@@ -290,7 +290,7 @@ async def cmd_next_track(self, player_id: str) -> None:
  if PlayerFeature.NEXT_PREVIOUS not in player.supported_features:
  msg = f"Player {player.display_name} does not support skipping to the next track."
  raise UnsupportedFeaturedException(msg)
- player_prov = self.mass.players.get_player_provider(player.player_id)
+ player_prov = self.get_player_provider(player.player_id)
  await player_prov.cmd_next(player.player_id)
 
  @api_command("players/cmd/previous")
@@ -305,7 +305,7 @@ async def cmd_previous_track(self, player_id: str) -> None:
  if PlayerFeature.NEXT_PREVIOUS not in player.supported_features:
  msg = f"Player {player.display_name} does not support skipping to the previous track."
  raise UnsupportedFeaturedException(msg)
- player_prov = self.mass.players.get_player_provider(player.player_id)
+ player_prov = self.get_player_provider(player.player_id)
  await player_prov.cmd_previous(player.player_id)
 
  @api_command("players/cmd/power")
@@ -565,15 +565,15 @@ async def play_media(self, player_id: str, media: PlayerMedia) -> None:
  # power on the player if needed
  if not player.powered:
  await self.cmd_power(player.player_id, True)
- player_prov = self.mass.players.get_player_provider(player.player_id)
+ player_prov = self.get_player_provider(player.player_id)
  await player_prov.play_media(
  player_id=player.player_id,
  media=media,
  )
 
  async def enqueue_next_media(self, player_id: str, media: PlayerMedia) -> None:
  """Handle enqueuing of a next media item on the player."""
- player_prov = self.mass.players.get_player_provider(player_id)
+ player_prov = self.get_player_provider(player_id)
  async with self._player_throttlers[player_id]:
  await player_prov.enqueue_next_media(player_id=player_id, media=media)
 
@@ -1058,7 +1058,7 @@ async def on_player_config_change(self, config: PlayerConfig, changed_keys: set[
  if player_provider := self.mass.get_provider(config.provider):
  with suppress(PlayerUnavailableError):
  await player_provider.on_player_config_change(config, changed_keys)
- if not (player := self.mass.players.get(config.player_id)):
+ if not (player := self.get(config.player_id)):
  return
  if player_disabled:
  # edge case: ensure that the player is powered off if the player gets disabled
@@ -1070,14 +1070,13 @@ async def on_player_config_change(self, config: PlayerConfig, changed_keys: set[
  # check for group memberships that need to be updated
  if player_disabled and player.active_group and player_provider:
  # try to remove from the group
- group_player = self.mass.players.get(player.active_group)
+ group_player = self.get(player.active_group)
  with suppress(UnsupportedFeaturedException, PlayerCommandFailed):
  await player_provider.set_members(
  player.active_group,
  [x for x in group_player.group_childs if x != player.player_id],
  )
  player.enabled = config.enabled
- self.mass.players.update(config.player_id, force_update=True)
 
  async def _play_announcement(
  self,

music_assistant/server/controllers/streams.py

@@ -58,7 +58,6 @@
  check_audio_support,
  crossfade_pcm_parts,
  get_chunksize,
- get_hls_radio_stream,
  get_hls_substream,
  get_icy_radio_stream,
  get_media_stream,
@@ -836,21 +835,21 @@ async def get_media_stream(
  )
  elif streamdetails.stream_type == StreamType.ICY:
  audio_source = get_icy_radio_stream(self.mass, streamdetails.path, streamdetails)
- elif streamdetails.stream_type == StreamType.HLS:
+ elif streamdetails.stream_type in (StreamType.HLS, StreamType.ENCRYPTED_HLS):
+ substream = await get_hls_substream(self.mass, streamdetails.path)
+ audio_source = substream.path
  if streamdetails.media_type == MediaType.RADIO:
  # Especially the BBC streams struggle when they're played directly
- # with ffmpeg, so we use our own HLS stream parser/logic
- audio_source = get_hls_radio_stream(self.mass, streamdetails.path, streamdetails)
- else:
- # normal tracks we just let ffmpeg deal with it
- substream = await get_hls_substream(self.mass, streamdetails.path)
- audio_source = substream.path
- elif streamdetails.stream_type == StreamType.ENCRYPTED_HTTP:
- audio_source = streamdetails.path
- extra_input_args += ["-decryption_key", streamdetails.decryption_key]
+ # with ffmpeg, where they just stop after some minutes,
+ # so we tell ffmpeg to loop around in this case.
+ extra_input_args += ["-stream_loop", "-1", "-re"]
  else:
  audio_source = streamdetails.path
 
+ # add support for decryption key provided in streamdetails
+ if streamdetails.decryption_key:
+ extra_input_args += ["-decryption_key", streamdetails.decryption_key]
+
  # handle seek support
  if (
  streamdetails.seek_position

music_assistant/server/helpers/audio.py

@@ -570,51 +570,9 @@ async def get_icy_radio_stream(
  streamdetails.stream_title = cleaned_stream_title
 
 
-async def get_hls_radio_stream(
- mass: MusicAssistant,
- url: str,
- streamdetails: StreamDetails,
-) -> AsyncGenerator[bytes, None]:
- """Get radio audio stream from HTTP HLS playlist."""
- logger = LOGGER.getChild("hls_stream")
- logger.debug("Start streaming HLS stream for url %s", url)
- # we simply select the best quality substream here
- # if we ever want to support adaptive stream selection based on bandwidth
- # we need to move the substream selection into the loop below and make it
- # bandwidth aware. For now we just assume domestic high bandwidth where
- # the user wants the best quality possible at all times.
- playlist_item = await get_hls_substream(mass, url)
- substream_url = playlist_item.path
- loops = 50 if streamdetails.media_type != MediaType.RADIO else 1
- while loops:
- logger.log(VERBOSE_LOG_LEVEL, "start streaming chunks from substream %s", substream_url)
- # We simply let ffmpeg deal with parsing the HLS playlist and stichting chunks together.
- # However we do not feed the playlist URL to ffmpeg directly to give us the possibility
- # to monitor the stream title and other metadata for radio streams in the future.
- # Also, we've seen cases where ffmpeg sometimes chokes in a stream and aborts, which is not
- # very useful for radio streams which you want to simply go on forever, so we need to loop
- # and restart ffmpeg in case of an error.
- input_format = AudioFormat(content_type=ContentType.UNKNOWN)
- audio_format_detected = False
- async for chunk in get_ffmpeg_stream(
- audio_input=substream_url,
- input_format=input_format,
- output_format=AudioFormat(content_type=ContentType.WAV),
- ):
- yield chunk
- if audio_format_detected:
- continue
- if input_format.content_type not in (ContentType.UNKNOWN, ContentType.WAV):
- # we need to determine the audio format from the first chunk
- streamdetails.audio_format = input_format
- audio_format_detected = True
- loops -= 1
-
-
 async def get_hls_substream(
  mass: MusicAssistant,
  url: str,
- allow_encrypted: bool = False,
 ) -> PlaylistItem:
  """Select the (highest quality) HLS substream for given HLS playlist/URL."""
  timeout = ClientTimeout(total=0, connect=30, sock_read=5 * 60)
@@ -627,9 +585,6 @@ async def get_hls_substream(
  raw_data = await resp.read()
  encoding = resp.charset or await detect_charset(raw_data)
  master_m3u_data = raw_data.decode(encoding)
- if not allow_encrypted and "EXT-X-KEY:METHOD=" in master_m3u_data:
- # for now we do not (yet) support encrypted HLS streams
- raise InvalidDataError("HLS stream is encrypted, not supported")
  substreams = parse_m3u(master_m3u_data)
  # There is a chance that we did not get a master playlist with subplaylists
  # but just a single master/sub playlist with the actual audio stream(s)

music_assistant/server/helpers/ffmpeg.py

@@ -218,16 +218,24 @@ def get_ffmpeg_args(
  if input_path.startswith("http"):
  # append reconnect options for direct stream from http
  input_args += [
- "-reconnect",
+ # Reconnect automatically when disconnected before EOF is hit.
+ "reconnect",
  "1",
- "-reconnect_streamed",
+ # Set the maximum delay in seconds after which to give up reconnecting.
+ "-reconnect_delay_max",
+ "30",
+ # If set then even streamed/non seekable streams will be reconnected on errors.
+ "reconnect_streamed",
  "1",
  ]
  if major_version > 4:
  # these options are only supported in ffmpeg > 5
  input_args += [
+ # Reconnect automatically in case of TCP/TLS errors during connect.
  "-reconnect_on_network_error",
  "1",
+ # A comma separated list of HTTP status codes to reconnect on.
+ # The list can include specific status codes (e.g. 503) or the strings 4xx / 5xx.
  "-reconnect_on_http_error",
  "5xx,4xx",
  ]