initial commit to repository

streaming-agent/gst_agent.py

@@ -0,0 +1,306 @@
+#!/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,
+        }))
+        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()

streaming-agent/package.json

@@ -0,0 +1,15 @@
+{
+  "name": "css-streaming-agent",
+  "version": "2.0.0",
+  "description": "Signaling and input forwarding for CSS stream",
+  "main": "agent.js",
+  "scripts": {
+    "agent": "node agent.js",
+    "signaling": "node signaling-server.js"
+  },
+  "dependencies": {
+    "node-fetch": "^2.7.0",
+    "ws": "^8.18.0",
+    "express": "^4.21.0"
+  }
+}

streaming-agent/requirements.txt

@@ -0,0 +1,2 @@
+requests
+websockets

streaming-agent/signaling-server.js

@@ -0,0 +1,155 @@
+/**
+ * Signaling Server
+ *
+ * Handles two tracks: video and game audio.
+ * All Cloudflare API calls happen server-side.
+ *
+ * Run: CF_APP_ID=xxx CF_APP_TOKEN=xxx node signaling-server.js
+ */
+
+'use strict';
+
+const express   = require('express');
+const http      = require('http');
+const WebSocket = require('ws');
+const fetch     = require('node-fetch');
+const path      = require('path');
+
+const PORT         = process.env.PORT         || 3000;
+const CF_APP_ID    = process.env.CF_APP_ID    || '';
+const CF_APP_TOKEN = process.env.CF_APP_TOKEN || '';
+const CF_BASE      = `https://rtc.live.cloudflare.com/v1/apps/${CF_APP_ID}`;
+
+if (!CF_APP_ID || !CF_APP_TOKEN) {
+  console.error('[sig] CF_APP_ID and CF_APP_TOKEN must be set');
+  process.exit(1);
+}
+
+async function cfPost(path, body) {
+  const res = await fetch(`${CF_BASE}${path}`, {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${CF_APP_TOKEN}` },
+    body: JSON.stringify(body),
+  });
+  const json = await res.json();
+  if (json.errorCode) throw new Error(`CF API error: ${json.errorDescription}`);
+  return json;
+}
+
+async function cfPut(path, body) {
+  const res = await fetch(`${CF_BASE}${path}`, {
+    method: 'PUT',
+    headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${CF_APP_TOKEN}` },
+    body: JSON.stringify(body),
+  });
+  const json = await res.json();
+  if (json.errorCode) throw new Error(`CF API error: ${json.errorDescription}`);
+  return json;
+}
+
+const app    = express();
+const server = http.createServer(app);
+const wss    = new WebSocket.Server({ server });
+
+app.use(express.json());
+app.get('/', (req, res) => res.sendFile(path.join(__dirname, '..', 'client.html')));
+
+const agentSessions  = new Map();
+const browserToAgent = new Map();
+
+wss.on('connection', ws => {
+  let role           = null;
+  let agentSessionId = null;
+  let browserSessId  = null;
+
+  ws.on('message', async raw => {
+    let msg;
+    try { msg = JSON.parse(raw); } catch { return; }
+
+    if (msg.type === 'register') {
+      role           = 'agent';
+      agentSessionId = msg.sessionId;
+      agentSessions.set(agentSessionId, {
+        agentSessionId: msg.sessionId,
+        videoTrackName: msg.videoTrackName,
+        audioTrackName: msg.audioTrackName,
+        agentWs:        ws,
+      });
+      console.log(`[sig] agent registered — sessionId=${agentSessionId} video=${msg.videoTrackName} audio=${msg.audioTrackName}`);
+    }
+
+    else if (msg.type === 'browser_connect') {
+      role = 'browser';
+      const entry = [...agentSessions.values()][0];
+      if (!entry) { ws.send(JSON.stringify({ type: 'waiting' })); return; }
+
+      let newSession;
+      try {
+        newSession = await cfPost('/sessions/new', {
+          sessionDescription: { type: 'offer', sdp: msg.sdp },
+        });
+      } catch (e) {
+        ws.send(JSON.stringify({ type: 'error', message: e.message }));
+        return;
+      }
+
+      browserSessId = newSession.sessionId;
+      browserToAgent.set(browserSessId, entry.agentSessionId);
+      ws.send(JSON.stringify({
+        type:      'session_created',
+        sessionId: browserSessId,
+        sdpAnswer: newSession.sessionDescription,
+      }));
+    }
+
+    else if (msg.type === 'pull_track') {
+      const agentId = browserToAgent.get(msg.sessionId);
+      const entry   = agentSessions.get(agentId);
+      if (!entry) { ws.send(JSON.stringify({ type: 'error', message: 'no agent session' })); return; }
+
+      let trackResult;
+      try {
+        trackResult = await cfPost(`/sessions/${msg.sessionId}/tracks/new`, {
+          tracks: [
+            { location: 'remote', sessionId: entry.agentSessionId, trackName: entry.videoTrackName },
+            { location: 'remote', sessionId: entry.agentSessionId, trackName: entry.audioTrackName },
+          ],
+        });
+      } catch (e) {
+        ws.send(JSON.stringify({ type: 'error', message: e.message }));
+        return;
+      }
+      ws.send(JSON.stringify({ type: 'track_ready', trackResult }));
+    }
+
+    else if (msg.type === 'renegotiate') {
+      try {
+        await cfPut(`/sessions/${msg.sessionId}/renegotiate`, {
+          sessionDescription: { type: 'answer', sdp: msg.sdp },
+        });
+      } catch (e) {
+        ws.send(JSON.stringify({ type: 'error', message: e.message }));
+      }
+    }
+
+    else if (msg.type === 'input') {
+      const agentId = browserToAgent.get(msg.sessionId);
+      const entry   = agentSessions.get(agentId);
+      if (entry && entry.agentWs && entry.agentWs.readyState === WebSocket.OPEN) {
+        entry.agentWs.send(JSON.stringify(msg.event));
+      }
+    }
+  });
+
+  ws.on('close', () => {
+    if (role === 'agent' && agentSessionId) {
+      agentSessions.delete(agentSessionId);
+      console.log(`[sig] agent disconnected — sessionId=${agentSessionId}`);
+    }
+    if (role === 'browser' && browserSessId) {
+      browserToAgent.delete(browserSessId);
+    }
+  });
+});
+
+server.listen(PORT, () => console.log(`[sig] signaling server listening on port ${PORT}`));