NeoTheFox/huion-keys
1
0
Fork 0
mirror of https://github.com/erjiang/huion-keys.git synced 2026-03-23 21:54:53 +03:00
Code Issues Projects Releases Wiki Activity

Merge pull request #12 from NeoTheFox/threaded-polling

Browse Source 
Threaded input implementation
This commit is contained in:
erjiang 2021-08-02 10:35:14 -07:00 committed by GitHub
commit b0f4cdc575
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 137 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

225
huion_keys.py
View File

@ -1,8 +1,8 @@
#!/usr/bin/env python3
import os
import time
import signal
import argparse
import threading
import configparser
from _xdo_cffi import ffi, lib
@ -17,10 +17,21 @@ TABLET_MODELS = {
BUTTON_BINDINGS = {}
BUTTON_BINDINGS_HOLD = {}
CYCLE_BUTTON = None
CYCLE_MODE = 1
CYCLE_MODES = 1
DIAL_MODES = {}
BUTTON_BITS = {
0x01: 1,
0x02: 2,
0x04: 3,
0x08: 4,
0x10: 5,
0x20: 6,
0x40: 7,
0x80: 8,
}
def main():
# Commandline arguments processing
parser = argparse.ArgumentParser(
@ -41,8 +52,6 @@ def main():
else:
CONFIG_FILE_PATH = os.path.expanduser(args.config)
global CYCLE_MODES, CYCLE_MODE, CYCLE_BUTTON
xdo = lib.xdo_new(ffi.NULL)
if os.path.isfile(CONFIG_FILE_PATH):
read_config(CONFIG_FILE_PATH)
else:
@ -50,73 +59,156 @@ def main():
create_default_config(CONFIG_FILE_PATH)
print("Created an example config file at " + CONFIG_FILE_PATH)
return 1
signal.signal(signal.SIGUSR1, handle_reload_signal) # Reload the config if recieved SIGUSR1
prev_button = None
hidraw_paths = []
while True:
hidraw_path = None
# search for a known tablet device
# search for a known tablet devices
for device_name, device_id in TABLET_MODELS.items():
hidraw_path = get_tablet_hidraw(device_id)
if hidraw_path is not None:
print("Found %s at %s" % (device_name, hidraw_path))
break
if hidraw_path is None:
print("Could not find tablet hidraw device")
time.sleep(2)
hidraw_paths = hidraw_paths + hidraw_path
if not hidraw_paths:
print("Could not find any tablet hidraw devices")
time.sleep(3)
continue
elif hidraw_paths:
threads = []
for hidraw_path in hidraw_paths:
thread = PollThread(hidraw_path)
# Do not let the threads to continue if main script is terminated
thread.daemon = True
threads.append(thread)
thread.start()
# TODO: Maybe should be reworked for the edge case of having more than one tablet connected at the same time.
for thread in threads:
thread.join()
hidraw_paths.clear()
continue
class PollThread(threading.Thread):
cycle_mode = None
scroll_state = None
hidraw_path = None
xdo = None
def __init__(self, hidraw_path):
super(PollThread, self).__init__()
self.xdo = lib.xdo_new(ffi.NULL)
self.hidraw_path = hidraw_path
self.cycle_mode = 1
def run(self):
global BUTTON_BINDINGS, BUTTON_BINDINGS_HOLD, CYCLE_MODES, CYCLE_BUTTON, DIAL_MODES
while True:
try:
hidraw = open(hidraw_path, 'rb')
hidraw = open(self.hidraw_path, 'rb')
break
except PermissionError as e:
print(e)
print("Trying again in 5 seconds...")
time.sleep(5)
continue
while True:
try:
btn = get_button_press(hidraw)
btn = self.get_button_press(hidraw)
except OSError as e:
print("Lost connection with the tablet - searching for tablet...")
time.sleep(3)
print("%s lost connection with the tablet..." % (self.name,))
break
print("Got button %s" % (btn,))
if btn == CYCLE_BUTTON and CYCLE_BUTTON is not None:
CYCLE_MODE = CYCLE_MODE + 1
if CYCLE_MODE > CYCLE_MODES:
CYCLE_MODE = 1
print("Cycling to mode %s" % (CYCLE_MODE,))
elif CYCLE_MODE in DIAL_MODES and btn in DIAL_MODES[CYCLE_MODE]:
print("Sending %s from Mode %d" % (DIAL_MODES[CYCLE_MODE][btn], CYCLE_MODE),)
self.cycle_mode = self.cycle_mode + 1
if self.cycle_mode > CYCLE_MODES:
self.cycle_mode = 1
print("Cycling to mode %s" % (self.cycle_mode,))
elif self.cycle_mode in DIAL_MODES and btn in DIAL_MODES[self.cycle_mode]:
print("Sending %s from Mode %d" % (DIAL_MODES[self.cycle_mode][btn], self.cycle_mode),)
lib.xdo_send_keysequence_window(
xdo, lib.CURRENTWINDOW, DIAL_MODES[CYCLE_MODE][btn], 1000)
self.xdo, lib.CURRENTWINDOW, DIAL_MODES[self.cycle_mode][btn], 1000)
elif btn in BUTTON_BINDINGS_HOLD:
print("Pressing %s" % (BUTTON_BINDINGS_HOLD[btn],))
lib.xdo_send_keysequence_window_down(xdo, lib.CURRENTWINDOW, BUTTON_BINDINGS_HOLD[btn], 12000)
get_button_release(hidraw)
lib.xdo_send_keysequence_window_down(self.xdo, lib.CURRENTWINDOW, BUTTON_BINDINGS_HOLD[btn], 12000)
self.get_button_release(hidraw)
print("Releasing %s" % (BUTTON_BINDINGS_HOLD[btn],))
lib.xdo_send_keysequence_window_up(xdo, lib.CURRENTWINDOW, BUTTON_BINDINGS_HOLD[btn], 12000)
lib.xdo_send_keysequence_window_up(self.xdo, lib.CURRENTWINDOW, BUTTON_BINDINGS_HOLD[btn], 12000)
elif btn in BUTTON_BINDINGS:
print("Sending %s" % (BUTTON_BINDINGS[btn],))
lib.xdo_send_keysequence_window(
xdo, lib.CURRENTWINDOW, BUTTON_BINDINGS[btn], 1000)
self.xdo, lib.CURRENTWINDOW, BUTTON_BINDINGS[btn], 1000)
def get_button_press(self, hidraw):
while True:
sequence = hidraw.read(12)
# 0xf7 is what my Kamvas Pro 22 reads
# another model seems to send 0x08
# Q620M reads as 0xf9
if sequence[0] != 0xf7 and sequence[0] != 0x08 and sequence[0] != 0xf9:
pass
if sequence[1] == 0xe0: # buttons
# doesn't seem like the tablet will let you push two buttons at once
if sequence[4] > 0:
return BUTTON_BITS[sequence[4]]
elif sequence[5] > 0:
# right-side buttons are 8-15, so add 8
return BUTTON_BITS[sequence[5]] + 8
else:
# must be button release (all zeros)
continue
elif sequence[1] == 0xf0: # scroll strip
scroll_pos = sequence[5]
if scroll_pos == 0:
# reset scroll state after lifting finger off scroll strip
self.scroll_state = None
elif self.scroll_state is not None:
# scroll strip is numbered from top to bottom so a greater new
# value means they scrolled down
if scroll_pos > self.scroll_state:
self.scroll_state = scroll_pos
return 'scroll_down'
elif scroll_pos < self.scroll_state:
self.scroll_state = scroll_pos
return 'scroll_up'
else:
self.scroll_state = scroll_pos
continue
elif sequence[1] == 0xf1: # dial on Q620M, practically 2 buttons
if sequence[5] == 0x1:
return 'dial_cw'
elif sequence[5] == 0xff:
return 'dial_ccw'
else:
continue
def get_button_release(self, hidraw):
while True:
sequence = hidraw.read(12)
if sequence[1] == 0xe0 and sequence[4] == 0 and sequence[5] == 0:
return True
def get_tablet_hidraw(device_id):
"""Finds the /dev/hidrawX file that belongs to the given device ID (in xxxx:xxxx format)."""
"""Finds the /dev/hidrawX file or files that belong to the given device ID (in xxxx:xxxx format)."""
# TODO: is this too fragile?
hidraws = os.listdir('/sys/class/hidraw')
inputs = []
for h in hidraws:
device_path = os.readlink(os.path.join('/sys/class/hidraw', h, 'device'))
if device_id.upper() in device_path:
# need to confirm that there's "input" because there are two hidraw
# files listed for the tablet, but only one of them carries the
# need to confirm that there's "input" because there are two or more hidraw
# files listed for the tablet, but only few of them carry the
# mouse/keyboard input
if os.path.exists(os.path.join('/sys/class/hidraw', h, 'device/input')):
return os.path.join('/dev', os.path.basename(h))
inputs.append(os.path.join('/dev', os.path.basename(h)))
if inputs:
return inputs
return None
def read_config(config_file):
global CYCLE_MODES, CYCLE_MODE, CYCLE_BUTTON
global CYCLE_MODES, CYCLE_BUTTON, BUTTON_BINDINGS, BUTTON_BINDINGS_HOLD
CONFIG = configparser.ConfigParser()
CONFIG.read(config_file)
# It is still better for performance to pre-encode these values
@ -159,16 +251,13 @@ def read_config(config_file):
DIAL_MODES[mode][binding] = CONFIG[key][binding].encode('utf-8')
def handle_reload_signal(signum, frame):
print("SIGUSR1 recieved - reloading config..")
read_config(CONFIG_FILE_PATH)
def make_rules():
for device_name, device_id in TABLET_MODELS.items():
print("# %s" % (device_name, ))
VID, PID = device_id.split(':')
print('KERNEL=="hidraw*", ATTRS{idVendor}=="%s", ATTRS{idProduct}=="%s", MODE="0660", TAG+="uaccess"' % (VID, PID, ))
def create_default_config(config_file):
with open(config_file, 'w') as config:
config.write("""
@ -199,69 +288,5 @@ dial_ccw=equal
""")
BUTTON_BITS = {
0x01: 1,
0x02: 2,
0x04: 3,
0x08: 4,
0x10: 5,
0x20: 6,
0x40: 7,
0x80: 8,
}
SCROLL_STATE=None
def get_button_press(hidraw):
global SCROLL_STATE
while True:
sequence = hidraw.read(12)
# 0xf7 is what my Kamvas Pro 22 reads
# another model seems to send 0x08
# Q620M reads as 0xf9
if sequence[0] != 0xf7 and sequence[0] != 0x08 and sequence[0] != 0xf9:
pass
if sequence[1] == 0xe0: # buttons
# doesn't seem like the tablet will let you push two buttons at once
if sequence[4] > 0:
return BUTTON_BITS[sequence[4]]
elif sequence[5] > 0:
# right-side buttons are 8-15, so add 8
return BUTTON_BITS[sequence[5]] + 8
else:
# must be button release (all zeros)
continue
elif sequence[1] == 0xf0: # scroll strip
scroll_pos = sequence[5]
if scroll_pos == 0:
# reset scroll state after lifting finger off scroll strip
SCROLL_STATE = None
elif SCROLL_STATE is not None:
# scroll strip is numbered from top to bottom so a greater new
# value means they scrolled down
if scroll_pos > SCROLL_STATE:
SCROLL_STATE = scroll_pos
return 'scroll_down'
elif scroll_pos < SCROLL_STATE:
SCROLL_STATE = scroll_pos
return 'scroll_up'
else:
SCROLL_STATE = scroll_pos
continue
elif sequence[1] == 0xf1: # dial on Q620M, practically 2 buttons
if sequence[5] == 0x1:
return 'dial_cw'
elif sequence[5] == 0xff:
return 'dial_ccw'
else:
continue
def get_button_release(hidraw):
while True:
sequence = hidraw.read(12)
if sequence[1] == 0xe0 and sequence[4] == 0 and sequence[5] == 0:
return True
if __name__ == "__main__":
main()