def read_frame_nowait(self) -> Optional[DataFrame]:
        """Read a single frame from the local buffer immediately.

        If no frames are available but the stream is still open, returns None.
        Otherwise, raises StreamConsumedError.
        """
        try:
            frame = self._data_frames.get_nowait()
        except asyncio.QueueEmpty:
            if self.closed:
                raise StreamConsumedError(self.id)
            return None
        self._data_frames.task_done()
        if frame is None:
            raise StreamConsumedError(self.id)
        return frame
Read a single frame from the local buffer immediately.

        If no frames are available but the stream is still open, returns None.
        Otherwise, raises StreamConsumedError.