#!/usr/bin/env python3 """ GStreamer WebRTC Agent for CSS Streaming Two tracks published to Cloudflare Calls: - Video: ximagesrc → VP8 → webrtcbin - Game audio: pulsesrc → Opus → webrtcbin Dependencies: apt: python3-gst-1.0 gstreamer1.0-plugins-bad gstreamer1.0-plugins-good gstreamer1.0-plugins-ugly gstreamer1.0-libav xdotool pip: requests websockets """ import asyncio import json import logging import os import subprocess import sys import threading import time import gi gi.require_version('Gst', '1.0') gi.require_version('GstWebRTC', '1.0') gi.require_version('GstSdp', '1.0') from gi.repository import Gst, GstWebRTC, GstSdp, GLib import requests import websockets logging.basicConfig(level=logging.INFO, format='[gst_agent] %(message)s', stream=sys.stderr) log = logging.getLogger(__name__) # ── Configuration ───────────────────────────────────────────────────────────── CF_APP_ID = os.environ.get('CF_APP_ID', '') CF_APP_TOKEN = os.environ.get('CF_APP_TOKEN', '') CF_BASE = f'https://rtc.live.cloudflare.com/v1/apps/{CF_APP_ID}' DISPLAY = os.environ.get('DISPLAY', ':99') CAP_W = int(os.environ.get('CAP_W', '800')) CAP_H = int(os.environ.get('CAP_H', '600')) CAP_FPS = int(os.environ.get('CAP_FPS', '60')) SIG_URL = os.environ.get('SIG_URL', 'ws://localhost:3000') # ── Cloudflare Calls REST helpers ───────────────────────────────────────────── def cf_post(path, body): r = requests.post(f'{CF_BASE}{path}', json=body, headers={ 'Authorization': f'Bearer {CF_APP_TOKEN}', 'Content-Type': 'application/json', }) data = r.json() if 'errorCode' in data: raise RuntimeError(f"CF API error: {data['errorDescription']}") return data def cf_put(path, body): r = requests.put(f'{CF_BASE}{path}', json=body, headers={ 'Authorization': f'Bearer {CF_APP_TOKEN}', 'Content-Type': 'application/json', }) data = r.json() if 'errorCode' in data: raise RuntimeError(f"CF API error: {data['errorDescription']}") return data # ── Input forwarding via xdotool ────────────────────────────────────────────── def xdotool(*args): env = {**os.environ, 'DISPLAY': DISPLAY} subprocess.Popen( ['xdotool'] + list(args), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) def handle_input(event): t = event.get('type') if t == 'mouse_move': dx = round(event.get('dx', 0)) dy = round(event.get('dy', 0)) if dx != 0 or dy != 0: xdotool('mousemove_relative', '--', str(dx), str(dy)) elif t == 'mouse_down': xdotool('mousedown', str(event.get('button', 1))) elif t == 'mouse_up': xdotool('mouseup', str(event.get('button', 1))) elif t == 'key_down': xdotool('keydown', event.get('key', '')) elif t == 'key_up': xdotool('keyup', event.get('key', '')) elif t == 'mouse_wheel': button = '4' if event.get('direction') == 'up' else '5' xdotool('click', button) # ── GStreamer Pipeline ──────────────────────────────────────────────────────── class CSSStreamAgent: def __init__(self): self.pipeline = None self.webrtcbin = None self.local_sdp = None self.loop = GLib.MainLoop() self._offer_ready = threading.Event() self._ice_ready = threading.Event() def build_pipeline(self): pipeline_str = ( # Video: ximagesrc → VP8 f'ximagesrc display-name={DISPLAY} use-damage=0 ! ' f'video/x-raw,framerate={CAP_FPS}/1,width={CAP_W},height={CAP_H} ! ' f'videoconvert ! ' f'vp8enc deadline=1 target-bitrate=1500000 keyframe-max-dist=60 ' f'error-resilient=partitions lag-in-frames=0 ! ' f'rtpvp8pay pt=96 ! ' f'application/x-rtp,media=video,encoding-name=VP8,payload=96 ! ' f'webrtcbin. ' # Game audio: pulsesrc → Opus f'pulsesrc ! ' f'audio/x-raw,rate=48000,channels=2 ! ' f'audioconvert ! ' f'opusenc ! ' f'rtpopuspay pt=97 ! ' f'application/x-rtp,media=audio,encoding-name=OPUS,payload=97 ! ' f'webrtcbin. ' # WebRTC f'webrtcbin name=webrtcbin bundle-policy=max-bundle ' f'stun-server=stun://stun.cloudflare.com:3478' ) log.info(f'Building pipeline {CAP_W}x{CAP_H}@{CAP_FPS}fps') self.pipeline = Gst.parse_launch(pipeline_str) self.webrtcbin = self.pipeline.get_by_name('webrtcbin') self.webrtcbin.connect('on-negotiation-needed', self._on_negotiation_needed) self.webrtcbin.connect('on-ice-candidate', self._on_ice_candidate) self.webrtcbin.connect('notify::ice-connection-state', self._on_ice_state) self.pipeline.set_state(Gst.State.READY) for i in range(2): try: t = self.webrtcbin.get_transceiver(i) if t: t.set_property('direction', GstWebRTC.WebRTCRTPTransceiverDirection.SENDONLY) except Exception: pass def _on_negotiation_needed(self, element): log.info('Creating offer...') promise = Gst.Promise.new_with_change_func(self._on_offer_created, element) element.emit('create-offer', None, promise) def _on_offer_created(self, promise, element): promise.wait() reply = promise.get_reply() offer = reply.get_value('offer') self.local_sdp = offer.sdp.as_text() set_promise = Gst.Promise.new() element.emit('set-local-description', offer, set_promise) set_promise.interrupt() log.info('Offer created and local description set') self._offer_ready.set() def _on_ice_candidate(self, element, mlineindex, candidate): log.debug(f'ICE candidate [{mlineindex}]: {candidate}') def _on_ice_state(self, element, prop): state = element.get_property('ice-connection-state') state_map = { GstWebRTC.WebRTCICEConnectionState.NEW: 'new', GstWebRTC.WebRTCICEConnectionState.CHECKING: 'checking', GstWebRTC.WebRTCICEConnectionState.CONNECTED: 'connected', GstWebRTC.WebRTCICEConnectionState.COMPLETED: 'completed', GstWebRTC.WebRTCICEConnectionState.DISCONNECTED: 'disconnected', GstWebRTC.WebRTCICEConnectionState.FAILED: 'failed', GstWebRTC.WebRTCICEConnectionState.CLOSED: 'closed', } log.info(f'ICE state: {state_map.get(state, state)}') if state in (GstWebRTC.WebRTCICEConnectionState.CONNECTED, GstWebRTC.WebRTCICEConnectionState.COMPLETED): self._ice_ready.set() elif state == GstWebRTC.WebRTCICEConnectionState.FAILED: log.error('ICE failed') self.loop.quit() def set_remote_description(self, sdp_text, sdp_type='answer'): _, sdp = GstSdp.SDPMessage.new_from_text(sdp_text) wtype = GstWebRTC.WebRTCSDPType.ANSWER if sdp_type == 'answer' else GstWebRTC.WebRTCSDPType.OFFER desc = GstWebRTC.WebRTCSessionDescription.new(wtype, sdp) promise = Gst.Promise.new() self.webrtcbin.emit('set-remote-description', desc, promise) promise.interrupt() log.info('Remote description set') def get_mids(self): mids = [] for line in self.local_sdp.split('\n'): line = line.strip() if line.startswith('a=mid:'): mids.append(line[6:].strip()) return mids def start(self): self.pipeline.set_state(Gst.State.PLAYING) log.info('Pipeline playing') def stop(self): if self.pipeline: self.pipeline.set_state(Gst.State.NULL) def run_glib_loop(self): try: self.loop.run() except Exception as e: log.error(f'GLib loop error: {e}') finally: self.stop() # ── Signaling ───────────────────────────────────────────────────────────────── async def run_signaling(session_id, video_track_name, audio_track_name): log.info(f'Connecting to signaling server {SIG_URL}') async with websockets.connect(SIG_URL) as ws: await ws.send(json.dumps({ 'type': 'register', 'sessionId': session_id, 'videoTrackName': video_track_name, 'audioTrackName': audio_track_name, 'containerName': os.environ.get('CONTAINER_NAME', 'css-test'), })) log.info('Registered with signaling server') async for raw in ws: try: msg = json.loads(raw) handle_input(msg) except Exception as e: log.error(f'Input error: {e}') # ── Main ────────────────────────────────────────────────────────────────────── def main(): if not CF_APP_ID or not CF_APP_TOKEN: log.error('CF_APP_ID and CF_APP_TOKEN must be set') sys.exit(1) Gst.init(None) log.info(f'Starting — {CAP_W}x{CAP_H}@{CAP_FPS}fps display={DISPLAY}') agent = CSSStreamAgent() agent.build_pipeline() agent.start() t = threading.Thread(target=agent.run_glib_loop, daemon=True) t.start() log.info('Waiting for SDP offer...') if not agent._offer_ready.wait(timeout=15): log.error('Timeout waiting for offer') sys.exit(1) log.info('Creating Cloudflare Calls session...') session_result = cf_post('/sessions/new', { 'sessionDescription': {'type': 'offer', 'sdp': agent.local_sdp} }) session_id = session_result['sessionId'] log.info(f'Session created: {session_id}') agent.set_remote_description(session_result['sessionDescription']['sdp']) log.info('Waiting for ICE...') if not agent._ice_ready.wait(timeout=15): log.error('Timeout waiting for ICE') sys.exit(1) log.info('ICE connected') mids = agent.get_mids() log.info(f'MIDs: {mids}') if len(mids) < 2: log.error(f'Expected 2 MIDs, got {len(mids)}: {mids}') sys.exit(1) ts = int(time.time() * 1000) video_track_name = f'css-video-{ts}' audio_track_name = f'css-audio-{ts}' track_result = cf_post(f'/sessions/{session_id}/tracks/new', { 'sessionDescription': {'type': 'offer', 'sdp': agent.local_sdp}, 'tracks': [ {'location': 'local', 'mid': mids[0], 'trackName': video_track_name}, {'location': 'local', 'mid': mids[1], 'trackName': audio_track_name}, ], }) if track_result.get('sessionDescription'): agent.set_remote_description(track_result['sessionDescription']['sdp']) log.info(f'Video track: {video_track_name}') log.info(f'Audio track: {audio_track_name}') log.info(f'Streaming — sessionId={session_id}') asyncio.run(run_signaling(session_id, video_track_name, audio_track_name)) if __name__ == '__main__': main()