Add data stream support #347
Conversation
…n-sdks into lukas/data-streams
| TOKEN = os.environ.get("LIVEKIT_TOKEN") | ||
| URL = os.environ.get("LIVEKIT_URL") | ||
|
|
||
|
|
There was a problem hiding this comment.
it'd be good to include some comments on what this example demonstrates, and what other components the user should prepare for in order to see a demo.
it seems like it's built to work with the JS example?
| attributes=dict(header.attributes), | ||
| name=header.byte_header.name, | ||
| ) | ||
| self._queue: RingQueue[proto_DataStream.Chunk | None] = RingQueue(capacity) |
There was a problem hiding this comment.
| self._queue: RingQueue[proto_DataStream.Chunk | None] = RingQueue(capacity) | |
| self._queue: RingQueue[Optional[proto_DataStream.Chunk]] = RingQueue(capacity) |
There was a problem hiding this comment.
this is how we also do it in other places in the python SDK, will leave it with | None for consistency
Co-authored-by: David Zhao <[email protected]>
Co-authored-by: David Zhao <[email protected]>
Co-authored-by: David Zhao <[email protected]>
…n-sdks into lukas/data-streams
| elif which == "stream_chunk_received": | ||
| asyncio.gather(self._handle_stream_chunk(event.stream_chunk_received.chunk)) | ||
| elif which == "stream_trailer_received": | ||
| asyncio.gather( |
There was a problem hiding this comment.
We shouldn't use asyncio.gather. The correct way to spawn tasks would be to use asyncio.create_task
We should also be careful about always keeping a reference to a task or it may get GC'ed and may not complete.
Let's also cancel them on disconnect
| mime_type: str = "application/octet-stream", | ||
| extensions: Optional[Dict[str, str]] = None, |
There was a problem hiding this comment.
Let's do like the send_text method
extensions: dict[str, str]
There was a problem hiding this comment.
I specifically changed this to optional because @longcw had concerns about defaulting a list to an empty list in method parameters
| mime_type: str = "application/octet-stream", | ||
| extensions: Optional[Dict[str, str]] = None, | ||
| stream_id: str | None = None, | ||
| destination_identities: Optional[List[str]] = None, |
There was a problem hiding this comment.
destination_identities: List[str] = [],
|
|
||
| async def send_file( | ||
| self, | ||
| file_path: str, |
| local_participant: LocalParticipant, | ||
| *, | ||
| topic: str = "", | ||
| extensions: Optional[Dict[str, str]] = {}, |
There was a problem hiding this comment.
In BaseStreamWriter it's renamed to attributes, but here is still extensions.
There was a problem hiding this comment.
thanks! fixed in #352
|
|
||
|
|
||
| @dataclass | ||
| class BaseStreamInfo(TypedDict): |
There was a problem hiding this comment.
This should be either a dataclass or a TypedDict, but not both. I prefer to make it a dataclass.
There was a problem hiding this comment.
Otherwise it raise error when trying to access the fields since info is a dict
Traceback (most recent call last):
File "/home/longc/data/code/agents/examples/avatar/plugin/worker.py", line 112, in _read_audio
async for frame in self._audio_receiver.stream():
File "/home/longc/data/code/agents/examples/avatar/plugin/io.py", line 208, in stream
sample_rate = int(reader.info.attributes["sample_rate"])
^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'dict' object has no attribute 'attributes'There was a problem hiding this comment.
thanks! updated in #352
No description provided.