extract desktop control interface

desktop_control_interface.py

@@ -0,0 +1,41 @@
+import pyautogui
+import numpy
+import evdev
+import keymap
+
+pyautogui.FAILSAFE = False
+
+class DesktopControlInterface():
+    def __init__(self, resolution):
+        self.resolution = resolution
+        self.ui = evdev.UInput()
+        self.valid_actions = ["keyboard", "click", "rightclick", "mousemove", "joystick", "paste"]
+
+    def handle_action(self, action, data):
+        if action == "mousemove":
+            x = numpy.interp(data["cursorPositionX"], (0, data["displayWidth"]), (0, self.resolution.width))
+            y = numpy.interp(data["cursorPositionY"], (0, data["displayHeight"]), (0, self.resolution.height))
+            pyautogui.moveTo(x, y, _pause=False)
+        # elif action == "joystick":
+        #     x = numpy.interp(data["x"], (-38, 38), (0, self.resolution.width))
+        #     y = numpy.interp(data["y"], (-38, 38), (self.resolution.height, 0))
+        #     print(f'{data["y"]} {self.resolution.height} {y}')
+        #     pyautogui.moveTo(x, y, _pause=False)
+        elif action == "click":
+            pyautogui.click()
+        elif action == "rightclick":
+            pyautogui.rightClick()
+        elif action == "keyboard":
+            try:
+                # keymap.reload()
+                osKey = keymap.iOStoLinux[data["key"]]
+                self.ui.write(evdev.ecodes.EV_KEY, osKey, data["direction"])
+                self.ui.syn()
+            except KeyError:
+                print(f"Unknown key: {data['key']}")
+        elif action == "paste":
+            # might as well support the secureput protocol completely
+            pyautogui.write(data["payload"]["string"])
+    
+    def stop(self):
+        self.ui.close()

desktop_stream_track.py

@@ -5,18 +5,14 @@ from aiortc import VideoStreamTrack
 import Xlib
 import Xlib.display
 import os
-import pyautogui
+from desktop_control_interface import DesktopControlInterface
-import numpy
-import evdev
-import keymap
-
-pyautogui.FAILSAFE = False
 
 # https://ffmpeg.org/ffmpeg-devices.html#x11grab
 class DesktopStreamTrack(VideoStreamTrack):
     def __init__(self):
         super().__init__()
         self.resolution = Xlib.display.Display(os.environ["DISPLAY"]).screen().root.get_geometry()
+        self.control_interface = DesktopControlInterface(self.resolution)
         options =  {
             'draw_mouse': '1',
             'i':':0.0+0,0',
@@ -24,8 +20,6 @@ class DesktopStreamTrack(VideoStreamTrack):
             'video_size': str(self.resolution.width) + "x" + str(self.resolution.height)
         }
         self.container = av.open(':0', format='x11grab', options=options)
-        self.ui = evdev.UInput()
-        self.valid_actions = ["keyboard", "click", "rightclick", "mousemove", "joystick", "paste"]
 
     async def recv(self):
         pts, time_base = await self.next_timestamp()
@@ -38,34 +32,11 @@ class DesktopStreamTrack(VideoStreamTrack):
         return frame
     
     def handle_action(self, action, data):
-        if action == "mousemove":
+        self.control_interface.handle_action(action, data)
-            x = numpy.interp(data["cursorPositionX"], (0, data["displayWidth"]), (0, self.resolution.width))
-            y = numpy.interp(data["cursorPositionY"], (0, data["displayHeight"]), (0, self.resolution.height))
-            pyautogui.moveTo(x, y, _pause=False)
-        # elif action == "joystick":
-        #     x = numpy.interp(data["x"], (-38, 38), (0, self.resolution.width))
-        #     y = numpy.interp(data["y"], (-38, 38), (self.resolution.height, 0))
-        #     print(f'{data["y"]} {self.resolution.height} {y}')
-        #     pyautogui.moveTo(x, y, _pause=False)
-        elif action == "click":
-            pyautogui.click()
-        elif action == "rightclick":
-            pyautogui.rightClick()
-        elif action == "keyboard":
-            try:
-                # keymap.reload()
-                osKey = keymap.iOStoLinux[data["key"]]
-                self.ui.write(evdev.ecodes.EV_KEY, osKey, data["direction"])
-                self.ui.syn()
-            except KeyError:
-                print(f"Unknown key: {data['key']}")
-        elif action == "paste":
-            # might as well support the secureput protocol completely
-            pyautogui.write(data["payload"]["string"])
             
     def stop(self) -> None:
         super().stop()
-        self.ui.close()
+        self.control_interface.stop()
 
 if __name__ == "__main__":
     from time import time_ns

server.py

@@ -164,7 +164,7 @@ async def signal(signaling):
                                 print(data)
                         elif data["type"] == "joystick":
                             webjoystick(data["x"], data["y"])
-                        elif desktop_track and data["type"] in desktop_track.valid_actions:
+                        elif desktop_track and data["type"] in desktop_track.control_interface.valid_actions:
                             desktop_track.handle_action(data["type"], data)
                         else:
                             print("ignored message")