Initial commit
This commit is contained in:
bepis
302
main.py
Normal file
302
main.py
Normal file
@@ -0,0 +1,302 @@
|
||||
#!/usr/bin/env python3
|
||||
import evdev
|
||||
from evdev import InputDevice, ecodes
|
||||
import os
|
||||
import subprocess
|
||||
import signal
|
||||
import threading
|
||||
import queue
|
||||
import time
|
||||
import shutil
|
||||
import argparse
|
||||
import glob
|
||||
|
||||
__version__ = "0.1.0"
|
||||
|
||||
BongoWindow = None
|
||||
stop_event = threading.Event()
|
||||
keyboard_stop_event = threading.Event()
|
||||
key_queue = queue.Queue()
|
||||
keyboard_threads = []
|
||||
keyboard_devices = []
|
||||
|
||||
TARGET_WINDOW_NAME = 'TapTapLoot'
|
||||
TARGET_DISPLAY = ':10'
|
||||
|
||||
def get_xauth_path():
|
||||
"""Dynamically find the current taptaploot xauth file"""
|
||||
matches = glob.glob('/tmp/taptaploot-xauth.*')
|
||||
if matches:
|
||||
return matches[0]
|
||||
return None
|
||||
|
||||
def get_xenv():
|
||||
"""Get environment dict for xdotool calls targeting the game display"""
|
||||
xauth = get_xauth_path()
|
||||
env = {**os.environ, 'DISPLAY': TARGET_DISPLAY}
|
||||
if xauth:
|
||||
env['XAUTHORITY'] = xauth
|
||||
return env
|
||||
|
||||
def get_xdotool_keyname(keycode):
|
||||
key_name = ecodes.KEY[keycode]
|
||||
if key_name.startswith('KEY_'):
|
||||
key_name = key_name[4:]
|
||||
elif key_name.startswith('BTN_'):
|
||||
key_name = key_name[4:]
|
||||
key_name = key_name.lower()
|
||||
special_keys = {
|
||||
'leftctrl': 'ctrl',
|
||||
'rightctrl': 'ctrl',
|
||||
'leftshift': 'shift',
|
||||
'rightshift': 'shift',
|
||||
'leftalt': 'alt',
|
||||
'rightalt': 'alt',
|
||||
'leftmeta': 'super',
|
||||
'rightmeta': 'super',
|
||||
'enter': 'Return',
|
||||
'esc': 'Escape',
|
||||
'backspace': 'BackSpace',
|
||||
'tab': 'Tab',
|
||||
'capslock': 'Caps_Lock',
|
||||
}
|
||||
return special_keys.get(key_name, key_name)
|
||||
|
||||
def find_game_window():
|
||||
"""Find the TapTapLoot window on the game display"""
|
||||
try:
|
||||
env = get_xenv()
|
||||
result = subprocess.getoutput(
|
||||
f"DISPLAY={TARGET_DISPLAY} XAUTHORITY={env.get('XAUTHORITY', '')} "
|
||||
f"xdotool search --name '{TARGET_WINDOW_NAME}'"
|
||||
)
|
||||
if not result.strip():
|
||||
return None
|
||||
|
||||
for window_id in result.strip().split('\n'):
|
||||
window_name = subprocess.getoutput(
|
||||
f"DISPLAY={TARGET_DISPLAY} XAUTHORITY={env.get('XAUTHORITY', '')} "
|
||||
f"xdotool getwindowname {window_id}"
|
||||
).strip()
|
||||
if window_name == TARGET_WINDOW_NAME:
|
||||
return window_id
|
||||
|
||||
# No exact match, return first result
|
||||
return result.strip().split('\n')[0]
|
||||
except:
|
||||
pass
|
||||
return None
|
||||
|
||||
def send_keys_to_game(keys):
|
||||
"""Send batched keypresses to the game window"""
|
||||
if not keys or not BongoWindow:
|
||||
return
|
||||
|
||||
env = get_xenv()
|
||||
batch_size = 4
|
||||
for i in range(0, len(keys), batch_size):
|
||||
batch = keys[i:i+batch_size]
|
||||
|
||||
cmd_keydown = ['xdotool', 'keydown', '--window', BongoWindow] + batch
|
||||
cmd_keyup = ['xdotool', 'keyup', '--window', BongoWindow] + batch
|
||||
|
||||
try:
|
||||
result = subprocess.run(cmd_keydown, capture_output=True, text=True, env=env)
|
||||
if result.returncode != 0:
|
||||
continue
|
||||
time.sleep(0.01)
|
||||
subprocess.run(cmd_keyup, capture_output=True, text=True, env=env)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def replace_duplicate_keys(keys):
|
||||
if not keys:
|
||||
return []
|
||||
replacement_pool = list('abcdefghijklmnopqrstuvwxyz')
|
||||
seen = set()
|
||||
result = []
|
||||
for key in keys:
|
||||
if key in seen:
|
||||
for replacement in replacement_pool:
|
||||
if replacement not in seen:
|
||||
result.append(replacement)
|
||||
seen.add(replacement)
|
||||
break
|
||||
else:
|
||||
result.append(key)
|
||||
else:
|
||||
result.append(key)
|
||||
seen.add(key)
|
||||
return result
|
||||
|
||||
def batch_sender():
|
||||
while not stop_event.is_set():
|
||||
time.sleep(0.2)
|
||||
keys = []
|
||||
while True:
|
||||
try:
|
||||
key = key_queue.get_nowait()
|
||||
keys.append(key)
|
||||
except queue.Empty:
|
||||
break
|
||||
if keys:
|
||||
replaced_keys = replace_duplicate_keys(keys)
|
||||
send_keys_to_game(replaced_keys)
|
||||
|
||||
def is_keyboard(device):
|
||||
capabilities = device.capabilities()
|
||||
if ecodes.EV_KEY not in capabilities:
|
||||
return False
|
||||
keys = capabilities[ecodes.EV_KEY]
|
||||
mouse_buttons = [
|
||||
ecodes.BTN_LEFT, ecodes.BTN_RIGHT, ecodes.BTN_MIDDLE,
|
||||
ecodes.BTN_SIDE, ecodes.BTN_EXTRA, ecodes.BTN_FORWARD,
|
||||
ecodes.BTN_BACK, ecodes.BTN_MOUSE,
|
||||
]
|
||||
for btn in mouse_buttons:
|
||||
if btn in keys:
|
||||
return False
|
||||
if ecodes.BTN_TOUCH in keys or ecodes.BTN_TOOL_FINGER in keys:
|
||||
return False
|
||||
keyboard_keys = [
|
||||
ecodes.KEY_A, ecodes.KEY_B, ecodes.KEY_C,
|
||||
ecodes.KEY_SPACE, ecodes.KEY_ENTER, ecodes.KEY_ESC,
|
||||
]
|
||||
for key in keyboard_keys:
|
||||
if key in keys:
|
||||
return True
|
||||
return False
|
||||
|
||||
def monitor_keyboard(device):
|
||||
try:
|
||||
for event in device.read_loop():
|
||||
if keyboard_stop_event.is_set() or stop_event.is_set():
|
||||
break
|
||||
if event.type == ecodes.EV_KEY and event.value == 1:
|
||||
try:
|
||||
key_name = get_xdotool_keyname(event.code)
|
||||
key_queue.put(key_name)
|
||||
except (KeyError, ValueError):
|
||||
pass
|
||||
except (OSError, IOError):
|
||||
pass
|
||||
|
||||
def start_keyboard_monitoring():
|
||||
global keyboard_threads, keyboard_devices
|
||||
keyboard_stop_event.clear()
|
||||
all_devices = [InputDevice(path) for path in evdev.list_devices()]
|
||||
keyboards = [device for device in all_devices if is_keyboard(device)]
|
||||
if not keyboards:
|
||||
print(" [Warning] No keyboard devices found")
|
||||
return False
|
||||
print(f" Found {len(keyboards)} keyboard device(s)")
|
||||
keyboard_devices = keyboards
|
||||
keyboard_threads = []
|
||||
for kb in keyboards:
|
||||
thread = threading.Thread(target=monitor_keyboard, args=(kb,), daemon=True)
|
||||
thread.start()
|
||||
keyboard_threads.append(thread)
|
||||
return True
|
||||
|
||||
def stop_keyboard_monitoring():
|
||||
global keyboard_threads, keyboard_devices
|
||||
keyboard_stop_event.set()
|
||||
for kb in keyboard_devices:
|
||||
try:
|
||||
kb.close()
|
||||
except Exception:
|
||||
pass
|
||||
for thread in keyboard_threads:
|
||||
thread.join(timeout=1.0)
|
||||
keyboard_threads = []
|
||||
keyboard_devices = []
|
||||
|
||||
def window_monitor():
|
||||
global BongoWindow
|
||||
last_window_state = None
|
||||
|
||||
while not stop_event.is_set():
|
||||
new_window = find_game_window()
|
||||
|
||||
if new_window and not last_window_state:
|
||||
BongoWindow = new_window
|
||||
print(f"\n✓ Connected to {TARGET_WINDOW_NAME} window (ID: {BongoWindow})")
|
||||
print(" Starting keyboard monitoring...")
|
||||
if start_keyboard_monitoring():
|
||||
print(" ✓ Keyboard monitoring started\n")
|
||||
last_window_state = True
|
||||
|
||||
elif not new_window and last_window_state:
|
||||
print(f"\n⚠ {TARGET_WINDOW_NAME} window closed")
|
||||
print(" Stopping keyboard monitoring...")
|
||||
stop_keyboard_monitoring()
|
||||
print(" ✓ Waiting for window to reappear...\n")
|
||||
BongoWindow = None
|
||||
last_window_state = False
|
||||
|
||||
elif not new_window and last_window_state is None:
|
||||
print(f"⚠ {TARGET_WINDOW_NAME} window not found")
|
||||
print(" Waiting... (checks every 10s)\n")
|
||||
BongoWindow = None
|
||||
last_window_state = False
|
||||
|
||||
elif new_window and last_window_state:
|
||||
if BongoWindow != new_window:
|
||||
BongoWindow = new_window
|
||||
print(f"\n✓ Window ID updated (ID: {BongoWindow})\n")
|
||||
|
||||
stop_event.wait(10)
|
||||
|
||||
def main():
|
||||
print(f"BongoCatXP - TapTapLoot edition\n")
|
||||
|
||||
def signal_handler(signum, frame):
|
||||
print("\nStopping...")
|
||||
stop_event.set()
|
||||
|
||||
signal.signal(signal.SIGINT, signal_handler)
|
||||
signal.signal(signal.SIGTERM, signal_handler)
|
||||
|
||||
xauth = get_xauth_path()
|
||||
if xauth:
|
||||
print(f"Using xauth: {xauth}")
|
||||
else:
|
||||
print("⚠ Warning: no taptaploot xauth file found in /tmp — is the game running?")
|
||||
|
||||
print("Starting window monitor (Ctrl+C to stop)\n")
|
||||
|
||||
sender_thread = threading.Thread(target=batch_sender, daemon=True)
|
||||
sender_thread.start()
|
||||
|
||||
monitor_thread = threading.Thread(target=window_monitor, daemon=True)
|
||||
monitor_thread.start()
|
||||
|
||||
try:
|
||||
stop_event.wait()
|
||||
finally:
|
||||
print("Stopping keyboard monitoring...")
|
||||
stop_keyboard_monitoring()
|
||||
print("Done")
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(
|
||||
prog='bongocatxp',
|
||||
description='BongoCat X11 Proxy - TapTapLoot edition'
|
||||
)
|
||||
parser.add_argument('--version', action='version', version=f'BongoCatXP {__version__}')
|
||||
args = parser.parse_args()
|
||||
|
||||
if not shutil.which('xdotool'):
|
||||
print("Error: xdotool not found — install with: sudo pacman -S xdotool")
|
||||
exit(1)
|
||||
|
||||
if not os.access('/dev/input/event0', os.R_OK):
|
||||
print("No permission to read input devices.")
|
||||
print("Run: sudo usermod -aG input $USER (then log out and back in)")
|
||||
print(f"Or temporarily: sudo python3 {__file__}")
|
||||
exit(1)
|
||||
|
||||
try:
|
||||
main()
|
||||
except Exception as e:
|
||||
print(f"\nError: {e}")
|
||||
Reference in New Issue
Block a user