From a504aba974326f78298ada0afa5507a27817a642 Mon Sep 17 00:00:00 2001 From: "Ilya (Marshal)" Date: Sat, 24 Oct 2026 15:17:30 +0200 Subject: [PATCH] Add browser server, Gemini Live and WHIP/WHEP examples --- docs/source/examples/browser_server.md | 18 + docs/source/examples/gemini_live.md | 38 ++ docs/source/examples/index.md | 32 +- docs/source/examples/whip_whep.md | 27 + docs/source/guides/connections.md | 5 +- docs/source/llms-txt.md | 8 +- examples/README.md | 24 + examples/browser_server.py | 528 +++++++++++++++++++ examples/gemini_live.py | 679 +++++++++++++++++++++++++ examples/openai_live.py | 7 +- examples/whip_whep.py | 419 +++++++++++++++ 11 files changed, 1773 insertions(+), 12 deletions(-) create mode 100644 docs/source/examples/browser_server.md create mode 100644 docs/source/examples/gemini_live.md create mode 100644 docs/source/examples/whip_whep.md create mode 100755 examples/browser_server.py create mode 100755 examples/gemini_live.py create mode 100755 examples/whip_whep.py diff --git a/docs/source/examples/browser_server.md b/docs/source/examples/browser_server.md new file mode 100644 index 0000000..b491bac --- /dev/null +++ b/docs/source/examples/browser_server.md @@ -0,0 +1,18 @@ +# Browser and Python server + +A browser page sends your camera to Python, which sends it back in grayscale, and Python answers a chat over a data +channel. Open the printed address and allow the camera: + +```bash +uv run https://raw.githubusercontent.com/MarshalX/python-webrtc/main/examples/browser_server.py +``` + +Python reads the video with a {obj}`~webrtc.MediaStreamTrackProcessor`, pipes it through a +{obj}`~webrtc.TransformStream` and writes it to a {obj}`~webrtc.VideoTrackGenerator`. Browsers allow the camera only on +localhost or over HTTPS, so the server listens on localhost. Behind a strict NAT, pass your own TURN server with +`--turn`. + +```{literalinclude} ../../../examples/browser_server.py +:language: python +:caption: examples/browser_server.py +``` diff --git a/docs/source/examples/gemini_live.md b/docs/source/examples/gemini_live.md new file mode 100644 index 0000000..b6358fb --- /dev/null +++ b/docs/source/examples/gemini_live.md @@ -0,0 +1,38 @@ +# Voice agent with Gemini Live + +Talks to Gemini from a browser page, with live transcripts, through a Python server in the middle: + +```text +browser <-- WebRTC --> Python server <-- WebSocket --> Gemini Live API +``` + +The server reads the microphone with a {obj}`~webrtc.MediaStreamTrackProcessor`, streams it to Gemini as 16 kHz PCM, +and writes Gemini's 24 kHz answer to a {obj}`~webrtc.MediaStreamTrackGenerator`. When you talk over Gemini, it drops +the rest of the answer. + +## Why a server in the middle + +Google's [Live API overview](https://ai.google.dev/gemini-api/docs/live-api) describes two ways to connect. A page can +connect to Gemini directly, with short-lived tokens in production, or go through your server. This example takes the +second way, with WebRTC to the browser, as the WebRTC partner integrations on that page do: + +- The API key stays on the server, which sees every message and can add tools, logging or limits. +- The browser gets WebRTC's echo cancellation, jitter buffer and NAT traversal. +- Gemini gets the WebSocket it's specified with. + +## Running it + +Get a key in [Google AI Studio](https://aistudio.google.com/apikey). The free tier works, but Google uses what you send +on it to improve its products. See [Gemini API pricing](https://ai.google.dev/gemini-api/docs/pricing). + +```bash +GEMINI_API_KEY=... uv run https://raw.githubusercontent.com/MarshalX/python-webrtc/main/examples/gemini_live.py +``` + +Open and press Talk. Anyone who can open the page talks on your key, so the server listens on +localhost. + +```{literalinclude} ../../../examples/gemini_live.py +:language: python +:caption: examples/gemini_live.py +``` diff --git a/docs/source/examples/index.md b/docs/source/examples/index.md index e558943..c2b3d38 100644 --- a/docs/source/examples/index.md +++ b/docs/source/examples/index.md @@ -23,11 +23,11 @@ Sends the received video back in grayscale. Writes received video and audio to raw files. ::: -:::{grid-item-card} {octicon}`broadcast;1em;sd-mr-1` Live stream in the terminal -:link: janus_streaming +:::{grid-item-card} {octicon}`browser;1em;sd-mr-1` Browser and Python server +:link: browser_server :link-type: doc -Watches a public stream, with the video as colored characters and the audio on your speakers. +Sends your camera back from Python in grayscale, with a chat over a data channel. ::: :::{grid-item-card} {octicon}`unmute;1em;sd-mr-1` Voice chat with OpenAI GPT-Live @@ -37,6 +37,27 @@ Watches a public stream, with the video as colored characters and the audio on y Talks to the model through your microphone and speakers. ::: +:::{grid-item-card} {octicon}`comment-discussion;1em;sd-mr-1` Voice agent with Gemini Live +:link: gemini_live +:link-type: doc + +Talks to Gemini from a browser page, through a Python server that keeps the API key. +::: + +:::{grid-item-card} {octicon}`broadcast;1em;sd-mr-1` Live stream in the terminal +:link: janus_streaming +:link-type: doc + +Watches a public stream, with the video as colored characters and the audio on your speakers. +::: + +:::{grid-item-card} {octicon}`broadcast;1em;sd-mr-1` WHIP and WHEP +:link: whip_whep +:link-type: doc + +Plays a WHEP stream in the terminal, or publishes one over WHIP. +::: + :::: ```{toctree} @@ -44,6 +65,9 @@ Talks to the model through your microphone and speakers. echo recorder -janus_streaming +browser_server openai_live +gemini_live +janus_streaming +whip_whep ``` diff --git a/docs/source/examples/whip_whep.md b/docs/source/examples/whip_whep.md new file mode 100644 index 0000000..ed8e082 --- /dev/null +++ b/docs/source/examples/whip_whep.md @@ -0,0 +1,27 @@ +# WHIP and WHEP + +Plays a live stream over [WHEP](https://www.rfc-editor.org/rfc/rfc9725) in the terminal, or publishes one over WHIP. +With no URL it plays the public demo stream of [Dolby OptiView](https://optiview.dolby.com/), with no account or key: + +```bash +uv run https://raw.githubusercontent.com/MarshalX/python-webrtc/main/examples/whip_whep.py +``` + +The demo stream is H.264. Where the OS has no H.264 codec, as on Linux and Windows, the example downloads Cisco's +OpenH264 binary once, and `--no-openh264` skips it. See [H.264](../guides/h264.md). + +OpenH264 Video Codec provided by Cisco Systems, Inc. + +To publish with `--whip` and play it back, a local [MediaMTX](https://github.com/bluenviron/mediamtx) server works: + +```bash +docker run -d --rm --name wrtc-mediamtx -p 8889:8889 -p 8189:8189/udp bluenviron/mediamtx:latest +uv run whip_whep.py --whip http://localhost:8889/wrtc/whip +uv run whip_whep.py http://localhost:8889/wrtc/whep +docker stop wrtc-mediamtx +``` + +```{literalinclude} ../../../examples/whip_whep.py +:language: python +:caption: examples/whip_whep.py +``` diff --git a/docs/source/guides/connections.md b/docs/source/guides/connections.md index b6700b1..8ae7887 100644 --- a/docs/source/guides/connections.md +++ b/docs/source/guides/connections.md @@ -54,8 +54,9 @@ async def on_candidate(event): await signaling.send(event.candidate.to_json()) ``` -To skip trickling, wait for `ice_gathering_state` to be `'complete'` and send `pc.local_description`, which then has -every candidate in it. +To skip trickling, send `pc.local_description` once the candidates you need have arrived: it includes them. Don't wait +for `ice_gathering_state` to be `'complete'`: with a STUN server, an address that can't reach it, like a VPN's, holds +gathering open for about 40 seconds. Stop at the first `srflx` candidate, or after a short pause with no new ones. ## Peers on one machine diff --git a/docs/source/llms-txt.md b/docs/source/llms-txt.md index ae0991b..2eaf5bb 100644 --- a/docs/source/llms-txt.md +++ b/docs/source/llms-txt.md @@ -54,9 +54,15 @@ of date. - {doc}`Echo peer `: sends back the video it receives, in grayscale. - {doc}`Recorder `: writes the video and audio a peer receives to files. -- {doc}`Live stream in the terminal `: plays a stream of the public Janus demo server. +- {doc}`Browser and Python server `: serves a page whose camera comes back from Python in + grayscale, with a chat over a data channel. - {doc}`Voice chat with OpenAI GPT-Live `: talks to the model through the microphone and speakers. +- {doc}`Voice agent with Gemini Live `: a browser talks to Gemini over WebRTC through a Python + server, which streams the audio to the Live API over a WebSocket. +- {doc}`Live stream in the terminal `: plays a stream of the public Janus demo server. +- {doc}`WHIP and WHEP `: plays a WHEP stream, a public demo by default, or publishes the synthetic + camera and microphone over WHIP. ## Optional diff --git a/examples/README.md b/examples/README.md index 35f89a9..72b8334 100644 --- a/examples/README.md +++ b/examples/README.md @@ -12,6 +12,14 @@ without worrying about copyrights. **Transforming video with a processor, a TransformStream and a VideoTrackGenerator.** Sends the received video back in grayscale. +### [browser_server.py](browser_server.py) + +**A Python WebRTC server for a browser page.** Your camera comes back from Python in grayscale, and Python answers a chat over a data channel. Open the printed address and allow the camera: + +```bash +uv run https://raw.githubusercontent.com/MarshalX/python-webrtc/main/examples/browser_server.py +``` + ### [openai_live.py](openai_live.py) **Voice chat with OpenAI GPT-Live.** Talks to the model through your microphone and speakers, with live transcripts in the terminal. No server: just an API key. Dependencies are declared inline, so [uv](https://docs.astral.sh/uv/) runs it with nothing to install: @@ -20,6 +28,14 @@ without worrying about copyrights. OPENAI_API_KEY=sk-... uv run https://raw.githubusercontent.com/MarshalX/python-webrtc/main/examples/openai_live.py ``` +### [gemini_live.py](gemini_live.py) + +**Voice agent with Gemini Live.** A browser page talks to Gemini through a Python server that keeps the API key, over WebRTC to the page and a WebSocket to the Live API: + +```bash +GEMINI_API_KEY=... uv run https://raw.githubusercontent.com/MarshalX/python-webrtc/main/examples/gemini_live.py +``` + ### [janus_streaming.py](janus_streaming.py) **Watching a live stream in the terminal.** Plays a stream of the public [Janus](https://janus.conf.meetecho.com/) demo server: the video drawn with colored characters, the audio on your speakers, if there are any, so it runs on a server too. No account or key: @@ -27,3 +43,11 @@ OPENAI_API_KEY=sk-... uv run https://raw.githubusercontent.com/MarshalX/python-w ```bash uv run https://raw.githubusercontent.com/MarshalX/python-webrtc/main/examples/janus_streaming.py ``` + +### [whip_whep.py](whip_whep.py) + +**Playing and publishing streams over WHIP and WHEP.** Plays the public demo stream of [Dolby OptiView](https://optiview.dolby.com/) in the terminal, or any WHEP endpoint, and `--whip` publishes instead. Where the OS has no H.264 codec, it downloads Cisco's OpenH264 once (OpenH264 Video Codec provided by Cisco Systems, Inc.). No account or key: + +```bash +uv run https://raw.githubusercontent.com/MarshalX/python-webrtc/main/examples/whip_whep.py +``` diff --git a/examples/browser_server.py b/examples/browser_server.py new file mode 100755 index 0000000..7a1435a --- /dev/null +++ b/examples/browser_server.py @@ -0,0 +1,528 @@ +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.9" +# dependencies = ["wrtc>=0.0.1", "aiohttp"] +# /// +# +# Copyright 2022-2026 Ilya (Marshal) . +# +# Dedicated to the public domain under CC0, see the LICENSE file of the examples. +# + +"""A WebRTC server for a browser page: your camera comes back from Python with an effect, and Python answers chat. + + uv run browser_server.py + uv run browser_server.py --effect negative + uv run browser_server.py --turn turn:turn.example.com:3478 --turn-username user --turn-credential secret + +Open the printed address and allow the camera, which browsers allow only on localhost or over HTTPS. A TURN server is +needed only behind a strict NAT, and the page gets its credentials too. Press Ctrl+C to stop. +""" + +from __future__ import annotations + +import argparse +import asyncio +import ipaddress +import json +import os +from dataclasses import dataclass, field +from typing import Callable + +from aiohttp import web + +import webrtc + +STUN = 'stun:stun.l.google.com:19302' +# gathering completes only once every STUN request is answered or times out, which can take tens of seconds +GATHERING_PAUSE = 0.4 # seconds without a new ICE candidate after which gathering counts as done +GATHERING_LIMIT = 3 # seconds to wait for ICE candidates at most + +Effect = Callable[[bytearray, list[webrtc.PlaneLayout]], None] + + +def grayscale(pixels: bytearray, layout: list[webrtc.PlaneLayout]) -> None: + """Sets the planes after the luma to 128, which leaves no color.""" + chroma = layout[1].offset + pixels[chroma:] = b'\x80' * (len(pixels) - chroma) + + +INVERTED = bytes(range(255, -1, -1)) + + +def negative(pixels: bytearray, _layout: list[webrtc.PlaneLayout]) -> None: + """Inverts every byte, which inverts the brightness and the colors.""" + pixels[:] = pixels.translate(INVERTED) + + +EFFECTS: dict[str, Effect] = {'gray': grayscale, 'negative': negative} + + +def video_frame(media: object) -> webrtc.VideoFrame: + """The media a processor of a video track reads, as a video frame.""" + if not isinstance(media, webrtc.VideoFrame): + msg = f'expected a video frame, not {media!r}' + raise TypeError(msg) + return media + + +async def with_effect(frame: webrtc.VideoFrame, effect: Effect) -> webrtc.VideoFrame: + """A copy of the visible pixels of a frame, in its own format, with the effect applied.""" + rect = frame.visible_rect + # only for the type checker: an open frame has both + if frame.format is None or rect is None: + msg = 'the frame is closed' + raise ValueError(msg) + + pixels = bytearray(frame.allocation_size()) + layout = await frame.copy_to(pixels) + effect(pixels, layout) + + init = webrtc.VideoFrameBufferInit( + format=frame.format, + coded_width=int(rect.width), + coded_height=int(rect.height), + timestamp=frame.timestamp, + layout=layout, + rotation=frame.rotation, + flip=frame.flip, + ) + return webrtc.VideoFrame(pixels, init) + + +@dataclass +class EffectTransformer: + """The transformer of a TransformStream that applies an effect to video frames.""" + + effect: Effect + + async def transform(self, media: object, controller: webrtc.TransformStreamDefaultController) -> None: + """Enqueues the frame with the effect.""" + with video_frame(media) as frame: + controller.enqueue(await with_effect(frame, self.effect)) + + +@dataclass +class Server: + """Answers the offers of the page, sends the camera back with an effect and answers chat messages.""" + + ice_servers: list[webrtc.RTCIceServer] + effect_name: str + wait_for_relay: bool + connections: set[webrtc.RTCPeerConnection] = field(default_factory=set) + pipes: set[asyncio.Future[None]] = field(default_factory=set) + count: int = 0 + + async def page(self, _request: web.Request) -> web.Response: + """The page, with the ICE servers and the effect filled in.""" + config = { + 'iceServers': [server.to_json() for server in self.ice_servers], + 'waitForRelay': self.wait_for_relay, + 'effect': self.effect_name, + } + # escaped, so a URL in the config can't end the script element + script_config = json.dumps(config).replace('<', '\\u003c') + return web.Response(text=PAGE.replace('/*CONFIG*/', script_config), content_type='text/html') + + async def offer(self, request: web.Request) -> web.Response: + """Answers an offer that has the ICE candidates of the page, with the candidates of the server.""" + # other sites can only post JSON after a CORS preflight, which this server doesn't allow + if request.content_type != 'application/json': + raise web.HTTPUnsupportedMediaType + + offer = webrtc.RTCSessionDescriptionInit.from_json(await request.json()) + self.count += 1 + name = f'Connection {self.count}' + pc = webrtc.RTCPeerConnection(webrtc.RTCConfiguration(ice_servers=self.ice_servers)) + self.connections.add(pc) + self.watch_state(pc, name) + self.serve_chat(pc, name) + + try: + await pc.set_remote_description(offer) + self.send_back_with_effect(pc) + await pc.set_local_description() + answer = await gathered(pc, wait_for_relay=self.wait_for_relay) + except Exception: + self.drop(pc) + raise + print(f'{name}: answered') + return web.json_response(answer.to_json()) + + def send_back_with_effect(self, pc: webrtc.RTCPeerConnection) -> None: + """Sends each received video track back through the effect, on the transceiver it came on.""" + for receiver in pc.get_receivers(): + if receiver.track.kind != 'video': + continue + + processor = webrtc.MediaStreamTrackProcessor(webrtc.MediaStreamTrackProcessorInit(receiver.track)) + transform = webrtc.TransformStream(EffectTransformer(EFFECTS[self.effect_name])) + generator = webrtc.VideoTrackGenerator() + + with_effect = processor.readable.pipe_through(transform) + pipe = with_effect.pipe_to(generator.writable) + self.pipes.add(pipe) + pipe.add_done_callback(self.pipe_ended) + + # the offered transceiver has no track to send yet, so it takes this one + pc.add_track(generator.track) + + def pipe_ended(self, pipe: asyncio.Future[None]) -> None: + """Forgets a finished pipe, and prints why it failed if it did.""" + self.pipes.discard(pipe) + if not pipe.cancelled() and pipe.exception() is not None: + print(f'The effect stopped: {pipe.exception()!r}') + + def watch_state(self, pc: webrtc.RTCPeerConnection, name: str) -> None: + """Prints the state of the connection, and drops it once it failed or closed.""" + + @pc.on('connectionstatechange') + def on_state(_event: webrtc.Event) -> None: + print(f'{name}: {pc.connection_state}') + if pc.connection_state in {'failed', 'closed'}: + self.drop(pc) + + def serve_chat(self, pc: webrtc.RTCPeerConnection, name: str) -> None: + """Replies to each text message on the chat channel, and drops the connection once the page closes it.""" + + @pc.on('datachannel') + def on_datachannel(event: webrtc.RTCDataChannelEvent) -> None: + reply_to_messages(event.channel, name) + + @event.channel.on('close') + def on_close(_event: webrtc.Event) -> None: + print(f'{name}: the page hung up') + self.drop(pc) + + def drop(self, pc: webrtc.RTCPeerConnection) -> None: + """Closes a connection, which ends its tracks and so its pipes.""" + self.connections.discard(pc) + pc.close() + + async def close(self, _app: web.Application) -> None: + """Closes every connection and waits for the pipes to end.""" + print('Closing the connections') + for pc in list(self.connections): + self.drop(pc) + await asyncio.wait_for(asyncio.gather(*self.pipes, return_exceptions=True), 10) + + +def reply_to_messages(channel: webrtc.RTCDataChannel, name: str) -> None: + """Replies to each text message of a channel.""" + + @channel.on('message') + def on_message(event: webrtc.MessageEvent) -> None: + if not isinstance(event.data, str): + return + print(f'{name}: the page says {event.data!r}') + channel.send(reply(event.data)) + + +def reply(text: str) -> str: + """Python's reply to a chat message.""" + words = len(text.split()) + return f'{text[::-1]} ({words} word{"" if words == 1 else "s"} backwards)' + + +def enough_candidates(pc: webrtc.RTCPeerConnection, *, wait_for_relay: bool) -> bool: + """Whether gathering completed, or found a relay candidate when there is a TURN server to wait for.""" + if pc.ice_gathering_state == 'complete': + return True + return wait_for_relay and pc.local_description is not None and ' typ relay' in pc.local_description.sdp + + +async def gathered(pc: webrtc.RTCPeerConnection, *, wait_for_relay: bool) -> webrtc.RTCSessionDescription: + """Waits for enough ICE candidates, GATHERING_LIMIT seconds at most, then returns the local description.""" + progress = asyncio.Event() + + def on_progress(_event: object) -> None: + progress.set() + + pc.on('icecandidate', on_progress) + pc.on('icegatheringstatechange', on_progress) + + loop = asyncio.get_running_loop() + deadline = loop.time() + GATHERING_LIMIT + # a relay candidate takes a few round trips, so a pause in gathering doesn't count while waiting for one + pause = GATHERING_LIMIT if wait_for_relay else GATHERING_PAUSE + try: + while not enough_candidates(pc, wait_for_relay=wait_for_relay) and loop.time() < deadline: + progress.clear() + try: + await asyncio.wait_for(progress.wait(), min(pause, deadline - loop.time())) + except asyncio.TimeoutError: + break + finally: + pc.off('icecandidate', on_progress) + pc.off('icegatheringstatechange', on_progress) + + if pc.local_description is None: + msg = 'no local description' + raise RuntimeError(msg) + return pc.local_description + + +def ice_servers(args: argparse.Namespace) -> list[webrtc.RTCIceServer]: + """Google's STUN server, and the TURN server of the options if there is one.""" + servers = [webrtc.RTCIceServer(urls=STUN)] + if args.turn: + servers.append( + webrtc.RTCIceServer(urls=args.turn, username=args.turn_username, credential=args.turn_credential) + ) + return servers + + +def parse_args() -> argparse.Namespace: + """The options of the command line.""" + parser = argparse.ArgumentParser( + description='Send the camera of a browser page back from Python with an effect, and chat with it.' + ) + parser.add_argument('--host', default='localhost', help='the address to listen on, default: localhost') + parser.add_argument('--port', type=int, default=8080, help='the port to listen on, default: 8080') + parser.add_argument('--effect', choices=sorted(EFFECTS), default='gray', help='the effect of Python, default: gray') + parser.add_argument('--turn', default=os.environ.get('TURN_URL'), help='a TURN server URL, default: $TURN_URL') + parser.add_argument('--turn-username', default=os.environ.get('TURN_USERNAME'), help='default: $TURN_USERNAME') + parser.add_argument( + '--turn-credential', default=os.environ.get('TURN_CREDENTIAL'), help='default: $TURN_CREDENTIAL' + ) + args = parser.parse_args() + + if args.turn and not (args.turn_username and args.turn_credential): + parser.error('a TURN server needs --turn-username and --turn-credential') + return args + + +def page_url(host: str, port: int) -> str: + """The address of the page, on localhost when the server listens on every address.""" + if is_every_address(host): + host = 'localhost' + return f'http://{host}:{port}' + + +def is_every_address(host: str) -> bool: + """Whether the host is 0.0.0.0 or ::, which mean every address of the machine.""" + try: + return ipaddress.ip_address(host).is_unspecified + except ValueError: # a host name + return False + + +def main() -> None: + """Serves the page until Ctrl+C.""" + args = parse_args() + server = Server(ice_servers(args), args.effect, wait_for_relay=bool(args.turn)) + + app = web.Application() + app.router.add_get('/', server.page) + app.router.add_post('/offer', server.offer) + app.on_shutdown.append(server.close) + + print(f'Open {page_url(args.host, args.port)} and allow the camera. Press Ctrl+C to stop.') + # handles Ctrl+C on every platform, then runs the shutdown handlers + web.run_app(app, host=args.host, port=args.port, print=None) + + +PAGE = """ + + + + +Browser and Python over WebRTC + + + +

Browser and Python over WebRTC

+

+ + +

+

+ Connection: new, + ICE: new, + chat: closed +

+
+
+ +
Your camera
+
+
+ +
Back from Python
+
+
+

Chat with Python

+
    +
    + + +
    + + + +""" + + +if __name__ == '__main__': + main() diff --git a/examples/gemini_live.py b/examples/gemini_live.py new file mode 100755 index 0000000..cbb574a --- /dev/null +++ b/examples/gemini_live.py @@ -0,0 +1,679 @@ +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.9" +# dependencies = ["wrtc>=0.0.1", "aiohttp>=3.9"] +# /// +# +# Copyright 2022-2026 Ilya (Marshal) . +# +# Dedicated to the public domain under CC0, see the LICENSE file of the examples. +# + +"""Talks to Gemini Live from a browser, through a Python server that keeps the API key. + + browser <-- WebRTC --> this server <-- WebSocket --> Gemini Live API + + export GEMINI_API_KEY=... + uv run gemini_live.py + uv run gemini_live.py --voice Kore --instructions "You are a pirate. Keep answers short." + +Open http://localhost:8080 and press Talk. Anyone who can open the page talks on your key, so the server listens on +localhost. Press Ctrl+C to stop. +""" + +from __future__ import annotations + +import argparse +import asyncio +import base64 +import json +import os +import sys +from array import array +from dataclasses import dataclass, field +from typing import TypedDict + +import aiohttp +from aiohttp import web + +import webrtc + +GEMINI_URL = 'wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent' +STUN = 'stun:stun.l.google.com:19302' +INPUT_RATE = 16000 # Gemini hears 16 kHz +OUTPUT_RATE = 24000 # and speaks at 24 kHz +INPUT_CHUNK = 0.04 # seconds of the microphone per message, Google advises 20 to 40 ms +OUTPUT_CHUNK = 0.02 # seconds of the answer written to the track at a time +EVENTS_CHANNEL = 0 # the id of the negotiated channel of the transcripts +SETUP_LIMIT = 10 # seconds for Gemini to accept a session +CONNECT_LIMIT = 15 # seconds for the page to connect once answered +# gathering completes only once every STUN request is answered or times out, which can take tens of seconds +GATHERING_PAUSE = 0.4 # seconds without a new ICE candidate after which gathering counts as done +GATHERING_LIMIT = 3 # seconds to wait for ICE candidates at most + + +class Call: + """A page connected to a Gemini session: its microphone goes to Gemini, Gemini's voice and transcripts to it.""" + + def __init__(self, name: str, gemini: Gemini, ice_servers: list[webrtc.RTCIceServer]) -> None: + self.name = name + self.gemini = gemini + self.pc = webrtc.RTCPeerConnection(webrtc.RTCConfiguration(ice_servers=ice_servers)) + self.generator = webrtc.MediaStreamTrackGenerator('audio') + self.playback = Playback(self.generator) + negotiated = webrtc.RTCDataChannelInit(negotiated=True, id=EVENTS_CHANNEL) + self.events = self.pc.create_data_channel('events', negotiated) + self.tasks: list[asyncio.Future[None]] = [] + self._ended = asyncio.Event() + + async def answer(self, offer: webrtc.RTCSessionDescriptionInit) -> webrtc.RTCSessionDescription: + """Answers the offer of the page, then starts passing the audio both ways.""" + self.pc.on('track', self._on_track) + self.pc.on('connectionstatechange', self._on_connection_state) + self.events.on('close', self._on_events_close) + + await self.pc.set_remote_description(offer) + # the offered transceiver has no track to send yet, so it takes this one + self.pc.add_track(self.generator) + await self.pc.set_local_description() + answer = await gathered(self.pc) + + self.tasks.append(asyncio.ensure_future(self.playback.run())) + self.tasks.append(asyncio.ensure_future(self._listen())) + asyncio.get_running_loop().call_later(CONNECT_LIMIT, self._hang_up_unless_connected) + return answer + + def hang_up(self) -> None: + """Ends the call: run() then closes everything. Ending an ended call does nothing.""" + self._ended.set() + + async def run(self) -> None: + """Waits for the call to end, then closes the connection and the session.""" + await self._ended.wait() + for task in self.tasks: + task.cancel() + results = await asyncio.gather(*self.tasks, return_exceptions=True) + for result in results: + if isinstance(result, Exception): + print(f'{self.name}: {result!r}') + + self.pc.close() + self.generator.stop() + await self.gemini.close() + print(f'{self.name}: ended') + + def _on_track(self, event: webrtc.RTCTrackEvent) -> None: + self.tasks.append(asyncio.ensure_future(self._send_microphone(event.track))) + + def _on_connection_state(self, _event: webrtc.Event) -> None: + state = self.pc.connection_state + print(f'{self.name}: {state}') + if state in {'failed', 'closed'}: + self.hang_up() + + def _on_events_close(self, _event: webrtc.Event) -> None: + print(f'{self.name}: the page hung up') + self.hang_up() + + def _hang_up_unless_connected(self) -> None: + if self.pc.connection_state in {'new', 'connecting'}: + print(f'{self.name}: the page did not connect in {CONNECT_LIMIT} s') + self.hang_up() + + async def _send_microphone(self, track: webrtc.MediaStreamTrack) -> None: + """Streams the microphone of the page to Gemini in chunks of INPUT_CHUNK seconds, until it ends.""" + processor = webrtc.MediaStreamTrackProcessor(webrtc.MediaStreamTrackProcessorInit(track)) + pending = array('h') + try: + async for media in processor.readable: + with audio_data(media) as data: + samples = mono_samples(data) + rate = int(data.sample_rate) + + sent_rate = gemini_rate(rate) + pending += downsample(samples, rate // sent_rate) + if len(pending) >= sent_rate * INPUT_CHUNK: + await self.gemini.send_audio(pending.tobytes(), sent_rate) + pending = array('h') + finally: + self.hang_up() + + async def _listen(self) -> None: + """Plays Gemini's voice and passes its transcripts to the page, until the session ends.""" + try: + while True: + message = await self.gemini.receive() + self._on_gemini_message(message) + except GeminiError as e: + print(f'{self.name}: {e}') + self._send_to_page({'type': 'note', 'text': str(e)}) + finally: + self.hang_up() + + def _on_gemini_message(self, message: ServerMessage) -> None: + content = message.get('serverContent') + if content is not None: + self._on_content(content) + + go_away = message.get('goAway') + if go_away is not None: + text = f'Gemini will end the session in {go_away.get("timeLeft", "a moment")}' + self._send_to_page({'type': 'note', 'text': text}) + + def _on_content(self, content: ServerContent) -> None: + if content.get('interrupted', False): + self.playback.clear() + self._send_to_page({'type': 'turn'}) + + for part in content.get('modelTurn', {}).get('parts', []): + blob = part.get('inlineData') + if blob is not None: + self.playback.add(base64.b64decode(blob.get('data', ''))) + + heard = content.get('inputTranscription', {}).get('text', '') + if heard != '': + self._send_to_page({'type': 'transcript', 'speaker': 'user', 'text': heard}) + + said = content.get('outputTranscription', {}).get('text', '') + if said != '': + self._send_to_page({'type': 'transcript', 'speaker': 'gemini', 'text': said}) + + if content.get('turnComplete', False): + self._send_to_page({'type': 'turn'}) + + def _send_to_page(self, message: dict[str, str]) -> None: + """Sends a transcript fragment, the end of a turn or a note to the page, if it's still there.""" + if self.events.ready_state == 'open': + self.events.send(json.dumps(message)) + + +class Blob(TypedDict, total=False): + """Base64 media of a message of Gemini.""" + + data: str + mimeType: str + + +class Part(TypedDict, total=False): + """A part of a turn of Gemini: the fields used here.""" + + inlineData: Blob + + +class Content(TypedDict, total=False): + """A turn of Gemini.""" + + parts: list[Part] + + +class Transcription(TypedDict, total=False): + """A fragment of a transcript.""" + + text: str + + +class ServerContent(TypedDict, total=False): + """What Gemini says: audio, transcripts and the state of its turn.""" + + modelTurn: Content + interrupted: bool + turnComplete: bool + inputTranscription: Transcription + outputTranscription: Transcription + + +class GoAway(TypedDict, total=False): + """The notice that Gemini will close the connection soon.""" + + timeLeft: str + + +class ServerMessage(TypedDict, total=False): + """A message of Gemini: the fields used here.""" + + setupComplete: dict[str, object] + serverContent: ServerContent + goAway: GoAway + + +class GeminiError(Exception): + """Gemini refused the session or closed it.""" + + +class Gemini: + """A session of the Live API, over a WebSocket.""" + + def __init__(self, http: aiohttp.ClientSession, ws: aiohttp.ClientWebSocketResponse[bool]) -> None: + self.http = http + self.ws = ws + + @classmethod + async def connect(cls, args: argparse.Namespace) -> Gemini: + """Opens a session, which Gemini has to accept in SETUP_LIMIT seconds.""" + http = aiohttp.ClientSession() + try: + return await asyncio.wait_for(cls._open(http, args), SETUP_LIMIT) + except asyncio.TimeoutError: + await http.close() + msg = f'Gemini did not accept the session in {SETUP_LIMIT} s' + raise GeminiError(msg) from None + except BaseException: + await http.close() + raise + + @classmethod + async def _open(cls, http: aiohttp.ClientSession, args: argparse.Namespace) -> Gemini: + try: + ws = await http.ws_connect(args.endpoint, headers={'x-goog-api-key': args.api_key}) + except aiohttp.ClientError as e: + msg = f'Could not connect to Gemini: {e}' + raise GeminiError(msg) from None + + session = cls(http, ws) + await ws.send_json({'setup': setup_message(args)}) + message = await session.receive() + if 'setupComplete' not in message: + msg = f'Gemini sent {message} instead of accepting the session' + raise GeminiError(msg) + return session + + async def receive(self) -> ServerMessage: + """The next message of Gemini, which sends JSON in text or binary frames.""" + frame = await self.ws.receive() + if frame.type in {aiohttp.WSMsgType.TEXT, aiohttp.WSMsgType.BINARY}: + message: ServerMessage = json.loads(frame.data) + return message + + reason = frame.extra if frame.type == aiohttp.WSMsgType.CLOSE else frame.type.name + msg = f'Gemini closed the session ({self.ws.close_code}): {reason}' + raise GeminiError(msg) + + async def send_audio(self, pcm: bytes, sample_rate: int) -> None: + """Sends 16-bit mono samples, unless the session has ended.""" + if self.ws.closed: + return + audio = {'data': base64.b64encode(pcm).decode(), 'mimeType': f'audio/pcm;rate={sample_rate}'} + await self.ws.send_json({'realtimeInput': {'audio': audio}}) + + async def close(self) -> None: + """Closes the WebSocket.""" + await self.ws.close() + await self.http.close() + + +def setup_message(args: argparse.Namespace) -> dict[str, object]: + """The first message of a session: the model, its voice and instructions, and the transcripts to send.""" + generation: dict[str, object] = {'responseModalities': ['AUDIO']} + if args.voice: + generation['speechConfig'] = {'voiceConfig': {'prebuiltVoiceConfig': {'voiceName': args.voice}}} + + setup: dict[str, object] = { + 'model': f'models/{args.model}', + 'generationConfig': generation, + 'inputAudioTranscription': {}, + 'outputAudioTranscription': {}, + } + if args.instructions: + setup['systemInstruction'] = {'parts': [{'text': args.instructions}]} + return setup + + +class Playback: + """Writes Gemini's voice to a track in real time, so an interruption can drop what's left.""" + + def __init__(self, generator: webrtc.MediaStreamTrackGenerator) -> None: + self.writer = generator.writable.get_writer() + self.queued = bytearray() + + def add(self, pcm: bytes) -> None: + """Queues 24 kHz 16-bit mono samples.""" + self.queued += pcm + + def clear(self) -> None: + """Drops the samples not played yet.""" + self.queued.clear() + + async def run(self) -> None: + """Writes a chunk every OUTPUT_CHUNK seconds, silence when nothing is queued.""" + loop = asyncio.get_running_loop() + frames = int(OUTPUT_RATE * OUTPUT_CHUNK) + size = frames * 2 + timestamp = 0 + next_write = loop.time() + while True: + chunk = bytes(self.queued[:size]).ljust(size, b'\0') + del self.queued[:size] + + init = webrtc.AudioDataInit( + format='s16', + sample_rate=OUTPUT_RATE, + number_of_frames=frames, + number_of_channels=1, + timestamp=timestamp, + data=chunk, + ) + await self.writer.write(webrtc.AudioData(init)) + + timestamp += int(OUTPUT_CHUNK * 1_000_000) + next_write += OUTPUT_CHUNK + await asyncio.sleep(max(0, next_write - loop.time())) + + +def audio_data(media: object) -> webrtc.AudioData: + """The media a processor of an audio track reads, as audio data.""" + if not isinstance(media, webrtc.AudioData): + msg = f'expected audio data, not {media!r}' + raise TypeError(msg) + return media + + +def mono_samples(data: webrtc.AudioData) -> array[int]: + """The 16-bit samples of the first channel.""" + options = webrtc.AudioDataCopyToOptions(plane_index=0, format='s16-planar') + buffer = bytearray(data.allocation_size(options)) + data.copy_to(buffer, options) + return array('h', buffer) + + +def gemini_rate(rate: int) -> int: + """The rate to send audio at: 16 kHz when the rate divides evenly, else the rate itself, which Gemini resamples.""" + return INPUT_RATE if rate % INPUT_RATE == 0 else rate + + +def downsample(samples: array[int], factor: int) -> array[int]: + """Averages every `factor` samples into one: a rough low-pass, enough for speech.""" + averages = (sum(samples[start : start + factor]) // factor for start in range(0, len(samples), factor)) + return array('h', averages) + + +async def gathered(pc: webrtc.RTCPeerConnection) -> webrtc.RTCSessionDescription: + """Waits for the ICE candidates until they pause, GATHERING_LIMIT seconds at most, then returns the description.""" + progress = asyncio.Event() + + def on_progress(_event: object) -> None: + progress.set() + + pc.on('icecandidate', on_progress) + pc.on('icegatheringstatechange', on_progress) + + loop = asyncio.get_running_loop() + deadline = loop.time() + GATHERING_LIMIT + try: + while pc.ice_gathering_state != 'complete' and loop.time() < deadline: + progress.clear() + try: + await asyncio.wait_for(progress.wait(), min(GATHERING_PAUSE, deadline - loop.time())) + except asyncio.TimeoutError: + break + finally: + pc.off('icecandidate', on_progress) + pc.off('icegatheringstatechange', on_progress) + + if pc.local_description is None: + msg = 'no local description' + raise RuntimeError(msg) + return pc.local_description + + +@dataclass +class Server: + """Serves the page, and connects each page that calls to a Gemini session of its own.""" + + args: argparse.Namespace + ice_servers: list[webrtc.RTCIceServer] + calls: dict[Call, asyncio.Future[None]] = field(default_factory=dict) + count: int = 0 + + async def page(self, _request: web.Request) -> web.Response: + """The page, with the ICE servers and the id of the transcripts channel filled in.""" + config = {'iceServers': [server.to_json() for server in self.ice_servers], 'eventsChannel': EVENTS_CHANNEL} + # escaped, so a URL in the config can't end the script element + script_config = json.dumps(config).replace('<', '\\u003c') + return web.Response(text=PAGE.replace('/*CONFIG*/', script_config), content_type='text/html') + + async def offer(self, request: web.Request) -> web.Response: + """Opens a Gemini session, then answers the offer of the page, which has its ICE candidates.""" + # other sites can only post JSON after a CORS preflight, which this server doesn't allow + if request.content_type != 'application/json': + raise web.HTTPUnsupportedMediaType + + offer = webrtc.RTCSessionDescriptionInit.from_json(await request.json()) + self.count += 1 + name = f'Call {self.count}' + print(f'{name}: connecting to Gemini') + try: + gemini = await Gemini.connect(self.args) + except GeminiError as e: + print(f'{name}: {e}') + return web.Response(status=502, text=str(e)) + + call = Call(name, gemini, self.ice_servers) + self.calls[call] = asyncio.ensure_future(self._serve(call)) + try: + answer = await call.answer(offer) + except Exception: + call.hang_up() + raise + return web.json_response(answer.to_json()) + + async def _serve(self, call: Call) -> None: + """Runs a call until it ends, then forgets it.""" + try: + await call.run() + finally: + del self.calls[call] + + async def close(self, _app: web.Application) -> None: + """Ends the calls when the server stops.""" + for call in self.calls: + call.hang_up() + await asyncio.gather(*self.calls.values(), return_exceptions=True) + + +def parse_args() -> argparse.Namespace: + """The options of the command line.""" + parser = argparse.ArgumentParser( + description='Voice chat with Gemini Live from a browser, through WebRTC and this server.', + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + parser.add_argument('--api-key', default=os.environ.get('GEMINI_API_KEY'), help='default: $GEMINI_API_KEY') + parser.add_argument('--model', default='gemini-3.8-live', help='the Live model') + parser.add_argument('--voice', help='Kore, Puck, Charon, ..., default: the one of the model') + parser.add_argument('--instructions', help='the system prompt') + parser.add_argument('--host', default='localhost', help='the address to listen on: whoever reaches it uses the key') + parser.add_argument('--port', type=int, default=8080, help='the port to listen on') + parser.add_argument('--endpoint', default=GEMINI_URL, help='the WebSocket URL of the Live API') + return parser.parse_args() + + +def main() -> None: + """Serves the page until Ctrl+C.""" + args = parse_args() + if not args.api_key: + print('No API key: set GEMINI_API_KEY or pass --api-key') + sys.exit(1) + + server = Server(args, [webrtc.RTCIceServer(urls=STUN)]) + app = web.Application() + app.router.add_get('/', server.page) + app.router.add_post('/offer', server.offer) + app.on_shutdown.append(server.close) + + print(f'Open http://{args.host}:{args.port} and press Talk. Press Ctrl+C to stop.') + # handles Ctrl+C on every platform, then runs the shutdown handlers + web.run_app(app, host=args.host, port=args.port, access_log=None, print=None) + + +PAGE = """ + + + + +Gemini Live over WebRTC + + + +

    Gemini Live

    +

    Your microphone goes to the Python server over WebRTC, the server talks to Gemini.

    +Not connected +
    + + + + +""" + + +if __name__ == '__main__': + main() diff --git a/examples/openai_live.py b/examples/openai_live.py index de9b733..6b4c32e 100755 --- a/examples/openai_live.py +++ b/examples/openai_live.py @@ -565,12 +565,9 @@ def _describe_http_error(response: httpx.Response) -> str: def parse_args() -> argparse.Namespace: """The options of the command line.""" - parser = argparse.ArgumentParser( - description='Voice chat with OpenAI GPT-Live over WebRTC.', - formatter_class=argparse.ArgumentDefaultsHelpFormatter, - ) + parser = argparse.ArgumentParser(description='Voice chat with OpenAI GPT-Live over WebRTC.') parser.add_argument('--api-key', default=os.environ.get('OPENAI_API_KEY'), help='default: $OPENAI_API_KEY') - parser.add_argument('--model', default='gpt-live-1', help='the GPT-Live model') + parser.add_argument('--model', default='gpt-live-1', help='the GPT-Live model, default: gpt-live-1') parser.add_argument('--instructions', help='the system prompt') parser.add_argument('--voice', help='quartz, ripple, vesper, willow, stone, gleam, meridian, bossa, tempo, ...') parser.add_argument('--input-device', help='microphone name or index (see --list-devices), default: the system one') diff --git a/examples/whip_whep.py b/examples/whip_whep.py new file mode 100755 index 0000000..e52067a --- /dev/null +++ b/examples/whip_whep.py @@ -0,0 +1,419 @@ +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.9" +# dependencies = ["wrtc>=0.0.1", "sounddevice", "httpx"] +# /// +# +# Copyright 2022-2026 Ilya (Marshal) . +# +# Dedicated to the public domain under CC0, see the LICENSE file of the examples. +# + +"""Plays a live stream over WHEP (RFC 9725) in the terminal, or publishes one over WHIP. + + uv run whip_whep.py + uv run whip_whep.py https://example.com/whep/stream --token SECRET + uv run whip_whep.py --whip http://localhost:8889/wrtc/whip + +With no URL it plays the public demo stream of Dolby OptiView, which is H.264: where the OS has no H.264 codec, it +downloads Cisco's OpenH264 once, and --no-openh264 skips it. The video is drawn with colored half blocks, so the +terminal needs 24-bit color. Press Ctrl+C to stop. +""" + +from __future__ import annotations + +import argparse +import asyncio +import contextlib +import shutil +import signal +import sys +import threading +from typing import TYPE_CHECKING + +import httpx + +import webrtc + +try: + import sounddevice +except OSError: + PORTAUDIO = False +else: + PORTAUDIO = True + +if TYPE_CHECKING: + from collections.abc import Generator + from types import TracebackType + + from typing_extensions import Self + +DEMO = 'https://director.millicast.com/api/whep/k9Mwad/multiview' +# RFC 9725 asks for one transport for all the media +CONFIGURATION = webrtc.RTCConfiguration(bundle_policy='max-bundle') +DECODING_GRACE = 5 # seconds of video packets without a decoded frame before the H.264 hint + +ENTER_ALTERNATE_SCREEN = '\033[?1049h' +LEAVE_ALTERNATE_SCREEN = '\033[?1049l' +HIDE_CURSOR = '\033[?25l' +SHOW_CURSOR = '\033[?25h' +RESET_COLORS = '\033[0m' + + +class EndpointError(Exception): + """The endpoint couldn't be reached or refused the offer.""" + + +class Session: + """A WHIP or WHEP session: the resource the server creates for an offer.""" + + def __init__(self, endpoint: str, token: str | None) -> None: + # a redirect to another host drops the token, as httpx doesn't send credentials across hosts + self.client = httpx.AsyncClient(timeout=30, follow_redirects=True) + self.endpoint = endpoint + self.headers = {} if token is None else {'Authorization': f'Bearer {token}'} + self.resource: httpx.URL | None = None + + async def __aenter__(self) -> Self: + return self + + async def __aexit__( + self, exc_type: type[BaseException] | None, exc: BaseException | None, traceback: TracebackType | None + ) -> None: + if self.resource is not None: + # the server ends the session on its own when it times out + with contextlib.suppress(httpx.HTTPError): + await self.client.delete(self.resource, headers=self.headers, timeout=5) + await self.client.aclose() + + async def post_offer(self, sdp: str) -> str: + """Posts the offer, returns the answer.""" + headers = {**self.headers, 'Content-Type': 'application/sdp'} + try: + response = await self.client.post(self.endpoint, content=sdp, headers=headers) + except httpx.HTTPError as e: + msg = f'{self.endpoint}: {e}' + raise EndpointError(msg) from e + + if not response.is_success: + msg = f'{self.endpoint} returned {response.status_code} {response.reason_phrase}: {response.text[:200]}' + raise EndpointError(msg) + + # relative to the endpoint, after any redirects + location = response.headers.get('Location') + if location is not None: + self.resource = response.url.join(location) + return response.text + + +async def gather_offer(pc: webrtc.RTCPeerConnection) -> str: + """Creates the offer with all the ICE candidates in the SDP, since there is no trickling.""" + gathered = asyncio.Event() + + @pc.on('icegatheringstatechange') + def on_gathering_state(_event: webrtc.Event) -> None: + if pc.ice_gathering_state == 'complete': + gathered.set() + + await pc.set_local_description() + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(gathered.wait(), 2) + if pc.local_description is None: + msg = 'no local description' + raise RuntimeError(msg) + return pc.local_description.sdp + + +async def connect(pc: webrtc.RTCPeerConnection, session: Session) -> None: + """Posts the offer and applies the answer.""" + sdp = await gather_offer(pc) + answer = await session.post_offer(sdp) + await pc.set_remote_description(webrtc.RTCSessionDescriptionInit('answer', answer)) + + +async def play(url: str, token: str | None, stop: asyncio.Event) -> None: + """Plays the stream of a WHEP endpoint until stopped.""" + has_speakers = speakers_found() + if not has_speakers: + print('No audio output, playing the video only') + + # the notice stays on screen while OpenH264 is in use + status = webrtc.openh264.NOTICE if webrtc.openh264.is_enabled() else '' + + pc = webrtc.RTCPeerConnection(CONFIGURATION) + tasks: list[asyncio.Future[None]] = [] + pc.add_transceiver('video', webrtc.RTCRtpTransceiverInit(direction='recvonly')) + pc.add_transceiver('audio', webrtc.RTCRtpTransceiverInit(direction='recvonly')) + + @pc.on('track') + def on_track(event: webrtc.RTCTrackEvent) -> None: + if event.track.kind == 'video': + tasks.append(asyncio.ensure_future(watch(event.track, status))) + elif has_speakers: + tasks.append(asyncio.ensure_future(listen(event.track))) + + @pc.on('connectionstatechange') + def on_connection_state(_event: webrtc.Event) -> None: + if pc.connection_state == 'failed': + stop.set() + + # closing inside ends the tracks before the normal screen is back, so no frame is drawn over it + with fullscreen(status): + try: + async with Session(url, token) as session: + await connect(pc, session) + problem = await wait_for_stop(pc, stop) + finally: + pc.close() + await asyncio.gather(*tasks) + if problem is not None: + print(problem) + + +async def wait_for_stop(pc: webrtc.RTCPeerConnection, stop: asyncio.Event) -> str | None: + """Waits until stopped, returns what went wrong, if anything; a video that doesn't decode is shown right away.""" + # the answer can't tell: the demo's server answers with the offered codecs, yet sends H.264 + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(stop.wait(), DECODING_GRACE) + problem = None + if not stop.is_set() and not h264_decodable() and await video_undecoded(pc): + problem = 'The video arrives but none of it decodes: if it is H.264, it is off (--no-openh264) or unavailable' + draw_status(problem) + + await stop.wait() + if pc.connection_state == 'failed': + return 'The connection failed: the stream ended or the network dropped it' + return problem + + +async def publish(url: str, token: str | None, stop: asyncio.Event) -> None: + """Publishes the synthetic camera and microphone to a WHIP endpoint until stopped.""" + stream = await webrtc.media_devices.get_user_media(webrtc.MediaStreamConstraints(audio=True, video=True)) + pc = webrtc.RTCPeerConnection(CONFIGURATION) + for track in stream.get_tracks(): + pc.add_transceiver(track, webrtc.RTCRtpTransceiverInit(direction='sendonly', streams=[stream])) + + @pc.on('connectionstatechange') + def on_connection_state(_event: webrtc.Event) -> None: + print(f'Connection: {pc.connection_state}') + if pc.connection_state == 'failed': + stop.set() + + try: + async with Session(url, token) as session: + await connect(pc, session) + print(f'Publishing to {session.resource or url}, press Ctrl+C to stop') + await stop.wait() + finally: + pc.close() + for track in stream.get_tracks(): + track.stop() + + +def h264_decodable() -> bool: + """Whether receivers can decode H.264: on macOS with VideoToolbox, elsewhere once OpenH264 is installed.""" + capabilities = webrtc.RTCRtpReceiver.get_capabilities('video') + if capabilities is None: + return False + return any(codec.mime_type.lower() == 'video/h264' for codec in capabilities.codecs) + + +async def video_undecoded(pc: webrtc.RTCPeerConnection) -> bool: + """Whether video packets arrived but no frame was decoded.""" + report = await pc.get_stats() + for stats in report.values(): + if isinstance(stats, webrtc.RTCInboundRtpStreamStats) and stats.kind == 'video': + received = stats.packets_received is not None and stats.packets_received > 0 + decoded = stats.frames_decoded is not None and stats.frames_decoded > 0 + return received and not decoded + return False + + +class Speakers: + """Plays 16-bit audio from a buffer that keeps half a second at most.""" + + def __init__(self) -> None: + self.buffer = bytearray() + self.lock = threading.Lock() + self.limit = 0 + self.stream: sounddevice.RawOutputStream | None = None + + def play(self, samples: bytes | bytearray, rate: int, channels: int) -> None: + """Queues samples, dropping the oldest ones over the limit, reopens the output when their format changes.""" + if self.stream is None or (self.stream.samplerate, self.stream.channels) != (rate, channels): + self.close() + self.buffer = bytearray() + # half a second of 16-bit samples + self.limit = rate * channels + self.stream = sounddevice.RawOutputStream(rate, channels=channels, dtype='int16', callback=self._on_need) + self.stream.start() + with self.lock: + self.buffer += samples + del self.buffer[: -self.limit] + + def _on_need(self, out: memoryview, *_: object) -> None: + with self.lock: + chunk = self.buffer[: len(out)] + del self.buffer[: len(out)] + out[:] = chunk.ljust(len(out), b'\0') + + def close(self) -> None: + """Stops the output.""" + if self.stream is not None: + self.stream.close() + + +def speakers_found() -> bool: + """Whether audio can be played: servers often lack the PortAudio library or an output device.""" + if not PORTAUDIO: + return False + try: + sounddevice.query_devices(kind='output') + except sounddevice.PortAudioError: + return False + return True + + +def move_to(row: int) -> str: + """The code that moves the cursor to the start of a row, counted from 1.""" + return f'\033[{row};1H' + + +def draw(rgbx: bytes | bytearray, width: int, height: int, *, status: str) -> None: + """Draws a frame above the status line with â–€, its foreground is the upper pixel, background the lower one.""" + columns, rows = shutil.get_terminal_size() + video_rows = rows - 1 + # a character is two pixels high + scale = max(width / columns, height / video_rows / 2) + shown_columns = int(width / scale) + shown_rows = int(height / scale / 2) + + def color(x: int, y: int) -> str: + """The red;green;blue of the pixel under a point of the screen.""" + i = (int(y * scale) * width + int(x * scale)) * 4 + red, green, blue = rgbx[i : i + 3] + return f'{red};{green};{blue}' + + def line(y: int) -> str: + """A row of the screen: two rows of pixels, padded to the width of the screen.""" + cells: list[str] = [] + for x in range(shown_columns): + upper = color(x, 2 * y) + lower = color(x, 2 * y + 1) + cells.append(f'\033[38;2;{upper}m\033[48;2;{lower}mâ–€') + return ''.join(cells) + RESET_COLORS + ' ' * (columns - shown_columns) + + lines = [line(y) for y in range(shown_rows)] + lines += [' ' * columns] * (video_rows - len(lines)) + sys.stdout.write(''.join(move_to(y + 1) + text for y, text in enumerate(lines))) + draw_status(status) + + +def draw_status(status: str) -> None: + """Writes the status on the last line of the screen.""" + columns, rows = shutil.get_terminal_size() + sys.stdout.write(move_to(rows) + status[:columns].ljust(columns)) + sys.stdout.flush() + + +@contextlib.contextmanager +def fullscreen(status: str) -> Generator[None, None, None]: + """Switches to the alternate screen, without the cursor, with the status on the last line.""" + sys.stdout.write(ENTER_ALTERNATE_SCREEN + HIDE_CURSOR) + draw_status(status) + try: + yield + finally: + sys.stdout.write(SHOW_CURSOR + LEAVE_ALTERNATE_SCREEN) + + +async def watch(track: webrtc.MediaStreamTrack, status: str) -> None: + """Draws the frames of the video track until it ends, with the status under them.""" + # a buffer of one frame drops the frames the terminal is too slow for + processor = webrtc.MediaStreamTrackProcessor(webrtc.MediaStreamTrackProcessorInit(track, max_buffer_size=1)) + options = webrtc.VideoFrameCopyToOptions(format='RGBX') + async for frame in processor.readable: + if not isinstance(frame, webrtc.VideoFrame): + msg = f'expected a video frame, not {frame!r}' + raise TypeError(msg) + + with frame: + rgbx = bytearray(frame.allocation_size(options)) + await frame.copy_to(rgbx, options) + size = frame.visible_rect + if size is None: + msg = 'the frame has no visible rect' + raise RuntimeError(msg) + + draw(rgbx, int(size.width), int(size.height), status=status) + + +async def listen(track: webrtc.MediaStreamTrack) -> None: + """Plays the audio track until it ends.""" + processor = webrtc.MediaStreamTrackProcessor(webrtc.MediaStreamTrackProcessorInit(track, max_buffer_size=50)) + options = webrtc.AudioDataCopyToOptions(plane_index=0, format='s16') + speakers = Speakers() + try: + async for data in processor.readable: + if not isinstance(data, webrtc.AudioData): + msg = f'expected audio data, not {data!r}' + raise TypeError(msg) + + with data: + samples = bytearray(data.allocation_size(options)) + data.copy_to(samples, options) + rate = int(data.sample_rate) + channels = data.number_of_channels + speakers.play(samples, rate, channels) + finally: + speakers.close() + + +async def enable_openh264() -> None: + """Enables H.264 with Cisco's OpenH264 binary, or goes on without it when it can't be had.""" + try: + await webrtc.openh264.install() + except (RuntimeError, OSError) as e: + print(f'Going on without H.264: {e}') + + +async def main(args: argparse.Namespace) -> None: + """Plays or publishes until Ctrl+C.""" + stop = asyncio.Event() + loop = asyncio.get_running_loop() + + def interrupt() -> None: + """Stops on the first Ctrl+C; the next one raises KeyboardInterrupt.""" + stop.set() + loop.remove_signal_handler(signal.SIGINT) + + # on Windows, Ctrl+C raises KeyboardInterrupt instead + with contextlib.suppress(NotImplementedError): + loop.add_signal_handler(signal.SIGINT, interrupt) + + # the codecs of a connection are fixed when it's created + if not args.no_openh264 and not h264_decodable(): + await enable_openh264() + + try: + if args.whip is not None: + await publish(args.whip, args.token, stop) + else: + await play(args.url, args.token, stop) + except EndpointError as e: + sys.exit(str(e)) + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description='Play a stream over WHEP in the terminal, or publish one over WHIP.') + parser.add_argument('url', nargs='?', default=DEMO, help='the WHEP endpoint to play, default: a public demo') + parser.add_argument('--whip', metavar='URL', help='publish the synthetic camera and microphone to this endpoint') + parser.add_argument('--token', help='a bearer token, if the endpoint needs one') + parser.add_argument( + '--no-openh264', + action='store_true', + help=f"don't download Cisco's OpenH264 for H.264 where the OS has none. {webrtc.openh264.NOTICE} " + 'License: https://www.openh264.org/BINARY_LICENSE.txt', + ) + with contextlib.suppress(KeyboardInterrupt): + asyncio.run(main(parser.parse_args()))