import copy import select import time from tooncher.actions import * try: import psutil except ImportError: psutil = False try: import Xlib.display from Xlib import X, XK except ImportError: Xlib = False EXTENDED_CONTROLS_DEFAULT_TOGGLE_KEYSYM_NAME = 'grave' TOONTOWN_WINDOW_NAME = 'Toontown Rewritten' if Xlib: EXTENDED_CONTROLS_DEFAULT_KEYSYM_MAPPINGS = { XK.XK_w: RewriteKeyEventAction(keysym=XK.XK_Up), XK.XK_a: RewriteKeyEventAction(keysym=XK.XK_Left), XK.XK_s: RewriteKeyEventAction(keysym=XK.XK_Down), XK.XK_d: RewriteKeyEventAction(keysym=XK.XK_Right), XK.XK_i: RewriteKeyEventAction( keysym=XK.XK_Up, target_engine=TargetEngine.NonPrimary, ), XK.XK_j: RewriteKeyEventAction( keysym=XK.XK_Left, target_engine=TargetEngine.NonPrimary, ), XK.XK_k: RewriteKeyEventAction( keysym=XK.XK_Down, target_engine=TargetEngine.NonPrimary, ), XK.XK_l: RewriteKeyEventAction( keysym=XK.XK_Right, target_engine=TargetEngine.NonPrimary, ), XK.XK_space: RewriteKeyEventAction( keysym=XK.XK_Control_L, target_engine=TargetEngine.All, ), XK.XK_e: SelectGagAction(factor_x=0.038, factor_y=-0.047), # elephant trunk XK.XK_f: SelectGagAction(factor_x=0.119, factor_y=-0.047), # foghorn } def x_find_window(parent_window, filter_callback): matching = [] for child_window in parent_window.query_tree().children: if filter_callback(child_window): matching.append(child_window) matching += x_find_window(child_window, filter_callback) return matching def x_find_window_by_pid(display, pid): pid_prop = display.intern_atom('_NET_WM_PID') def filter_callback(window): prop = window.get_full_property(pid_prop, X.AnyPropertyType) return prop and prop.value.tolist() == [pid] return x_find_window(display.screen().root, filter_callback) def x_wait_for_event(xdisplay, timeout_seconds): """ Wait up to `timeout_seconds` seconds for a xevent. Return True, if a xevent is available. Return False, if the timeout was reached. """ rlist = select.select( [xdisplay.display.socket], # rlist [], # wlist [], # xlist timeout_seconds, # timeout [seconds] )[0] return len(rlist) > 0 class ExtendedControls: def __init__(self, engine_pid, toggle_keysym_name=EXTENDED_CONTROLS_DEFAULT_TOGGLE_KEYSYM_NAME): if not psutil: raise Exception('\n'.join([ 'Extended keyboard controls require the python lib psutil to be installed.', 'Depending on your system run', '\t$ sudo apt-get install python3-psutil', 'or', '\t$ pip3 install --user psutil', ])) if not Xlib: raise Exception('\n'.join([ 'Extended keyboard controls require xlib for python to be installed.', 'Depending on your system run', '\t$ sudo apt-get install python3-xlib', 'or', '\t$ pip3 install --user xlib', ])) self._engine_pid = engine_pid self._xdisplay = Xlib.display.Display() self._toggle_keysym = XK.string_to_keysym(toggle_keysym_name) if self._toggle_keysym == X.NoSymbol: raise Exception("Extended keyboard controls toggle:" + " Unknown keysym name '{}'".format(toggle_keysym_name)) self._keysym_mappings = copy.deepcopy( EXTENDED_CONTROLS_DEFAULT_KEYSYM_MAPPINGS, ) if self._toggle_keysym in self._keysym_mappings: print("INFO Extended Controls:" + " Ignoring mapping for toggle key '{}'".format(toggle_keysym_name)) self._keysym_mappings[self._toggle_keysym] \ = ToggleExtendedControlsAction() self._default_action = ForwardKeyEventAction() self._engine_window = None self._other_engine_windows = None self._active_key_registry = {} self._enabled = False @property def engine_running(self): return psutil.pid_exists(self._engine_pid) @property def xdisplay(self): return self._xdisplay @property def engine_window(self): return self._engine_window def _wait_for_engine_window(self, timeout_seconds=20, search_interval_seconds=2): start_epoch = time.time() while self.engine_running and (time.time() - start_epoch) <= timeout_seconds: windows = x_find_window_by_pid( self._xdisplay, self._engine_pid, ) assert len(windows) <= 1 if len(windows) == 1: return windows[0] time.sleep(search_interval_seconds) return None def run(self): self._engine_window = self._wait_for_engine_window() if not self._engine_window: raise Exception('Could not find the game\'s window.') self._grab_key( self._xdisplay.keysym_to_keycode(self._toggle_keysym), ) if not self.enabled: keysym_name = XK.keysym_to_string(self._toggle_keysym) print("INFO Extended Controls are currently disabled." + " Press key '{}' to enable.".format(keysym_name)) while self.engine_running: while self.xdisplay.pending_events(): self._handle_xevent(self._xdisplay.next_event()) self._check_active_key_registry() # keep timeout low for _check_active_key_registry() # to be called frequently x_wait_for_event(self.xdisplay, timeout_seconds=0.05) def _handle_xevent(self, xevent): if isinstance(xevent, Xlib.protocol.event.KeyPress) \ or isinstance(xevent, Xlib.protocol.event.KeyRelease): self._handle_xkeyevent(xevent) def _handle_xkeyevent(self, xkeyevent): self._update_active_key_registry(xkeyevent) keysym_in = self._xdisplay.keycode_to_keysym( xkeyevent.detail, index=0, ) if keysym_in in self._keysym_mappings: action = self._keysym_mappings[keysym_in] else: action = self._default_action action.execute(self, xkeyevent) def enable(self): for keysym in self._keysym_mappings.keys(): if keysym != self._toggle_keysym: self._grab_key( self._xdisplay.keysym_to_keycode(keysym), ) self._enabled = True # reset cache self._other_engine_windows = None print("INFO Enabled Extended Controls") def disable(self): for keysym in self._keysym_mappings.keys(): if keysym != self._toggle_keysym: self._ungrab_key( self._xdisplay.keysym_to_keycode(keysym), ) self._enabled = False print("INFO Disabled Extended Controls") @property def enabled(self): return self._enabled def toggle(self): if self.enabled: self.disable() else: self.enable() def _grab_key(self, keycode): self._engine_window.grab_key( keycode, X.AnyModifier, # owner_events # https://stackoverflow.com/questions/32122360/x11-will-xgrabpointer-prevent-other-apps-from-any-mouse-event # False, True, X.GrabModeAsync, X.GrabModeAsync, ) def _ungrab_key(self, keycode): self._engine_window.ungrab_key(keycode, X.AnyModifier) def find_other_engine_windows(self): return x_find_window( self.xdisplay.screen().root, lambda w: w.get_wm_name() == TOONTOWN_WINDOW_NAME and w.id != self.engine_window.id, ) @property def other_engine_windows(self): if not self._other_engine_windows: self._other_engine_windows = self.find_other_engine_windows() return self._other_engine_windows def _update_active_key_registry(self, xkeyevent): # see self._check_active_key_registry keycode = xkeyevent.detail if isinstance(xkeyevent, Xlib.protocol.event.KeyPress): self._active_key_registry[keycode] = xkeyevent elif keycode in self._active_key_registry: del self._active_key_registry[keycode] def _check_active_key_registry(self): """ WORKAROUND For an unknown reason some key release events don't get queued when multiple keys are being released simultaneously. So we keep a hashmap of supposedly currently pressed keys and periodically compare it with xdispaly.query_keymap(). ref: https://stackoverflow.com/q/18160792/5894777 """ # https://tronche.com/gui/x/xlib/input/XQueryKeymap.html keymap = self.xdisplay.query_keymap() missed_releases = [] for keycode, press_event in self._active_key_registry.items(): if not (1 << (keycode & ((1 << 3) - 1))) & keymap[keycode >> 3]: print("WARN missed release event of key {}".format(keycode)) missed_releases.append(Xlib.protocol.event.KeyRelease( window=press_event.window, detail=press_event.detail, state=press_event.state, root_x=press_event.root_x, root_y=press_event.root_y, event_x=press_event.event_x, event_y=press_event.event_y, child=press_event.child, root=press_event.root, time=X.CurrentTime, same_screen=press_event.same_screen, )) for release_event in missed_releases: self._handle_xkeyevent(release_event)