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)
    # When the encoder was last told of a drop, on the clock the capture stamps
    # encode instants with, and the frames held since then that it coded after
    # that or that carry no encode instant.
    self._told_at: Optional[float] = None
    self._held = 0
    self.dropped = 0
    self.invalidated = 0
    self._budget_bps: Optional[float] = None
    self._budget_bytes = 0.0
    # Budget drops in a row, and whether the encoder has sent an anchor (`BUDGET_RUN_MAX`).
    self._budget_run = 0
    self._anchored = False
    # Delta frames naming their reference let through, beside `over_budget`, the ones the budget dropped.
    self.passed = 0
    self._budget_at = 0.0
    self.over_budget = 0
    # Frames the encoder coded since the last keyframe, as they arrive.
    self._since_key = 0

def set_budget(self, bps: Optional[float]) -> None:
    """Hold the frames let through to `bps`, the display's steered rate;
    None lets every frame through."""
    if not bps:
        self._budget_bps = None
        self._budget_bytes = 0.0
        return
    if self._budget_bps is None:
        self._budget_at = self._clock()
    self._budget_bps = float(bps)

def _budget_level(self) -> float:
    """The bytes the frames let through still stand at, drained to now."""
    now = self._clock()
    self._budget_bytes = max(0.0, self._budget_bytes - (now - self._budget_at) * self._budget_bps / 8.0)
    self._budget_at = now
    return self._budget_bytes

def _budget_full(self) -> bool:
    """Whether the frames let through stand past BUDGET_WINDOW_S of the rate."""
    return (self._budget_bps is not None
            and self._budget_level() > self._budget_bps / 8.0 * BUDGET_WINDOW_S)

def _charge(self, item: Any) -> None:
    if self._budget_bps is not None:
        self._budget_bytes = self._budget_level() + _frame_bytes(item)

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()
        self._held = 0
        self._since_key = 0
        self._budget_run = 0
        return
    self._since_key += 1
    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._hold(frame_id, getattr(data, "timing", None))
            return
        kept = (str(getattr(data, "codec", "") or "").lower() == "video/h264"
                and not self._since_key % FRAME_NUM_WRAP)
        self._anchored = self._anchored or bool(getattr(data, "anchor", False))
        if (self._since_key % FRAME_NUM_WRAP and self._budget_full()
                and (self._anchored or self._budget_run < BUDGET_RUN_MAX)):
            self.over_budget += 1
            self._budget_run += 1
            self._drop(data)
            return
        self._budget_run = 0
        if queue.full():
            if self._queued_keyframe:
                self._drop(data)
                return
            self._drop(queue.get_nowait())
            if reference in self._lost:
                self._hold(frame_id, getattr(data, "timing", None))
                return
        queue.put_nowait(data)
        self._charge(data)
        self.passed += 1
        self._queued_keyframe = kept
        self._held = 0
        if self._lost and self._told_at is not None:
            encoded = _encode_start(getattr(data, "timing", None))
            if encoded is not None and encoded > self._told_at:
                self._lost.clear()
        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._held = 0
    self._told_at = self._clock()
    self._lost.append(item.dependency[0])
    self._invalidate(item.dependency[0])

def _hold(self, frame_id: int, timing: Optional[tuple]) -> None:
    """Hold back a frame predicting from one already let go, and ask for a
    keyframe when the encoder coded it after it was told of the drop."""
    self.dropped += 1
    self._lost.append(frame_id)
    encoded = _encode_start(timing)
    if encoded is None:
        self._held += 1
        unheard = self._held > LOST_CHAIN_FRAMES
    elif self._told_at is not None and encoded > self._told_at:
        self._held += 1
        unheard = self._held > 1
    else:
        unheard = False
    if unheard:
        self._ask(self._clock())

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()