jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-25 22:27:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Extract resolve announcement media ID for AssistSatelliteEntity (#134917)

Browse Source
This commit is contained in:
Paulus Schoutsen 2025-01-10 18:33:49 -08:00 committed by GitHub
parent 619dee5d93
commit ab8af033c0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 47 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
homeassistant/components/assist_satellite/entity.py
View File

@ -187,47 +187,10 @@ class AssistSatelliteEntity(entity.Entity):
""" """
await self._cancel_running_pipeline() await self._cancel_running_pipeline()
media_id_source: Literal["url", "media_id", "tts"] | None = None
if message is None: if message is None:
message = "" message = ""
if not media_id: announcement = await self._resolve_announcement_media_id(message, media_id)
media_id_source = "tts"
# Synthesize audio and get URL
pipeline_id = self._resolve_pipeline()
pipeline = async_get_pipeline(self.hass, pipeline_id)
tts_options: dict[str, Any] = {}
if pipeline.tts_voice is not None:
tts_options[tts.ATTR_VOICE] = pipeline.tts_voice
if self.tts_options is not None:
tts_options.update(self.tts_options)
media_id = tts_generate_media_source_id(
self.hass,
message,
engine=pipeline.tts_engine,
language=pipeline.tts_language,
options=tts_options,
)
if media_source.is_media_source_id(media_id):
if not media_id_source:
media_id_source = "media_id"
media = await media_source.async_resolve_media(
self.hass,
media_id,
None,
)
media_id = media.url
if not media_id_source:
media_id_source = "url"
# Resolve to full URL
media_id = async_process_play_media_url(self.hass, media_id)
if self._is_announcing: if self._is_announcing:
raise SatelliteBusyError raise SatelliteBusyError
@ -237,9 +200,7 @@ class AssistSatelliteEntity(entity.Entity):
try: try:
# Block until announcement is finished # Block until announcement is finished
await self.async_announce( await self.async_announce(announcement)
AssistSatelliteAnnouncement(message, media_id, media_id_source)
)
finally: finally:
self._is_announcing = False self._is_announcing = False
self._set_state(AssistSatelliteState.IDLE) self._set_state(AssistSatelliteState.IDLE)
@ -428,3 +389,48 @@ class AssistSatelliteEntity(entity.Entity):
vad_sensitivity = vad.VadSensitivity(vad_sensitivity_state.state) vad_sensitivity = vad.VadSensitivity(vad_sensitivity_state.state)
return vad.VadSensitivity.to_seconds(vad_sensitivity) return vad.VadSensitivity.to_seconds(vad_sensitivity)
async def _resolve_announcement_media_id(
self, message: str, media_id: str | None
) -> AssistSatelliteAnnouncement:
"""Resolve the media ID."""
media_id_source: Literal["url", "media_id", "tts"] | None = None
if not media_id:
media_id_source = "tts"
# Synthesize audio and get URL
pipeline_id = self._resolve_pipeline()
pipeline = async_get_pipeline(self.hass, pipeline_id)
tts_options: dict[str, Any] = {}
if pipeline.tts_voice is not None:
tts_options[tts.ATTR_VOICE] = pipeline.tts_voice
if self.tts_options is not None:
tts_options.update(self.tts_options)
media_id = tts_generate_media_source_id(
self.hass,
message,
engine=pipeline.tts_engine,
language=pipeline.tts_language,
options=tts_options,
)
if media_source.is_media_source_id(media_id):
if not media_id_source:
media_id_source = "media_id"
media = await media_source.async_resolve_media(
self.hass,
media_id,
None,
)
media_id = media.url
if not media_id_source:
media_id_source = "url"
# Resolve to full URL
media_id = async_process_play_media_url(self.hass, media_id)
return AssistSatelliteAnnouncement(message, media_id, media_id_source)