Update app.py

app.py

@@ -5,36 +5,176 @@ import shutil
 import subprocess
 import time
 import logging
+import concurrent.futures
 import threading
+import numpy as np
 import uuid
-import uvloop
-
-# Enable high-performance event loop
-asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
-
+import pty
+import fcntl
+import termios
+import struct
 from aiohttp import web
-from aiortc import RTCPeerConnection, RTCSessionDescription, RTCIceServer, RTCConfiguration
-from aiortc.contrib.media import MediaPlayer
+from aiortc import RTCPeerConnection, RTCSessionDescription, VideoStreamTrack, RTCIceServer, RTCConfiguration
+from av import VideoFrame
+import mss
 
-# --- Configuration ---
 HOST = "0.0.0.0"
 PORT = 7860
 DISPLAY_NUM = ":99"
 
-# Resolution Config
-# Note: Changing resolution dynamically with x11grab is complex. 
-# We lock capture to these dimensions for stability.
-WIDTH = 1280
-HEIGHT = 720
+MAX_WIDTH = 3840
+MAX_HEIGHT = 2160
+
+DEFAULT_WIDTH = 1280
+DEFAULT_HEIGHT = 720
 
-# Cloudflare TURN Credentials
 TURN_USER = "g08abe68c81a07f098bb5f0914549bb32440e5aad0b216c7fba2b61e76fd62c6"
 TURN_PASS = "aed1a10dd10eba9401ad9d99e5c66036d8a970eab5ba8e6dc9845ab57c771a7d"
 
-logging.basicConfig(level=logging.WARNING)
-logger = logging.getLogger("WebRTC-Fast")
+logging.basicConfig(level=logging.WARNING) 
+logger = logging.getLogger("WebRTC-System")
+
+executor = concurrent.futures.ThreadPoolExecutor(max_workers=16)
+thread_local_storage = threading.local()
+
+config = {
+    "width": DEFAULT_WIDTH,
+    "height": DEFAULT_HEIGHT
+}
+
+class ClipboardManager:
+    def __init__(self):
+        self.last_content = ""
+        self.channels = set()
+        self.lock = threading.Lock()
+        self.running = True
+        self.thread = threading.Thread(target=self.monitor_loop, daemon=True)
+        self.thread.start()
+
+    def add_channel(self, channel):
+        with self.lock:
+            self.channels.add(channel)
+
+    def remove_channel(self, channel):
+        with self.lock:
+            self.channels.discard(channel)
+
+    def update_x11(self, content):
+        if content == self.last_content: 
+            return
+        
+        self.last_content = content
+        try:
+            env = os.environ.copy()
+            env["DISPLAY"] = DISPLAY_NUM
+            
+            p = subprocess.Popen(
+                ['xclip', '-i', '-selection', 'clipboard'], 
+                stdin=subprocess.PIPE, 
+                env=env
+            )
+            p.communicate(input=content.encode('utf-8'))
+        except Exception as e:
+            logger.error(f"Clipboard Write Error: {e}")
+
+    def monitor_loop(self):
+        while self.running:
+            try:
+                if not os.environ.get("DISPLAY"):
+                    os.environ["DISPLAY"] = DISPLAY_NUM
+
+                try:
+                    curr = subprocess.check_output(
+                        ['xclip', '-o', '-selection', 'clipboard'], 
+                        stderr=subprocess.DEVNULL,
+                        timeout=0.5
+                    ).decode('utf-8')
+                except:
+                    curr = ""
+
+                if curr and curr != self.last_content:
+                    self.last_content = curr
+                    self.broadcast(curr)
+
+            except Exception:
+                pass
+            time.sleep(1.0)
+
+    def broadcast(self, content):
+        msg = json.dumps({"type": "clipboard", "content": content})
+        with self.lock:
+            for ch in list(self.channels):
+                try:
+                    if ch.readyState == "open":
+                        ch.send(msg)
+                    else:
+                        self.channels.discard(ch)
+                except:
+                    self.channels.discard(ch)
+
+clipboard_manager = ClipboardManager()
+
+class Terminal:
+    def __init__(self, channel):
+        self.channel = channel
+        self.master_fd, self.slave_fd = pty.openpty()
+        self.pid = os.fork()
+
+        if self.pid == 0:
+            os.close(self.master_fd)
+            os.setsid()
+            os.dup2(self.slave_fd, 0)
+            os.dup2(self.slave_fd, 1)
+            os.dup2(self.slave_fd, 2)
+            if self.slave_fd > 2:
+                os.close(self.slave_fd)
+            
+            os.environ["TERM"] = "xterm-256color"
+            os.environ["DISPLAY"] = DISPLAY_NUM
+            os.environ["HOME"] = "/home/user"
+            
+            shell = "/bin/bash"
+            os.execl(shell, shell)
+        else:
+            os.close(self.slave_fd)
+            self.loop = asyncio.get_running_loop()
+            self.loop.add_reader(self.master_fd, self.read_output)
+            self.channel.on("close", self.close)
+
+    def read_output(self):
+        try:
+            data = os.read(self.master_fd, 4096)
+            if data:
+                self.channel.send(data.decode('utf-8', 'ignore'))
+        except OSError:
+            self.close()
+
+    def write_input(self, data):
+        if self.master_fd:
+            try:
+                if data.startswith('{"type":"resize"'):
+                    try:
+                        cmd = json.loads(data)
+                        self.resize(cmd.get("rows", 24), cmd.get("cols", 80))
+                    except: pass
+                else:
+                    os.write(self.master_fd, data.encode('utf-8'))
+            except Exception as e:
+                logger.error(f"PTY Write Error: {e}")
+
+    def resize(self, rows, cols):
+        try:
+            winsize = struct.pack("HHHH", rows, cols, 0, 0)
+            fcntl.ioctl(self.master_fd, termios.TIOCSWINSZ, winsize)
+        except: pass
+
+    def close(self):
+        try:
+            self.loop.remove_reader(self.master_fd)
+            os.close(self.master_fd)
+            os.kill(self.pid, 9)
+        except: pass
 
-# --- High Performance Input Manager ---
 class InputManager:
     def __init__(self):
         self.process = None
@@ -44,33 +184,28 @@ class InputManager:
     def start_process(self):
         if not os.environ.get("DISPLAY"): return
         try:
-            # - is stdin
-            self.process = subprocess.Popen(
-                ['xdotool', '-'], 
-                stdin=subprocess.PIPE, 
-                encoding='utf-8', 
-                bufsize=0 
-            )
-        except Exception as e:
-            logger.error(f"Failed to start xdotool: {e}")
+            self.process = subprocess.Popen(['xdotool', '-'], stdin=subprocess.PIPE, encoding='utf-8', bufsize=0)
+        except Exception as e: logger.error(f"Failed to start xdotool: {e}")
 
     def _send_raw(self, command):
-        if self.process is None or self.process.poll() is not None:
-            self.start_process()
-        
+        if self.process is None or self.process.poll() is not None: self.start_process()
         if self.process:
             try:
                 self.process.stdin.write(command + "\n")
                 self.process.stdin.flush()
-            except Exception:
+            except:
                 try: self.process.kill()
                 except: pass
                 self.process = None
 
     def send(self, command):
-        with self.lock:
-            self._send_raw(command)
+        with self.lock: self._send_raw(command)
 
+    def mouse_move(self, x, y): self.send(f"mousemove {x} {y}")
+    def mouse_down(self, btn): self.send(f"mousedown {btn}")
+    def mouse_up(self, btn): self.send(f"mouseup {btn}")
+    def key_down(self, key): self.send(f"keydown {key}")
+    def key_up(self, key): self.send(f"keyup {key}")
     def scroll(self, dy):
         with self.lock:
             self.scroll_accum += dy
@@ -82,103 +217,85 @@ class InputManager:
                 self._send_raw("click 4")
                 self.scroll_accum += THRESHOLD
 
-    def mouse_move(self, x, y): self.send(f"mousemove {x} {y}")
-    def mouse_down(self, btn): self.send(f"mousedown {btn}")
-    def mouse_up(self, btn): self.send(f"mouseup {btn}")
-    def key_down(self, key): self.send(f"keydown {key}")
-    def key_up(self, key): self.send(f"keyup {key}")
-
 input_manager = InputManager()
 
-# --- System Management ---
 def start_system():
     os.environ["DISPLAY"] = DISPLAY_NUM
-
     if not shutil.which("Xvfb"): raise FileNotFoundError("Xvfb missing")
     
     logger.warning(f"Starting Xvfb on {DISPLAY_NUM}...")
-    # Start Xvfb
-    subprocess.Popen([
-        "Xvfb", DISPLAY_NUM, 
-        "-screen", "0", f"{WIDTH}x{HEIGHT}x24",
-        "-ac", "-noreset"
-    ])
+    subprocess.Popen(["Xvfb", DISPLAY_NUM, "-screen", "0", f"{MAX_WIDTH}x{MAX_HEIGHT}x24", "-ac", "-noreset"])
+    time.sleep(2)
+    input_manager.start_process()
+    set_resolution(DEFAULT_WIDTH, DEFAULT_HEIGHT)
     
-    time.sleep(2) # Wait for X11
-
-    # Start Window Manager
     if shutil.which("matchbox-window-manager"):
         subprocess.Popen("matchbox-window-manager -use_titlebar no", shell=True)
 
-    # Initialize Input
-    input_manager.start_process()
+def set_resolution(w, h):
+    try:
+        if w % 2 != 0: w += 1
+        if h % 2 != 0: h += 1
+        config["width"] = w
+        config["height"] = h
+    except: pass
+
+class VirtualScreenTrack(VideoStreamTrack):
+    kind = "video"
+    def __init__(self):
+        super().__init__()
+        self.last_frame_time = 0
 
-    # Start Browser
-    threading.Thread(target=keep_chrome_alive, daemon=True).start()
-
-def keep_chrome_alive():
-    chrome_cmd = (
-        "google-chrome "
-        "--no-sandbox "
-        "--start-maximized "
-        "--user-data-dir=/home/user/chrome-data "
-        "--disable-infobars "
-        "--disable-dev-shm-usage "
-        "--disable-gpu " # Software rendering is often more stable for pure Xvfb capture
-        "--no-first-run "
-        "--no-default-browser-check "
-        "--window-position=0,0 "
-        f"--window-size={WIDTH},{HEIGHT}" 
-    )
-    while True:
+    def _capture(self):
         try:
-            subprocess.run(chrome_cmd, shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-            time.sleep(1)
-        except: time.sleep(2)
-
-# --- Input Mapping ---
-def map_key(key):
-    if key == " ": return "space"
-    k = key.lower()
-    charmap = {
-        "control": "ctrl", "shift": "shift", "alt": "alt", "meta": "super", "cmd": "super",
-        "enter": "Return", "backspace": "BackSpace", "tab": "Tab", "escape": "Escape",
-        "arrowup": "Up", "arrowdown": "Down", "arrowleft": "Left", "arrowright": "Right",
-        "delete": "Delete", "insert": "Insert",
-        "!": "exclam", "@": "at", "#": "numbersign", "$": "dollar", "%": "percent",
-        "^": "asciicircum", "&": "ampersand", "*": "asterisk", "(": "parenleft",
-        ")": "parenright", "-": "minus", "_": "underscore", "=": "equal", "+": "plus",
-        "[": "bracketleft", "{": "braceleft", "]": "bracketright", "}": "braceright",
-        ";": "semicolon", ":": "colon", "'": "apostrophe", "\"": "quotedbl",
-        ",": "comma", "<": "less", ".": "period", ">": "greater", "/": "slash",
-        "?": "question", "\\": "backslash", "|": "bar", "`": "grave", "~": "asciitilde",
-        " ": "space"
-    }
-    return charmap.get(k, k)
+            if not hasattr(thread_local_storage, "sct"):
+                thread_local_storage.sct = mss.mss()
+            monitor = {"top": 0, "left": 0, "width": config["width"], "height": config["height"]}
+            return np.array(thread_local_storage.sct.grab(monitor))[..., :3]
+        except: return None
+
+    async def recv(self):
+        FPS = 30
+        FRAME_TIME = 1.0 / FPS
+        pts, time_base = await self.next_timestamp()
+        
+        current_time = time.time()
+        wait = FRAME_TIME - (current_time - self.last_frame_time)
+        if wait > 0: await asyncio.sleep(wait)
+        self.last_frame_time = time.time()
+
+        frame = await asyncio.get_event_loop().run_in_executor(executor, self._capture)
+        if frame is None:
+            blank = np.zeros((config["height"], config["width"], 3), dtype=np.uint8)
+            av_frame = VideoFrame.from_ndarray(blank, format="bgr24")
+        else:
+            av_frame = VideoFrame.from_ndarray(frame, format="bgr24")
+        
+        av_frame.pts = pts
+        av_frame.time_base = time_base
+        return av_frame
 
 def process_input(data):
     try:
         msg = json.loads(data)
         t = msg.get("type")
+        w, h = config["width"], config["height"]
         
-        # Mapping normalized coordinates (0.0 - 1.0) back to pixels
-        if t == "mousemove":
-            input_manager.mouse_move(int(msg["x"] * WIDTH), int(msg["y"] * HEIGHT))
-        elif t == "mousedown":
+        if t == "clipboard":
+            clipboard_manager.update_x11(msg.get("content", ""))
+        elif t == "mousemove": 
+            input_manager.mouse_move(int(msg["x"]*w), int(msg["y"]*h))
+        elif t == "mousedown": 
             input_manager.mouse_down({0:1, 1:2, 2:3}.get(msg.get("button"), 1))
-        elif t == "mouseup":
+        elif t == "mouseup": 
             input_manager.mouse_up({0:1, 1:2, 2:3}.get(msg.get("button"), 1))
+        elif t == "keydown": 
+            input_manager.key_down(msg.get("key"))
+        elif t == "keyup": 
+            input_manager.key_up(msg.get("key"))
         elif t == "wheel":
             input_manager.scroll(msg.get("deltaY", 0))
-        elif t == "keydown":
-            k = map_key(msg.get("key"))
-            if k: input_manager.key_down(k)
-        elif t == "keyup":
-            k = map_key(msg.get("key"))
-            if k: input_manager.key_up(k)
-    except Exception: pass
-
-# --- WebRTC Routes ---
+    except: pass
 
 async def offer(request):
     try:
@@ -186,80 +303,45 @@ async def offer(request):
         offer = RTCSessionDescription(sdp=params["sdp"], type=params["type"])
     except: return web.Response(status=400)
 
-    # 1. Create Peer Connection
     pc = RTCPeerConnection(RTCConfiguration(iceServers=[
         RTCIceServer(urls=["turns:turn.cloudflare.com:443?transport=tcp", "turn:turn.cloudflare.com:3478?transport=udp"], username=TURN_USER, credential=TURN_PASS),
-        RTCIceServer(urls=["stun:stun.l.google.com:19302"]) 
+        RTCIceServer(urls=["stun:stun.l.google.com:19302"])
     ]))
-    
-    # 2. Configure FFmpeg Options for Low Latency
-    # These flags tell FFmpeg to capture X11 directly and encode immediately
-    options = {
-        "framerate": "30",
-        "video_size": f"{WIDTH}x{HEIGHT}",
-        "probesize": "32",          # Fast startup
-        "analyzeduration": "0",     # No delay for analysis
-        "fflags": "nobuffer",       # Reduce buffer latency
-        "preset": "ultrafast",      # Fastest encoding (trades CPU for bitrate, but low lag)
-        "tune": "zerolatency",      # Critical for realtime
-        "threads": "4"              # Use multi-threading
-    }
-
-    # 3. Create MediaPlayer (Replaces manual VideoTrack)
-    # This runs in a separate C-thread, bypassing Python GIL issues
-    player = MediaPlayer(DISPLAY_NUM, format="x11grab", options=options)
-    
-    # Add track to PC
-    pc.addTrack(player.video)
     pcs.add(pc)
 
-    # Lifecycle management
     @pc.on("connectionstatechange")
     async def on_state():
         if pc.connectionState in ["failed", "closed"]:
             await pc.close()
-            # Ensure ffmpeg process is killed
-            if player:
-                player.video.stop()
             pcs.discard(pc)
 
     @pc.on("datachannel")
     def on_dc(channel):
-        # Offload input processing to default thread executor to avoid blocking async loop
-        channel.on("message", lambda m: asyncio.get_event_loop().run_in_executor(None, process_input, m))
-
-    # SDP Negotiation
+        if channel.label == "input":
+            clipboard_manager.add_channel(channel)
+            channel.on("message", lambda m: asyncio.get_event_loop().run_in_executor(executor, process_input, m))
+            channel.on("close", lambda: clipboard_manager.remove_channel(channel))
+        elif channel.label == "terminal":
+            term = Terminal(channel)
+            channel.on("message", term.write_input)
+
+    pc.addTrack(VirtualScreenTrack())
     await pc.setRemoteDescription(offer)
     answer = await pc.createAnswer()
     await pc.setLocalDescription(answer)
+    return web.Response(content_type="application/json", text=json.dumps({"sdp": pc.localDescription.sdp, "type": pc.localDescription.type}), headers={"Access-Control-Allow-Origin": "*"})
 
-    # Clean SDP for faster connection
-    sdp = "\r\n".join([l for l in pc.localDescription.sdp.splitlines() if "a=candidate" not in l or "typ relay" in l]) + "\r\n"
-    
-    return web.Response(
-        content_type="application/json", 
-        text=json.dumps({"sdp": sdp, "type": pc.localDescription.type}), 
-        headers={"Access-Control-Allow-Origin": "*"}
-    )
-
-async def index(r): return web.Response(text="WebRTC X11 Streamer Ready")
+async def index(r): return web.Response(text="WebRTC Backend Running")
 async def options(r): return web.Response(headers={"Access-Control-Allow-Origin": "*", "Access-Control-Allow-Methods": "POST, OPTIONS", "Access-Control-Allow-Headers": "Content-Type"})
 
-# --- Cleanup ---
 pcs = set()
-async def on_shutdown(app):
-    coros = [pc.close() for pc in pcs]
-    await asyncio.gather(*coros)
+async def on_shutdown(app): await asyncio.gather(*[pc.close() for pc in pcs])
 
-# --- Entry Point ---
 if __name__ == "__main__":
     start_system()
-    
     app = web.Application()
     app.on_shutdown.append(on_shutdown)
     app.router.add_get("/", index)
     app.router.add_post("/offer", offer)
     app.router.add_options("/offer", options)
-    
-    logger.warning(f"Server starting on {HOST}:{PORT}")
     web.run_app(app, host=HOST, port=PORT)