signal voice session termination reason to client

zehnm · zehnm · commit a449a097f584 · 2025-12-12T14:29:46.000+01:00
- For abnormal terminations (timeout, remote abort, local abort, error),
  raise specific exceptions from the iterator.
- expose metadata about the end (reason, error) via properties for post-
  loop inspection.
- Exception hierarchy:
  - `VoiceSessionClosed(Exception)` base with optional fields (reason, error).
  - `VoiceSessionTimeout(VoiceSessionClosed)`
  - `VoiceSessionRemoteEnd(VoiceSessionClosed)`
  - `VoiceSessionLocalEnd(VoiceSessionClosed)`
  - `VoiceSessionError(VoiceSessionClosed)`
- End reason type:
  - `VoiceEndReason` Enum: `NORMAL`, `TIMEOUT`, `REMOTE`, `LOCAL`, `ERROR`
diff --git a/examples/voice.py b/examples/voice.py
@@ -6,7 +6,13 @@
 from typing import Any
 
 import ucapi
-from ucapi import AssistantEvent, AssistantEventType, VoiceAssistant
+from ucapi import (
+    AssistantError,
+    AssistantErrorCode,
+    AssistantEvent,
+    AssistantEventType,
+    VoiceAssistant,
+)
 from ucapi.api_definitions import AssistantSttResponse, AssistantTextResponse
 from ucapi.voice_assistant import Attributes as VAAttr
 from ucapi.voice_assistant import (
@@ -18,6 +24,7 @@
     SampleFormat,
     VoiceAssistantEntityOptions,
 )
+from ucapi.voice_stream import VoiceEndReason, VoiceSessionClosed
 
 loop = asyncio.new_event_loop()
 api = ucapi.IntegrationAPI(loop)
@@ -83,34 +90,57 @@ async def on_voice_session(session):
     global session_id
 
     total = 0
-    async for frame in session:  # frame is bytes
-        total += len(frame)
-        # feed frame into your voice assistant / LLM here
-        print(f"Got {len(frame)} bytes of audio data")
-    print(f"Voice stream ended: session={session.session_id}, bytes={total}")
+    try:
+        async for frame in session:  # frame is bytes
+            total += len(frame)
+            # feed frame into your voice assistant / LLM here
+            print(f"Got {len(frame)} bytes of audio data")
+        print(f"Voice stream ended: session={session.session_id}, bytes={total}")
+
+        event = AssistantEvent(
+            type=AssistantEventType.STT_RESPONSE,
+            entity_id=session.entity_id,
+            session_id=session_id,
+            data=AssistantSttResponse(
+                text="I'm just a demo and I don't know what you said."
+            ),
+        )
+        await api.broadcast_assistant_event(event)
 
-    event = AssistantEvent(
-        type=AssistantEventType.STT_RESPONSE,
-        entity_id=session.entity_id,
-        session_id=session_id,
-        data=AssistantSttResponse(
-            text="I'm just a demo and I don't know what you said."
-        ),
-    )
-    await api.broadcast_assistant_event(event)
+        await sleep(1)
+        event = AssistantEvent(
+            type=AssistantEventType.TEXT_RESPONSE,
+            entity_id=session.entity_id,
+            session_id=session_id,
+            data=AssistantTextResponse(
+                success=True, text=f"You have sent {total} bytes of audio data"
+            ),
+        )
+        await api.broadcast_assistant_event(event)
 
-    await sleep(1)
-    event = AssistantEvent(
-        type=AssistantEventType.TEXT_RESPONSE,
-        entity_id=session.entity_id,
-        session_id=session_id,
-        data=AssistantTextResponse(
-            success=True, text=f"You have sent {total} bytes of audio data"
-        ),
-    )
-    await api.broadcast_assistant_event(event)
+        await sleep(1)
+    except VoiceSessionClosed as ex:
+        print(
+            f"Voice stream session {session.session_id} closed (bytes={total})! Reason: {ex.reason}, exception: {ex.error}"
+        )
+        if ex.reason == VoiceEndReason.REMOTE:
+            return  # Remote disconnected
+        event = AssistantEvent(
+            type=AssistantEventType.ERROR,
+            entity_id=session.entity_id,
+            session_id=session_id,
+            data=AssistantError(
+                code=(
+                    AssistantErrorCode.TIMEOUT
+                    if ex.reason == VoiceEndReason.TIMEOUT
+                    else AssistantErrorCode.UNEXPECTED_ERROR
+                ),
+                message=f"Reason: {ex.reason}, exception: {ex.error}",
+            ),
+        )
+        await api.broadcast_assistant_event(event)
 
-    await sleep(1)
+    # final event
     event = AssistantEvent(
         type=AssistantEventType.FINISHED,
         entity_id=session.entity_id,
diff --git a/ucapi/api.py b/ucapi/api.py
@@ -46,7 +46,7 @@
     AudioConfiguration,
 )
 from .voice_assistant import Commands as VaCommands
-from .voice_stream import VoiceSession, VoiceStreamHandler
+from .voice_stream import VoiceEndReason, VoiceSession, VoiceStreamHandler
 
 try:
     from ._version import version as __version__
@@ -559,7 +559,7 @@ async def _cleanup_voice_session(self, session_id: int) -> None:
         # End and remove session
         session = self._voice_sessions.pop(session_id, None)
         if session is not None and not session.closed:
-            session.end()
+            session.end()  # normal close
 
         # Remove ownership mappings
         try:
@@ -609,18 +609,20 @@ async def _run_voice_handler(self, session: VoiceSession) -> None:
         handler = self._voice_handler
         if handler is None:
             # Handler cleared after session start; just end the session
-            session.end()
+            session.end(VoiceEndReason.LOCAL)
             return
         try:
             result = handler(session)
             if asyncio.iscoroutine(result):
                 await result
-        except Exception:  # pylint: disable=W0718
+            if not session.closed:
+                session.end()
+        except Exception as ex:  # pylint: disable=W0718
             _LOG.exception("Voice handler failed for session %s", session.session_id)
+            if not session.closed:
+                session.end(VoiceEndReason.ERROR, ex)
         finally:
             # Ensure iterator is unblocked and session is cleaned up
-            if not session.closed:
-                session.end()
             await self._cleanup_voice_session(session.session_id)
 
     def _schedule_voice_timeout(self, session_id: int) -> None:
@@ -671,7 +673,7 @@ async def _voice_session_timeout_task(self, session_id: int) -> None:
                 )
 
         # End and cleanup
-        session.end()
+        session.end(VoiceEndReason.TIMEOUT)
         await self._cleanup_voice_session(session_id)
 
     async def _handle_ws_request_msg(
diff --git a/ucapi/voice_stream.py b/ucapi/voice_stream.py
@@ -13,6 +13,7 @@
 import asyncio
 import logging
 from asyncio import AbstractEventLoop
+from enum import Enum, auto
 from typing import AsyncIterator, Awaitable, Callable, Optional
 
 from ucapi.voice_assistant import AudioConfiguration
@@ -28,6 +29,53 @@
 """
 
 
+class VoiceEndReason(Enum):
+    """Reasons for ending a voice session."""
+
+    NORMAL = auto()
+    """Normal session end."""
+    TIMEOUT = auto()
+    """Closed due to timeout."""
+    REMOTE = auto()
+    """Closed remotely, for example by the client disconnecting."""
+    LOCAL = auto()
+    """Closed locally in integration, for example by a new session request."""
+    ERROR = auto()
+    """Closed due to an error."""
+
+
+class VoiceSessionClosed(Exception):
+    """Base class for voice session-related exceptions."""
+
+    def __init__(self, reason: VoiceEndReason, error: Exception | None = None):
+        """Create a voice session-related exception instance."""
+        super().__init__(str(reason))
+        self.reason = reason
+        self.error = error
+
+
+class VoiceSessionTimeout(VoiceSessionClosed):
+    """Raised when a voice session times out."""
+
+
+class VoiceSessionRemoteEnd(VoiceSessionClosed):
+    """Raised when a voice session ends remotely."""
+
+
+class VoiceSessionLocalEnd(VoiceSessionClosed):
+    """Raised when a voice session ends locally."""
+
+
+class VoiceSessionError(VoiceSessionClosed):
+    """Raised when a voice session ends due to an error."""
+
+
+class _EndSignal:
+    def __init__(self, reason: VoiceEndReason, error: Exception | None = None):
+        self.reason = reason
+        self.error = error
+
+
 class VoiceSession:
     """Represents a single remote voice capture session.
 
@@ -50,9 +98,11 @@ def __init__(
         self.entity_id = entity_id
         self.config = config
         self._loop = loop
-        self._q: asyncio.Queue[bytes | None] = asyncio.Queue(maxsize=max_queue)
+        self._q: asyncio.Queue[bytes | _EndSignal] = asyncio.Queue(maxsize=max_queue)
         self._closed = False
         self._drops_logged = 0
+        self.ended_by: VoiceEndReason | None = None
+        self.end_error: Exception | None = None
 
     def __aiter__(self) -> AsyncIterator[bytes]:
         """Return an async iterator over audio frames."""
@@ -68,13 +118,27 @@ async def frames(self) -> AsyncIterator[bytes]:
 
         :yield: Bytes : The next frame of data retrieved from the queue.
         :rtype: AsyncIterator[bytes]
+        :raises VoiceSessionTimeout: If the session times out.
+        :raises VoiceSessionRemoteEnd: If the session ends remotely.
+        :raises VoiceSessionLocalEnd: If the session ends locally.
+        :raises VoiceSessionError: If the session ends due to an error.
         """
         while True:
             item = await self._q.get()
-            if item is None:
+            if isinstance(item, _EndSignal):
                 self._closed = True
-                return
-            yield item
+                self.ended_by = item.reason
+                self.end_error = item.error
+                if item.reason is VoiceEndReason.NORMAL:
+                    return  # StopAsyncIteration
+                if item.reason is VoiceEndReason.TIMEOUT:
+                    raise VoiceSessionTimeout(item.reason)
+                if item.reason is VoiceEndReason.REMOTE:
+                    raise VoiceSessionRemoteEnd(item.reason)
+                if item.reason is VoiceEndReason.LOCAL:
+                    raise VoiceSessionLocalEnd(item.reason)
+                raise VoiceSessionError(item.reason, item.error)
+            yield item  # bytes
 
     def feed(self, chunk: bytes) -> None:
         """Feed an audio chunk into the session.
@@ -85,6 +149,8 @@ def feed(self, chunk: bytes) -> None:
         :param chunk: Audio data bytes.
         """
         try:
+            if self._closed:
+                return
             self._q.put_nowait(chunk)
         except asyncio.QueueFull:
             # Drop newest if consumer lags; throttle debug logging.
@@ -95,23 +161,31 @@ def feed(self, chunk: bytes) -> None:
                 )
             self._drops_logged += 1
 
-    def end(self) -> None:
+    def end(
+        self,
+        reason: VoiceEndReason = VoiceEndReason.NORMAL,
+        error: Exception | None = None,
+    ) -> None:
         """
         Signal the end of the voice session.
 
-        Enqueues a sentinel (None) to stop the consumer iterator.
+        Enqueues a sentinel (_EndSignal) to stop the consumer iterator.
         If the queue is full, attempts to make space for the sentinel.
         """
-        if not self._closed:
+        if self._closed:
+            return
+        try:
+            self._q.put_nowait(_EndSignal(reason, error))
+        except asyncio.QueueFull:
+            # Clear one and try to enqueue sentinel
             try:
-                self._q.put_nowait(None)
-            except asyncio.QueueFull:
-                # Clear one and try to enqueue sentinel
-                try:
-                    _ = self._q.get_nowait()
-                    self._q.put_nowait(None)
-                except Exception:  # pylint: disable=W0718
-                    self._closed = True
+                _ = self._q.get_nowait()
+                self._q.put_nowait(_EndSignal(reason, error))
+            except Exception:  # pylint: disable=W0718
+                # _should_ not happen, but just in case
+                pass
+        finally:
+            self._closed = True
 
     @property
     def closed(self) -> bool:
