Source code for viam.media.media

import sys

if sys.version_info >= (3, 9):
    from collections.abc import AsyncIterator
else:
    from typing import AsyncIterator

from typing import Protocol, TypeVar

MediaType = TypeVar("MediaType", covariant=True)


[docs]class MediaStream(Protocol[MediaType]):
[docs] async def next(self) -> MediaType: ...
[docs] def __aiter__(self) -> AsyncIterator: return self
[docs] async def __anext__(self) -> MediaType: return await self.next()
[docs]class MediaReader(Protocol[MediaType]):
[docs] async def read(self) -> MediaType: ...
[docs]class MediaSource(Protocol[MediaType]):
[docs] async def stream(self) -> MediaStream[MediaType]: ...
[docs]class MediaStreamWithIterator(MediaStream[MediaType]): _stream: AsyncIterator[MediaType] def __init__(self, stream: AsyncIterator[MediaType]): self._stream = stream
[docs] async def next(self) -> MediaType: return await self._stream.__anext__()
[docs] def __aiter__(self): return self._stream
[docs] async def __anext__(self) -> MediaType: return await self._stream.__anext__()