StreamWatch

def __init__(self, display_id: str,
             publish: Callable[[str, Dict[str, Any]], Awaitable[None]]) -> None:
    self.display_id = display_id
    self.info: Optional[Dict[str, Any]] = None
    self._publish = publish
    self._task: Optional["asyncio.Task[None]"] = None
    self._module: Any = None
    self._totals: Optional[Dict[str, int]] = None
    self._totals_at = 0.0
    # The window's extremes: pixelflux's, merged over every read since the
    # window began, and the frames sent: count, total, least, most, in ms.
    self._peaks: Dict[str, int] = {}
    self._sends: Optional[List[float]] = None
    self._buckets: Deque[List[int]] = collections.deque()

def follow(self, module: Any, encoder: str, use_cpu: bool) -> None:
    """Start following `module`, the capture that now serves the display with
    this encoder and software-encoding choice."""
    self.stop()
    if not hasattr(module, "stream_info"):
        return
    self._module = module
    self._task = asyncio.ensure_future(self._watch(module, encoder, use_cpu))

def stop(self) -> None:
    """Stop following; the last description is forgotten with the capture."""
    if self._task is not None:
        self._task.cancel()
        self._task = None
    self._module = None
    self._totals = None
    self.info = None
    self._restart_window()

def note_send(self, capture_ns: int, size: int) -> None:
    """One frame on its way to the display's controller: `size` bytes captured at
    `capture_ns` (`time.monotonic_ns()`, zero where the capture stamped none),
    noted when the bytes are handed to the socket over WebSockets and when the
    frame's last packet leaves the pacer over WebRTC. Runs on the event loop."""
    now = time.monotonic_ns()
    if capture_ns > 0:
        ms = (now - capture_ns) / 1e6
        sends = self._sends
        if sends is None:
            self._sends = [1, ms, ms, ms]
        else:
            sends[0] += 1
            sends[1] += ms
            if ms < sends[2]:
                sends[2] = ms
            if ms > sends[3]:
                sends[3] = ms
    start = now - now % PEAK_BUCKET_NS
    buckets = self._buckets
    if buckets and buckets[-1][0] == start:
        buckets[-1][1] += size
        return
    buckets.append([start, size])
    while buckets[0][0] <= start - PEAK_WINDOW_S * 1e9:
        buckets.popleft()

def _restart_window(self) -> None:
    self._peaks = {}
    self._sends = None

def _merge_peaks(self, totals: Dict[str, int]) -> None:
    """Fold one read's extremes into the window's: pixelflux starts its own again
    on every read, so a read that does not end the window must keep them."""
    peaks = self._peaks
    low = totals.get("pipeline_min_ns", 0)
    if low and (not peaks.get("pipeline_min_ns") or low < peaks["pipeline_min_ns"]):
        peaks["pipeline_min_ns"] = low
    for key in ("pipeline_max_ns", "frame_max_bytes"):
        if totals.get(key, 0) > peaks.get(key, 0):
            peaks[key] = totals[key]

def sent_mbps(self) -> Dict[str, float]:
    """The video sent over the last `PEAK_WINDOW_S`, in Mbit/s: `video_mbps` on
    average, over as much of the window as the stream has run, at least a second,
    and `peak_mbps`, its fullest `PEAK_BUCKET_NS`. Video alone, which is what the
    encoder's rate is set for and the page's received figure, audio and all, is
    not. Empty where nothing was sent in it."""
    now = time.monotonic_ns()
    recent = [b for b in self._buckets if b[0] > now - PEAK_WINDOW_S * 1e9]
    if not recent:
        return {}
    bits = sum(b[1] for b in recent) * 8
    span = max(1.0, min(PEAK_WINDOW_S, (now - recent[0][0]) / 1e9))
    return {"video_mbps": round(bits / span / 1e6, 2),
            "peak_mbps": round(max(b[1] for b in recent) * 8 / (PEAK_BUCKET_NS / 1e9) / 1e6, 2)}

async def _watch(self, module: Any, encoder: str, use_cpu: bool) -> None:
    gpu = gpu_present()
    delay = SETTLE_S
    while True:
        await asyncio.sleep(delay)
        try:
            info = await asyncio.to_thread(module.stream_info)
        except Exception as e:
            logger.debug(f"Stream description of '{self.display_id}' unreadable: {e}")
            return
        if not info or not info.get("encoder"):
            continue
        delay = WATCH_S
        info["gpu_present"] = gpu
        info["hardware_expected"] = hardware_expected(encoder, use_cpu, gpu, settings.encoder_backends())
        if info == self.info:
            continue
        self.info = info
        logger.debug(f"Display '{self.display_id}' stream description: {info}")
        await self._publish(self.display_id, info)

def rates(self) -> Dict[str, float]:
    """The encode's rate and cost since the last call: frames a second, the
    milliseconds a frame spent encoding and from capture to the end of its
    encode, on average and at the extremes, the largest frame, and the
    milliseconds from capture to the wire (`note_send`). Empty without a
    capture that counts, on the first call after a start or a spell
    unwatched, which only takes the baseline, and within `RATE_WINDOW_MIN_S`
    of the last call, which leaves the window growing."""
    module = self._module
    if module is None or not hasattr(module, "stream_stats"):
        return {}
    try:
        totals = module.stream_stats()
    except Exception:
        return {}
    if totals:
        self._merge_peaks(totals)
    now = time.monotonic()
    last, last_at = self._totals, self._totals_at
    elapsed = now - last_at
    if totals and last and totals["frames"] >= last["frames"] and elapsed < RATE_WINDOW_MIN_S:
        return {}
    self._totals, self._totals_at = totals, now
    peaks, sends = self._peaks, self._sends
    self._restart_window()
    if not totals or not last or totals["frames"] < last["frames"] or not 0 < elapsed <= RATE_WINDOW_MAX_S:
        return {}
    frames = totals["frames"] - last["frames"]
    rates = {"encoded_fps": round(frames / elapsed, 1)}
    if frames:
        rates["encode_ms"] = round((totals["encode_ns"] - last["encode_ns"]) / frames / 1e6, 2)
        rates["pipeline_ms"] = round((totals["pipeline_ns"] - last["pipeline_ns"]) / frames / 1e6, 2)
        if peaks.get("pipeline_max_ns"):
            rates["pipeline_min_ms"] = round(peaks.get("pipeline_min_ns", 0) / 1e6, 2)
            rates["pipeline_max_ms"] = round(peaks["pipeline_max_ns"] / 1e6, 2)
        if peaks.get("frame_max_bytes"):
            rates["frame_max_kb"] = round(peaks["frame_max_bytes"] / 1000, 1)
    if sends:
        rates["send_ms"] = round(sends[1] / sends[0], 2)
        rates["send_min_ms"] = round(sends[2], 2)
        rates["send_max_ms"] = round(sends[3], 2)
    rates.update(self.sent_mbps())
    return rates