SelkiesStreamingApp

def __init__(
    self,
    async_event_loop: asyncio.AbstractEventLoop,
    framerate: int,
    encoder: str,
    data_streaming_server: Optional["DataStreamingServer"] = None,
    mode: str = "websockets",
) -> None:
    self.server_enable_resize = ENABLE_RESIZE
    self.mode = mode
    self.display_width = 1024
    self.display_height = 768
    if settings.manual_resolution[0]:
        manual_w = int(settings.manual_width or 0)
        manual_h = int(settings.manual_height or 0)
        if manual_w > 0 and manual_h > 0:
            self.display_width = manual_w - (manual_w % 2)
            self.display_height = manual_h - (manual_h % 2)
    self.pipeline_running = False
    self.async_event_loop = async_event_loop
    self.audio_channels = int(getattr(settings, 'audio_channels', AUDIO_CHANNELS_DEFAULT)
                              or AUDIO_CHANNELS_DEFAULT)
    self.gpu_id = GPU_ID_DEFAULT
    self.audio_bitrate = AUDIO_BITRATE_DEFAULT
    self.encoder = encoder
    self.framerate = framerate
    self.last_cursor_sent = None
    self.data_streaming_server = data_streaming_server

async def send_ws_clipboard_data(
    self,
    data: Union[str, bytes],
    mime_type: str = "text/plain",
    reply_to: Optional[str] = None,
    conn_id: Optional[int] = None,
) -> None:
    """Send clipboard data to the session's clients, multipart when large.

    Args:
        data: Clipboard text (str) or binary payload (bytes).
        mime_type: The payload's MIME type; anything but "text/plain" is
            treated as binary and gated on enable_binary_clipboard.
        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. A `clipboard_reply,<verb>` frame then
            precedes the payload frames on the same ordered socket, so
            clients can treat the payload cache-only without time
            heuristics. Legacy clients route the unknown verb to their
            input module, which ignores it.
        conn_id: Connection that asked for this payload. An answer goes to
            that client alone: every other one already holds the content or
            is about to be told of a change, and a tagged reply they did
            not ask for is read as their own fetch and cached without ever
            reaching their clipboard.

    Payload frames get the bulk tolerance the data channel's drain allows
    (a slow link is not a dead client); control frames keep the liveness
    bound, since one stalled client must not wedge clipboard delivery for
    all.
    """
    if not (self.data_streaming_server and self.data_streaming_server.clients):
        data_logger.warning("Cannot send clipboard: no clients or server not ready.")
        return
    try:
        is_binary = mime_type != "text/plain"
        # Markup travels on the binary verbs because it carries a mime, but
        # it is text: the image gate is not its gate.
        if is_binary and not mime_type.startswith("text/") \
                and mime_type != CLIPBOARD_FLAVOURS_MIME \
                and not self.data_streaming_server.enable_binary_clipboard:
            data_logger.warning(
                f"Attempted to send binary clipboard data ({mime_type}) but feature is disabled on server."
            )
            return
        data_bytes = data.encode('utf-8') if not is_binary and isinstance(data, str) else data
        total_size = len(data_bytes)
        if total_size:
            audit.emit("clipboard.send", mime_type=mime_type, size_bytes=total_size)
        clients = self.data_streaming_server.clients
        # One payload at a time per client: the start/data/finish frames
        # carry no transfer id, so a send racing another (a clipboard
        # change during the connect push) would interleave two payloads'
        # chunks into one assembly -- and a reply tag must precede its own
        # payload, nothing else's.
        locks = self.__dict__.setdefault("_clipboard_send_locks", {})
        live = {id(c) for c in list(clients)}
        for gone in [k for k in locks if k not in live]:
            del locks[gone]
        small = total_size < CLIPBOARD_CHUNK_SIZE
        if small:
            encoded_data = base64.b64encode(data_bytes).decode('ascii')
            if is_binary:
                message = f"clipboard_binary,{mime_type},{encoded_data}"
            else:
                message = f"clipboard,{encoded_data}"
        else:
            data_logger.debug(f"Sending large clipboard data ({mime_type}, {total_size} bytes) via multipart.")
            start_message = f"clipboard_start,{mime_type},{total_size}"

        async def deliver(client: Any) -> None:
            """One pipeline per client, a chunk at a time: a slow link
            paces only its own transfer and a dead one drops out of the
            set without touching anyone else's. Each chunk is held
            twice, so the audio and input sharing the connection are
            never queued behind more than one of them: for this host's
            own send queue to drain (`_await_bulk_window`) and for the
            rate the whole path has been taking (`_bulk_pace`), which
            is the only one of the two a buffer in front can hide."""
            cid = id(client)
            async with locks.setdefault(cid, asyncio.Lock()):
                if reply_to:
                    await _broadcast_to_clients(clients, f"clipboard_reply,{reply_to}",
                                                per_client_timeout=2.0, only=cid)
                if small:
                    await _broadcast_to_clients(clients, message,
                                                per_client_timeout=BULK_DRAIN_TIMEOUT_S,
                                                only=cid)
                    return
                if await _broadcast_to_clients(clients, start_message,
                                               per_client_timeout=2.0, only=cid):
                    return
                offset = 0
                loop = asyncio.get_running_loop()
                gauge = socket_gauge(client)
                pacer = TransferPacer(adaptive=True)
                while offset < total_size:
                    chunk = data_bytes[offset:offset + CLIPBOARD_CHUNK_SIZE]
                    data_message = "clipboard_data," + base64.b64encode(chunk).decode('ascii')
                    await _bulk_pace(gauge, pacer, len(data_message))
                    await _await_bulk_window(client, loop.time() + BULK_DRAIN_TIMEOUT_S)
                    if await _broadcast_to_clients(clients, data_message,
                                                   per_client_timeout=BULK_DRAIN_TIMEOUT_S, only=cid):
                        return
                    offset += len(chunk)
                    await asyncio.sleep(0)
                await _broadcast_to_clients(clients, "clipboard_finish",
                                            per_client_timeout=2.0, only=cid)

        recipients = [c for c in list(clients) if conn_id is None or id(c) == conn_id]
        await asyncio.gather(*(deliver(c) for c in recipients))
        if not small:
            data_logger.debug("Finished sending multi-part clipboard data.")
    except Exception as e:
        data_logger.error(f"Failed to send clipboard data: {e}", exc_info=True)

def send_ws_cursor_data(self, data: dict) -> None:
    """Broadcast a cursor-change payload to all clients.

    Thread-safe: called from pixelflux's cursor thread, so the broadcast is
    scheduled onto the event loop via run_coroutine_threadsafe rather than
    awaited. The payload is also cached as last_cursor_sent so late-joining
    clients receive the current cursor at connect.
    """
    self.last_cursor_sent = data
    if (
        self.data_streaming_server
        and hasattr(self.data_streaming_server, "clients")
        and self.data_streaming_server.clients
        and self.async_event_loop
        and self.async_event_loop.is_running()
    ):

        msg_str = json.dumps(data)
        msg_to_broadcast = f"cursor,{msg_str}"
        clients_ref = self.data_streaming_server.clients

        async def _broadcast_cursor_helper():
            """Bounded: cursor changes arrive at high rate, and a stalled
            client would otherwise accumulate one blocked coroutine each."""
            await _broadcast_to_clients(clients_ref, msg_to_broadcast, per_client_timeout=2.0)

        asyncio.run_coroutine_threadsafe(
            _broadcast_cursor_helper(), self.async_event_loop
        )
    else:
        data_logger.warning("Cannot broadcast cursor data: no clients connected or server not ready.")

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

    Addressed to the connection that asked for it when one is named, as an
    answer that belongs to one client rather than to the session;
    broadcast otherwise.
    """
    if (
        self.data_streaming_server
        and getattr(self.data_streaming_server, "clients", None)
        and self.async_event_loop
        and self.async_event_loop.is_running()
    ):
        msg = "system," + json.dumps({"action": action})
        clients_ref = self.data_streaming_server.clients

        async def _broadcast_system_helper():
            await _broadcast_to_clients(clients_ref, msg, per_client_timeout=2.0,
                                        only=conn_id)

        asyncio.run_coroutine_threadsafe(
            _broadcast_system_helper(), self.async_event_loop
        )

async def stop_pipeline(self) -> None:
    """Stop all pipelines by reconciling displays against current state."""
    logger_app.debug("Stopping pipelines (generic call)...")
    if self.data_streaming_server:
        await self.data_streaming_server.reconfigure_displays()
    self.pipeline_running = False
    logger_app.debug("Pipelines stop signal processed.")

def set_framerate(self, framerate: Union[int, float]) -> None:
    """Store the session default framerate; applies at the next pipeline (re)start."""
    self.framerate = int(framerate)
    logger_app.debug(
        f"Framerate for {self.encoder} set to {self.framerate}. Restart pipeline if active."
    )