CaptureDemand

def __init__(self) -> None:
    self._ports: List[Any] = []
    self._told: Optional[Tuple[Any, Any]] = None
    self._wanted = False
    self._reader: Optional[str] = None
    self._idle_since: Optional[float] = None
    self._idle_polls = 0
    self._task: Optional[Any] = None
    self._lock: Optional[asyncio.Lock] = None

async def prepare(self) -> bool:
    """Brings the device up, since nothing can read a device that is not there."""
    raise NotImplementedError

async def reader(self) -> Optional[str]:
    """What reads the device, or None when nothing does or nothing can say."""
    raise NotImplementedError

def attach(self, port: Any) -> None:
    """Adds a transport; the first one starts the watch."""
    if port not in self._ports:
        self._ports.append(port)
    if self._task is None or self._task.done():
        self._task = asyncio.ensure_future(self._run())

def detach(self, port: Any) -> None:
    """Drops a transport; the watch ends with the last one and the device outlives it."""
    if port in self._ports:
        self._ports.remove(port)
    if self._told is not None and self._told[0] is port:
        self._told = None

async def route(self) -> None:
    """Re-addresses the answer after a transport's candidates changed.

    The page asked last is released when it may no longer capture, and the page that may
    is asked if it was not; a send that fails hands the device to the next candidate.
    """
    if self._lock is None:
        self._lock = asyncio.Lock()
    async with self._lock:
        candidates = [(p, page) for p in self._ports for page in p.capture_candidates()]
        target = candidates[0] if candidates else None
        if self._told is not None and (not self._wanted or self._told != target):
            port, page = self._told
            self._told = None
            await port.tell_capture(page, self.subject, False)
        if not self._wanted or self._told is not None:
            return
        for port, page in candidates:
            if await port.tell_capture(page, self.subject, True):
                self._told = (port, page)
                return

async def _run(self) -> None:
    if not await self.prepare():
        logger.error("The %s cannot be watched for readers: its device did not start.",
                     self.subject)
        return
    while self._ports:
        await asyncio.sleep(POLL_SECONDS)
        try:
            await self._poll()
        except Exception:
            logger.exception("The %s demand poll failed.", self.subject)
    self._wanted, self._idle_since, self._idle_polls = False, None, 0

async def _poll(self) -> None:
    if not any(port.capture_candidates() for port in self._ports):
        # Nobody to ask, so no reading is taken: it would be stale by the time a page arrives.
        self._idle_since, self._idle_polls = None, 0
        await self._publish(False)
        return
    name = await self.reader()
    if name is not None:
        self._idle_since, self._idle_polls = None, 0
        self._reader = name
        await self._publish(True)
        return
    self._idle_polls += 1
    if self._idle_since is None:
        self._idle_since = time.monotonic()
    elif (self._idle_polls >= MIN_IDLE_POLLS
          and time.monotonic() - self._idle_since >= self.release_seconds):
        await self._publish(False)

async def _publish(self, wanted: bool) -> None:
    if wanted != self._wanted:
        self._wanted = wanted
        logger.info("The %s is %s.", self.subject,
                    f"read by {self._reader}; the page is asked for it" if wanted else "released")
        audit.emit("capture.demand", subject=self.subject,
                   action="ask" if wanted else "release",
                   reader=self._reader if wanted else None)
    await self.route()