jyapayne/nxbt
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Added direct input functionality

Browse source
This commit is contained in:
Brikwerk 2020-08-30 20:22:43 -07:00
commit 2eaacadbe4
6 changed files with 590 additions and 153 deletions
Download patch file Download diff file Expand all files Collapse all files

24
nxbt/cli.py
View file

@ -112,6 +112,19 @@ def check_bluetooth_address(address):
raise ValueError("Invalid Bluetooth address")
def get_reconnect_target():
if args.reconnect:
reconnect_target = find_devices_by_alias("Nintendo Switch")
elif args.address:
check_bluetooth_address(args.address)
reconnect_target = args.address
else:
reconnect_target = None
return reconnect_target
def demo():
"""Loops over all available Bluetooth adapters
and creates controllers on each. The last available adapter
@ -153,13 +166,7 @@ def macro():
print("to load a macro string from.")
return
if args.reconnect:
reconnect_target = find_devices_by_alias("Nintendo Switch")
elif args.address:
check_bluetooth_address(args.address)
reconnect_target = args.address
else:
reconnect_target = None
reconnect_target = get_reconnect_target()
nx = Nxbt(debug=args.debug, log_to_file=args.logfile)
print("Creating controller...")
@ -204,7 +211,8 @@ def main():
elif args.command == 'macro':
macro()
elif args.command == 'tui':
tui = InputTUI()
reconnect_target = get_reconnect_target()
tui = InputTUI(reconnect_target=reconnect_target)
tui.start()
elif args.command == 'addresses':
list_switch_addresses()

145
nxbt/controller/input.py
View file

@ -1,4 +1,55 @@
from time import perf_counter
from json import dumps
DIRECT_INPUT_IDLE_PACKET = {
# Sticks
"L_STICK": {
"PRESSED": False,
"X_VALUE": 0,
"Y_VALUE": 0,
# Keyboard position calculation values
"LS_UP": False,
"LS_LEFT": False,
"LS_RIGHT": False,
"LS_DOWN": False
},
"R_STICK": {
"PRESSED": False,
"X_VALUE": 0,
"Y_VALUE": 0,
# Keyboard position calculation values
"RS_UP": False,
"RS_LEFT": False,
"RS_RIGHT": False,
"RS_DOWN": False
},
# Dpad
"DPAD_UP": False,
"DPAD_LEFT": False,
"DPAD_RIGHT": False,
"DPAD_DOWN": False,
# Triggers
"L": False,
"ZL": False,
"R": False,
"ZR": False,
# Joy-Con Specific Buttons
"JCL_SR": False,
"JCL_SL": False,
"JCR_SR": False,
"JCR_SL": False,
# Meta buttons
"PLUS": False,
"MINUS": False,
"HOME": False,
"CAPTURE": False,
# Buttons
"Y": False,
"X": False,
"B": False,
"A": False
}
class InputParser():
@ -85,7 +136,7 @@ class InputParser():
return
def clear_macros(self, state=None):
def clear_macros(self):
self.current_macro = None
self.current_macro_id = None
@ -102,7 +153,8 @@ class InputParser():
def set_protocol_input(self, state=None):
if self.controller_input:
# Act on direct input if we're not getting idle packets
if dumps(self.controller_input) != dumps(DIRECT_INPUT_IDLE_PACKET):
self.parse_controller_input(self.controller_input)
self.controller_input = None
@ -140,6 +192,91 @@ class InputParser():
def parse_controller_input(self, controller_input):
# Check for input validity
if type(controller_input) != dict:
return
# Check if the Grip/Order menu would be closed
if not self.exited_grip_order_menu and (
controller_input["A"] or controller_input["B"] or controller_input["HOME"]):
self.exited_grip_order_menu = True
# Arrays representing the 3 button bytes in the
# standard input report as binary.
upper = ['0'] * 8
shared = ['0'] * 8
lower = ['0'] * 8
# Upper Byte
if controller_input["Y"]:
upper[7] = '1'
if controller_input["X"]:
upper[6] = '1'
if controller_input["B"]:
upper[5] = '1'
if controller_input["A"]:
upper[4] = '1'
if controller_input["JCL_SR"]:
upper[3] = '1'
if controller_input["JCL_SL"]:
upper[2] = '1'
if controller_input["R"]:
upper[1] = '1'
if controller_input["ZR"]:
upper[0] = '1'
# Shared byte
if controller_input["MINUS"]:
shared[7] = '1'
if controller_input["PLUS"]:
shared[6] = '1'
if controller_input["R_STICK"]["PRESSED"]:
shared[5] = '1'
if controller_input["L_STICK"]["PRESSED"]:
shared[4] = '1'
if controller_input["HOME"]:
shared[3] = '1'
if controller_input["CAPTURE"]:
shared[2] = '1'
# Lower byte
if controller_input["DPAD_DOWN"]:
lower[7] = '1'
if controller_input["DPAD_UP"]:
lower[6] = '1'
if controller_input["DPAD_RIGHT"]:
lower[5] = '1'
if controller_input["DPAD_LEFT"]:
lower[4] = '1'
if controller_input["JCR_SR"]:
lower[3] = '1'
if controller_input["JCR_SL"]:
lower[2] = '1'
if controller_input["L"]:
lower[1] = '1'
if controller_input["ZL"]:
lower[0] = '1'
# Analog Stick Positions
stick_left = self.stick_ratio_to_calibrated_position(
controller_input["L_STICK"]["X_VALUE"] / 100,
controller_input["L_STICK"]["Y_VALUE"] / 100,
"L_STICK"
)
stick_right = self.stick_ratio_to_calibrated_position(
controller_input["R_STICK"]["X_VALUE"] / 100,
controller_input["R_STICK"]["Y_VALUE"] / 100,
"R_STICK"
)
# Converting binary strings to ints
upper_byte = int("".join(upper), 2)
shared_byte = int("".join(shared), 2)
lower_byte = int("".join(lower), 2)
self.protocol.set_button_inputs(upper_byte, shared_byte, lower_byte)
self.protocol.set_left_stick_inputs(stick_left)
self.protocol.set_right_stick_inputs(stick_right)
return controller_input
def parse_macro(self, macro):
@ -233,9 +370,9 @@ class InputParser():
upper[0] = '1'
# Shared byte
elif button == "-":
elif button == "MINUS":
shared[7] = '1'
elif button == "+":
elif button == "PLUS":
shared[6] = '1'
elif button == "R_STICK_PRESS":
shared[5] = '1'

2
nxbt/controller/protocol.py
View file

@ -3,7 +3,7 @@ import random
from time import perf_counter
from .controller import ControllerTypes
from .utils import replace_subarray, format_msg_controller
from .utils import replace_subarray
class SwitchResponses(Enum):

37
nxbt/controller/server.py
View file

@ -28,7 +28,8 @@ class ControllerServer():
self.state = {
"state": "",
"finished_macros": [],
"errors": None
"errors": None,
"direct_input": None
}
self.task_queue = task_queue
@ -114,21 +115,26 @@ class ControllerServer():
# Getting any inputs from the task queue
if self.task_queue:
try:
msg = self.task_queue.get_nowait()
if msg and msg["type"] == "macro":
self.input.buffer_macro(
msg["macro"], msg["macro_id"])
elif msg and msg["type"] == "stop":
self.input.stop_macro(
msg["macro_id"], state=self.state)
elif msg and msg["type"] == "clear":
self.input.clear_macros(
state=self.state)
while True:
msg = self.task_queue.get_nowait()
if msg and msg["type"] == "macro":
self.input.buffer_macro(
msg["macro"], msg["macro_id"])
elif msg and msg["type"] == "stop":
self.input.stop_macro(
msg["macro_id"], state=self.state)
elif msg and msg["type"] == "clear":
self.input.clear_macros()
except queue.Empty:
pass
# Set Direct Input
if self.state["direct_input"]:
self.input.set_controller_input(self.state["direct_input"])
self.protocol.process_commands(reply)
self.input.set_protocol_input(state=self.state)
msg = self.protocol.get_report()
if self.logger_level <= logging.DEBUG and reply and len(reply) > 45:
@ -331,16 +337,17 @@ class ControllerServer():
ctrl = None
if type(reconnect_address) == list:
for address in reconnect_address:
test_itr, test_ctrl = recreate_sockets()
try:
test_itr, test_ctrl = recreate_sockets
# Setting up HID interrupt/control sockets
test_ctrl.connect((reconnect_address, 17))
test_itr.connect((reconnect_address, 19))
test_ctrl.connect((address, 17))
test_itr.connect((address, 19))
itr = test_itr
ctrl = test_ctrl
except OSError:
test_itr.close()
test_ctrl.close()
pass
elif type(reconnect_address) == str:
test_itr, test_ctrl = recreate_sockets()

100
nxbt/nxbt.py
View file

@ -6,6 +6,7 @@ import signal
import os
import sys
import time
import json
import dbus
@ -22,6 +23,56 @@ JOYCON_R = ControllerTypes.JOYCON_R
PRO_CONTROLLER = ControllerTypes.PRO_CONTROLLER
DIRECT_INPUT_PACKET = {
# Sticks
"L_STICK": {
"PRESSED": False,
"X_VALUE": 0,
"Y_VALUE": 0,
# Keyboard position calculation values
"LS_UP": False,
"LS_LEFT": False,
"LS_RIGHT": False,
"LS_DOWN": False
},
"R_STICK": {
"PRESSED": False,
"X_VALUE": 0,
"Y_VALUE": 0,
# Keyboard position calculation values
"RS_UP": False,
"RS_LEFT": False,
"RS_RIGHT": False,
"RS_DOWN": False
},
# Dpad
"DPAD_UP": False,
"DPAD_LEFT": False,
"DPAD_RIGHT": False,
"DPAD_DOWN": False,
# Triggers
"L": False,
"ZL": False,
"R": False,
"ZR": False,
# Joy-Con Specific Buttons
"JCL_SR": False,
"JCL_SL": False,
"JCR_SR": False,
"JCR_SL": False,
# Meta buttons
"PLUS": False,
"MINUS": False,
"HOME": False,
"CAPTURE": False,
# Buttons
"Y": False,
"X": False,
"B": False,
"A": False
}
class Buttons():
"""The button object containing the button string constants.
"""
@ -34,8 +85,8 @@ class Buttons():
JCL_SL = 'JCL_SL'
R = 'R'
ZR = 'ZR'
MINUS = '-'
PLUS = '+'
MINUS = 'MINUS'
PLUS = 'PLUS'
R_STICK_PRESS = 'R_STICK_PRESS'
L_STICK_PRESS = 'L_STICK_PRESS'
HOME = 'HOME'
@ -206,8 +257,9 @@ class Nxbt():
cm.clear_macros(
msg["arguments"]["controller_index"])
elif msg["command"] == NxbtCommands.REMOVE_CONTROLLER:
cm.clear_macros(
msg["arguments"]["controller_index"])
index = msg["arguments"]["controller_index"]
cm.clear_macros(index)
cm.remove_controller(index)
finally:
cm.shutdown()
@ -385,6 +437,9 @@ class Nxbt():
"""Clears all running and queued macros on a specified
controller.
WARNING: Any blocking macro calls will continue to run
forever if this command is run.
:param controller_index: The index of a given controller
:type controller_index: int
:raises ValueError: If the controller_index does not exist
@ -408,6 +463,39 @@ class Nxbt():
for controller in self.manager_state.keys():
self.clear_macros(controller)
def set_controller_input(self, controller_index, input_packet):
"""Sets the controllers buttons and analog sticks for 1 cycle.
This means that exactly 1 packet will be sent to the Switch with
input specified with this method. To keep a continuous input
stream of a desired input, packets must be set at a rate that
roughly matches the set controller. Eg: An emulated Pro Controller's
input must be set at roughly 120Hz and a Joy-Con's at 60Hz.
:param controller_index: The index of the emulated controller
:type controller_index: int
:param input_packet: The input packet with the desired input. This
*must* be an instance of the create_input_packet method.
:type input_packet: dict
:raises ValueError: On bad controller index
"""
if controller_index not in self.manager_state.keys():
raise ValueError("Specified controller does not exist")
self.manager_state[controller_index]["direct_input"] = input_packet
def create_input_packet(self):
"""Creates an input packet that is used to specify the input
of a controller for a single cycle.
:return: An input packet dictionary
:rtype: dict
"""
# Create a copy of the direct input packet in a thread safe manner.
# NOTE: Using the copy.deepcopy method of copying dicts IS NOT thread safe.
return json.loads(json.dumps(DIRECT_INPUT_PACKET))
def create_controller(self, controller_type, adapter_path=None,
colour_body=None, colour_buttons=None,
reconnect_address=None):
@ -578,6 +666,9 @@ class Nxbt():
A list of UUIDs
"errors":
A string with the crash error
"direct_input":
A dictionary that represents all inputs
being directly input into the controller.
}
}
@ -638,6 +729,7 @@ class _ControllerManager():
controller_state["state"] = "initializing"
controller_state["finished_macros"] = []
controller_state["errors"] = False
controller_state["direct_input"] = json.loads(json.dumps(DIRECT_INPUT_PACKET))
self._controller_queues[index] = controller_queue

435
nxbt/tui.py
View file

@ -2,6 +2,7 @@ import os
import time
import psutil
from collections import deque
import multiprocessing
from blessed import Terminal
@ -49,10 +50,10 @@ class ControllerTUI():
"RS_LEFT": "(",
"RS_RIGHT": ")",
"RS_DOWN": "`─'",
"DP_UP": "△",
"DP_LEFT": "◁",
"DP_RIGHT": "▷",
"DP_DOWN": "▽",
"DPAD_UP": "△",
"DPAD_LEFT": "◁",
"DPAD_RIGHT": "▷",
"DPAD_DOWN": "▽",
"MINUS": "◎",
"PLUS": "◎",
"HOME": "□",
@ -74,6 +75,30 @@ class ControllerTUI():
for control in self.CONTROL_RELEASE_TIMERS.keys():
self.CONTROL_RELEASE_TIMERS[control] = False
self.auto_keypress_deactivation = True
self.remote_connection = False
def toggle_auto_keypress_deactivation(self, toggle):
"""Toggles whether or not the ControllerTUI should deactivate
a control after a period of time.
:param toggle: A True/False value that toggles auto keypress
deactivation
:type toggle: bool
"""
self.auto_keypress_deactivation = toggle
def set_remote_connection_status(self, status):
"""Sets whether or not the controller should render
with remote connection specific controls.
:param status: The status of the remote connection
:type status: bool
"""
self.remote_connection = status
def activate_control(self, key, activated_text=None):
if activated_text:
@ -82,7 +107,8 @@ class ControllerTUI():
self.CONTROLS[key] = self.term.bold_black_on_white(self.CONTROLS[key])
# Keep track of when the key was pressed so we can release later
self.CONTROL_RELEASE_TIMERS[key] = time.perf_counter()
if self.auto_keypress_deactivation:
self.CONTROL_RELEASE_TIMERS[key] = time.perf_counter()
def deactivate_control(self, key):
@ -90,12 +116,13 @@ class ControllerTUI():
def render_controller(self):
# Release any overdue timers
for control in self.CONTROL_RELEASE_TIMERS.keys():
pressed_time = self.CONTROL_RELEASE_TIMERS[control]
current_time = time.perf_counter()
if pressed_time is not False and current_time - pressed_time > 0.25:
self.deactivate_control(control)
if self.auto_keypress_deactivation:
# Release any overdue timers
for control in self.CONTROL_RELEASE_TIMERS.keys():
pressed_time = self.CONTROL_RELEASE_TIMERS[control]
current_time = time.perf_counter()
if pressed_time is not False and current_time - pressed_time > 0.25:
self.deactivate_control(control)
ZL = self.CONTROLS['ZL']
L = self.CONTROLS['L']
@ -109,10 +136,10 @@ class ControllerTUI():
RL = self.CONTROLS['RS_LEFT']
RR = self.CONTROLS['RS_RIGHT']
RD = self.CONTROLS['RS_DOWN']
DU = self.CONTROLS['DP_UP']
DL = self.CONTROLS['DP_LEFT']
DR = self.CONTROLS['DP_RIGHT']
DD = self.CONTROLS['DP_DOWN']
DU = self.CONTROLS['DPAD_UP']
DL = self.CONTROLS['DPAD_LEFT']
DR = self.CONTROLS['DPAD_RIGHT']
DD = self.CONTROLS['DPAD_DOWN']
MN = self.CONTROLS['MINUS']
PL = self.CONTROLS['PLUS']
HM = self.CONTROLS['HOME']
@ -122,6 +149,11 @@ class ControllerTUI():
X = self.CONTROLS['X']
Y = self.CONTROLS['Y']
if self.remote_connection:
lr_press = "L + R - - - - - - - - -▷ E"
else:
lr_press = " "
print(self.term.home + self.term.move_y((self.term.height // 2) - 9))
print(self.term.center(f" {ZL} {ZR} "))
print(self.term.center(f" ─{L}──────────{R}─ ┌─────────────┬────────────┐"))
@ -135,45 +167,113 @@ class ControllerTUI():
print(self.term.center("│░░░░╲ ──────────────── ╱░░░░│ L/ZL ─ ─ ─ ─ ─ ─ ─ ─ ▷ 1/2 "))
print(self.term.center("│░░░░╱ ╲░░░░│ R/ZR ─ ─ ─ ─ ─ ─ ─ ─ ▷ 8/9 "))
print(self.term.center("│░░╱ ╲░░│ Right Stick - - - ▷ Arrows "))
print(self.term.center("│╱ ╲│ "))
print(self.term.center(f"│╱ ╲│ {lr_press} "))
class InputTUI():
CONTROLS = {
"ZL": "◿□□□□",
"L": "◿□□□□",
"ZR": "□□□□◺",
"R": "□□□□◺",
"LS_UP": ".─.",
"LS_LEFT": "(",
"LS_RIGHT": ")",
"LS_DOWN": "`─'",
"RS_UP": ".─.",
"RS_LEFT": "(",
"RS_RIGHT": ")",
"RS_DOWN": "`─'",
"DP_UP": "△",
"DP_LEFT": "◁",
"DP_RIGHT": "▷",
"DP_DOWN": "▽",
"MINUS": "◎",
"PLUS": "◎",
"HOME": "□",
"CAPTURE": "□",
"A": "○",
"B": "○",
"X": "○",
"Y": "○",
KEYMAP = {
# Left Stick Mapping
"w": {
"control": "LS_UP",
"stick_data": {
"stick_name": "L_STICK",
"x": "+000",
"y": "+100"
}
},
"a": {
"control": "LS_LEFT",
"stick_data": {
"stick_name": "L_STICK",
"x": "-100",
"y": "+000"
}
},
"d": {
"control": "LS_RIGHT",
"stick_data": {
"stick_name": "L_STICK",
"x": "+100",
"y": "+000"
}
},
"s": {
"control": "LS_DOWN",
"stick_data": {
"stick_name": "L_STICK",
"x": "+000",
"y": "-100"
}
},
# Right Stick Mapping
"KEY_UP": {
"control": "RS_UP",
"stick_data": {
"stick_name": "R_STICK",
"x": "+000",
"y": "+100"
}
},
"KEY_LEFT": {
"control": "RS_LEFT",
"stick_data": {
"stick_name": "R_STICK",
"x": "-100",
"y": "+000"
}
},
"KEY_RIGHT": {
"control": "RS_RIGHT",
"stick_data": {
"stick_name": "R_STICK",
"x": "+100",
"y": "+000"
}
},
"KEY_DOWN": {
"control": "RS_DOWN",
"stick_data": {
"stick_name": "R_STICK",
"x": "+000",
"y": "-100"
}
},
# Dpad Mapping
"g": "DPAD_UP",
"v": "DPAD_LEFT",
"n": "DPAD_RIGHT",
"b": "DPAD_DOWN",
# Button Mapping
"6": "MINUS",
"7": "PLUS",
"[": "CAPTURE",
"]": "HOME",
"i": "X",
"j": "Y",
"l": "A",
"k": "B",
# Triggers
"1": "L",
"2": "ZL",
"8": "R",
"9": "ZR",
}
def __init__(self, reconnect_target=None):
def __init__(self, reconnect_target=None, debug=False, logfile=False):
self.reconnect_target = reconnect_target
self.term = Terminal()
self.remote_connection = self.detect_remote_connection()
self.controller = ControllerTUI(self.term)
self.debug = debug
self.logfile = logfile
def detect_remote_connection(self):
"""Traverse up the parent processes and check if any
have their parent as a remote daemon. If so, the python
@ -210,7 +310,10 @@ class InputTUI():
def mainloop(self, term):
# Initializing a controller
self.nx = Nxbt(disable_logging=True)
if not self.debug:
self.nx = Nxbt(disable_logging=True)
else:
self.nx = Nxbt(debug=self.debug, logfile=self.logfile)
self.controller_index = self.nx.create_controller(
PRO_CONTROLLER,
reconnect_address=self.reconnect_target)
@ -269,6 +372,8 @@ class InputTUI():
def remote_input_loop(self, term):
self.controller.set_remote_connection_status(True)
inp = term.inkey(timeout=0)
while inp != chr(113): # Checking for q press
# Cutoff large buffered input from the deque
@ -287,85 +392,27 @@ class InputTUI():
elif inp:
pressed_key = inp
if pressed_key == 'w':
self.controller.activate_control('LS_UP')
self.nx.macro(self.controller_index, "L_STICK@+000+100 0.1s")
elif pressed_key == 'a':
self.controller.activate_control('LS_LEFT')
self.nx.macro(self.controller_index, "L_STICK@-100+000 0.1s")
elif pressed_key == 'd':
self.controller.activate_control('LS_RIGHT')
self.nx.macro(self.controller_index, "L_STICK@+100+000 0.1s")
elif pressed_key == 's':
self.controller.activate_control('LS_DOWN')
self.nx.macro(self.controller_index, "L_STICK@+000-100 0.1s")
elif pressed_key == 'g':
self.controller.activate_control('DP_UP')
self.nx.macro(self.controller_index, "DPAD_UP 0.1s")
elif pressed_key == 'v':
self.controller.activate_control('DP_LEFT')
self.nx.macro(self.controller_index, "DPAD_LEFT 0.1s")
elif pressed_key == 'n':
self.controller.activate_control('DP_RIGHT')
self.nx.macro(self.controller_index, "DPAD_RIGHT 0.1s")
elif pressed_key == 'b':
self.controller.activate_control('DP_DOWN')
self.nx.macro(self.controller_index, "DPAD_DOWN 0.1s")
elif pressed_key == '[':
self.controller.activate_control('CAPTURE')
self.nx.macro(self.controller_index, "CAPTURE 0.1s")
elif pressed_key == ']':
self.controller.activate_control('HOME')
self.nx.macro(self.controller_index, "HOME 0.1s")
elif pressed_key == '6':
self.controller.activate_control('MINUS')
self.nx.macro(self.controller_index, "- 0.1s")
elif pressed_key == '7':
self.controller.activate_control('PLUS')
self.nx.macro(self.controller_index, "+ 0.1s")
elif pressed_key == 'i':
self.controller.activate_control('X')
self.nx.macro(self.controller_index, "X 0.1s")
elif pressed_key == 'j':
self.controller.activate_control('Y')
self.nx.macro(self.controller_index, "Y 0.1s")
elif pressed_key == 'l':
self.controller.activate_control('A')
self.nx.macro(self.controller_index, "A 0.1s")
elif pressed_key == 'k':
self.controller.activate_control('B')
self.nx.macro(self.controller_index, "B 0.1s")
elif pressed_key == '1':
if pressed_key == 'e':
self.controller.activate_control('L')
self.nx.macro(self.controller_index, "L 0.1s")
elif pressed_key == '2':
self.controller.activate_control('ZL')
self.nx.macro(self.controller_index, "ZL 0.1s")
elif pressed_key == '8':
self.controller.activate_control('R')
self.nx.macro(self.controller_index, "R 0.1s")
elif pressed_key == '9':
self.controller.activate_control('ZR')
self.nx.macro(self.controller_index, "ZR 0.1s")
self.nx.macro(self.controller_index, "L R 0.1s")
else:
try:
control_data = self.KEYMAP[pressed_key]
if type(control_data) == dict and "stick_data" in control_data.keys():
x_value = control_data['stick_data']['x']
y_value = control_data['stick_data']['y']
stick_name = control_data['stick_data']['stick_name']
elif pressed_key == 'KEY_UP':
self.controller.activate_control('RS_UP')
self.nx.macro(self.controller_index, "L_STICK@+000+100 0.1s")
elif pressed_key == 'KEY_LEFT':
self.controller.activate_control('RS_LEFT')
self.nx.macro(self.controller_index, "L_STICK@-100+000 0.1s")
elif pressed_key == 'KEY_RIGHT':
self.controller.activate_control('RS_RIGHT')
self.nx.macro(self.controller_index, "L_STICK@+100+000 0.1s")
elif pressed_key == 'KEY_DOWN':
self.controller.activate_control('RS_DOWN')
self.nx.macro(self.controller_index, "L_STICK@+000-100 0.1s")
self.controller.activate_control(control_data["control"])
self.nx.macro(
self.controller_index,
f"{stick_name}@{x_value}{y_value} 0.1s")
else:
self.controller.activate_control(control_data)
self.nx.macro(self.controller_index, f"{control_data} 0.05s")
except KeyError:
pass
self.controller.render_controller()
@ -373,7 +420,147 @@ class InputTUI():
def direct_input_loop(self, term):
pass
# pynput must be imported here since earlier imports
# will cause errors on remote connections
from pynput import keyboard
self.controller.toggle_auto_keypress_deactivation(False)
self.exit_tui = False
self.capture_input = True
# Create a packet that is accessible from a multiprocessing Process
# and from within threads
packet_manager = multiprocessing.Manager()
input_packet = packet_manager.dict()
input_packet["packet"] = self.nx.create_input_packet()
print(term.move_y(term.height - 5))
print(term.center(term.bold_black_on_white(" <Press esc to toggle input capture> ")))
def on_press(key):
# Parse the key press event
pressed_key = None
try:
pressed_key = key.char
except AttributeError:
pressed_key = str(key).replace(".", "_").upper()
if not self.capture_input: # If we're not capturing input, pass
pass
else:
try:
control_data = self.KEYMAP[pressed_key]
packet = input_packet["packet"]
if type(control_data) == dict and "stick_data" in control_data.keys():
stick_name = control_data['stick_data']['stick_name']
self.controller.activate_control(control_data["control"])
packet[stick_name][control_data["control"]] = True
else:
self.controller.activate_control(control_data)
packet[control_data] = True
input_packet["packet"] = packet
except KeyError:
pass
def on_release(key):
# Parse the key release event
released_key = None
try:
released_key = key.char
except AttributeError:
released_key = str(key).replace(".", "_").upper()
# If the esc key is released, toggle input capturing
if released_key == "KEY_ESC":
self.capture_input = not self.capture_input
# Exit on q key press
if released_key == 'q':
self.exit_tui = True
return False
if not self.capture_input: # If we're not capturing input, pass
pass
else:
try:
control_data = self.KEYMAP[released_key]
packet = input_packet["packet"]
if type(control_data) == dict and "stick_data" in control_data.keys():
stick_name = control_data['stick_data']['stick_name']
self.controller.deactivate_control(control_data["control"])
packet[stick_name][control_data["control"]] = False
else:
self.controller.deactivate_control(control_data)
packet[control_data] = False
input_packet["packet"] = packet
except KeyError:
pass
def input_worker(nxbt, controller_index, input_packet):
while True:
packet = input_packet["packet"]
# Calculating left x/y stick values
ls_x_value = 0
ls_y_value = 0
if packet["L_STICK"]["LS_LEFT"]:
ls_x_value -= 100
if packet["L_STICK"]["LS_RIGHT"]:
ls_x_value += 100
if packet["L_STICK"]["LS_UP"]:
ls_y_value += 100
if packet["L_STICK"]["LS_DOWN"]:
ls_y_value -= 100
packet["L_STICK"]["X_VALUE"] = ls_x_value
packet["L_STICK"]["Y_VALUE"] = ls_y_value
# Calculating right x/y stick values
rs_x_value = 0
rs_y_value = 0
if packet["R_STICK"]["RS_LEFT"]:
rs_x_value -= 100
if packet["R_STICK"]["RS_RIGHT"]:
rs_x_value += 100
if packet["R_STICK"]["RS_UP"]:
rs_y_value += 100
if packet["R_STICK"]["RS_DOWN"]:
rs_y_value -= 100
packet["R_STICK"]["X_VALUE"] = rs_x_value
packet["R_STICK"]["Y_VALUE"] = rs_y_value
nxbt.set_controller_input(controller_index, packet)
time.sleep(1/120)
input_process = multiprocessing.Process(
target=input_worker, args=(self.nx, self.controller_index, input_packet))
input_process.start()
# Start a non-blocking keyboard event listener
listener = keyboard.Listener(
on_press=on_press,
on_release=on_release)
listener.start()
# Main TUI Loop
while True:
if self.exit_tui:
packet_manager.shutdown()
input_process.terminate()
break
if not self.capture_input:
print(term.home + term.move_y((term.height // 2) - 4))
print(term.bold_black_on_white(term.center("")))
print(term.bold_black_on_white(term.center(
"<Input Paused. Press ESC Again to Begin Capturing Input>"
)))
print(term.bold_black_on_white(term.center("")))
else:
self.controller.render_controller()
self.check_for_disconnect(term)
time.sleep(1/30)
def render_start_screen(self, term, loading_text):
@ -415,7 +602,13 @@ class InputTUI():
print(term.bold_black_on_red(term.center("")))
print(term.bold_black_on_red(term.center(state.title())))
print(term.bold_black_on_red(term.center("")))
self.nx.wait_for_connection(self.controller_index)
while True:
inp = term.inkey(1/30)
if inp == chr(113):
exit(1)
elif self.nx.state[self.controller_index]["state"] == 'connected':
break
def main():