Expose VoiceSessionKey in VoiceSession

Author: zehnm

ucapi/api.py

@@ -46,7 +46,12 @@
     AudioConfiguration,
 )
 from .voice_assistant import Commands as VaCommands
-from .voice_stream import VoiceEndReason, VoiceSession, VoiceStreamHandler
+from .voice_stream import (
+    VoiceEndReason,
+    VoiceSession,
+    VoiceSessionKey,
+    VoiceStreamHandler,
+)
 
 try:
     from ._version import version as __version__
@@ -57,10 +62,6 @@
 _LOG.setLevel(logging.DEBUG)
 
 
-VoiceSessionKey = tuple[Any, int]
-"""Tuple of (websocket, session_id)"""
-
-
 @dataclass(slots=True)
 class _VoiceSessionContext:
     session: VoiceSession

ucapi/voice_stream.py

@@ -14,14 +14,17 @@
 import logging
 from asyncio import AbstractEventLoop
 from enum import Enum, auto
-from typing import AsyncIterator, Awaitable, Callable, Optional
+from typing import Any, AsyncIterator, Awaitable, Callable, Optional
 
 from .api_definitions import AssistantEvent
 from .voice_assistant import AudioConfiguration
 
 _LOG = logging.getLogger(__name__)
 
 
+VoiceSessionKey = tuple[Any, int]
+"""Tuple of (websocket, session_id)"""
+
 VoiceStreamHandler = Callable[["VoiceSession"], Optional[Awaitable[None]]]
 """Public type alias for the handler callable.
 
@@ -113,6 +116,11 @@ def __aiter__(self) -> AsyncIterator[bytes]:
         """Return an async iterator over audio frames."""
         return self.frames()
 
+    @property
+    def key(self) -> VoiceSessionKey:
+        """Return the session key to uniquely identify a session."""
+        return self._websocket, self.session_id
+
     async def send_event(self, event: AssistantEvent):
         """Send an assistant event to the session initiator client."""
         await self._api.send_assistant_event(self._websocket, event)