import re
import struct
from collections.abc import AsyncGenerator
+from contextlib import suppress
from io import BytesIO
from time import time
from typing import TYPE_CHECKING
import aiofiles
from aiohttp import ClientTimeout
-from music_assistant.common.models.errors import AudioError, MediaNotFoundError, MusicAssistantError
+from music_assistant.common.models.errors import (
+ AudioError,
+ InvalidDataError,
+ MediaNotFoundError,
+ MusicAssistantError,
+)
from music_assistant.common.models.media_items import (
AudioFormat,
ContentType,
CONF_VOLUME_NORMALIZATION_TARGET,
ROOT_LOGGER_NAME,
)
+from music_assistant.server.helpers.playlists import fetch_playlist
from .process import AsyncProcess, check_output
from .util import create_tempfile
mass.create_task(analyze_audio(mass, streamdetails))
+async def resolve_radio_stream(mass: MusicAssistant, url: str) -> tuple[str, bool]:
+ """
+ Resolve a streaming radio URL.
+
+ Unwraps any playlists if needed.
+ Determines if the stream supports ICY metadata.
+
+ Returns unfolded URL and a bool if the URL supports ICY metadata.
+ """
+ cache_key = f"resolved_radio_url_{url}"
+ if cache := await mass.cache.get(cache_key):
+ return cache
+ # handle playlisted radio urls
+ is_mpeg_dash = False
+ supports_icy = False
+ if ".m3u" in url or ".pls" in url:
+ # url is playlist, try to figure out how to handle it
+ with suppress(InvalidDataError, IndexError):
+ playlist = await fetch_playlist(mass, url)
+ if len(playlist) > 1 or ".m3u" in playlist[0] or ".pls" in playlist[0]:
+ # if it is an mpeg-dash stream, let ffmpeg handle that
+ is_mpeg_dash = True
+ url = playlist[0]
+ if not is_mpeg_dash:
+ # determine ICY metadata support by looking at the http headers
+ headers = {"Icy-MetaData": "1", "User-Agent": "VLC/3.0.2.LibVLC/3.0.2"}
+ timeout = ClientTimeout(total=0, connect=10, sock_read=5)
+ async with mass.http_session.head(
+ url, headers=headers, allow_redirects=True, timeout=timeout
+ ) as resp:
+ headers = resp.headers
+ supports_icy = int(headers.get("icy-metaint", "0")) > 0
+
+ result = (url, supports_icy)
+ await mass.cache.set(cache_key, result)
+ return result
+
+
async def get_radio_stream(
mass: MusicAssistant, url: str, streamdetails: StreamDetails
) -> AsyncGenerator[bytes, None]:
"""Get radio audio stream from HTTP, including metadata retrieval."""
- headers = {"Icy-MetaData": "1"}
+ headers = {"Icy-MetaData": "1", "User-Agent": "VLC/3.0.2.LibVLC/3.0.2"}
timeout = ClientTimeout(total=0, connect=30, sock_read=60)
async with mass.http_session.get(url, headers=headers, timeout=timeout) as resp:
headers = resp.headers
from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import LinkType, ProviderFeature
-from music_assistant.common.models.errors import InvalidDataError
from music_assistant.common.models.media_items import (
AudioFormat,
BrowseFolder,
SearchResults,
StreamDetails,
)
-from music_assistant.server.helpers.audio import get_radio_stream
-from music_assistant.server.helpers.playlists import fetch_playlist
+from music_assistant.server.helpers.audio import get_radio_stream, resolve_radio_stream
from music_assistant.server.models.music_provider import MusicProvider
SUPPORTED_FEATURES = (ProviderFeature.SEARCH, ProviderFeature.BROWSE)
async def get_stream_details(self, item_id: str) -> StreamDetails:
"""Get streamdetails for a radio station."""
stream = await self.radios.station(uuid=item_id)
- url_resolved = stream.url_resolved
await self.radios.station_click(uuid=item_id)
- direct = None
- if ".m3u" in url_resolved or ".pls" in url_resolved:
- # url is playlist, try to figure out how to handle it
- # if it is an mpeg-dash stream, let ffmpeg handle that
- try:
- playlist = await fetch_playlist(self.mass, url_resolved)
- if len(playlist) > 1 or ".m3u" in playlist[0] or ".pls" in playlist[0]:
- direct = playlist[0]
- elif playlist:
- url_resolved = playlist[0]
- except (InvalidDataError, IndexError):
- # empty playlist ?!
- direct = url_resolved
+ url_resolved, supports_icy = await resolve_radio_stream(self.mass, stream.url_resolved)
return StreamDetails(
provider=self.domain,
item_id=item_id,
),
media_type=MediaType.RADIO,
data=url_resolved,
- direct=direct,
+ direct=url_resolved if not supports_icy else None,
expires=time() + 24 * 3600,
)
StreamDetails,
)
from music_assistant.constants import CONF_USERNAME
-from music_assistant.server.helpers.audio import get_radio_stream
-from music_assistant.server.helpers.playlists import fetch_playlist
+from music_assistant.server.helpers.audio import get_radio_stream, resolve_radio_stream
from music_assistant.server.helpers.tags import parse_tags
from music_assistant.server.models.music_provider import MusicProvider
if stream["media_type"] != media_type:
continue
# check if the radio stream is not a playlist
- url = stream["url"]
- direct = None
- direct = None
- if ".m3u" in url or ".pls" in url or stream.get("playlist_type"):
- # url is playlist, try to figure out how to handle it
- # if it is an mpeg-dash stream, let ffmpeg handle that
- try:
- playlist = await fetch_playlist(self.mass, url)
- if len(playlist) > 1 or ".m3u" in playlist[0] or ".pls" in playlist[0]:
- direct = playlist[0]
- elif playlist:
- url_resolved = playlist[0]
- except (InvalidDataError, IndexError):
- # empty playlist ?!
- direct = url_resolved
+ url_resolved, supports_icy = await resolve_radio_stream(self.mass, stream["url"])
return StreamDetails(
provider=self.domain,
item_id=item_id,
content_type=ContentType(stream["media_type"]),
),
media_type=MediaType.RADIO,
- data=url,
+ data=url_resolved,
expires=time() + 24 * 3600,
- direct=direct,
+ direct=url_resolved if not supports_icy else None,
)
raise MediaNotFoundError(f"Unable to retrieve stream details for {item_id}")
StreamDetails,
Track,
)
-from music_assistant.server.helpers.audio import get_file_stream, get_http_stream, get_radio_stream
+from music_assistant.server.helpers.audio import (
+ get_file_stream,
+ get_http_stream,
+ get_radio_stream,
+ resolve_radio_stream,
+)
from music_assistant.server.helpers.playlists import fetch_playlist
from music_assistant.server.helpers.tags import AudioTags, parse_tags
from music_assistant.server.models.music_provider import MusicProvider
"""Get streamdetails for a track/radio."""
item_id, url, media_info = await self._get_media_info(item_id)
is_radio = media_info.get("icy-name") or not media_info.duration
- # we let ffmpeg handle with mpeg dash streams
- mpeg_dash_stream = ".m3u" in url or ".pls" in url
+ if is_radio:
+ url, supports_icy = await resolve_radio_stream(self.mass, url)
return StreamDetails(
provider=self.instance_id,
item_id=item_id,
bit_depth=media_info.bits_per_sample,
),
media_type=MediaType.RADIO if is_radio else MediaType.TRACK,
- direct=None if is_radio and not mpeg_dash_stream else url,
+ direct=None if is_radio and supports_icy else url,
data=url,
)
projects / music-assistant-server.git / commitdiff