diff --git a/nxbt/cli.py b/nxbt/cli.py index 4220d20..e7075f5 100644 --- a/nxbt/cli.py +++ b/nxbt/cli.py @@ -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() diff --git a/nxbt/controller/input.py b/nxbt/controller/input.py index fa39340..b726161 100644 --- a/nxbt/controller/input.py +++ b/nxbt/controller/input.py @@ -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' diff --git a/nxbt/controller/protocol.py b/nxbt/controller/protocol.py index 8bdb4c0..34b794a 100644 --- a/nxbt/controller/protocol.py +++ b/nxbt/controller/protocol.py @@ -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): diff --git a/nxbt/controller/server.py b/nxbt/controller/server.py index dae0c1f..0ffbf79 100644 --- a/nxbt/controller/server.py +++ b/nxbt/controller/server.py @@ -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() diff --git a/nxbt/nxbt.py b/nxbt/nxbt.py index 75248d8..dd097fd 100644 --- a/nxbt/nxbt.py +++ b/nxbt/nxbt.py @@ -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 diff --git a/nxbt/tui.py b/nxbt/tui.py index 2cc26a1..054428d 100644 --- a/nxbt/tui.py +++ b/nxbt/tui.py @@ -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(" "))) + + 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( + "" + ))) + 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():