PipelineBridge

def __init__(self, maxsize: int = 1,
             request_keyframe: Optional[Callable[[], None]] = None,
             clock: Callable[[], float] = time.monotonic,
             invalidate_reference: Optional[Callable[[int], None]] = None) -> None:
    """Initializes the bridge.

    Args:
        maxsize: Queue depth; 1 means latest-wins, larger is drop-oldest.
        request_keyframe: Asks the display's encoder for a keyframe, on
            the loop thread, at most once per IDR_REQUEST_FLOOR_S while
            the gate is closed. None leaves every item ungated: audio
            samples are self-contained.
        clock: Monotonic time source.
        invalidate_reference: Tells the display's encoder a frame id was
            dropped, so the frames after it stop predicting from it. None
            gates every drop behind a keyframe.

    Attributes:
        dropped: Items discarded since construction, evicted for a newer
            one or held back behind a closed gate. None of them spent a
            sequence number, so they appear in no loss statistic on either
            side; this counter is what separates a lagging sender from a
            lossy link.
        invalidated: The share of those the encoder was told to predict
            past, the rest being frames that predicted from one already
            dropped and so needed no word of their own.
    """
    self._queue: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
    self._request_keyframe = request_keyframe
    self._invalidate = invalidate_reference
    self._clock = clock
    self._queued_keyframe = False
    self._gated_at: Optional[float] = None
    self._last_request: Optional[float] = None
    # Frame ids dropped recently, which nothing delivered may predict from.
    self._lost: deque = deque(maxlen=LOST_FRAME_MEMORY)
    self.dropped = 0
    self.invalidated = 0

def set_data(self, data: Any, keyframe: bool = True) -> None:
    """Enqueue an item, dropping the oldest one when the queue is full.

    Synchronous, no lock: the checks and the put have no await, so the
    single-threaded loop runs them without interleaving (all access is on
    the loop thread). A full queue means the consumer is lagging.

    Args:
        data: The item.
        keyframe: Whether the item decodes on its own. A delta frame needs
            the frame it predicts from on the wire: one that names it is
            held back only while that frame was dropped here, and one that
            does not is held back behind any drop until the next keyframe.
    """
    queue = self._queue
    if self._request_keyframe is None:
        if queue.full():
            queue.get_nowait()
            self.dropped += 1
        queue.put_nowait(data)
        return
    if keyframe:
        if queue.full():
            queue.get_nowait()
            self.dropped += 1
        queue.put_nowait(data)
        self._queued_keyframe = True
        self._gated_at = None
        self._lost.clear()
        return
    dependency = data.dependency if self._invalidate is not None else None
    if dependency is not None:
        frame_id, reference = dependency
        if reference in self._lost:
            self._lost.append(frame_id)
            self.dropped += 1
            return
        if queue.full():
            if self._queued_keyframe:
                self._drop(data)
                return
            self._drop(queue.get_nowait())
            if reference in self._lost:
                self._lost.append(frame_id)
                self.dropped += 1
                return
        queue.put_nowait(data)
        self._queued_keyframe = False
        return
    now = self._clock()
    if self._gated_at is not None:
        if now - self._gated_at < GATE_TIMEOUT_S:
            self.dropped += 1
            self._ask(now)
            return
        logger.warning("Video bridge: no keyframe within %.1fs of a drop; "
                       "sending delta frames again", GATE_TIMEOUT_S)
        self._gated_at = None
    if queue.full():
        self.dropped += 1
        if not self._queued_keyframe:
            queue.get_nowait()
            self.dropped += 1
        self._gated_at = now
        self._ask(now)
        return
    queue.put_nowait(data)
    self._queued_keyframe = False

def _ask(self, now: float) -> None:
    if self._last_request is not None and now - self._last_request < IDR_REQUEST_FLOOR_S:
        return
    self._last_request = now
    self._request_keyframe()

def _drop(self, item: Any) -> None:
    """Let a frame go and tell the encoder, so nothing later predicts from it."""
    self.dropped += 1
    self.invalidated += 1
    self._lost.append(item.dependency[0])
    self._invalidate(item.dependency[0])

def empty(self) -> bool:
    return self._queue.empty()

async def get_data(self) -> Any:
    """Wait until an item is available in the queue and return it."""
    return await self._queue.get()