CONF_ANNOUNCE_VOLUME: Final[str] = "announce_volume"
CONF_ANNOUNCE_VOLUME_MIN: Final[str] = "announce_volume_min"
CONF_ANNOUNCE_VOLUME_MAX: Final[str] = "announce_volume_max"
+CONF_PRE_ANNOUNCE_CHIME_URL: Final[str] = "pre_announcement_chime_url"
CONF_ICON: Final[str] = "icon"
CONF_LANGUAGE: Final[str] = "language"
CONF_SAMPLE_RATES: Final[str] = "sample_rates"
CONF_ENTRY_ANNOUNCE_VOLUME_MAX_HIDDEN = ConfigEntry.from_dict(
{**CONF_ENTRY_ANNOUNCE_VOLUME_MAX.to_dict(), "hidden": True}
)
+
+
HIDDEN_ANNOUNCE_VOLUME_CONFIG_ENTRIES = (
CONF_ENTRY_ANNOUNCE_VOLUME_HIDDEN,
CONF_ENTRY_ANNOUNCE_VOLUME_MIN_HIDDEN,
from music_assistant_models.player_control import PlayerControl # noqa: TC002
from music_assistant.constants import (
+ ANNOUNCE_ALERT_FILE,
ATTR_ANNOUNCEMENT_IN_PROGRESS,
ATTR_FAKE_MUTE,
ATTR_FAKE_POWER,
CONF_ENTRY_ANNOUNCE_VOLUME_MAX,
CONF_ENTRY_ANNOUNCE_VOLUME_MIN,
CONF_ENTRY_ANNOUNCE_VOLUME_STRATEGY,
+ CONF_ENTRY_TTS_PRE_ANNOUNCE,
CONF_PLAYER_DSP,
CONF_PLAYERS,
- CONF_TTS_PRE_ANNOUNCE,
+ CONF_PRE_ANNOUNCE_CHIME_URL,
)
+from music_assistant.controllers.streams import AnnounceData
from music_assistant.helpers.api import api_command
from music_assistant.helpers.tags import async_parse_tags
from music_assistant.helpers.throttle_retry import Throttler
-from music_assistant.helpers.util import TaskManager
+from music_assistant.helpers.util import TaskManager, validate_announcement_chime_url
from music_assistant.models.core_controller import CoreController
from music_assistant.models.player import Player, PlayerMedia, PlayerState
from music_assistant.models.player_provider import PlayerProvider
self,
player_id: str,
url: str,
- use_pre_announce: bool | None = None,
+ pre_announce: bool | str | None = None,
volume_level: int | None = None,
+ pre_announce_url: str | None = None,
) -> None:
- """Handle playback of an announcement (url) on given player."""
+ """
+ Handle playback of an announcement (url) on given player.
+
+ - player_id: player_id of the player to handle the command.
+ - url: URL of the announcement to play.
+ - pre_announce: optional bool if pre-announce should be used.
+ - volume_level: optional volume level to set for the announcement.
+ - pre_announce_url: optional custom URL to use for the pre-announce chime.
+ """
player = self.get(player_id, True)
assert player is not None # for type checking
if not url.startswith("http"):
raise PlayerCommandFailed("Only URLs are supported for announcements")
+ if (
+ pre_announce
+ and pre_announce_url
+ and not validate_announcement_chime_url(pre_announce_url)
+ ):
+ raise PlayerCommandFailed("Invalid pre-announce chime URL specified.")
# prevent multiple announcements at the same time to the same player with a lock
if player_id not in self._announce_locks:
self._announce_locks[player_id] = lock = asyncio.Lock()
PlayerFeature.PLAY_ANNOUNCEMENT in player.supported_features
)
# determine pre-announce from (group)player config
- if use_pre_announce is None and "tts" in url:
- use_pre_announce = await self.mass.config.get_player_config_value(
+ if pre_announce is None and "tts" in url:
+ conf_pre_announce = self.mass.config.get_raw_player_config_value(
player_id,
- CONF_TTS_PRE_ANNOUNCE,
+ CONF_ENTRY_TTS_PRE_ANNOUNCE.key,
+ CONF_ENTRY_TTS_PRE_ANNOUNCE.default_value,
)
+ pre_announce = cast("bool", conf_pre_announce)
+ if pre_announce_url is None:
+ if conf_pre_announce_url := self.mass.config.get_raw_player_config_value(
+ player_id,
+ CONF_PRE_ANNOUNCE_CHIME_URL,
+ ):
+ # player default custom chime url
+ pre_announce_url = cast("str", conf_pre_announce_url)
+ else:
+ # use global default chime url
+ pre_announce_url = ANNOUNCE_ALERT_FILE
# if player type is group with all members supporting announcements,
# we forward the request to each individual player
if player.type == PlayerType.GROUP and (
self.play_announcement(
group_member,
url=url,
- use_pre_announce=use_pre_announce,
+ pre_announce=pre_announce,
volume_level=volume_level,
+ pre_announce_url=pre_announce_url,
)
)
return
self.logger.info(
"Playback announcement to player %s (with pre-announce: %s): %s",
player.display_name,
- use_pre_announce,
+ pre_announce,
url,
)
# create a PlayerMedia object for the announcement so
# we can send a regular play-media call downstream
+ announce_data = AnnounceData(
+ announcement_url=url,
+ pre_announce=pre_announce,
+ pre_announce_url=pre_announce_url,
+ )
announcement = PlayerMedia(
- uri=self.mass.streams.get_announcement_url(player_id, url, use_pre_announce),
+ uri=self.mass.streams.get_announcement_url(player_id, url, announce_data),
media_type=MediaType.ANNOUNCEMENT,
title="Announcement",
- custom_data={"url": url, "use_pre_announce": use_pre_announce},
+ custom_data=announce_data,
)
# handle native announce support
if native_announce_support:
import urllib.parse
from collections.abc import AsyncGenerator
from dataclasses import dataclass, field
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, TypedDict
from aiofiles.os import wrap
from aiohttp import web
get_free_space_percentage,
get_ip_addresses,
select_free_port,
- try_parse_bool,
)
from music_assistant.helpers.webserver import Webserver
from music_assistant.models.core_controller import CoreController
from music_assistant_models.player_queue import PlayerQueue
from music_assistant_models.queue_item import QueueItem
+ from music_assistant.mass import MusicAssistant
from music_assistant.models.player import Player
session_id: str | None = None
+class AnnounceData(TypedDict):
+ """Announcement data."""
+
+ announcement_url: str
+ pre_announce: bool
+ pre_announce_url: str
+
+
class StreamsController(CoreController):
"""Webserver Controller to stream audio to players."""
domain: str = "streams"
- def __init__(self, *args, **kwargs) -> None:
+ def __init__(self, mass: MusicAssistant) -> None:
"""Initialize instance."""
- super().__init__(*args, **kwargs)
+ super().__init__(mass)
self._server = Webserver(self.logger, enable_dynamic_routes=True)
self.register_dynamic_route = self._server.register_dynamic_route
self.unregister_dynamic_route = self._server.unregister_dynamic_route
"streaming audio to players on the local network."
)
self.manifest.icon = "cast-audio"
- self.announcements: dict[str, str] = {}
+ self.announcements: dict[str, AnnounceData] = {}
# prefer /tmp/.audio as audio cache dir
self._audio_cache_dir = os.path.join("/tmp/.audio") # noqa: S108
self.allow_cache_default = "auto"
# crossfade is not supported on this player due to missing gapless playback
self.logger.warning(
"Crossfade disabled: Player %s does not support gapless playback",
- queue_player.display_name,
+ queue_player.display_name if queue_player else "Unknown Player",
)
crossfade = False
reason="OK",
headers=headers,
)
- http_profile: str = await self.mass.config.get_player_config_value(
+ http_profile_value = await self.mass.config.get_player_config_value(
queue_id, CONF_HTTP_PROFILE
)
+ http_profile = str(http_profile_value) if http_profile_value is not None else "default"
if http_profile == "forced_content_length":
# just set an insane high content length to make sure the player keeps playing
resp.content_length = get_chunksize(output_format, 12 * 3600)
player = self.mass.player_queues.get(player_id)
if not player:
raise web.HTTPNotFound(reason=f"Unknown Player: {player_id}")
- if player_id not in self.announcements:
+ if not (announce_data := self.announcements.get(player_id)):
raise web.HTTPNotFound(reason=f"No pending announcements for Player: {player_id}")
- announcement_url = self.announcements[player_id]
- use_pre_announce = try_parse_bool(request.query.get("pre_announce"))
# work out output format/details
- fmt = request.match_info.get("fmt", announcement_url.rsplit(".")[-1])
+ fmt = request.match_info["fmt"]
audio_format = AudioFormat(content_type=ContentType.try_parse(fmt))
- http_profile: str = await self.mass.config.get_player_config_value(
+ http_profile_value = await self.mass.config.get_player_config_value(
player_id, CONF_HTTP_PROFILE
)
+ http_profile = str(http_profile_value) if http_profile_value is not None else "default"
if http_profile == "forced_content_length":
# given the fact that an announcement is just a short audio clip,
# just send it over completely at once so we have a fixed content length
data = b""
async for chunk in self.get_announcement_stream(
- announcement_url=announcement_url,
+ announcement_url=announce_data["announcement_url"],
output_format=audio_format,
- use_pre_announce=use_pre_announce,
+ pre_announce=announce_data["pre_announce"],
+ pre_announce_url=announce_data["pre_announce_url"],
):
data += chunk
return web.Response(
# all checks passed, start streaming!
self.logger.debug(
"Start serving audio stream for Announcement %s to %s",
- announcement_url,
+ announce_data["announcement_url"],
player.display_name,
)
async for chunk in self.get_announcement_stream(
- announcement_url=announcement_url,
+ announcement_url=announce_data["announcement_url"],
output_format=audio_format,
- use_pre_announce=use_pre_announce,
+ pre_announce=announce_data["pre_announce"],
+ pre_announce_url=announce_data["pre_announce_url"],
):
try:
await resp.write(chunk)
self.logger.debug(
"Finished serving audio stream for Announcement %s to %s",
- announcement_url,
+ announce_data["announcement_url"],
player.display_name,
)
output_format = await self.get_output_format(
output_format_str=request.match_info["fmt"],
player=player,
- content_sample_rate=plugin_source.audio_format.sample_rate,
- content_bit_depth=plugin_source.audio_format.bit_depth,
+ content_sample_rate=plugin_source.audio_format.sample_rate
+ if plugin_source.audio_format
+ else 44100,
+ content_bit_depth=plugin_source.audio_format.bit_depth
+ if plugin_source.audio_format
+ else 16,
)
headers = {
**DEFAULT_STREAM_HEADERS,
headers=headers,
)
resp.content_type = f"audio/{output_format.output_format_str}"
- http_profile: str = await self.mass.config.get_player_config_value(
+ http_profile_value = await self.mass.config.get_player_config_value(
player_id, CONF_HTTP_PROFILE
)
+ http_profile = str(http_profile_value) if http_profile_value is not None else "default"
if http_profile == "forced_content_length":
# guess content length based on duration
resp.content_length = get_chunksize(output_format, 12 * 3600)
def get_announcement_url(
self,
player_id: str,
- announcement_url: str,
- use_pre_announce: bool = False,
+ announce_data: AnnounceData,
content_type: ContentType = ContentType.MP3,
) -> str:
"""Get the url for the special announcement stream."""
- self.announcements[player_id] = announcement_url
+ self.announcements[player_id] = announce_data
# use stream server to host announcement on local network
# this ensures playback on all players, including ones that do not
# like https hosts and it also offers the pre-announce 'bell'
- return f"{self.base_url}/announcement/{player_id}.{content_type.value}?pre_announce={use_pre_announce}" # noqa: E501
+ return f"{self.base_url}/announcement/{player_id}.{content_type.value}"
async def get_queue_flow_stream(
self,
self,
announcement_url: str,
output_format: AudioFormat,
- use_pre_announce: bool = False,
+ pre_announce: bool | str = False,
+ pre_announce_url: str = ANNOUNCE_ALERT_FILE,
) -> AsyncGenerator[bytes, None]:
"""Get the special announcement stream."""
filter_params = ["loudnorm=I=-10:LRA=11:TP=-2"]
- if use_pre_announce:
+ if pre_announce:
# Note: TTS URLs might take a while to load cause the actual data are often generated
# asynchronously by the TTS provider. If we ask ffmpeg to mix the pre-announce, it will
# wait until it reads the TTS data, so the whole stream will be delayed. It is much
# So far players seem to tolerate this, but it might break some player in the future.
async for chunk in get_ffmpeg_stream(
- audio_input=ANNOUNCE_ALERT_FILE,
- input_format=AudioFormat(content_type=ContentType.try_parse(ANNOUNCE_ALERT_FILE)),
+ audio_input=pre_announce_url,
+ input_format=AudioFormat(content_type=ContentType.try_parse(pre_announce_url)),
output_format=output_format,
filter_params=filter_params,
):
) -> AsyncGenerator[bytes, None]:
"""Get the audio stream for a single queue item with crossfade to the next item."""
queue = self.mass.player_queues.get(queue_item.queue_id)
+ if not queue:
+ raise RuntimeError(f"Queue {queue_item.queue_id} not found")
+
streamdetails = queue_item.streamdetails
assert streamdetails
crossfade_duration = self.mass.config.get_raw_player_config_value(
self.logger.debug(
"Start Streaming queue track: %s (%s) for queue %s - crossfade: %s",
- queue_item.streamdetails.uri,
+ queue_item.streamdetails.uri if queue_item.streamdetails else "Unknown URI",
queue_item.name,
queue.display_name,
f"{crossfade_duration} seconds",
if cache_enabled == "disabled":
max_cache_size = 0.001
- def _clean_old_files(foldersize: float):
+ def _clean_old_files(foldersize: float) -> None:
files: list[os.DirEntry] = [x for x in os.scandir(self._audio_cache_dir) if x.is_file()]
files.sort(key=lambda x: x.stat().st_atime)
for _file in files:
return int(100 * float(part) / float(whole))
+def validate_announcement_chime_url(url: str) -> bool:
+ """Validate announcement chime URL format."""
+ if not url or not url.strip():
+ return True # Empty URL is valid
+
+ try:
+ parsed = urlparse(url.strip())
+
+ if parsed.scheme not in ("http", "https"):
+ return False
+
+ if not parsed.netloc:
+ return False
+
+ path_lower = parsed.path.lower()
+ audio_extensions = (".mp3", ".wav", ".flac", ".ogg", ".m4a", ".aac")
+
+ return any(path_lower.endswith(ext) for ext in audio_extensions)
+
+ except Exception:
+ return False
+
+
class TaskManager:
"""
Helper class to run many tasks at once.
CONF_MUTE_CONTROL,
CONF_OUTPUT_CODEC,
CONF_POWER_CONTROL,
+ CONF_PRE_ANNOUNCE_CHIME_URL,
CONF_SAMPLE_RATES,
CONF_VOLUME_CONTROL,
)
-from music_assistant.helpers.util import get_changed_dataclass_values
+from music_assistant.helpers.util import (
+ get_changed_dataclass_values,
+ validate_announcement_chime_url,
+)
if TYPE_CHECKING:
from .player_provider import PlayerProvider
+CONF_ENTRY_PRE_ANNOUNCE_CUSTOM_CHIME_URL = ConfigEntry(
+ key=CONF_PRE_ANNOUNCE_CHIME_URL,
+ type=ConfigEntryType.STRING,
+ label="Custom (pre)announcement chime URL",
+ description="URL to a custom audio file to play before announcements.\n"
+ "Leave empty to use the default chime.\n"
+ "Supports http:// and https:// URLs pointing to "
+ "audio files (.mp3, .wav, .flac, .ogg, .m4a, .aac).\n"
+ "Example: http://homeassistant.local:8123/local/audio/custom_chime.mp3",
+ category="announcements",
+ required=False,
+ depends_on=CONF_ENTRY_TTS_PRE_ANNOUNCE.key,
+ depends_on_value=True,
+ validate=lambda val: validate_announcement_chime_url(cast("str", val)),
+)
BASE_CONFIG_ENTRIES = [
# config entries that are valid for all player types
CONF_ENTRY_OUTPUT_LIMITER,
CONF_ENTRY_VOLUME_NORMALIZATION_TARGET,
CONF_ENTRY_TTS_PRE_ANNOUNCE,
+ CONF_ENTRY_PRE_ANNOUNCE_CUSTOM_CHIME_URL,
CONF_ENTRY_HTTP_PROFILE,
]
assert media.custom_data
input_format = AIRPLAY_PCM_FORMAT
audio_source = self.mass.streams.get_announcement_stream(
- media.custom_data["url"],
+ media.custom_data["announcement_url"],
output_format=AIRPLAY_PCM_FORMAT,
- use_pre_announce=media.custom_data["use_pre_announce"],
+ pre_announce=media.custom_data["pre_announce"],
+ pre_announce_url=media.custom_data["pre_announce_url"],
)
elif media.media_type == MediaType.PLUGIN_SOURCE:
# special case: plugin source stream
input_format = DEFAULT_SNAPCAST_FORMAT
assert announcement.custom_data is not None # for type checking
audio_source = self.mass.streams.get_announcement_stream(
- announcement.custom_data["url"],
+ announcement.custom_data["announcement_url"],
output_format=DEFAULT_SNAPCAST_FORMAT,
- use_pre_announce=announcement.custom_data["use_pre_announce"],
+ pre_announce=announcement.custom_data["pre_announce"],
+ pre_announce_url=announcement.custom_data["pre_announce_url"],
)
# stream the audio, wait for it to finish (play_announcement should return after the
if media.media_type == MediaType.ANNOUNCEMENT:
# special case: stream announcement
audio_source = self.mass.streams.get_announcement_stream(
- media.custom_data["url"],
+ media.custom_data["announcement_url"],
output_format=master_audio_format,
- use_pre_announce=media.custom_data["use_pre_announce"],
+ pre_announce=media.custom_data["pre_announce"],
+ pre_announce_url=media.custom_data["pre_announce_url"],
)
elif media.media_type == MediaType.PLUGIN_SOURCE:
# special case: plugin source stream
if media.media_type == MediaType.ANNOUNCEMENT and media.custom_data:
# special case: stream announcement
audio_source = self.mass.streams.get_announcement_stream(
- media.custom_data["url"],
+ media.custom_data["announcement_url"],
output_format=UGP_FORMAT,
- use_pre_announce=media.custom_data["use_pre_announce"],
+ pre_announce=media.custom_data["pre_announce"],
+ pre_announce_url=media.custom_data["pre_announce_url"],
)
elif media.media_type == MediaType.PLUGIN_SOURCE and media.custom_data:
# special case: plugin source stream
"ifaddr==0.2.0",
"mashumaro==3.16",
"music-assistant-frontend==2.15.4",
- "music-assistant-models==1.1.55",
+ "music-assistant-models==1.1.56",
"mutagen==1.47.0",
"orjson==3.11.3",
"pillow==11.3.0",
lyricsgenius==3.6.5
mashumaro==3.16
music-assistant-frontend==2.15.4
-music-assistant-models==1.1.55
+music-assistant-models==1.1.56
mutagen==1.47.0
orjson==3.11.3
pillow==11.3.0
projects / music-assistant-server.git / commitdiff