342 lines
13 KiB
Python
Executable File
342 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
GStreamer WebRTC Agent for CSS Streaming
|
|
|
|
Two tracks published to Cloudflare Calls:
|
|
- Video: ximagesrc (Xwayland :99) → VP8 → webrtcbin
|
|
- Game audio: pipewiresrc → Opus → webrtcbin
|
|
|
|
CSS renders via Wine Wayland driver → Weston (Vulkan/DXVK) → Xwayland → X11
|
|
ximagesrc captures from Xwayland X11 display :99
|
|
|
|
Dependencies:
|
|
apt: python3-gst-1.0 gstreamer1.0-plugins-bad gstreamer1.0-plugins-good
|
|
gstreamer1.0-plugins-ugly gstreamer1.0-libav gstreamer1.0-pipewire
|
|
xdotool weston xwayland wireplumber pipewire pipewire-pulse
|
|
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,
|
|
)
|
|
|
|
# Track held keys so we can release them all on controller change
|
|
_held_keys = set()
|
|
|
|
def release_all_keys():
|
|
# Release all currently held keys and reset X11 modifier state
|
|
for key in list(_held_keys):
|
|
try:
|
|
xdotool('keyup', key)
|
|
except Exception:
|
|
pass
|
|
_held_keys.clear()
|
|
# Reset X11 modifier state (clears stuck shift/ctrl/alt etc.)
|
|
xdotool('key', '--clearmodifiers', 'Escape')
|
|
|
|
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':
|
|
key = event.get('key', '')
|
|
_held_keys.add(key)
|
|
xdotool('keydown', key)
|
|
elif t == 'key_up':
|
|
key = event.get('key', '')
|
|
_held_keys.discard(key)
|
|
xdotool('keyup', key)
|
|
elif t == 'mouse_wheel':
|
|
button = '4' if event.get('direction') == 'up' else '5'
|
|
xdotool('click', button)
|
|
elif t == 'release_keys':
|
|
release_all_keys()
|
|
|
|
# ── 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 → convert → scale → rate-enforce → VP8
|
|
# videoconvert must come before caps negotiation;
|
|
# videoscale enforces width/height; videorate enforces framerate.
|
|
f'ximagesrc display-name={DISPLAY} use-damage=0 ! '
|
|
f'videoconvert ! '
|
|
f'videoscale ! '
|
|
f'videorate ! '
|
|
f'video/x-raw,framerate={CAP_FPS}/1,width={CAP_W},height={CAP_H} ! '
|
|
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: pipewiresrc → Opus
|
|
# target-object=game_output pins capture to the game audio sink.
|
|
# Without this, pipewiresrc defaults to the system capture device
|
|
# (virtual_mic / microphone) instead of the game audio monitor.
|
|
# stream.capture.sink=true tells PipeWire to use game_output's
|
|
# monitor ports rather than treating it as a regular source node.
|
|
f'pipewiresrc target-object=game_output '
|
|
f'stream-properties="props,stream.capture.sink=(boolean)true" ! '
|
|
f'audio/x-raw,format=F32LE,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 via Xwayland')
|
|
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()
|