Dup Ver Goto 📝

X11 Key Sender/Receiver in Python v1

PT2/lang/python/qt/pyside6 python qt osc x11 does not exist
To
263 lines, 699 words, 6349 chars Page 'KeySender1' does not exist.

This is a simple pair of scripts to send keys from a Qt window to an OSC server which, in turn, uses either xdotool via subprocess.run, or else python-libxdo.

One use case is remote-controlling a desktop on a remote machine. When the key sender window is focused, key events are translated into xkeysyms and forwarded to a server on the remote machine using OSC, which in turn uses xdotool or libxdo in order to send the keypresses. Things like switching between the video view and the playlist (Control-l) can't be done via the telnet interface to vlc, and the telnet interface has greater latency than doing it via osc and xdo. (I would love for VLC to have OSC support for remote control, and perhaps JSONRPC.)

Overview

The libxdo version uses this line:

import xdo
x11 = xdo.Xdo()
...
      ks = " ".join(args).encode() # note that the string to pass to xdo must be bytes not str
      x11.send_keysequence_window(xdo.CURRENTWINDOW,ks,0)

whereas the subprocess.run(xdotool... version does this:

import subprocess
...
    cmd = ["xdotool","key"]+list(args)
    m = subprocess.run(cmd)

Otherwise the two OSC servers are identical.

Key Sender

Modify to e.g. take the host/port from the command line or environment.

#!/usr/bin/env python

from icecream import ic; ic.configureOutput(includeContext=True)

from PySide6.QtCore import *
from PySide6.QtGui import *
from PySide6.QtWidgets import *
from PySide6.QtNetwork import *

import Xlib.display
from pythonosc import udp_client

app = QApplication()
display = Xlib.display.Display()
lookup = {
    " ": "space"
}

host = HOST
port = PORT
class OscClient:
  def __init__(self):
    self.client = udp_client.SimpleUDPClient(host,port)
  def send(self,addr,*params):
    self.client.send_message(addr,params)

class A(QWidget):
  def __init__(self,client):
    super().__init__()
    self.client = client
  def keyPressEvent(self,e):
    nativeScanCode = e.nativeScanCode()
    nativeVirtualKey = e.nativeVirtualKey()
    ks = display.lookup_string(nativeVirtualKey)
    modifiers = e.modifiers()
    if ks is not None:
      t = str(ks)
      if t in lookup:
        t = lookup[t]
      if modifiers & Qt.ShiftModifier:
        print("Shift is pressed")
      if modifiers & Qt.ControlModifier:
        print("Control is pressed")
        t = "Control+" + t
      if modifiers & Qt.AltModifier:
        print("Alt is pressed")
        t = "Alt+" + t
      if modifiers & Qt.MetaModifier:
        print("Meta (Command on Mac) is pressed")
        t = "Meta+" + t
      ic(t)
      self.client.send("/keys",t)

osc = OscClient()
a = A(osc)
a.show()

app.exec()

Using xdotool

#!/usr/bin/env python

import sys
import os
import subprocess

from pythonosc.dispatcher import Dispatcher
from pythonosc.osc_server import BlockingOSCUDPServer

if not "DISPLAY" in os.environ:
  os.environ["DISPLAY"] = ":0.0"

class XKeys:
  def __init__(self):
    pass
  def __call__(self,address,*args):
    print("osc",address,args)
    method_name = "handle"+address.replace("/","_")
    try:
      method = getattr(self,method_name)
    except AttributeError as e:
      print("No method",method_name,"for",address,args)
      return
    return method(args)
  def handle_keys(self,args):
    cmd = ["xdotool","key"]+list(args)
    m = subprocess.run(cmd)
    print(f"{cmd} returned {m.returncode}")

def default_handler(address, *args):
    print(f"recv {address}: {args}")

dispatcher = Dispatcher()
xkeys = XKeys()
dispatcher.set_default_handler(xkeys)

ip = os.getenv("HOST","0.0.0.0")
port = int(os.getenv("PORT",1069))

def helpexit(rv=1):
  printhelp()
  exit(rv)
def printhelp(rv=1):
  print(f"{sys.argv[0]} [-h host] [-p port] [--help]")

args = sys.argv[1:]
while len(args) > 0:
  a = args[0]
  match a:
    case "--help":
      helpexit(0)
    case "-p":
      try:
        b = args[1]
        port = int(b)
        args = args[2:]
      except ValueError:
        print(f"-p takes an integer")
        helpexit()
      except IndexError:
        print(f"-p missing arg")
        helpexit()
    case "-h":
      try:
        ip = args[1]
        args = args[2:]
      except IndexError:
        print(f"-h missing arg")
        helpexit()

server = BlockingOSCUDPServer((ip, port), dispatcher)
try:
  print(f"Listening on {ip}:{port}")
  server.serve_forever()  # Blocks forever
except KeyboardInterrupt:
  print("Ctrl-C")
  exit()

Using python-libxdo

#!/usr/bin/env python

import sys
import os
import subprocess
from icecream import ic; ic.configureOutput(includeContext=True)



from pythonosc.dispatcher import Dispatcher
from pythonosc.osc_server import BlockingOSCUDPServer

import xdo

if not "DISPLAY" in os.environ:
  os.environ["DISPLAY"] = ":0.0"

x11 = xdo.Xdo()

class XKeys:
  def __init__(self):
    pass
  def __call__(self,address,*args):
    print("osc",address,args)
    method_name = "handle"+address.replace("/","_")
    try:
      method = getattr(self,method_name)
    except AttributeError as e:
      print("No method",method_name,"for",address,args)
      return
    return method(args)
  def handle_keys(self,args):
    ks = " ".join(args).encode()
    try:
      x11.send_keysequence_window(xdo.CURRENTWINDOW,ks,0)
    except Exception as e:
      ic(e)

def default_handler(address, *args):
    print(f"recv {address}: {args}")

dispatcher = Dispatcher()
xkeys = XKeys()
dispatcher.set_default_handler(xkeys)

ip = os.getenv("HOST","0.0.0.0")
port = int(os.getenv("PORT",1068))

def helpexit(rv=1):
  printhelp()
  exit(rv)
def printhelp(rv=1):
  print(f"{sys.argv[0]} [-h host] [-p port] [--help]")

args = sys.argv[1:]
while len(args) > 0:
  a = args[0]
  match a:
    case "--help":
      helpexit(0)
    case "-p":
      try:
        b = args[1]
        port = int(b)
        args = args[2:]
      except ValueError:
        print(f"-p takes an integer")
        helpexit()
      except IndexError:
        print(f"-p missing arg")
        helpexit()
    case "-h":
      try:
        ip = args[1]
        args = args[2:]
      except IndexError:
        print(f"-h missing arg")
        helpexit()

server = BlockingOSCUDPServer((ip, port), dispatcher)
try:
  print(f"Listening on {ip}:{port}")
  server.serve_forever()  # Blocks forever
except KeyboardInterrupt:
  print("Ctrl-C")
  exit()