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"
        if is_binary 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
        if reply_to:
            await _broadcast_to_clients(
                self.data_streaming_server.clients,
                f"clipboard_reply,{reply_to}", per_client_timeout=2.0,
                only=conn_id)
        data_bytes = data.encode('utf-8') if not is_binary and isinstance(data, str) else data
        total_size = len(data_bytes)
        if total_size < CLIPBOARD_CHUNK_SIZE:
            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}"
            await _broadcast_to_clients(self.data_streaming_server.clients, message,
                                        per_client_timeout=BULK_DRAIN_TIMEOUT_S, only=conn_id)
        else:
            data_logger.info(f"Sending large clipboard data ({mime_type}, {total_size} bytes) via multipart.")
            start_message = f"clipboard_start,{mime_type},{total_size}"
            clients = self.data_streaming_server.clients

            async def deliver(cid: int) -> 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."""
                if await _broadcast_to_clients(clients, start_message,
                                               per_client_timeout=2.0, only=cid):
                    return
                offset = 0
                while offset < total_size:
                    chunk = data_bytes[offset:offset + CLIPBOARD_CHUNK_SIZE]
                    data_message = "clipboard_data," + base64.b64encode(chunk).decode('ascii')
                    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 = [id(c) for c in list(clients) if conn_id is None or id(c) == conn_id]
            await asyncio.gather(*(deliver(cid) for cid in recipients))
            data_logger.info("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) -> None:
    """Broadcast a system action (e.g. ``command_error,<text>``) to clients."""
    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)

        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.info("Stopping pipelines (generic call)...")
    if self.data_streaming_server:
        await self.data_streaming_server.reconfigure_displays()
    self.pipeline_running = False
    logger_app.info("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.info(
        f"Framerate for {self.encoder} set to {self.framerate}. Restart pipeline if active."
    )