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

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

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)
        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, and the
    milliseconds a frame spent encoding and from capture to the end of its
    encode. Empty without a capture that counts, and on the
    first call after a start or a spell unwatched, which only takes the baseline."""
    module = self._module
    if module is None or not hasattr(module, "stream_stats"):
        return {}
    try:
        totals = module.stream_stats()
    except Exception:
        return {}
    now = time.monotonic()
    last, last_at = self._totals, self._totals_at
    self._totals, self._totals_at = totals, now
    elapsed = now - last_at
    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)
    return rates