Pass satellite id through assist pipeline (home-assistant#151992)

Author: arturpragacz

homeassistant/components/assist_pipeline/__init__.py

@@ -103,6 +103,7 @@ async def async_pipeline_from_audio_stream(
     wake_word_settings: WakeWordSettings | None = None,
     audio_settings: AudioSettings | None = None,
     device_id: str | None = None,
+    satellite_id: str | None = None,
     start_stage: PipelineStage = PipelineStage.STT,
     end_stage: PipelineStage = PipelineStage.TTS,
     conversation_extra_system_prompt: str | None = None,
@@ -115,6 +116,7 @@ async def async_pipeline_from_audio_stream(
         pipeline_input = PipelineInput(
             session=session,
             device_id=device_id,
+            satellite_id=satellite_id,
             stt_metadata=stt_metadata,
             stt_stream=stt_stream,
             wake_word_phrase=wake_word_phrase,

homeassistant/components/assist_pipeline/pipeline.py

@@ -583,6 +583,9 @@ class PipelineRun:
     _device_id: str | None = None
     """Optional device id set during run start."""
 
+    _satellite_id: str | None = None
+    """Optional satellite id set during run start."""
+
     _conversation_data: PipelineConversationData | None = None
     """Data tied to the conversation ID."""
 
@@ -636,16 +639,21 @@ def process_event(self, event: PipelineEvent) -> None:
             return
         pipeline_data.pipeline_debug[self.pipeline.id][self.id].events.append(event)
 
-    def start(self, conversation_id: str, device_id: str | None) -> None:
+    def start(
+        self, conversation_id: str, device_id: str | None, satellite_id: str | None
+    ) -> None:
         """Emit run start event."""
         self._device_id = device_id
+        self._satellite_id = satellite_id
         self._start_debug_recording_thread()
 
         data: dict[str, Any] = {
             "pipeline": self.pipeline.id,
             "language": self.language,
             "conversation_id": conversation_id,
         }
+        if satellite_id is not None:
+            data["satellite_id"] = satellite_id
         if self.runner_data is not None:
             data["runner_data"] = self.runner_data
         if self.tts_stream:
@@ -1057,7 +1065,6 @@ async def recognize_intent(
         self,
         intent_input: str,
         conversation_id: str,
-        device_id: str | None,
         conversation_extra_system_prompt: str | None,
     ) -> str:
         """Run intent recognition portion of pipeline. Returns text to speak."""
@@ -1088,7 +1095,8 @@ async def recognize_intent(
                     "language": input_language,
                     "intent_input": intent_input,
                     "conversation_id": conversation_id,
-                    "device_id": device_id,
+                    "device_id": self._device_id,
+                    "satellite_id": self._satellite_id,
                     "prefer_local_intents": self.pipeline.prefer_local_intents,
                 },
             )
@@ -1099,7 +1107,8 @@ async def recognize_intent(
                 text=intent_input,
                 context=self.context,
                 conversation_id=conversation_id,
-                device_id=device_id,
+                device_id=self._device_id,
+                satellite_id=self._satellite_id,
                 language=input_language,
                 agent_id=self.intent_agent.id,
                 extra_system_prompt=conversation_extra_system_prompt,
@@ -1269,6 +1278,7 @@ async def tts_input_stream_generator() -> AsyncGenerator[str]:
                         text=user_input.text,
                         conversation_id=user_input.conversation_id,
                         device_id=user_input.device_id,
+                        satellite_id=user_input.satellite_id,
                         context=user_input.context,
                         language=user_input.language,
                         agent_id=user_input.agent_id,
@@ -1567,10 +1577,15 @@ class PipelineInput:
     device_id: str | None = None
     """Identifier of the device that is processing the input/output of the pipeline."""
 
+    satellite_id: str | None = None
+    """Identifier of the satellite that is processing the input/output of the pipeline."""
+
     async def execute(self) -> None:
         """Run pipeline."""
         self.run.start(
-            conversation_id=self.session.conversation_id, device_id=self.device_id
+            conversation_id=self.session.conversation_id,
+            device_id=self.device_id,
+            satellite_id=self.satellite_id,
         )
         current_stage: PipelineStage | None = self.run.start_stage
         stt_audio_buffer: list[EnhancedAudioChunk] = []
@@ -1656,7 +1671,6 @@ async def buffer_then_audio_stream() -> AsyncGenerator[
                     tts_input = await self.run.recognize_intent(
                         intent_input,
                         self.session.conversation_id,
-                        self.device_id,
                         self.conversation_extra_system_prompt,
                     )
                     if tts_input.strip():

homeassistant/components/assist_satellite/entity.py

@@ -522,6 +522,7 @@ async def async_accept_pipeline_from_satellite(
                         pipeline_id=self._resolve_pipeline(),
                         conversation_id=session.conversation_id,
                         device_id=device_id,
+                        satellite_id=self.entity_id,
                         tts_audio_output=self.tts_options,
                         wake_word_phrase=wake_word_phrase,
                         audio_settings=AudioSettings(

homeassistant/components/conversation/agent_manager.py

@@ -71,6 +71,7 @@ async def async_converse(
     language: str | None = None,
     agent_id: str | None = None,
     device_id: str | None = None,
+    satellite_id: str | None = None,
     extra_system_prompt: str | None = None,
 ) -> ConversationResult:
     """Process text and get intent."""
@@ -97,6 +98,7 @@ async def async_converse(
         context=context,
         conversation_id=conversation_id,
         device_id=device_id,
+        satellite_id=satellite_id,
         language=language,
         agent_id=agent_id,
         extra_system_prompt=extra_system_prompt,

homeassistant/components/conversation/default_agent.py

@@ -470,6 +470,7 @@ async def _async_process_intent_result(
                 language,
                 assistant=DOMAIN,
                 device_id=user_input.device_id,
+                satellite_id=user_input.satellite_id,
                 conversation_agent_id=user_input.agent_id,
             )
         except intent.MatchFailedError as match_error:

homeassistant/components/conversation/http.py

@@ -201,6 +201,7 @@ async def websocket_hass_agent_debug(
             context=connection.context(msg),
             conversation_id=None,
             device_id=msg.get("device_id"),
+            satellite_id=None,
             language=msg.get("language", hass.config.language),
             agent_id=agent.entity_id,
         )

homeassistant/components/conversation/models.py

@@ -37,6 +37,9 @@ class ConversationInput:
     device_id: str | None
     """Unique identifier for the device."""
 
+    satellite_id: str | None
+    """Unique identifier for the satellite."""
+
     language: str
     """Language of the request."""
 
@@ -53,6 +56,7 @@ def as_dict(self) -> dict[str, Any]:
             "context": self.context.as_dict(),
             "conversation_id": self.conversation_id,
             "device_id": self.device_id,
+            "satellite_id": self.satellite_id,
             "language": self.language,
             "agent_id": self.agent_id,
             "extra_system_prompt": self.extra_system_prompt,

homeassistant/components/conversation/trigger.py

@@ -100,6 +100,7 @@ async def call_action(
                 entity_name: entity["value"] for entity_name, entity in details.items()
             },
             "device_id": user_input.device_id,
+            "satellite_id": user_input.satellite_id,
             "user_input": user_input.as_dict(),
         }
 

homeassistant/helpers/intent.py

@@ -114,6 +114,7 @@ async def async_handle(
     language: str | None = None,
     assistant: str | None = None,
     device_id: str | None = None,
+    satellite_id: str | None = None,
     conversation_agent_id: str | None = None,
 ) -> IntentResponse:
     """Handle an intent."""
@@ -138,6 +139,7 @@ async def async_handle(
         language=language,
         assistant=assistant,
         device_id=device_id,
+        satellite_id=satellite_id,
         conversation_agent_id=conversation_agent_id,
     )
 
@@ -1276,6 +1278,7 @@ class Intent:
         "intent_type",
         "language",
         "platform",
+        "satellite_id",
         "slots",
         "text_input",
     ]
@@ -1291,6 +1294,7 @@ def __init__(
         language: str,
         assistant: str | None = None,
         device_id: str | None = None,
+        satellite_id: str | None = None,
         conversation_agent_id: str | None = None,
     ) -> None:
         """Initialize an intent."""
@@ -1303,6 +1307,7 @@ def __init__(
         self.language = language
         self.assistant = assistant
         self.device_id = device_id
+        self.satellite_id = satellite_id
         self.conversation_agent_id = conversation_agent_id
 
     @callback

tests/components/assist_pipeline/snapshots/test_init.ambr

@@ -45,6 +45,7 @@
         'intent_input': 'test transcript',
         'language': 'en',
         'prefer_local_intents': False,
+        'satellite_id': None,
       }),
       'type': <PipelineEventType.INTENT_START: 'intent-start'>,
     }),
@@ -145,6 +146,7 @@
         'intent_input': 'test transcript',
         'language': 'en-US',
         'prefer_local_intents': False,
+        'satellite_id': None,
       }),
       'type': <PipelineEventType.INTENT_START: 'intent-start'>,
     }),
@@ -245,6 +247,7 @@
         'intent_input': 'test transcript',
         'language': 'en-US',
         'prefer_local_intents': False,
+        'satellite_id': None,
       }),
       'type': <PipelineEventType.INTENT_START: 'intent-start'>,
     }),
@@ -369,6 +372,7 @@
         'intent_input': 'test transcript',
         'language': 'en',
         'prefer_local_intents': False,
+        'satellite_id': None,
       }),
       'type': <PipelineEventType.INTENT_START: 'intent-start'>,
     }),