diff --git a/modules/config/schemas_v1.py b/modules/config/schemas_v1.py index 76b0fab55..1cfbecd26 100644 --- a/modules/config/schemas_v1.py +++ b/modules/config/schemas_v1.py @@ -5,7 +5,7 @@ from typing import Literal from confz import BaseConfig -from pydantic import ConfigDict, Field +from pydantic import ConfigDict, Field, model_validator from pydantic.types import Annotated, ClassVar, NonNegativeInt, PositiveInt @@ -211,6 +211,7 @@ class HTTP(BaseConfig): filename: ClassVar = "http.yml" http_server: HTTPServer = Field(default_factory=lambda: HTTPServer()) + webrtc: WebRTC = Field(default_factory=lambda: WebRTC()) class OBS(BaseConfig): @@ -242,6 +243,29 @@ class HTTPServer(BaseConfig): port: Annotated[int, Field(gt=0, lt=65536)] = 8888 +class WebRTC(BaseConfig): + """Schema for the webrtc section in the HTTP config.""" + + enabled: bool = False + turn: WebRTCTurn | None = None + + +class WebRTCTurn(BaseConfig): + """Schema for the webrtc.turn section in the HTTP config.""" + + urls: list[str] + secret: str + + @model_validator(mode="after") + def check_turn_settings(self) -> "WebRTCTurn": + if len(self.urls) == 0: + raise ValueError("`urls` must contain at least one TURN server URL.") + for url in self.urls: + if not url.startswith(("turn:", "turns:")): + raise ValueError(f"TURN server URL '{url}' must start with 'turn:' or 'turns:'.") + return self + + class ProfileMetadata(BaseConfig): """Schema for the metadata configuration file part of profiles.""" diff --git a/modules/config/templates/http.yml b/modules/config/templates/http.yml index 98189345a..9cfd5a5e3 100644 --- a/modules/config/templates/http.yml +++ b/modules/config/templates/http.yml @@ -4,3 +4,18 @@ http_server: enable: false ip: 127.0.0.1 port: 8888 + +# WebRTC can be used to stream video and audio data to the web UI. It also has a +# channel for inputs, so the bot can be controlled remotely. +# Requires the `aiortc` package. +webrtc: + enabled: false + # Only needed if browsers cannot reach the bot's IP address directly, e.g. if the bot runs + # in a Docker container or in a NAT environment. + turn: null + #turn: + # urls: + # - turn:bot.example.com:3478?transport=udp + # - turn:bot.example.com:3478?transport=tcp + # # Shared secret, the same as coturn's `static-auth-secret` + # secret: long-random-string diff --git a/modules/web/http.py b/modules/web/http.py index b73c61f1d..ca1d07312 100644 --- a/modules/web/http.py +++ b/modules/web/http.py @@ -1,13 +1,10 @@ import asyncio -import fractions import io -import json import queue import re -import time -from queue import Queue +import sys from threading import Thread -from typing import Callable, TypeVar, Union +from typing import Callable, TypeVar import aiohttp.client_exceptions from aiohttp import web @@ -17,24 +14,6 @@ from modules.daycare import get_daycare_data from modules.game import get_current_game_data -try: - from aiortc import ( - MediaStreamTrack, - VideoStreamTrack, - RTCConfiguration, - RTCPeerConnection, - RTCSessionDescription, - ) - from aiortc.contrib.media import MediaRelay - from aiortc.mediastreams import MediaStreamError - from av import VideoFrame, Packet, AudioFrame, AudioResampler - from PIL import Image - from av.frame import Frame - - webrtc_available = True -except ImportError: - webrtc_available = False - from modules.console import console from modules.context import context from modules.items import get_item_bag, get_item_storage @@ -48,11 +27,20 @@ from modules.pokedex import get_pokedex from modules.pokemon_party import get_party from modules.pokemon_storage import get_pokemon_storage -from modules.runtime import get_base_path +from modules.runtime import get_base_path, is_bundled_app from modules.state_cache import StateCacheItem from modules.version import pokebot_version, pokebot_name from modules.web.http_stream import add_subscriber +try: + from modules.web import webrtc +except ModuleNotFoundError as error: + # WebRTC is optional, so it's fine if its packages are missing (see `_get_webrtc_error()`.) + # Any other missing module is a bug though. + if error.name is None or error.name.split(".")[0] not in ("aiortc", "av"): + raise + webrtc = None + custom_state: dict = {} T = TypeVar("T") @@ -138,11 +126,32 @@ def update_inputs(): work_queue.put_nowait(update_inputs) +def _get_webrtc_error() -> str | None: + """ + :return: A message explaining why WebRTC cannot be used, or `None` if it can. + """ + if not context.config.http.webrtc.enabled: + return "WebRTC is disabled. Set `webrtc.enabled` in `http.yml` to enable it." + + if webrtc is None: + if is_bundled_app(): + return "WebRTC is not available in the bundled version of the bot." + return ( + "WebRTC requires the `aiortc` package, which is not installed. You can install it with:\n" + f'"{sys.executable}" -m pip install "aiortc~=1.10.0"' + ) + + return None + + def http_server(host: str, port: int) -> web.AppRunner: """ Run Flask server to make bot data available via HTTP requests. """ + if context.config.http.webrtc.enabled and (webrtc_error := _get_webrtc_error()) is not None: + console.print(webrtc_error, style="yellow", markup=False, highlight=False, soft_wrap=True) + server = web.Application() route = web.RouteTableDef() @@ -808,238 +817,120 @@ async def http_get_events_stream(request: web.Request): return response - if webrtc_available: - rtc_connections: set[RTCPeerConnection] = set() + @route.get("/rtc/config") + async def http_get_rtc_config(request: web.Request): + """ + --- + get: + description: > + Returns the ICE configuration (TURN servers and short-lived credentials) that a + browser should use for its `RTCPeerConnection` before calling `/rtc`. + responses: + 200: + content: + application/json: {} + 503: + description: WebRTC is disabled or not installed. + tags: + - streams + """ + if (error := _get_webrtc_error()) is not None: + return web.Response(text=error, status=503, content_type="text/plain") - class EmuVideo(VideoStreamTrack): - def __init__(self): - super().__init__() - self._black_image: Image.Image = Image.new("RGB", (240, 160)) + return web.json_response(webrtc.get_client_config(), headers={"Cache-Control": "no-store"}) - async def recv(self) -> Union[Frame, Packet]: - pts, time_base = await self.next_timestamp() + @route.post("/rtc") + async def http_post_rtc(request: web.Request): + """ + --- + post: + description: > + Answers a browser's WebRTC offer and sets up a connection that streams video and audio, and + receives input via a data channel labelled `input`. The response contains the answer's `sdp` + and `type`, as well as an `id` for the new connection that can be passed to `/rtc/close`. + requestBody: + description: The offer's session description. + content: + application/json: + schema: {} + responses: + 200: + content: + application/json: {} + 503: + description: WebRTC is disabled or not installed. + tags: + - streams + """ + if (error := _get_webrtc_error()) is not None: + return web.Response(text=error, status=503, content_type="text/plain") - # If video is disabled, mGBA does not render anything, so the screen buffer - # would contain rubbish. Send a black image instead. - if context.video: - image = context.emulator.get_current_screen_image() - else: - image = self._black_image - - # The same frame object is passed to the encoders of all connected clients, which - # run in separate threads. The encoder would convert the frame to YUV if it isn't - # already, but `reformat()` is not thread-safe and crashes the process if several - # threads call it on the same frame at once. So convert it once, here. - frame = VideoFrame.from_image(image).reformat(format="yuv420p") - frame.pts = pts - frame.time_base = time_base - - return frame - - class EmuAudio(MediaStreamTrack): - """ - Streams the emulator's audio output. - - The emulator's sample rate depends on the host's audio device and may change at - runtime (e.g. when that device disappears), but aiortc's encoder cannot handle a - change of sample rate mid-stream. So all audio is resampled to a fixed rate here. - """ - - kind = "audio" - output_sample_rate = 48000 - - # If the timestamp of the audio lags behind the wall clock by more than this - # many seconds, it is assumed that the emulator has not been producing audio - # for a while (because it was paused or running unthrottled.) - maximum_lag = 0.25 - - def __init__(self): - super().__init__() - self._queue: Queue[bytes] = context.emulator.get_last_audio_data() - self._resampler: AudioResampler | None = None - self._resampler_input_rate: int | None = None - self._start: float | None = None - self._timestamp: int = 0 - - async def _get_audio_data(self) -> bytes: - while True: - if self.readyState != "live": - raise MediaStreamError - try: - data = self._queue.get_nowait() - if len(data) > 0: - return data - except queue.Empty: - await asyncio.sleep(1 / 480) - - def _resample(self, data: bytes) -> bytes: - input_sample_rate = context.emulator.get_sample_rate() - if self._resampler is None or self._resampler_input_rate != input_sample_rate: - self._resampler = AudioResampler(format="s16", layout="stereo", rate=self.output_sample_rate) - self._resampler_input_rate = input_sample_rate - - input_frame = AudioFrame(format="s16", layout="stereo", samples=len(data) // 4) - input_frame.planes[0].update(data) - input_frame.sample_rate = input_sample_rate - - return b"".join( - bytes(output_frame.planes[0])[: output_frame.samples * 4] - for output_frame in self._resampler.resample(input_frame) - ) - - async def recv(self) -> Union[Frame, Packet]: - if self.readyState != "live": - raise MediaStreamError - - if self._start is None: - # Discard audio that has been queued up before anyone was listening. - while not self._queue.empty(): - try: - self._queue.get_nowait() - except queue.Empty: - break - - # The resampler may hold back some samples, so it is possible that we - # need to feed it more than one chunk before we get any output. - data = b"" - while len(data) == 0: - data = self._resample(await self._get_audio_data()) - - now = time.time() - if self._start is None: - self._start = now - elif now - (self._start + self._timestamp / self.output_sample_rate) > self.maximum_lag: - # Skip the timestamp ahead so that the browser treats this as a gap in - # the audio rather than as audio that has been delayed. - self._timestamp = int((now - self._start) * self.output_sample_rate) - - frame = AudioFrame(format="s16", layout="stereo", samples=len(data) // 4) - frame.planes[0].update(data) - frame.sample_rate = self.output_sample_rate - frame.time_base = fractions.Fraction(1, self.output_sample_rate) - frame.pts = self._timestamp - - self._timestamp += frame.samples - - return frame - - relay = MediaRelay() - emu_audio = None - emu_video = None - - def stop_streams_if_unused(): - """ - The relay keeps reading from the source tracks even after all subscribers are - gone. So once the last connection has closed, we stop the tracks (which ends the - relay's reading task) and create new ones for the next client. - """ - nonlocal emu_video, emu_audio - if len(rtc_connections) == 0: - if emu_video is not None: - emu_video.stop() - emu_video = None - if emu_audio is not None: - emu_audio.stop() - emu_audio = None - - @route.post("/rtc") - async def http_post_rtc(request: web.Request): - nonlocal emu_video, emu_audio - - params = await request.json() - offer = RTCSessionDescription(sdp=params["sdp"], type=params["type"]) - - connection = RTCPeerConnection(RTCConfiguration(iceServers=[])) - rtc_connections.add(connection) - - @connection.on("datachannel") - def on_datachannel(channel): - if channel.label == "input": - last_message_time = time.monotonic() - buttons_held = False - - # Each message contains the full list of buttons that should be held down, - # in the same format as `POST /input`. - @channel.on("message") - def on_input_message(message): - nonlocal last_message_time, buttons_held - last_message_time = time.monotonic() - try: - buttons = json.loads(message) - except (TypeError, ValueError): - return - if isinstance(buttons, list): - _set_held_buttons(buttons) - buttons_held = len(buttons) > 0 - - # The client repeats its current input state every second. Some browsers do not - # tell us when a tab is closed, in which case it would take ~30 seconds for the - # connection to time out. So if the client has gone quiet for a few seconds, we - # assume that it has gone away and release any buttons that it held down. - async def release_buttons_if_client_is_gone(): - nonlocal buttons_held - while True: - await asyncio.sleep(0.5) - if buttons_held and time.monotonic() - last_message_time > 3: - _set_held_buttons([]) - buttons_held = False - - watchdog = asyncio.ensure_future(release_buttons_if_client_is_gone()) - - # This also gets called if the connection fails or the client goes away, so - # this makes sure that no buttons remain held down in that case. - @channel.on("close") - def on_input_close(): - watchdog.cancel() - _set_held_buttons([]) - - else: - - @channel.on("message") - def on_message(message): - if isinstance(message, str) and message.startswith("ping"): - channel.send("pong" + message[4:]) - - @connection.on("connectionstatechange") - async def on_connection_state_change(): - if connection.connectionState in ("failed", "closed"): - await connection.close() - rtc_connections.discard(connection) - stop_streams_if_unused() - - if emu_video is None: - emu_video = EmuVideo() - - if emu_audio is None: - emu_audio = EmuAudio() - - # Video is not buffered so that a slow client always gets the most recent frame - # rather than lagging further and further behind. Audio needs to be buffered because - # dropping parts of it would be audible. - connection.addTrack(relay.subscribe(emu_video, buffered=False)) - connection.addTrack(relay.subscribe(emu_audio)) + params = await request.json() + answer = await webrtc.answer_offer(params["sdp"], params["type"], _set_held_buttons) + return web.json_response(answer) - try: - await connection.setRemoteDescription(offer) - answer = await connection.createAnswer() - await connection.setLocalDescription(answer) - except Exception: - await connection.close() - rtc_connections.discard(connection) - stop_streams_if_unused() - raise + @route.post("/rtc/close") + async def http_post_rtc_close(request: web.Request): + """ + --- + post: + description: > + Closes WebRTC connections, given the IDs that `/rtc` returned for them. Any buttons held + down by those connections are released. Returns the IDs of the connections that have + been closed; unknown IDs (e.g. of connections that have already been closed) are ignored. + requestBody: + description: JSON array of connection IDs + content: + application/json: + schema: {} + examples: + close_one_connection: + summary: Close one connection + value: ["0b7c3f4e-5d2a-4c8e-9f1b-2a6d8e4c1f3a"] + responses: + 200: + content: + application/json: {} + 422: + description: The payload is not a JSON array of strings. + 503: + description: WebRTC is disabled or not installed. + tags: + - streams + """ + if (error := _get_webrtc_error()) is not None: + return web.Response(text=error, status=503, content_type="text/plain") - return web.json_response({"sdp": connection.localDescription.sdp, "type": connection.localDescription.type}) + try: + connection_ids = await request.json() + except ValueError: + connection_ids = None + if not isinstance(connection_ids, list) or not all(isinstance(id, str) for id in connection_ids): + return web.Response(text="This endpoint expects a JSON array of connection IDs as its payload.", status=422) - else: + return web.json_response(await webrtc.close_connections(connection_ids)) - @route.post("/rtc") - async def http_post_rtc(request: web.Request): - return web.Response( - text="WebRTC is not available because aiortc was not installed.", - status=503, - headers={"Content-Type": "text/plain"}, - ) + @route.post("/rtc/close_all") + async def http_post_rtc_close_all(request: web.Request): + """ + --- + post: + description: > + Closes all WebRTC connections and releases any buttons they held down. Returns the IDs of + the connections that have been closed. + responses: + 200: + content: + application/json: {} + 503: + description: WebRTC is disabled or not installed. + tags: + - streams + """ + if (error := _get_webrtc_error()) is not None: + return web.Response(text=error, status=503, content_type="text/plain") + + return web.json_response(await webrtc.close_all_connections()) @route.get("/stream_video") async def http_get_video_stream(request: web.Request): diff --git a/modules/web/static/index.html b/modules/web/static/index.html index 381cfa6e5..845f15668 100644 --- a/modules/web/static/index.html +++ b/modules/web/static/index.html @@ -23,8 +23,7 @@
- - +
Start Video
@@ -297,8 +296,6 @@

Event Log

}; eventSource.onopen = () => { document.getElementById("screen_connection_lost").style.display = "none"; - document.getElementById("gba_video").src = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII="; - document.getElementById("gba_video").src = "/stream_video?fps=30&cache_buster=" + (new Date).toString(); }; }); @@ -800,15 +797,29 @@

Current Encounter

document.getElementById("gba_video_placeholder").remove(); const video = document.getElementById("gba_video"); video.style.display = "block"; - const audio = document.getElementById("gba_audio"); - const rtcConnection = new RTCPeerConnection(); + // If the server has a TURN server configured (because browsers cannot reach its IP address + // directly, e.g. when it runs in a Docker container), this contains that server. + const configResponse = await fetch("/rtc/config"); + if (!configResponse.ok) { + // E.g. if WebRTC is disabled, or the packages it needs are not installed. + showVideoError(await configResponse.text()); + return; + } + const rtcConfig = await configResponse.json(); + const useTurn = rtcConfig.iceServers.length > 0; + + const rtcConnection = new RTCPeerConnection({ + iceServers: rtcConfig.iceServers, + iceTransportPolicy: rtcConfig.iceTransportPolicy, + }); + // The server sends audio and video as a single stream, so this fires twice with the same + // stream. Both tracks play in the video element, so its controls (mute etc.) apply to both. rtcConnection.addEventListener("track", (event) => { - if (event.track.kind === "video") { - video.srcObject = new MediaStream([event.track]); - } else if (event.track.kind === "audio") { - audio.srcObject = new MediaStream([event.track]); + if (video.srcObject !== event.streams[0]) { + video.srcObject = event.streams[0]; + playVideo(video); } }); @@ -832,26 +843,101 @@

Current Encounter

if (inputChannel === channel) { inputChannel = null; } + + // If the bot closes the connection (see `/rtc/close`), the data channel closes while the + // connection itself still counts as connected. If the network fails instead, the + // connection's state changes first. + if (rtcConnection.connectionState === "connected") { + rtcConnection.close(); + video.srcObject = null; + showVideoError("The connection has been closed by the bot."); + } }); const offer = await rtcConnection.createOffer(); await rtcConnection.setLocalDescription(offer); - // The offer is deliberately sent before any ICE candidates have been gathered. - // Browsers usually hide their local IP addresses behind random mDNS hostnames that - // the server cannot resolve, and some (Firefox) also mark the list of candidates as - // complete -- at which point the server would give up without even trying. + // Without a TURN server, the offer is deliberately sent before any ICE candidates have + // been gathered. Browsers usually hide their local IP addresses behind random mDNS + // hostnames that the server cannot resolve, and some (Firefox) also mark the list of + // candidates as complete -- at which point the server would give up without even trying. // Without any candidates, the server instead waits for the browser to contact one // of the server's (real) addresses, which are included in the answer. + // + // With a TURN server, the browser only gathers relay candidates (`iceTransportPolicy` is + // 'relay'), which the server can use. So in that case we wait for them and send the + // updated local description, which contains them. + let description = offer; + if (useTurn) { + await waitForIceGathering(rtcConnection, 3000); + description = rtcConnection.localDescription; + if (!description.sdp.includes(" typ relay")) { + rtcConnection.close(); + showVideoError("Could not connect to the TURN server. Check the `webrtc.turn` settings in `http.yml`."); + return; + } + } + const response = await fetch("/rtc", { method: "POST", headers: {"Content-Type": "application/json"}, - body: JSON.stringify({sdp: offer.sdp, type: offer.type}), + body: JSON.stringify({sdp: description.sdp, type: description.type}), }); + if (!response.ok) { + rtcConnection.close(); + showVideoError(await response.text()); + return; + } + const answer = await response.json(); - await rtcConnection.setRemoteDescription(answer); + await rtcConnection.setRemoteDescription({sdp: answer.sdp, type: answer.type}); + } + + async function playVideo(video) { + try { + await video.play(); + } catch (error) { + if (error.name !== "NotAllowedError") { + throw error; + } + // Browsers may refuse to play audio without a recent click (the one on 'Start Video' may + // have expired while connecting.) Playing muted is always allowed, and the user can + // unmute using the video controls. + video.muted = true; + await video.play(); + } + } + + function showVideoError(message) { + document.getElementById("gba_video").style.display = "none"; + const placeholder = document.createElement("div"); + placeholder.id = "gba_video_placeholder"; + placeholder.textContent = message + "\n\nClick to try again."; + placeholder.addEventListener("click", startVideo); + document.getElementById("video_container").appendChild(placeholder); + } + + function waitForIceGathering(rtcConnection, timeoutMs) { + return new Promise((resolve) => { + if (rtcConnection.iceGatheringState === "complete") { + resolve(); + return; + } + const done = () => { + window.clearTimeout(timeout); + rtcConnection.removeEventListener("icegatheringstatechange", onChange); + resolve(); + }; + const onChange = () => { + if (rtcConnection.iceGatheringState === "complete") { + done(); + } + }; + const timeout = window.setTimeout(done, timeoutMs); + rtcConnection.addEventListener("icegatheringstatechange", onChange); + }); } @@ -935,6 +1021,11 @@

Current Encounter

justify-content: center; align-items: center; height: 100%; + padding: 1rem; + box-sizing: border-box; + text-align: center; + white-space: pre-line; + overflow-wrap: anywhere; } #gba_video_placeholder:hover { diff --git a/modules/web/webrtc.py b/modules/web/webrtc.py new file mode 100644 index 000000000..00e20bc09 --- /dev/null +++ b/modules/web/webrtc.py @@ -0,0 +1,324 @@ +""" +WebRTC streaming of the emulator's video and audio output (and input from the browser.) + +This needs aiortc, which is an optional dependency. `modules.web.http` only imports this +module if it is installed. +""" + +import asyncio +import base64 +import fractions +import hashlib +import hmac +import json +import queue +import time +import uuid +from queue import Queue +from typing import Callable, Union + +from aiortc import ( + MediaStreamTrack, + VideoStreamTrack, + RTCConfiguration, + RTCPeerConnection, + RTCSessionDescription, +) +from aiortc.contrib.media import MediaRelay +from aiortc.mediastreams import MediaStreamError +from av import VideoFrame, Packet, AudioFrame, AudioResampler +from av.frame import Frame +from PIL import Image + +from modules.context import context + +# How long the TURN credentials handed out to the browser remain valid, in seconds. The TURN +# server only checks this when a connection is set up, so it does not limit how long a stream +# can run. And the browser fetches new credentials before every connection attempt. +turn_credential_lifetime = 3600 + + +def get_client_config() -> dict: + """ + :return: The ICE configuration for the browser's `RTCPeerConnection`. If a TURN server has been + configured, the browser is told to only use that (see `startVideo()` in `index.html`.) + """ + turn = context.config.http.webrtc.turn + if turn is None: + return {"iceServers": [], "iceTransportPolicy": "all"} + + # This is the 'TURN REST API' scheme that coturn's `use-auth-secret` option expects, + # so the secret itself never has to be handed to the browser. + username = f"{int(time.time()) + turn_credential_lifetime}:pokebot" + digest = hmac.new(turn.secret.encode("utf-8"), username.encode("utf-8"), hashlib.sha1).digest() + credential = base64.b64encode(digest).decode("ascii") + + return { + "iceServers": [{"urls": turn.urls, "username": username, "credential": credential}], + "iceTransportPolicy": "relay", + } + + +class EmuVideo(VideoStreamTrack): + def __init__(self): + super().__init__() + self._black_image: Image.Image = Image.new("RGB", (240, 160)) + + async def recv(self) -> Union[Frame, Packet]: + pts, time_base = await self.next_timestamp() + + # If video is disabled, mGBA does not render anything, so the screen buffer + # would contain rubbish. Send a black image instead. + if context.video: + image = context.emulator.get_current_screen_image() + else: + image = self._black_image + + # The same frame object is passed to the encoders of all connected clients, which + # run in separate threads. The encoder would convert the frame to YUV if it isn't + # already, but `reformat()` is not thread-safe and crashes the process if several + # threads call it on the same frame at once. So convert it once, here. + frame = VideoFrame.from_image(image).reformat(format="yuv420p") + frame.pts = pts + frame.time_base = time_base + + return frame + + +class EmuAudio(MediaStreamTrack): + """ + Streams the emulator's audio output. + + The emulator's sample rate depends on the host's audio device and may change at + runtime (e.g. when that device disappears), but aiortc's encoder cannot handle a + change of sample rate mid-stream. So all audio is resampled to a fixed rate here. + """ + + kind = "audio" + output_sample_rate = 48000 + + # If the timestamp of the audio lags behind the wall clock by more than this + # many seconds, it is assumed that the emulator has not been producing audio + # for a while (because it was paused or running unthrottled.) + maximum_lag = 0.25 + + def __init__(self): + super().__init__() + self._queue: Queue[bytes] = context.emulator.get_last_audio_data() + self._resampler: AudioResampler | None = None + self._resampler_input_rate: int | None = None + self._start: float | None = None + self._timestamp: int = 0 + + async def _get_audio_data(self) -> bytes: + while True: + if self.readyState != "live": + raise MediaStreamError + try: + data = self._queue.get_nowait() + if len(data) > 0: + return data + except queue.Empty: + await asyncio.sleep(1 / 480) + + def _resample(self, data: bytes) -> bytes: + input_sample_rate = context.emulator.get_sample_rate() + if self._resampler is None or self._resampler_input_rate != input_sample_rate: + self._resampler = AudioResampler(format="s16", layout="stereo", rate=self.output_sample_rate) + self._resampler_input_rate = input_sample_rate + + input_frame = AudioFrame(format="s16", layout="stereo", samples=len(data) // 4) + input_frame.planes[0].update(data) + input_frame.sample_rate = input_sample_rate + + return b"".join( + bytes(output_frame.planes[0])[: output_frame.samples * 4] + for output_frame in self._resampler.resample(input_frame) + ) + + async def recv(self) -> Union[Frame, Packet]: + if self.readyState != "live": + raise MediaStreamError + + if self._start is None: + # Discard audio that has been queued up before anyone was listening. + while not self._queue.empty(): + try: + self._queue.get_nowait() + except queue.Empty: + break + + # The resampler may hold back some samples, so it is possible that we + # need to feed it more than one chunk before we get any output. + data = b"" + while len(data) == 0: + data = self._resample(await self._get_audio_data()) + + now = time.time() + if self._start is None: + self._start = now + elif now - (self._start + self._timestamp / self.output_sample_rate) > self.maximum_lag: + # Skip the timestamp ahead so that the browser treats this as a gap in + # the audio rather than as audio that has been delayed. + self._timestamp = int((now - self._start) * self.output_sample_rate) + + frame = AudioFrame(format="s16", layout="stereo", samples=len(data) // 4) + frame.planes[0].update(data) + frame.sample_rate = self.output_sample_rate + frame.time_base = fractions.Fraction(1, self.output_sample_rate) + frame.pts = self._timestamp + + self._timestamp += frame.samples + + return frame + + +# Open connections, by the ID that `answer_offer()` returned for them. +rtc_connections: dict[str, RTCPeerConnection] = {} +relay = MediaRelay() +emu_audio: EmuAudio | None = None +emu_video: EmuVideo | None = None + + +def stop_streams_if_unused() -> None: + """ + The relay keeps reading from the source tracks even after all subscribers are + gone. So once the last connection has closed, we stop the tracks (which ends the + relay's reading task) and create new ones for the next client. + """ + global emu_video, emu_audio + if len(rtc_connections) == 0: + if emu_video is not None: + emu_video.stop() + emu_video = None + if emu_audio is not None: + emu_audio.stop() + emu_audio = None + + +async def answer_offer(sdp: str, type: str, set_held_buttons: Callable[[list], None]) -> dict: + """ + Sets up a connection to a browser that wants to receive the stream. + + :param sdp: SDP of the browser's offer. + :param type: Type of the browser's session description (i.e. 'offer'.) + :param set_held_buttons: Function that replaces the set of buttons held down in the emulator, + for input that the browser sends via the 'input' data channel. + :return: The answer's session description, to be sent back to the browser, plus the ID + of the new connection (which can be passed to `close_connections()`.) + """ + global emu_video, emu_audio + + offer = RTCSessionDescription(sdp=sdp, type=type) + + connection = RTCPeerConnection(RTCConfiguration(iceServers=[])) + connection_id = str(uuid.uuid4()) + rtc_connections[connection_id] = connection + + @connection.on("datachannel") + def on_datachannel(channel): + if channel.label == "input": + last_message_time = time.monotonic() + buttons_held = False + + # Each message contains the full list of buttons that should be held down, + # in the same format as `POST /input`. + @channel.on("message") + def on_input_message(message): + nonlocal last_message_time, buttons_held + last_message_time = time.monotonic() + try: + buttons = json.loads(message) + except (TypeError, ValueError): + return + if isinstance(buttons, list): + set_held_buttons(buttons) + buttons_held = len(buttons) > 0 + + # The client repeats its current input state every second. Some browsers do not + # tell us when a tab is closed, in which case it would take ~30 seconds for the + # connection to time out. So if the client has gone quiet for a few seconds, we + # assume that it has gone away and release any buttons that it held down. + async def release_buttons_if_client_is_gone(): + nonlocal buttons_held + while True: + await asyncio.sleep(0.5) + if buttons_held and time.monotonic() - last_message_time > 3: + set_held_buttons([]) + buttons_held = False + + watchdog = asyncio.ensure_future(release_buttons_if_client_is_gone()) + + # This also gets called if the connection fails or the client goes away, so + # this makes sure that no buttons remain held down in that case. + @channel.on("close") + def on_input_close(): + watchdog.cancel() + set_held_buttons([]) + + else: + + @channel.on("message") + def on_message(message): + if isinstance(message, str) and message.startswith("ping"): + channel.send("pong" + message[4:]) + + @connection.on("connectionstatechange") + async def on_connection_state_change(): + if connection.connectionState in ("failed", "closed"): + await connection.close() + rtc_connections.pop(connection_id, None) + stop_streams_if_unused() + + if emu_video is None: + emu_video = EmuVideo() + + if emu_audio is None: + emu_audio = EmuAudio() + + # Video is not buffered so that a slow client always gets the most recent frame + # rather than lagging further and further behind. Audio needs to be buffered because + # dropping parts of it would be audible. + connection.addTrack(relay.subscribe(emu_video, buffered=False)) + connection.addTrack(relay.subscribe(emu_audio)) + + try: + await connection.setRemoteDescription(offer) + answer = await connection.createAnswer() + await connection.setLocalDescription(answer) + except Exception: + await connection.close() + rtc_connections.pop(connection_id, None) + stop_streams_if_unused() + raise + + return {"id": connection_id, "sdp": connection.localDescription.sdp, "type": connection.localDescription.type} + + +async def close_connections(connection_ids: list[str]) -> list[str]: + """ + Closes connections to browsers. This also closes their input data channels, which + releases any buttons they held down. + + :param connection_ids: IDs of the connections to close, as returned by `answer_offer()`. + IDs of connections that do not exist (anymore) are ignored. + :return: IDs of the connections that have actually been closed. + """ + connections = {} + for connection_id in connection_ids: + if connection_id in rtc_connections: + connections[connection_id] = rtc_connections.pop(connection_id) + + await asyncio.gather(*(connection.close() for connection in connections.values())) + stop_streams_if_unused() + + return list(connections) + + +async def close_all_connections() -> list[str]: + """ + Closes the connections to all browsers. + + :return: IDs of the connections that have been closed. + """ + return await close_connections(list(rtc_connections)) diff --git a/requirements.py b/requirements.py index 620a0b762..4ae5ece8f 100644 --- a/requirements.py +++ b/requirements.py @@ -31,7 +31,6 @@ "darkdetect~=0.8.0", "show-in-file-manager~=1.1.4", "aiohttp~=3.10.9", - "aiortc~=1.10.0", ] if platform.system() == "Windows":