RTCApp

def __init__(
    self,
    async_event_loop: asyncio.AbstractEventLoop,
    encoder: str,
    stun_servers: Optional[List[str]] = None,
    turn_servers: Optional[List[str]] = None
) -> None:
    self.peer_connections: Dict[str, Any] = {}
    self.async_event_loop = async_event_loop
    self.stun_servers = stun_servers
    self.turn_servers = turn_servers
    self.ice_udp_mux: Optional[UdpMux] = None
    self.ice_tcp_mux: Optional[TcpMux] = None
    self.encoder = encoder
    self.last_cursor_sent = None

    self.displays: Dict[str, Dict[str, Any]] = {}
    self.media_pipeline: Optional[MediaPipeline] = None
    self.start_display_media = self._default_start_display_media
    self.stop_display_media = self._default_stop_display_media
    self.get_encoder_for_display = lambda display_id: self.encoder
    self.get_fullcolor_for_display = lambda display_id: bool(app_settings.video_fullcolor[0])
    self.get_use_cpu_for_display = lambda display_id: bool(app_settings.use_cpu[0])
    self.on_video_codec_declined = None
    self.on_fullcolor_declined = None

    self.on_data_open = lambda channel=None: logger.warning('unhandled on_data_open')
    self.on_data_close = lambda: logger.warning('unhandled on_data_close')
    self.on_data_error = lambda e=None: logger.warning('unhandled on_data_error')
    self.on_data_message = lambda msg, display_id='primary', conn_id=None: logger.warning('unhandled on_data_message')
    self.on_peer_gone = None

    self.on_ice = lambda ice, client_peer_id: logger.warning('unhandled ice event')
    self.on_sdp = lambda sdp_type, sdp, client_peer_id: logger.warning('unhandled sdp event')

    self.request_idr_frame = lambda display_id='primary': logger.warning('unhandled request_idr_frame')
    self.invalidate_reference = lambda display_id, frame_id: logger.warning('unhandled invalidate_reference')

    self.on_video_consumer_active = None
    self.on_audio_consumer_active = None
    self.on_consumers_changed = None

    self.provision_virtual_mic = None

async def set_sdp(self, sdp_type: str, sdp: str, client_peer_id: str) -> None:
    """Apply the remote SDP answer received from a peer over signaling.

    Answers for peers already in a closed/failed state are ignored rather
    than raised: teardown routinely races late signaling messages.

    Args:
        sdp_type: Must be "answer"; the server always offers.
        sdp: The remote session description text.
        client_peer_id: Registered peer the answer belongs to.

    Raises:
        RTCAppError: On a non-answer type, missing sdp/peer id, or an
            unknown peer.
    """
    if sdp_type != 'answer':
        raise RTCAppError('ERROR: sdp type is not "answer"')
    if sdp is None:
        raise RTCAppError("ERROR: sdp can't be None")
    if not client_peer_id:
        raise RTCAppError("ERROR: client_peer_id is required to set sdp")

    peer_obj = self.peer_connections.get(client_peer_id, None)
    if peer_obj is None:
        raise RTCAppError(f"ERROR: peer connection for client_peer_id: {client_peer_id} not found")

    peer_conn = peer_obj["peer_conn"]
    if peer_conn.connectionState in ["closed", "failed"]:
        logger.warning(
            f"Ignoring remote SDP: peer connection in {peer_conn.connectionState} state",
            extra={'client_peer_id': client_peer_id, 'client_type': peer_obj.get('client_type')}
        )
        return

    desc = RTCSessionDescription(sdp=sdp, type=sdp_type)
    await peer_conn.setRemoteDescription(desc)
    await self._settle_video_codec(client_peer_id, peer_obj)

async def set_ice(self, ice: Dict, client_peer_id: str) -> None:
    """Add an ICE candidate received from the signaling server.

    An empty candidate string is the end-of-candidates marker and is
    forwarded as None.

    Args:
        ice: Candidate dict with `candidate` and `sdpMid` or
            `sdpMLineIndex` keys, as sent by the browser.
        client_peer_id: Registered peer the candidate belongs to.

    Raises:
        RTCAppError: On a missing peer id, an unknown peer, or a
            candidate that fails to parse.
    """
    if not client_peer_id:
        raise RTCAppError("ERROR: client_peer_id is required to set sdp")

    peer_obj = self.peer_connections.get(client_peer_id, None)
    if peer_obj is None:
        raise RTCAppError(f"ERROR: peer connection for client_peer_id: {client_peer_id} not found")

    peer_conn = peer_obj["peer_conn"]
    if peer_conn.connectionState in ["closed", "failed"]:
        logger.warning(
            f"Ignoring adding ICE candidate: peer connection in {peer_conn.connectionState} state",
            extra={'client_peer_id': client_peer_id, 'client_type': peer_obj.get('client_type')}
        )
        return

    if ice.get('candidate') == "":
        await peer_conn.addIceCandidate(None)
        return

    obj = Candidate.from_sdp(ice.get('candidate'))
    icecandidate = candidate_from_aioice(obj)

    sdp_mid = ice.get('sdpMid')
    if sdp_mid is not None:
        icecandidate.sdpMid = sdp_mid
    else:
        icecandidate.sdpMLineIndex = ice.get('sdpMLineIndex')

    if isinstance(icecandidate, RTCIceCandidate):
        await peer_conn.addIceCandidate(icecandidate)
    else:
        raise RTCAppError("ERROR: ice candidate is not an instance of RTCIceCandidate")

async def send_clipboard_data(self, data: Union[str, bytes], mime_type: str = "text/plain",
                              reply_to: Optional[str] = None,
                              peer_id: Optional[str] = None) -> None:
    """Send clipboard data over the data channel, chunked when large.

    Chunk sends are paced against each channel's SCTP queue (see
    `drain_data_channel`) so a multi-MB clipboard neither buffers
    unboundedly in memory nor starves input/cursor/stats behind it on the
    one ordered stream, and the per-chunk gzip runs off the event loop.
    Each channel runs its own pipeline a chunk at a time, so a slow peer
    paces only its own transfer; one that cannot drain a chunk within the
    bulk timeout is left out of the rest of this payload (it keeps its
    stream, and the client discards the partial transfer at the next
    start) instead of wedging the others. Chunk payloads are built once
    and shared by every channel; an entry is dropped once the slowest
    channel has passed it, so the cache holds what is in flight, not the
    whole payload. An empty payload is sent only as a tagged reply,
    settling a client fetch against an empty server clipboard.

    Args:
        data: Clipboard payload; str is UTF-8 encoded before sending.
        mime_type: Payload MIME type; "text/plain" marks it as text.
        reply_to: Set to the requesting verb (e.g. "cr") when this send
            answers a client fetch rather than announcing a server-side
            clipboard change — the websockets `clipboard_reply,<verb>`
            contract, carried here as an extra field on the clipboard-msg
            / clipboard-msg-start payload so the client can treat the
            payload cache-only without time heuristics (old clients ignore
            the unknown field).
        peer_id: Peer that asked for this payload, addressed alone the way
            `send_system_action` addresses a requester: the other peers
            already hold the content, and a reply they did not ask for is
            read as their own fetch and cached rather than pasted. A
            requester whose channel has closed receives nothing.
    """
    if not data and not reply_to:
        return

    is_text = mime_type == "text/plain"
    data_bytes: bytes = data.encode() if isinstance(data, str) else data
    clipboard_chunk_size = get_adjusted_chunk_size(self.peer_connections)
    requester = None
    if peer_id is not None:
        peer_obj = self.peer_connections.get(peer_id)
        channel = peer_obj.get("data_channel") if peer_obj else None
        if channel is None or channel.readyState != "open":
            return
        requester = channel
    if data_bytes and (requester is not None
                       or next(self._iter_open_data_channels(), None) is not None):
        audit.emit("clipboard.send", mime_type=mime_type, size_bytes=len(data_bytes))

    def send_typed(msg_type: str, payload: Any) -> None:
        if requester is not None:
            self.send_message_to_channel(requester, msg_type, payload)
        else:
            self.__send_data_channel_message(msg_type, payload)

    if len(data_bytes) <= clipboard_chunk_size:
        b64data = base64.b64encode(data_bytes).decode('utf-8')
        payload = {
            "content": b64data,
            "mime_type": mime_type,
            "is_binary_data": not is_text,
            "total_size": len(data_bytes)
        }
        if reply_to:
            payload["reply_to"] = reply_to
        send_typed("clipboard-msg", payload)
    else:
        start_payload = {
            "mime_type": mime_type,
            "is_binary_data": not is_text,
            "total_size": len(data_bytes),
        }
        if reply_to:
            start_payload["reply_to"] = reply_to
        channels = ([requester] if requester is not None
                    else list(self._iter_open_data_channels()))
        # One payload at a time per channel: start/data/end carry no
        # transfer id, so a send racing another would interleave two
        # payloads' chunks into one assembly.
        locks = self.__dict__.setdefault("_clipboard_send_locks", {})
        live = {id(c) for c in self._iter_open_data_channels()}
        live.update(id(c) for c in channels)
        for gone in [k for k in locks if k not in live]:
            del locks[gone]
        offsets = list(range(0, len(data_bytes), clipboard_chunk_size))
        prepared: dict = {}
        prepare_lock = asyncio.Lock()
        progress = {id(c): 0 for c in channels}

        async def chunk_for(offset: int, want_gz: bool) -> tuple:
            async with prepare_lock:
                entry = prepared.get(offset)
                if entry is None:
                    chunk = data_bytes[offset:offset + clipboard_chunk_size]
                    entry = (json.dumps({
                        "type": "clipboard-msg-data",
                        "data": {"content": base64.b64encode(chunk).decode("utf-8")},
                    }), None)
                if want_gz and entry[1] is None:
                    entry = (entry[0], await asyncio.to_thread(
                        gzip.compress, entry[0].encode("utf-8"), 6))
                prepared[offset] = entry
                return entry

        def passed(channel: Any, offset: int) -> None:
            progress[id(channel)] = offset + clipboard_chunk_size
            floor = min(progress.values())
            for done in [o for o in prepared if o + clipboard_chunk_size <= floor]:
                prepared.pop(done, None)

        async def deliver(channel: Any) -> None:
            want_gz = bool(getattr(channel, "_selkies_gz_tx", False))
            async with locks.setdefault(id(channel), asyncio.Lock()):
                self.send_message_to_channel(channel, "clipboard-msg-start", start_payload)
                for offset in offsets:
                    if channel.readyState != "open":
                        progress.pop(id(channel), None)
                        return
                    payload, gz_payload = await chunk_for(offset, want_gz)
                    self._send_prepared_to_channel(
                        channel, "clipboard-msg-data", payload, gz_payload)
                    if not await drain_data_channel(channel):
                        logger.warning(
                            "Data channel did not drain a clipboard chunk within "
                            f"{BULK_DRAIN_TIMEOUT_S:.0f}s; leaving it out of this payload.")
                        progress.pop(id(channel), None)
                        return
                    passed(channel, offset)
                self.send_message_to_channel(channel, "clipboard-msg-end", {})

        await asyncio.gather(*(deliver(c) for c in channels), return_exceptions=True)

    logger.debug(f"Sent clipboard data of length {len(data_bytes)} with mime type {mime_type}")

def send_cursor_data(self, data: Any) -> None:
    """Broadcast a cursor update, remembering it for late-joining peers."""
    self.last_cursor_sent = data
    self.__send_data_channel_message(
        "cursor", data)

def _controller_channels(self, display_id: Optional[str] = None,
                         subscribed: bool = False) -> Iterator[Tuple[str, RTCDataChannel]]:
    """Yield `(display id, open data channel)` of each connected controller,
    narrowed to one display and to the peers whose page has its stats open
    (`stream_stats` module docstring). A viewer is never among them."""
    for peer_obj in self.peer_connections.values():
        if peer_obj.get("client_type") != ClientType.CONTROLLER:
            continue
        did = peer_obj.get("display_id") or "primary"
        if display_id is not None and did != display_id:
            continue
        if subscribed and not peer_obj.get("stats"):
            continue
        peer_conn = peer_obj.get("peer_conn")
        channel = peer_obj.get("data_channel")
        if (peer_conn is not None and channel is not None
                and peer_conn.connectionState == "connected"
                and channel.readyState == "open"):
            yield did, channel

def stats_displays(self) -> List[str]:
    """The displays a connected controller has its stats open on."""
    return sorted({did for did, _ in self._controller_channels(subscribed=True)})

def send_stream_info(self, display_id: str, info: Dict[str, Any],
                     channel: Optional[RTCDataChannel] = None) -> None:
    """Tell a display's controllers, or the one on `channel`, what its capture
    streams and how."""
    channels = [channel] if channel is not None else [
        ch for _, ch in self._controller_channels(display_id)]
    for ch in channels:
        self.send_message_to_channel(ch, "stream_info", info)

def send_stream_stats(self, display_id: str, stats: Dict[str, Any]) -> None:
    """Send one second's figures to the display's controllers watching them."""
    for _, channel in self._controller_channels(display_id, subscribed=True):
        self.send_message_to_channel(channel, "stream_stats", stats)

def send_system_action(self, action: str, peer_id: Optional[str] = None) -> None:
    """Send a system action (e.g. ``command_error,<text>``) to clients.

    With a `peer_id` whose channel is still open, only that peer is
    addressed (requester-scoped feedback); otherwise — including a
    requester that reconnected under a new peer id — the action is
    broadcast, and shared-mode viewers suppress it client-side.
    """
    if peer_id is not None:
        peer_obj = self.peer_connections.get(peer_id)
        channel = peer_obj.get("data_channel") if peer_obj else None
        if channel is not None and channel.readyState == "open":
            self.send_message_to_channel(channel, "system", {"action": action})
            return
    self.__send_data_channel_message("system", {"action": action})

def send_print_document(self, name: str, size: int,
                        channel: Optional[RTCDataChannel] = None) -> None:
    """Tell the controller pages, or the one peer whose `channel` is
    given, that a printed document waits in the spool."""
    payload = {"name": name, "size_bytes": size}
    channels = [channel] if channel is not None else [
        obj.get("data_channel") for obj in self.peer_connections.values()
        if obj.get("client_type") == ClientType.CONTROLLER
        and (obj.get("display_id") or "primary") == "primary"
        and obj.get("peer_conn") is not None
        and obj["peer_conn"].connectionState == "connected"]
    for open_channel in channels:
        if open_channel is not None and open_channel.readyState == "open":
            self.send_message_to_channel(open_channel, "print_document", payload)

async def sessions(self) -> List[Dict[str, Any]]:
    """The pages on this transport, with the round trip the peer's RTCP
    receiver reports carry for the video it is sent."""
    out = []
    for peer_id, obj in list(self.peer_connections.items()):
        peer_conn, rtt = obj.get("peer_conn"), None
        if peer_conn is not None and peer_conn.connectionState == "connected":
            try:
                report = await asyncio.wait_for(peer_conn.getStats(), 1.0)
                rtts = [s.roundTripTime for s in report.values()
                        if getattr(s, "type", "") == "remote-inbound-rtp" and s.roundTripTime]
                rtt = round(max(rtts) * 1000, 1) if rtts else None
            except Exception:
                rtt = None
        out.append({"id": peer_id, "transport": "webrtc",
                    "role": "controller" if obj.get("client_type") is ClientType.CONTROLLER else "viewer",
                    "slot": obj.get("client_slot"), "display": obj.get("display_id") or "primary",
                    "connected_at": audit.rfc3339(obj["connected_at"]) if obj.get("connected_at") else None,
                    "rtt_ms": rtt})
    return out

async def disconnect_peer(self, peer_id: str) -> bool:
    """Close the peer `peer_id` names; the state change then reaps it."""
    obj = self.peer_connections.get(peer_id)
    if not obj:
        return False
    peer_conn = obj.get("peer_conn")
    if peer_conn is not None:
        await peer_conn.close()
    return True

def capture_candidates(self) -> List[RTCDataChannel]:
    """The open channels of the peers that may capture a device: controllers on the primary
    display, in connection order. A shared viewer never captures."""
    return [peer["data_channel"] for peer in self.peer_connections.values()
            if peer.get("display_id") == "primary"
            and peer.get("client_type") is not ClientType.VIEWER
            and peer.get("data_channel") is not None
            and peer["data_channel"].readyState == "open"]

async def tell_capture(self, channel: RTCDataChannel, subject: str, wanted: bool) -> bool:
    """Sends one capture demand to one peer; False where its channel has closed."""
    if channel.readyState != "open":
        return False
    self.send_message_to_channel(
        channel, "system", {"action": f"capture_demand,{subject},{int(wanted)}"})
    return True

def send_framerate(self, framerate: int) -> None:
    """Broadcast the current framerate to all peers."""
    logger.debug("sending framerate")
    self.__send_data_channel_message(
        "system", {"action": "videoFramerate," + str(framerate)})

def send_video_bitrate(self, bitrate: int) -> None:
    """Broadcast the current video bitrate to all peers."""
    logger.debug("sending video bitrate")
    self.__send_data_channel_message(
        "system", {"action": "video_bitrate," + str(bitrate)})

def send_audio_bitrate(self, bitrate: int) -> None:
    """Broadcast the current audio bitrate to all peers."""
    logger.debug("sending audio bitrate")
    self.__send_data_channel_message(
        "system", {"action": "audio_bitrate,%d" % bitrate})

def send_encoder(self, encoder: str) -> None:
    """Broadcast the active encoder name to all peers."""
    logger.debug("sending encoder: " + encoder)
    self.__send_data_channel_message(
        "system", {"action": "encoder,%s" % encoder})

def send_resize_enabled(self, resize_enabled: bool) -> None:
    """Broadcast the current resize-enabled state to all peers."""
    logger.debug("sending resize enabled state")
    self.__send_data_channel_message(
        "system", {"action": "resize," + str(resize_enabled)})

def send_remote_resolution(self, res: str, display_id: str = "primary") -> None:
    """Send the realized remote resolution to the clients of `display_id`.

    Display-scoped: the websockets transport tags its stream_resolution
    with the display id and addresses only that page, so a secondary's
    realized size must never rescale the primary page.
    """
    logger.debug("sending remote resolution of: " + res)
    sent = False
    for peer_obj in self.peer_connections.values():
        if (peer_obj.get("display_id") or "primary") != display_id:
            continue
        peer_conn = peer_obj.get("peer_conn")
        channel = peer_obj.get("data_channel")
        if (
            peer_conn is not None
            and channel is not None
            and peer_conn.connectionState == "connected"
            and channel.readyState == "open"
        ):
            self.send_message_to_channel(
                channel, "system", {"action": "resolution," + res})
            sent = True
    if not sent:
        logger.debug("skipping remote resolution because no data channel is ready")

def send_ping(self, t: float) -> None:
    """Send a ping request to the PRIMARY controller only, while its page
    has its stats open.

    Latency is measured against one shared ping_start, and the websockets
    transport likewise derives its reported latency from the primary
    client.
    """
    state, data_channel = self.get_data_channel()
    if not state or not (self.get_controller_instance() or {}).get("stats"):
        return
    self.send_message_to_channel(
        data_channel, "ping", {"start_time": float("%.3f" % t)})

def send_latency_time(self, latency: float) -> None:
    """Send the measured latency in milliseconds to the controllers watching
    their stats."""
    for _, channel in self._controller_channels(subscribed=True):
        self.send_message_to_channel(channel, "latency_measurement", {"latency_ms": latency})

def get_data_channel(self) -> Tuple[bool, Optional[RTCDataChannel]]:
    """Return the controller's data channel and whether it is usable.

    Returns:
        A `(ready, channel)` pair: ready is True only when the controller's
        connection is connected and its channel open; channel is None when
        no controller exists.
    """
    state = False
    peer_obj = self.get_controller_instance()
    if not peer_obj:
        return state, None

    conn_state = peer_obj.get("peer_conn").connectionState
    data_channel_state = peer_obj.get("data_channel").readyState
    return conn_state == "connected" and data_channel_state == "open", peer_obj.get("data_channel")

def _iter_open_data_channels(self) -> Iterator[RTCDataChannel]:
    """Yield every connected peer's open data channel — controllers and
    viewers, all displays."""
    for peer_obj in self.peer_connections.values():
        peer_conn = peer_obj.get("peer_conn")
        channel = peer_obj.get("data_channel")
        if (
            peer_conn is not None
            and channel is not None
            and peer_conn.connectionState == "connected"
            and channel.readyState == "open"
        ):
            yield channel

def send_message_to_channel(self, channel: RTCDataChannel, msg_type: str,
                            data: Any) -> None:
    """Send one typed message to one specific peer's data channel.

    Payloads of 512 bytes or more (cursor PNGs, settings, clipboard, stats)
    are gzipped for a channel that completed the `_gz` handshake; smaller
    ones are not worth the CPU or the risk to input latency.
    """
    payload = json.dumps({"type": msg_type, "data": data})
    gz_payload = None
    if getattr(channel, "_selkies_gz_tx", False) and len(payload) >= 512:
        gz_payload = gzip.compress(payload.encode("utf-8"), 6)
    self._send_prepared_to_channel(channel, msg_type, payload, gz_payload)

def _send_prepared_to_channel(self, channel: RTCDataChannel, msg_type: str,
                              payload: str,
                              gz_payload: Optional[bytes]) -> None:
    """Guarded raw send of an already-serialized (and possibly
    pre-compressed) message; bulk senders reuse one compression across
    channels instead of re-gzipping per peer.

    A message oversized for the peer's negotiated max-message-size
    (`ValueError`) is dropped and logged, which beats the peer
    hard-closing the channel; one that finds the channel no longer open
    (`InvalidStateError`, a close racing the sender) is dropped like any
    not-ready channel's.
    """
    try:
        if gz_payload is not None and getattr(channel, "_selkies_gz_tx", False):
            channel.send(gz_payload)
        else:
            channel.send(payload)
    except ValueError as e:
        logger.error("dropping oversized data channel message '%s': %s", msg_type, e)
    except InvalidStateError:
        logger.debug("skipping message because data channel closed mid-send: %s" % msg_type)

def __send_data_channel_message(self, msg_type: str, data: Any) -> None:
    """Broadcast a typed message to every connected peer.

    All display controllers and viewers receive it — the websockets
    transport broadcasts cursor, stats, and clipboard to all of its
    clients, so the channel path must too. Channels that are not open are
    skipped.
    """
    if not self.peer_connections:
        return
    sent = False
    for channel in self._iter_open_data_channels():
        self.send_message_to_channel(channel, msg_type, data)
        sent = True
    if not sent:
        logger.debug("skipping message because no data channel is ready: %s" % msg_type)

def send_media_data_over_channel(self, msg_type: str, data: Any) -> None:
    """Broadcast a media-related message to all peers."""
    self.__send_data_channel_message(msg_type, data)

async def close_display_peers(self, display_id: str) -> None:
    """Close every peer attached to `display_id`.

    Each close is finished by the connection-state handler (channel
    consumers, media graph, display registration), which runs as its own
    task.
    """
    for peer_obj in [obj for obj in self.peer_connections.values()
                     if (obj.get("display_id") or "primary") == display_id]:
        peer_conn = peer_obj.get("peer_conn")
        if peer_conn is not None:
            try:
                await peer_conn.close()
            except Exception as e:
                logger.warning(f"Error closing a '{display_id}' peer: {e}")

def get_controller_instance(self) -> Optional[Dict[str, Any]]:
    """Return the peer entry for the controller client, if one exists.

    With multiple display controllers connected, the PRIMARY display's
    controller is the authoritative one (latency pings and
    controller-directed replies).
    """
    controllers = [
        obj for obj in self.peer_connections.values()
        if obj.get("client_type") == ClientType.CONTROLLER
    ]
    if not controllers:
        return None
    return next(
        (obj for obj in controllers if obj.get("display_id", "primary") == "primary"),
        controllers[0],
    )

def munge_sdp(self, sdp: str, encoder: Optional[str] = None,
              fullcolor: Optional[bool] = None,
              use_cpu: Optional[bool] = None) -> str:
    """Rewrite the local offer SDP for optimal streaming behavior.

    Injects a 125 ms rtx-time, `sps-pps-idr-in-keyframe=1` for H.264/H.265,
    the Opus ptime, and generous video bandwidth ceilings
    (`_munge_video_bandwidth`). Displays can run different encoders,
    chroma formats and software-encoding flags; the caller passes the ones
    this offer's display is using (defaults: the primary/global encoder and
    the configured full-color and software-encoding settings).

    Full color is a 4:4:4 bitstream, so the H.264 profile-level-id of
    the display's own video section is rewritten to High 4:4:4 (`f4001f`)
    rather than handing the decoder a 4:2:0 baseline profile that cannot
    match what it receives; 4:2:0 keeps `42e01f`, the profile Firefox
    negotiates. The recvonly webcam section, which the browser encodes,
    keeps the profiles it offered. A session known to encode on
    OpenH264 (the software encoder of a GPL-free pixelflux build, forced
    onto the CPU) is excluded: it always emits limited-range 4:2:0, and a
    4:4:4 profile makes decoders misread its color range (visibly darker
    output). A full-color VP9 display offers profile 1, the 4:4:4 profile,
    in place of profile 0; the client asks for full color only where its
    receiver takes that profile.

    The Opus ptime advertises the real frame duration pcmflux emits
    (`audio_frame_duration_ms`) so the client keys its minptime munge off
    it, rounded to whole milliseconds for browser SDP parsers.

    Args:
        sdp: The local offer SDP text.
        encoder: Encoder the display is running; None means the global one.
        fullcolor: Whether the display emits a 4:4:4 bitstream; None reads
            the configured setting.
        use_cpu: Whether the display forces software encoding; None reads
            the configured setting.

    Returns:
        The munged SDP text.
    """
    encoder = encoder or self.encoder
    if fullcolor is None:
        fullcolor = bool(app_settings.video_fullcolor[0])
    if use_cpu is None:
        use_cpu = bool(app_settings.use_cpu[0])
    software_path = software_video_path(
        encoder, use_cpu or str(getattr(app_settings, "gpu_id", "")).strip() == "-1")
    sdp_text = sdp
    if 'rtx-time' not in sdp_text:
        logger.warning("injecting rtx-time to SDP")
        sdp_text = re.sub(r'(apt=\d+)', r'\1;rtx-time=125', sdp_text)
    elif 'rtx-time=125' not in sdp_text:
        logger.warning("injecting modified rtx-time to SDP")
        sdp_text = re.sub(r'rtx-time=\d+', r'rtx-time=125', sdp_text)
    # The codec rewrites describe the display's own stream. The webcam
    # m-section is recvonly here and the browser's to encode: a profile it
    # has no encoder for makes it reject that section, and the camera's
    # sender is stopped with it.
    sections = re.split(r'(?m)(?=^m=)', sdp_text)
    for i, section in enumerate(sections):
        if not section.startswith('m=video') or 'a=recvonly' in section:
            continue
        if "h264" in encoder or "x264" in encoder or "h265" in encoder or "x265" in encoder:
            if 'sps-pps-idr-in-keyframe' not in section:
                section = section.replace('packetization-mode=', 'sps-pps-idr-in-keyframe=1;packetization-mode=')
            else:
                section = re.sub(r'sps-pps-idr-in-keyframe=\d+', 'sps-pps-idr-in-keyframe=1', section)
            if ("h264" in encoder or "x264" in encoder) and fullcolor \
                    and not (software_path and software_encoders().get("h264") == "openh264"):
                section = re.sub(r'profile-level-id=[0-9A-Fa-f]{6}', 'profile-level-id=f4001f', section)
        if "vp9" in encoder and fullcolor:
            section = re.sub(r'\bprofile-id=0\b', 'profile-id=1', section)
        sections[i] = section
    sdp_text = ''.join(sections)
    if "opus/" in sdp_text.lower():
        frame_ms = float(getattr(app_settings, 'audio_frame_duration_ms', '10') or 10)
        # A 2.5 ms frame advertises 3; pcmflux keeps the real frame.
        ptime = int(frame_ms + 0.5)
        # a=ptime is media-level: it must sit in the audio m-section, not after
        # the video sprop lines, and audio-less offers must skip it.
        if f"a=ptime:{ptime}" not in sdp_text:
            sections = re.split(r'(?m)(?=^m=)', sdp_text)
            for i, section in enumerate(sections):
                if section.startswith('m=audio'):
                    lines = section.split('\r\n')
                    lines.insert(1, f'a=ptime:{ptime}')
                    sections[i] = '\r\n'.join(lines)
                    break
            sdp_text = ''.join(sections)

    sdp_text = self._munge_video_bandwidth(sdp_text)

    return sdp_text

def _munge_video_bandwidth(self, sdp_text: str) -> str:
    """Raise the bandwidth ceiling of every video m-section in the SDP.

    A generous `b=AS` keeps the browser's REMB from throttling a
    high-bitrate desktop stream (it is a cap hint, not a target), and
    `x-google-max-bitrate` mirrors it on the Chrome receive side. Both are
    scoped to each video m-section: the `b=AS` presence check looks only
    inside that section (a `b=AS` elsewhere says nothing about this
    section's ceiling), and the line goes after the section's own `c=`,
    or right after the `m=video` line when the section inherits the
    session-level `c=` (RFC 4566). The x-google hints go on the fmtp of
    every video rtpmap payload type except RTX, so VP8/VP9 get them too;
    a codec without an fmtp gets one carrying just the hints.
    """
    XGOOGLE = "x-google-max-bitrate=300000;x-google-min-bitrate=0"
    lines = sdp_text.split("\r\n")
    out: List[str] = []
    i = 0
    n = len(lines)
    while i < n:
        line = lines[i]
        if not line.startswith("m=video"):
            out.append(line)
            i += 1
            continue

        section = [line]
        i += 1
        while i < n and not lines[i].startswith("m="):
            section.append(lines[i])
            i += 1

        if not any(s.startswith("b=AS:") for s in section):
            c_idx = next(
                (idx for idx, s in enumerate(section) if s.startswith("c=")),
                None,
            )
            insert_at = (c_idx + 1) if c_idx is not None else 1
            section.insert(insert_at, "b=AS:300000")

        video_pts = []
        for s in section:
            m = re.match(r'a=rtpmap:(\d+)\s+(\S+)', s)
            if m and not m.group(2).lower().startswith("rtx/"):
                video_pts.append(m.group(1))

        for pt in video_pts:
            fmtp_idx = next(
                (idx for idx, s in enumerate(section)
                 if s.startswith("a=fmtp:{} ".format(pt))),
                None,
            )
            if fmtp_idx is not None:
                if "x-google-max-bitrate" not in section[fmtp_idx]:
                    section[fmtp_idx] = re.sub(
                        r'^(a=fmtp:{} )'.format(pt),
                        r'\g<1>' + XGOOGLE + ';',
                        section[fmtp_idx],
                    )
            else:
                rtpmap_idx = next(
                    (idx for idx, s in enumerate(section)
                     if s.startswith("a=rtpmap:{} ".format(pt))),
                    None,
                )
                if rtpmap_idx is not None:
                    section.insert(rtpmap_idx + 1, "a=fmtp:{} {}".format(pt, XGOOGLE))

        out.extend(section)

    return "\r\n".join(out)

def consume_data(self, buf: Any, pts: Optional[int], kind: str,
                 keyframe: bool = True, display_id: str = "primary",
                 timing: Optional[tuple] = None,
                 dependency: Optional[tuple] = None) -> None:
    """Feed one encoded frame from the capture side into a display's bridge.

    Synchronous: scheduled via `loop.call_soon_threadsafe` from the capture
    thread, since `set_data` does not await — no per-frame Future/Task.
    `EncodedPacket` references the encoder buffer without copying and the
    packers walk it as a memoryview, so no per-frame FFmpeg object is
    allocated and no whole-frame copy is taken.

    Args:
        buf: Buffer-protocol object holding the encoded sample.
        pts: Presentation timestamp in the stream's clock, or None.
        kind: "video" or "audio".
        keyframe: Whether the sample decodes on its own; a video delta
            frame needs the one before it.
        display_id: Display whose media graph receives the sample.
        timing: The frame's capture and encode instants as the capture
            library stamped them, for the video-timing extension.
        dependency: The frame's id and the id of the frame it predicts
            from, where the encoder tracks them, for the dependency
            descriptor and the bridge's drops.
    """
    graph = self.displays.get(display_id or "primary")
    if graph is None:
        return
    if kind == "video":
        if buf:
            try:
                RTP_VIDEO_CLOCK_RATE = 90000
                packet = EncodedPacket(buf, pts, Fraction(1, RTP_VIDEO_CLOCK_RATE), keyframe, timing, dependency)
                bridge = graph.get("video_bridge")
                if bridge is not None:
                    bridge.set_data(packet, keyframe)
            except Exception as e:
                logger.error(f"error processing video sample: {e}")
    elif kind == "audio":
        if buf:
            try:
                packet = EncodedPacket(buf, pts, Fraction(1, 48000))
                bridge = graph.get("audio_bridge")
                if bridge is not None:
                    bridge.set_data(packet)
            except Exception as e:
                logger.error(f"error processing audio sample: {e}")

def bridge_drops(self) -> Dict[str, tuple]:
    """Frames each display's video bridge has dropped, and how many of those the
    encoder was told to predict past, by display id.

    A drop happens before the sender packetizes, so no sequence number is
    spent and neither `packetsLost` nor the pacer's counters move. Rising
    here with those flat is a lagging sender, not a lossy link; the cost is
    frame rate rather than a smear, since a stream whose encoder names its
    references carries on from the next frame and one that does not holds
    the picture still until the keyframe the bridge asks for.
    """
    return {did: (graph["video_bridge"].dropped, graph["video_bridge"].invalidated)
            for did, graph in self.displays.items()
            if graph.get("video_bridge") is not None}

def update_rtc_config(self, stun_servers: List[str], turn_servers: List[str]) -> None:
    """Update the STUN/TURN servers used for every NEW peer connection.

    get_rtc_config() reads these at peer-creation time, so a refresh (typically
    rotated TURN REST credentials) takes effect for every subsequent connection.
    Live sessions deliberately keep their established ICE: their TURN allocations
    stay valid, and forcing an ICE restart on refresh would drop working streams.
    """
    changed = (stun_servers, turn_servers) != (self.stun_servers, self.turn_servers)
    self.stun_servers = stun_servers
    self.turn_servers = turn_servers
    if changed:
        logger.debug(
            "RTC ICE servers updated; applies to new connections "
            "(established sessions keep their current ICE)."
        )

def format_turn_servers(self, turn_servers: List[str]) -> List[Dict[str, Optional[str]]]:
    """Parse turn:// or turns:// URL strings into ICE server dicts.

    Non-TURN or unparsable entries are skipped; missing ports fall back to
    the scheme default, bare IPv6 hosts are bracketed, and URL-encoded
    credentials are decoded.

    Returns:
        One dict per valid server with `urls` and, when the URL carried
        them, `username` / `credential` keys.
    """
    formatted_servers: List[Dict[str, Optional[str]]] = []
    for server in turn_servers or []:
        if not isinstance(server, str):
            continue

        lower_server = server.lower()
        if not (lower_server.startswith("turn://") or lower_server.startswith("turns://")):
            continue

        parsed = urllib.parse.urlparse(server)
        if not parsed.hostname:
            continue

        scheme = 'turns' if parsed.scheme.lower() == 'turns' else 'turn'
        try:
            port = parsed.port or (443 if scheme == 'turns' else 3478)
        except ValueError:
            port = 443 if scheme == 'turns' else 3478

        host = parsed.hostname
        if host and ":" in host and not (host.startswith("[") and host.endswith("]")):
            host = f"[{host}]"

        query = f"?{parsed.query}" if parsed.query else ""
        turn_entry: Dict[str, Optional[str]] = {
            'urls': f'{scheme}:{host}:{port}{query}'
        }

        if parsed.username is not None and parsed.password is not None:
            turn_entry['username'] = urllib.parse.unquote(parsed.username)
            turn_entry['credential'] = urllib.parse.unquote(parsed.password)

        formatted_servers.append(turn_entry)
    return formatted_servers

def format_stun_servers(self, stun_servers: List[str]) -> List[str]:
    """Strip the URL scheme separator from each STUN server string."""
    formatted_servers: List[str] = []
    for stun in stun_servers:
        server = stun.split("//")
        formatted_servers.append("".join(server))
    return formatted_servers

@staticmethod
def ice_lite_enabled() -> bool:
    """Whether `webrtc_ice_lite` runs every peer's ICE agent as ICE-lite."""
    return bool(getattr(app_settings, "webrtc_ice_lite", (False, False))[0])

async def open_ice_muxes(self) -> None:
    """Bind the shared ICE ports the settings name, ahead of the first peer.

    `webrtc_udp_mux_port` binds one UDP socket per host address and
    `webrtc_tcp_mux_port` one TCP listener, once for the service; every
    peer's gatherer then shares them (`ice.mux`). A port in use fails here,
    at startup, rather than on a session. An address that appears later is
    bound on first use by the gatherer.

    Raises:
        OSError: A mux port could not be bound; nothing stays bound.
    """
    udp_port = int(getattr(app_settings, "webrtc_udp_mux_port", 0) or 0)
    tcp_port = int(getattr(app_settings, "webrtc_tcp_mux_port", 0) or 0)
    if not udp_port and not tcp_port:
        return
    addresses = get_host_addresses(use_ipv4=True, use_ipv6=True)
    listed = ", ".join(addresses) or "no host address yet"
    if udp_port:
        mux = UdpMux(udp_port)
        try:
            await mux.open(addresses)
        except OSError as exc:
            logger.error(f"WebRTC UDP mux: cannot bind port {udp_port}: {exc}")
            raise
        self.ice_udp_mux = mux
        logger.info(f"WebRTC UDP mux: every session's host candidates share UDP port {udp_port} on {listed}")
        if (getattr(app_settings, "webrtc_port_range", "") or "").strip():
            logger.warning("webrtc_port_range is unused while webrtc_udp_mux_port is set: the mux port is the only UDP port")
    if tcp_port:
        mux = TcpMux(tcp_port)
        try:
            await mux.open(addresses)
        except OSError as exc:
            logger.error(f"WebRTC TCP mux: cannot bind port {tcp_port}: {exc}")
            await self.close_ice_muxes()
            raise
        self.ice_tcp_mux = mux
        logger.info(f"WebRTC TCP mux: ICE-TCP accepted on TCP port {tcp_port} on {listed}")

async def close_ice_muxes(self) -> None:
    """Release the shared ICE ports, ending every peer's ICE that rode them."""
    muxes = [mux for mux in (self.ice_udp_mux, self.ice_tcp_mux) if mux is not None]
    self.ice_udp_mux = None
    self.ice_tcp_mux = None
    for mux in muxes:
        await mux.close()

def get_rtc_config(self) -> RTCConfiguration:
    """Build the RTCConfiguration for a new peer from the current servers.

    Operator-configured public addresses (`webrtc_public_ip`, comma- or
    space-separated IPv4/IPv6) are advertised in host ICE candidates for
    hosts behind static 1:1 NAT; each family maps to its own host
    candidates. The shared mux sockets, when opened, and the ICE-lite
    choice travel the same way; an ICE-lite agent gathers host candidates
    only, so the STUN and TURN servers are left out of its configuration
    and serve the client side alone.
    """
    formatted_turn_servers = self.format_turn_servers(self.turn_servers)
    formatted_stun_servers = self.format_stun_servers(self.stun_servers)
    logger.debug(f"stun servers: {formatted_stun_servers}")
    logger.debug(f"turn servers: {formatted_turn_servers}")

    ice_servers = []
    if self.stun_servers:
        ice_servers.append(RTCIceServer(urls=formatted_stun_servers))
    for turn in formatted_turn_servers:
        turn_kwargs: Dict[str, Any] = {
            'urls': turn.get('urls', [])
        }
        if turn.get('username') is not None:
            turn_kwargs['username'] = turn.get('username')
        if turn.get('credential') is not None:
            turn_kwargs['credential'] = turn.get('credential')
        ice_servers.append(RTCIceServer(**turn_kwargs))
    public_ips = (
        getattr(app_settings, "webrtc_public_ip", "") or ""
    ).replace(",", " ").split()
    ice_lite = self.ice_lite_enabled()
    config = RTCConfiguration(
        iceServers=[] if ice_lite else ice_servers,
        bundlePolicy=RTCBundlePolicy.MAX_BUNDLE,
        iceHostPublicIps=public_ips or None,
        icePortRange=parse_webrtc_port_range(
            getattr(app_settings, "webrtc_port_range", "") or ""
        ),
        iceUdpMux=self.ice_udp_mux,
        iceTcpMux=self.ice_tcp_mux,
        iceLite=ice_lite,
    )
    return config

def prefer_codec(self, pc: RTCPeerConnection, sender: RTCRtpSender,
                 preferred_mime: str) -> None:
    """Order a sender's codec preferences: one MIME type first, the other
    video codecs down the fallback ladder behind it, RTX.

    Every codec matching the MIME type stays eligible — H.264 appears once
    per advertised profile. Behind it come the encoder menu's other codecs
    in `encoder_rung` order, the host's hardware codecs, most efficient first,
    before its software ones by encode time, so a browser that declines the codec
    answers with the first of them it decodes (`_settle_video_codec` reads
    which one it took and moves the display there); the codecs off the menu
    follow, negotiated so a live encoder switch (`switch_display_codec`)
    changes the payload type without renegotiating. FlexFEC rides along when the
    receiver supports it (Chrome family); a receiver without it answers
    without the codec and the sender emits no repair stream.

    Args:
        pc: Peer connection owning the sender's transceiver.
        sender: RTP sender whose transceiver is being ordered.
        preferred_mime: MIME type (e.g. "video/VP9") to put first.

    Raises:
        ValueError: When the codec or its RTX companion is not in the
            sender capabilities.
    """
    kind = sender.track.kind
    capabilities = RTCRtpSender.getCapabilities(kind)
    logger.debug(f"Current capabilities for {kind}: {capabilities}")

    chosen_codec = []
    for codec in capabilities.codecs:
        if codec.mimeType == preferred_mime:
            chosen_codec.append(codec)

    if not chosen_codec:
        raise ValueError(f"Codec {preferred_mime} not found in capabilities")
    # The menu's other codecs down the ladder, so a declined codec lands on the next
    # one the peer decodes and the display can move to, then the codecs off the menu,
    # negotiated so a later encoder switch finds its payload type in place.
    menu = next(d for d in SETTING_DEFINITIONS if d["name"] == "encoder")["meta"]["allowed"]
    backends = app_settings.encoder_backends()
    ladder = sorted(WEBRTC_ENCODER_CHOICES, key=lambda enc: (enc not in menu, encoder_rung(enc, backends)))
    for mime in (self.get_mime_by_encoder(enc) for enc in ladder):
        if mime == preferred_mime:
            continue
        chosen_codec += [c for c in capabilities.codecs
                         if c.mimeType == mime and c not in chosen_codec]

    rtx_codec = None
    for codec in capabilities.codecs:
        if codec.mimeType.lower() == f"{kind}/rtx":
            rtx_codec = codec
            break

    if not rtx_codec:
        raise ValueError(f"RTX codec for {preferred_mime} not found")

    flexfec_codec = next(
        (
            codec
            for codec in capabilities.codecs
            if codec.mimeType.lower() == f"{kind}/flexfec-03"
        ),
        None,
    )
    preferences = [*chosen_codec, rtx_codec]
    if flexfec_codec is not None:
        preferences.append(flexfec_codec)

    transceiver = next(t for t in pc.getTransceivers() if t.sender == sender)
    logger.debug(f"Codec preferences: {preferences}")
    transceiver.setCodecPreferences(preferences)

def switch_display_codec(self, display_id: str, encoder: str) -> None:
    """Move every peer of a display to the codec its new encoder streams.

    A peer whose answer took the codec sends it from the capture's next
    frame under its own payload type; one that did not is told, and its
    video pauses until an encoder it took is chosen. A peer still
    negotiating settles at its answer.
    """
    mime = self.get_mime_by_encoder(encoder)
    for peer_id, peer in list(self.peer_connections.items()):
        if (peer.get("display_id") or "primary") != display_id:
            continue
        sender = peer.get("video_sender")
        if sender is None:
            continue
        peer["video_mime"] = mime
        if sender.switch_codec(mime):
            logger.info(f"Video for peer {peer_id} on display '{display_id}' switched to {mime}")
        else:
            logger.error(f"Peer {peer_id} did not negotiate {mime}: its video pauses "
                         f"until display '{display_id}' runs an encoder it takes.")

async def _settle_fullcolor(self, client_peer_id: str, display_id: str, encoder: str,
                            fullcolor_codecs: List[str]) -> bool:
    """Whether the offer to a peer may describe 4:4:4.

    The peer's hello named the codecs it decodes at 4:4:4. When the display's
    codec is not among them, full color goes off for the display through
    `on_fullcolor_declined` before the offer is built, so the stream is 4:2:0
    from its first frame rather than a profile the peer paints nothing of;
    a display whose full color the operator holds keeps it, and the peer is
    told so in the log.

    Args:
        client_peer_id: The joining peer.
        display_id: The display it joins.
        encoder: The display's encoder.
        fullcolor_codecs: The codec names the peer decodes at 4:4:4.

    Returns:
        Whether the display still emits 4:4:4.
    """
    codec = FULLCOLOR_CODECS.get(encoder)
    if codec is None or codec in fullcolor_codecs:
        return True
    # An encoder this host carries no 4:4:4 for streams 4:2:0 whatever is asked, so the
    # offer describes that and the setting stays for one that does carry it.
    try:
        use_cpu = bool(self.get_use_cpu_for_display(display_id))
    except Exception:
        use_cpu = bool(app_settings.use_cpu[0])
    if app_settings.encoder_fullcolor(encoder, use_cpu) is False:
        return False
    moved = False
    if self.on_fullcolor_declined is not None:
        try:
            moved = bool(await self.on_fullcolor_declined(display_id))
        except Exception:
            logger.warning("on_fullcolor_declined failed", exc_info=True)
    if moved:
        logger.info(f"Peer {client_peer_id} decodes no {codec} 4:4:4: display '{display_id}' streams 4:2:0.")
        return False
    logger.error(f"Peer {client_peer_id} decodes no {codec} 4:4:4 and the full color of "
                 f"display '{display_id}' is held: it will paint nothing of this stream.")
    return True

async def _settle_video_codec(self, client_peer_id: str, peer_obj: Dict[str, Any]) -> None:
    """Take the video codec a peer's answer settled on.

    The offer put the display's codec first and the menu's other codecs
    behind it down the ladder, so a browser that declines the codec answers
    with the next one it decodes. The display then moves to that encoder
    through `on_video_codec_declined`, or, where the display's full color
    is held, to the first of the answer's menu codecs the peer decodes at
    4:4:4 too or that this host encodes 4:2:0 anyway, the step the
    WebSocket ladder takes; a display whose encoder the operator holds
    stops sending video to this peer instead, since one codec's bitstream
    must never be packed as another's, and tells the page so once its
    channel is open.
    """
    sender = peer_obj.get("video_sender")
    wanted = peer_obj.get("video_mime")
    if sender is None or wanted is None:
        return
    transceiver = next(
        (t for t in peer_obj["peer_conn"].getTransceivers() if t.sender is sender), None)
    if transceiver is None:
        return
    answered = [c.mimeType for c in transceiver._codecs
                if not c.mimeType.lower().endswith(("/rtx", "/flexfec-03"))]
    if not answered:
        return
    display_id = peer_obj.get("display_id") or "primary"
    logger.info(f"Video for peer {client_peer_id} on display '{display_id}' negotiated {answered[0]}")
    if answered[0].lower() == wanted.lower():
        return
    # The answer lists the codecs this peer decodes in the offer's order, the menu's first,
    # so its first is the rung the display moves to. Under a held full color the rung is the
    # first menu codec whose 4:4:4 the peer decodes too, or that this host encodes 4:2:0
    # anyway, and the first stands where there is none.
    by_mime = {self.get_mime_by_encoder(enc).lower(): enc for enc in WEBRTC_ENCODER_CHOICES}
    candidates = [by_mime[mime.lower()] for mime in answered if mime.lower() in by_mime] or ["h264enc"]
    fullcolor = peer_obj.get("fullcolor_codecs")
    display_fullcolor = fullcolor is not None and bool(self.get_fullcolor_for_display(display_id))
    if display_fullcolor and bool(app_settings.video_fullcolor[1]):
        try:
            use_cpu = bool(self.get_use_cpu_for_display(display_id))
        except Exception:
            use_cpu = bool(app_settings.use_cpu[0])
        menu = next(d for d in SETTING_DEFINITIONS if d["name"] == "encoder")["meta"]["allowed"]
        fits = [enc for enc in candidates if enc in menu and (
            FULLCOLOR_CODECS.get(enc) is None or FULLCOLOR_CODECS[enc] in fullcolor
            or app_settings.encoder_fullcolor(enc, use_cpu) is False)]
        candidates = fits or candidates
    taken = candidates[0]
    # Full color was settled for the codec offered; the one taken may carry a 4:4:4 this
    # peer decodes no better, so it is settled again before the display moves.
    if display_fullcolor:
        await self._settle_fullcolor(client_peer_id, display_id, taken, fullcolor)
    moved = False
    if self.on_video_codec_declined is not None:
        try:
            moved = bool(await self.on_video_codec_declined(display_id, wanted, taken))
        except Exception:
            logger.warning("on_video_codec_declined failed", exc_info=True)
    if not moved:
        logger.error(f"Peer {client_peer_id} declined {wanted} and the encoder of "
                     f"display '{display_id}' is held: its video stays off for this peer.")
        sender._enabled = False
        peer_obj["video_declined"] = wanted
        self._send_video_declined(peer_obj["data_channel"], client_peer_id)