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

Added more functionality

Browse source 
Added previously connected Switch finding

Cleaned up demo CLI command

Added ability to stop  macros

Fixed reconnecting and timing quirks

Bugfixes and polish
This commit is contained in:
Brikwerk 2020-08-01 23:11:47 -07:00
commit 7e99a36675
10 changed files with 745 additions and 182 deletions
Download patch file Download diff file Expand all files Collapse all files

95
demo.py
View file

@ -2,46 +2,40 @@ import time
from random import randint
import nxbt
from nxbt import Buttons
MACRO = """
B 0.1s
0.1s
B 0.1s
0.1s
B 0.1s
0.1s
B 0.1s
LOOP 12
B 0.1s
0.1s
1.5s
DPAD_RIGHT 0.075s
0.075s
A 0.1s
1.5s
DPAD_DOWN 1.0s
DPAD_DOWN 1.00s
A 0.1s
0.25s
DPAD_DOWN 0.8s
DPAD_DOWN 0.95s
A 0.1s
0.25s
L_STICK_PRESS 0.1s
1.0s
L_STICK@-100+000 0.75s
L_STICK@+000+100 0.75s
L_STICK@+100+000 0.75s
L_STICK@+000-100 0.75s
L_STICK@-100+000 0.5s
L_STICK@+000+100 0.5s
L_STICK@+100+000 0.5s
L_STICK@+000-100 0.5s
B 0.1s
0.25s
R_STICK_PRESS 0.1s
1.0s
R_STICK@-100+000 0.75s
R_STICK@+000+100 0.75s
R_STICK@+100+000 0.75s
R_STICK@+000-100 0.75s
B 0.1s
0.1s
B 0.1s
0.1s
B 0.1s
0.1s
R_STICK@-100+000 0.5s
R_STICK@+000+100 0.5s
R_STICK@+100+000 0.5s
R_STICK@+000-100 0.5s
LOOP 4
B 0.1s
0.1s
B 0.1s
0.4s
DPAD_LEFT 0.1s
@ -49,7 +43,7 @@ DPAD_LEFT 0.1s
A 0.1s
1.5s
A 0.1s
0.1s
5.0s
"""
@ -64,11 +58,16 @@ def random_colour():
if __name__ == "__main__":
# Init NXBT
nx = nxbt.Nxbt()
# Get a list of all available Bluetooth adapters
adapters = nx.get_available_adapters()
# Prepare a list to store the indexes of the
# created controllers.
controller_idxs = []
# Loop over all Bluetooth adapters and create
# Switch Pro Controllers
nx = nxbt.Nxbt()
adapters = nx.get_available_adapters()
controller_idxs = []
for i in range(0, len(adapters)):
index = nx.create_controller(
nxbt.PRO_CONTROLLER,
@ -77,16 +76,38 @@ if __name__ == "__main__":
colour_buttons=random_colour())
controller_idxs.append(index)
# Select the last controller for input
controller_idx = controller_idxs[-1]
time.sleep(4)
# Wait for the switch to connect to the controller
nx.wait_for_connection(controller_idx)
# Run a macro on the last controller
# and don't wait for the macro to complete
nx.macro(controller_idxs[-1], MACRO, block=False)
print("Macro Started")
macro_id = nx.macro(controller_idx, MACRO, block=False)
time.sleep(3)
# Stop the macro
print("Stopping Macro")
nx.stop_macro(controller_idx, macro_id)
print("Stopped Macro")
# Check the state
while True:
time.sleep(1)
for key in nx.state.keys():
state = nx.state[key]
if not state["errors"]:
print(state)
else:
print(state["errors"])
# Return to the "Change Grip/Order Screen"
time.sleep(2)
nx.press_buttons(controller_idx, [Buttons.A])
time.sleep(2)
nx.press_buttons(controller_idx, [Buttons.A])
time.sleep(2)
# Enter the same macro, but block this time
print("Macro Started")
macro_id = nx.macro(controller_idx, MACRO)
print("Macro finished, going to the home screen...")
nx.press_buttons(controller_idx, [Buttons.B])
time.sleep(2)
nx.press_buttons(controller_idx, [Buttons.B])
time.sleep(2)
print("Exiting...")

1
nxbt/__init__.py
View file

@ -5,6 +5,7 @@ from .controller import SwitchResponses
from .controller import Controller
from .bluez import *
from .nxbt import Nxbt
from .nxbt import Buttons
from .nxbt import JOYCON_L
from .nxbt import JOYCON_R
from .nxbt import PRO_CONTROLLER

44
nxbt/bluez.py
View file

@ -154,6 +154,46 @@ def toggle_input_plugin(toggle):
time.sleep(0.5)
def find_devices_by_alias(alias):
"""Finds the Bluetooth addresses of devices
that have a specified Bluetooth alias. Aliases
are converted to uppercase before comparison
as BlueZ usually converts aliases to uppercase.
:param address: The Bluetooth MAC address
:type address: string
:return: The path to the D-Bus object or None
:rtype: string or None
"""
bus = dbus.SystemBus()
# Find all connected/paired/discovered devices
devices = find_objects(
bus,
SERVICE_NAME,
DEVICE_INTERFACE)
addresses = []
for path in devices:
# Get the device's address and paired status
device_props = dbus.Interface(
bus.get_object(SERVICE_NAME, path),
"org.freedesktop.DBus.Properties")
device_alias = device_props.Get(
DEVICE_INTERFACE,
"Alias").upper()
device_addr = device_props.Get(
DEVICE_INTERFACE,
"Address").upper()
# Check for an address match
if device_alias.upper() == alias.upper():
addresses.append(device_addr)
bus.close()
return addresses
class BlueZ():
"""Exposes the BlueZ D-Bus API as a Python object.
"""
@ -580,8 +620,8 @@ class BlueZ():
DEVICE_INTERFACE)
try:
device.Connect()
except dbus.exceptions.DBusException:
print("Here1")
except dbus.exceptions.DBusException as e:
print("Error: ", e)
def remove_device(self, path):
"""Removes a device that's been either discovered, paired,

111
nxbt/cli.py
View file

@ -12,66 +12,53 @@ parser.add_argument('command', default=False, choices=['start', 'demo'],
args = parser.parse_args()
MACRO = """
LOOP 100
B 0.1s
0.1s
B 0.1s
0.1s
B 0.1s
0.1s
B 0.1s
1.5s
DPAD_RIGHT 0.075s
0.075s
A 0.1s
1.5s
DPAD_DOWN 1.0s
A 0.1s
0.25s
DPAD_DOWN 0.8s
A 0.1s
0.25s
L_STICK_PRESS 0.1s
1.0s
L_STICK@-100+000 0.75s
L_STICK@+000+100 0.75s
L_STICK@+100+000 0.75s
L_STICK@+000-100 0.75s
B 0.1s
0.25s
R_STICK_PRESS 0.1s
1.0s
R_STICK@-100+000 0.75s
R_STICK@+000+100 0.75s
R_STICK@+100+000 0.75s
R_STICK@+000-100 0.75s
B 0.1s
0.1s
B 0.1s
0.1s
B 0.1s
0.1s
B 0.1s
0.4s
DPAD_LEFT 0.1s
0.1s
A 0.1s
1.5s
A 0.1s
20.0s
"""
MACRO = """
B 0.1s
3.0s
0.1s
B 0.1s
0.1s
B 0.1s
0.1s
B 0.1s
1.5s
DPAD_RIGHT 0.075s
0.075s
A 0.1s
3.0s
20.0s
1.5s
DPAD_DOWN 1.0s
A 0.1s
0.25s
DPAD_DOWN 0.95s
A 0.1s
0.25s
L_STICK_PRESS 0.1s
1.0s
L_STICK@-100+000 0.75s
L_STICK@+000+100 0.75s
L_STICK@+100+000 0.75s
L_STICK@+000-100 0.75s
B 0.1s
LOOP 100
B 0.1s
0.1s
0.25s
R_STICK_PRESS 0.1s
1.0s
R_STICK@-100+000 0.75s
R_STICK@+000+100 0.75s
R_STICK@+100+000 0.75s
R_STICK@+000-100 0.75s
B 0.1s
0.1s
B 0.1s
0.1s
B 0.1s
0.1s
B 0.1s
0.4s
DPAD_LEFT 0.1s
0.1s
A 0.1s
1.5s
A 0.1s
5.0s
"""
@ -103,17 +90,7 @@ def demo():
# Run a macro on the last controller
# and don't wait for the macro to complete
nx.macro(controller_idxs[-1], MACRO, block=False)
# Check the state
while True:
time.sleep(1)
for key in nx.state.keys():
state = nx.state[key]
if not state["errors"]:
print(state)
else:
print(state["errors"])
nx.macro(controller_idxs[-1], MACRO)
def main():

50
nxbt/controller/input.py
View file

@ -55,6 +55,43 @@ class InputParser():
self.macro_buffer.append([macro, macro_id])
def stop_macro(self, macro_id, state=None):
# Check if the macro is being input currently
if macro_id == self.current_macro_id:
# If so, reset the current macro
self.current_macro = None
self.current_macro_id = None
self.current_macro_commands = None
self.macro_timer_length = 0
self.macro_timer_start = 0
else:
# Check if the macro is still in the buffer
for i in range(0, len(self.macro_buffer)):
if macro_id == self.macro_buffer[i][1]:
del self.macro_buffer[i]
# Ensure the stopped macro is added to the finished
# macros so that any blocking parties listening can
# continue.
if state:
finished = state["finished_macros"]
finished.append(macro_id)
state["finished_macros"] = finished
return
def clear_macros(self, state=None):
self.current_macro = None
self.current_macro_id = None
self.current_macro_commands = None
self.macro_timer_length = 0
self.macro_timer_start = 0
self.macro_buffer = []
return
def set_controller_input(self, controller_input):
self.controller_input = controller_input
@ -85,8 +122,7 @@ class InputParser():
self.macro_timer_length = float(timer_length)
self.macro_timer_start = perf_counter()
print(self.current_macro_commands)
self.parse_macro_input(self.current_macro_commands)
self.set_macro_input(self.current_macro_commands)
# Check if we're done inputting the current command
time_delta = perf_counter() - self.macro_timer_start
@ -153,7 +189,7 @@ class InputParser():
return parsed
def parse_macro_input(self, macro_input):
def set_macro_input(self, macro_input):
# Checking if this is a wait macro command
if len(macro_input) < 2:
@ -178,9 +214,9 @@ class InputParser():
upper[5] = '1'
elif button == "A":
upper[4] = '1'
elif button == "SR":
elif button == "JCL_SR":
upper[3] = '1'
elif button == "SL":
elif button == "JCL_SL":
upper[2] = '1'
elif button == "R":
upper[1] = '1'
@ -210,9 +246,9 @@ class InputParser():
lower[5] = '1'
elif button == "DPAD_LEFT":
lower[4] = '1'
elif button == "SR":
elif button == "JCR_SR":
lower[3] = '1'
elif button == "SL":
elif button == "JCR_SL":
lower[2] = '1'
elif button == "L":
lower[1] = '1'

45
nxbt/controller/server.py
View file

@ -70,6 +70,7 @@ class ControllerServer():
self.controller.setup()
if reconnect_address:
print("reconnecting")
itr, ctrl = self.reconnect(reconnect_address)
else:
itr, ctrl = self.connect()
@ -92,6 +93,9 @@ class ControllerServer():
# Mainloop
while True:
# Start timing the command processing
timer_start = time.perf_counter()
# Attempt to get output from Switch
try:
reply = itr.recv(50)
@ -104,9 +108,15 @@ class ControllerServer():
if self.task_queue:
try:
msg = self.task_queue.get_nowait()
if msg:
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)
except queue.Empty:
pass
@ -114,8 +124,8 @@ class ControllerServer():
self.input.set_protocol_input(state=self.state)
msg = self.protocol.get_report()
# if reply:
# print(format_msg_controller(msg))
if reply and len(reply) > 45:
print(format_msg_controller(msg))
try:
itr.sendall(msg)
@ -125,12 +135,20 @@ class ControllerServer():
# Attempt to reconnect to the Switch
itr, ctrl = self.save_connection(e)
# Figure out how long it took to process commands
timer_end = time.perf_counter()
elapsed_time = (timer_end - timer_start)
# Respond at 120Hz for Pro Controller
# or 60Hz for Joy-Cons
# or 60Hz for Joy-Cons.
# Sleep timers are compensated with the elapsed command
# processing time.
if self.controller_type == ControllerTypes.PRO_CONTROLLER:
time.sleep(1/120)
if elapsed_time < 1/120:
time.sleep(1/120 - elapsed_time)
else:
time.sleep(1/60)
if elapsed_time < 1/60:
time.sleep(1/60 - elapsed_time)
def save_connection(self, error, state=None):
@ -169,6 +187,11 @@ class ControllerServer():
colour_buttons=self.colour_buttons)
self.input.reassign_protocol(self.protocol)
# Since we were forced to attempt a reconnection
# we need to press the L and R buttons before
# we can proceed with any input.
self.input.current_macro_commands = "L R 0.0s".strip(" ").split(" ")
if self.lock:
self.lock.acquire()
try:
@ -221,9 +244,9 @@ class ControllerServer():
msg = self.protocol.get_report()
itr.sendall(msg)
# Setting interrupt connection as non-blocking
# Setting interrupt connection as non-blocking.
# In this case, non-blocking means it throws a "BlockingIOError"
# for sending and receiving, instead of blocking
# for sending and receiving, instead of blocking.
fcntl.fcntl(itr, fcntl.F_SETFL, os.O_NONBLOCK)
# Mainloop
@ -247,8 +270,10 @@ class ControllerServer():
except BlockingIOError:
continue
# Exit pairing loop on set player lights
if reply and len(reply) > 45 and reply[11] == 0x30:
# Exit pairing loop when player lights have been set and
# vibration has been enabled
if (reply and len(reply) > 45 and
self.protocol.vibration_enabled and self.protocol.player_number):
break
# Switch responds to packets slower during pairing

265
nxbt/nxbt.py
View file

@ -1,8 +1,8 @@
from multiprocessing import Process, Lock
from multiprocessing import Queue, Manager
from multiprocessing import Process, Lock, Queue, Manager
import queue
from enum import Enum
import atexit
import signal
import os
import dbus
@ -10,6 +10,7 @@ import dbus
from .controller import ControllerServer
from .controller import ControllerTypes
from .bluez import find_objects, toggle_input_plugin
from .bluez import find_devices_by_alias
from .bluez import SERVICE_NAME, ADAPTER_INTERFACE
@ -18,10 +19,41 @@ JOYCON_R = ControllerTypes.JOYCON_R
PRO_CONTROLLER = ControllerTypes.PRO_CONTROLLER
class Buttons():
Y = 'Y'
X = 'X'
B = 'B'
A = 'A'
JCL_SR = 'JCL_SR'
JCL_SL = 'JCL_SL'
R = 'R'
ZR = 'ZR'
MINUS = '-'
PLUS = '+'
R_STICK_PRESS = 'R_STICK_PRESS'
L_STICK_PRESS = 'L_STICK_PRESS'
HOME = 'HOME'
CAPTURE = 'CAPTURE'
DPAD_DOWN = 'DPAD_DOWN'
DPAD_UP = 'DPAD_UP'
DPAD_RIGHT = 'DPAD_RIGHT'
DPAD_LEFT = 'DPAD_LEFT'
JCR_SR = 'JCR_SR'
JCR_SL = 'JCR_SL'
L = 'L'
ZL = 'ZL'
class NxbtCommands(Enum):
CREATE_CONTROLLER = 0
INPUT_MACRO = 1
STOP_MACRO = 2
CLEAR_MACROS = 3
CLEAR_ALL_MACROS = 4
REMOVE_CONTROLLER = 5
QUIT = 6
class Nxbt():
@ -32,7 +64,7 @@ class Nxbt():
self.task_queue = Queue()
# Sychronizes bluetooth actions
self.__bluetooth_lock = Lock()
self.__bluetooth_lock__ = Lock()
# Creates/manages shared resources
self.resource_manager = Manager()
@ -44,9 +76,10 @@ class Nxbt():
# Shared, controller management properties.
# The controller lock is used to sychronize use.
self.__controller_lock = Lock()
self.__controller_counter = 0
self.__adapters_in_use = []
self.__controller_lock__ = Lock()
self.__controller_counter__ = 0
self.__adapters_in_use__ = {}
self.__controller_adapter_lookup__ = {}
# Disable the BlueZ input plugin so we can use the
# HID control/interrupt Bluetooth ports
@ -57,7 +90,7 @@ class Nxbt():
# Starting the nxbt worker process
self.controllers = Process(
target=self.__command_manager,
target=self.__command_manager__,
args=((self.task_queue), (self.manager_state)))
# Disabling daemonization since we need to spawn
# other controller processes, however, this means
@ -72,32 +105,53 @@ class Nxbt():
if hasattr(self, "controllers") and self.controllers.is_alive():
self.controllers.terminate()
self.resource_manager.shutdown()
# Re-enable the BlueZ input plugin
toggle_input_plugin(True)
def __command_manager(self, task_queue, state):
def __command_manager__(self, task_queue, state):
cm = ControllerManager(state, self.__bluetooth_lock)
cm = ControllerManager(state, self.__bluetooth_lock__)
# Ensure a SystemExit exception is raised on SIGTERM
# so that we can gracefully shutdown.
signal.signal(signal.SIGTERM, lambda sigterm_handler: quit())
while True:
try:
msg = task_queue.get_nowait()
except queue.Empty:
msg = None
try:
while True:
try:
msg = task_queue.get_nowait()
except queue.Empty:
msg = None
if msg:
if msg["command"] == NxbtCommands.CREATE_CONTROLLER:
cm.create_controller(
msg["arguments"]["controller_index"],
msg["arguments"]["controller_type"],
msg["arguments"]["adapter_path"],
msg["arguments"]["colour_body"],
msg["arguments"]["colour_buttons"])
elif msg["command"] == NxbtCommands.INPUT_MACRO:
cm.input_macro(
msg["arguments"]["controller_index"],
msg["arguments"]["macro"],
msg["arguments"]["macro_id"])
if msg:
if msg["command"] == NxbtCommands.CREATE_CONTROLLER:
cm.create_controller(
msg["arguments"]["controller_index"],
msg["arguments"]["controller_type"],
msg["arguments"]["adapter_path"],
msg["arguments"]["colour_body"],
msg["arguments"]["colour_buttons"],
msg["arguments"]["reconnect_address"])
elif msg["command"] == NxbtCommands.INPUT_MACRO:
cm.input_macro(
msg["arguments"]["controller_index"],
msg["arguments"]["macro"],
msg["arguments"]["macro_id"])
elif msg["command"] == NxbtCommands.STOP_MACRO:
cm.stop_macro(
msg["arguments"]["controller_index"],
msg["arguments"]["macro_id"])
elif msg["command"] == NxbtCommands.CLEAR_MACROS:
cm.clear_macros(
msg["arguments"]["controller_index"])
elif msg["command"] == NxbtCommands.REMOVE_CONTROLLER:
cm.clear_macros(
msg["arguments"]["controller_index"])
finally:
cm.shutdown()
quit()
def macro(self, controller_index, macro, block=True):
@ -125,31 +179,101 @@ class Nxbt():
return macro_id
def press_buttons(self, controller_index, buttons, up=0.1, down=0.1, block=True):
if controller_index not in self.manager_state.keys():
raise ValueError("Specified controller does not exist")
macro_buttons = " ".join(buttons)
macro_times = f"{up}s \n{down}s"
macro = macro_buttons + " " + macro_times
# Get a unique ID to identify the button press
# so we can check when the controller is done inputting it
macro_id = os.urandom(24).hex()
self.task_queue.put({
"command": NxbtCommands.INPUT_MACRO,
"arguments": {
"controller_index": controller_index,
"macro": macro,
"macro_id": macro_id,
}
})
if block:
while True:
finished = (self.manager_state
[controller_index]["finished_macros"])
if macro_id in finished:
break
return macro_id
def stop_macro(self, controller_index, macro_id, block=True):
if controller_index not in self.manager_state.keys():
raise ValueError("Specified controller does not exist")
self.task_queue.put({
"command": NxbtCommands.STOP_MACRO,
"arguments": {
"controller_index": controller_index,
"macro_id": macro_id,
}
})
if block:
while True:
finished = (self.manager_state
[controller_index]["finished_macros"])
if macro_id in finished:
break
def clear_macros(self, controller_index):
if controller_index not in self.manager_state.keys():
raise ValueError("Specified controller does not exist")
self.task_queue.put({
"command": NxbtCommands.CLEAR_MACROS,
"arguments": {
"controller_index": controller_index,
}
})
def clear_all_macros(self):
for controller in self.manager_state.keys():
self.clear_macros(controller)
def create_controller(self, controller_type, adapter_path,
colour_body=None, colour_buttons=None):
colour_body=None, colour_buttons=None,
reconnect_address=None):
if adapter_path not in self.get_available_adapters():
raise ValueError("Specified adapter is unavailable")
if adapter_path in self.__adapters_in_use:
if adapter_path in self.__adapters_in_use__.keys():
raise ValueError("Specified adapter in use")
controller_index = None
try:
self.__controller_lock.acquire()
self.__controller_lock__.acquire()
self.task_queue.put({
"command": NxbtCommands.CREATE_CONTROLLER,
"arguments": {
"controller_index": self.__controller_counter,
"controller_index": self.__controller_counter__,
"controller_type": controller_type,
"adapter_path": adapter_path,
"colour_body": colour_body,
"colour_buttons": colour_buttons,
"reconnect_address": reconnect_address,
}
})
controller_index = self.__controller_counter
self.__controller_counter += 1
self.__adapters_in_use.append(adapter_path)
controller_index = self.__controller_counter__
self.__controller_counter__ += 1
self.__adapters_in_use__[adapter_path] = controller_index
self.__controller_adapter_lookup__[controller_index] = adapter_path
# Block until the controller is ready
# This needs to be done to prevent race conditions
@ -162,10 +286,34 @@ class Nxbt():
state["state"] == "reconnecting"):
break
finally:
self.__controller_lock.release()
self.__controller_lock__.release()
return controller_index
def remove_controller(self, controller_index):
if controller_index not in self.manager_state.keys():
raise ValueError("Specified controller does not exist")
self.__controller_lock__.acquire()
try:
adapter_path = self.__controller_adapter_lookup__.pop(controller_index, None)
self.__adapters_in_use__.pop(adapter_path, None)
finally:
self.__controller_lock__.release()
self.task_queue.put({
"command": NxbtCommands.REMOVE_CONTROLLER,
"arguments": {
"controller_index": controller_index,
}
})
def wait_for_connection(self, controller_index):
while not self.state[controller_index]["state"] == "connected":
pass
def get_available_adapters(self):
bus = dbus.SystemBus()
@ -173,6 +321,10 @@ class Nxbt():
return adapters
def get_switch_addresses(self):
return (find_devices_by_alias("Nintendo Switch"))
@property
def state(self):
@ -186,10 +338,12 @@ class ControllerManager():
self.state = state
self.lock = lock
self.controller_resources = Manager()
self.__controller_queues = {}
self.__controller_queues__ = {}
self.__children__ = {}
def create_controller(self, index, controller_type, adapter_path,
colour_body=None, colour_buttons=None):
colour_body=None, colour_buttons=None,
reconnect_address=None):
controller_queue = Queue()
@ -198,7 +352,7 @@ class ControllerManager():
controller_state["finished_macros"] = []
controller_state["errors"] = False
self.__controller_queues[index] = controller_queue
self.__controller_queues__[index] = controller_queue
self.state[index] = controller_state
@ -209,18 +363,41 @@ class ControllerManager():
task_queue=controller_queue,
colour_body=colour_body,
colour_buttons=colour_buttons)
controller = Process(target=server.run)
controller = Process(target=server.run, args=(reconnect_address,))
controller.daemon = True
self.__children__[index] = controller
controller.start()
def input_macro(self, index, macro, macro_id):
# finished = self.state[index]["finished_macros"]
# finished.append(macro_id)
# self.state[index]["finished_macros"] = finished
self.__controller_queues[index].put({
self.__controller_queues__[index].put({
"type": "macro",
"macro": macro,
"macro_id": macro_id
})
def stop_macro(self, index, macro_id):
self.__controller_queues__[index].put({
"type": "stop",
"macro_id": macro_id,
})
def clear_macros(self, index):
self.__controller_queues__[index].put({
"type": "clear",
})
def remove_controller(self, index):
self.__children__[index].kill()
def shutdown(self):
# Loop over children and kill all
for index in self.__children__.keys():
child = self.__children__[index]
child.terminate()
self.controller_resources.shutdown()

12
scripts/proxy.py
View file

@ -3,6 +3,10 @@ This is a quick and dirty script for recording input from a controller
and dumping it into a "messages.txt" file. You'll need to input the
device's Bluetooth MAC address manually and specify the type of
controller before this script works.
Note: If you get an Invalid Exchange error when running this script, this means
that the Switch has paired to the controller, invalidating the original pairing
key we created. You'll need to remove the controller before continuing.
"""
import socket
@ -93,7 +97,7 @@ def write_to_buffer(buffer, message, message_type):
if __name__ == "__main__":
# Switch Controller Bluetooth MAC Address goes here
jc_MAC = "7C:BB:8A:FA:41:3D"
jc_MAC = "98:B6:E9:B0:05:E7"
# Specify the type of controller here
controller_type = PRO_CONTROLLER
if controller_type == JOYCON_L:
@ -203,11 +207,11 @@ if __name__ == "__main__":
jc_itr.sendall(reply)
# Sending Switch the proxy's device info
if controller_type == ControllerTypes.JOYCON_R:
if controller_type == JOYCON_R:
client_interrupt.sendall(REPLY)
elif controller_type == ControllerTypes.JOYCON_L:
elif controller_type == JOYCON_L:
client_interrupt.sendall(REPLY)
elif controller_type == ControllerTypes.PRO_CONTROLLER:
elif controller_type == PRO_CONTROLLER:
client_interrupt.sendall(REPLY)
# Waste some cycles here until we get the controllers info.

291
scripts/reconnect_proxy.py Normal file
View file

@ -0,0 +1,291 @@
"""
This is a quick and dirty script for recording input from a controller
and dumping it into a "messages.txt" file. You'll need to input the
device's Bluetooth MAC address manually and specify the type of
controller before this script works.
Note: If you get an Invalid Exchange error when running this script, this means
that the Switch has paired to the controller, invalidating the original pairing
key we created. You'll need to remove the controller before continuing.
"""
import socket
import sys
import os
import time
import fcntl
from time import perf_counter
from nxbt import toggle_input_plugin
from nxbt import BlueZ
from nxbt import Controller
from nxbt import JOYCON_L, JOYCON_R, PRO_CONTROLLER
JCL_REPLY02 = b'\xA2\x21\x05\x8E\x84\x00\x12\x01\x18\x80\x01\x18\x80\x80\x82\x02\x03\x48\x01\x02\xDC\xA6\x32\x16\x4A\x7C\x01\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
PRO_REPLY02 = b'\xA2\x21\x1A\x40\x00\x00\x00\x02\x20\x00\x01\x00\x00\x00\x82\x02\x03\x48\x03\x02\xDC\xA6\x32\x16\x4A\x7C\x01\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
JCR_REPLY02 = b'\xA2\x21\x05\x8E\x84\x00\x12\x01\x18\x80\x01\x18\x80\x80\x82\x02\x03\x48\x02\x02\xDC\xA6\x32\x16\x4A\x7C\x01\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
def format_message(data, split, name):
"""Formats a given byte message in hex format split
into payload and subcommand sections.
:param data: A series of bytes
:type data: bytes
:param split: The location of the payload/subcommand split
:type split: integer
:param name: The name featured in the start/end messages
:type name: string
:return: The formatted data
:rtype: string
"""
payload = ""
subcommand = ""
for i in range(0, len(data)):
data_byte = str(hex(data[i]))[2:].upper()
if len(data_byte) < 2:
data_byte = "0" + data_byte
if i <= split:
payload += "0x" + data_byte + " "
else:
subcommand += "0x" + data_byte + " "
formatted = (
f"--- {name} Msg ---\n" +
f"Payload: {payload}\n" +
f"Subcommand: {subcommand}")
return formatted
def print_msg_controller(data):
"""Prints a formatted message from a controller
:param data: The bytes from the controller message
:type data: bytes
"""
print(format_message(data, 13, "Controller"))
def print_msg_switch(data):
"""Prints a formatted message from a Switch
:param data: The bytes from the Switch message
:type data: bytes
"""
print(format_message(data, 10, "Switch"))
def write_to_buffer(buffer, message, message_type):
formatted_message = None
if message_type == "switch":
formatted_message = format_message(message, 10, "Switch")
elif message_type == "controller":
formatted_message = format_message(message, 13, "Controller")
elif message_type == "comment":
formatted_message = "### " + message + " ###"
else:
raise ValueError("Unspecified or wrong message type")
buffer.append(formatted_message)
if __name__ == "__main__":
# Switch Controller Bluetooth MAC Address goes here
jc_MAC = "98:B6:E9:B0:05:E7"
switch_MAC = "7C:BB:8A:D9:91:5A"
# Specify the type of controller here
controller_type = PRO_CONTROLLER
if controller_type == JOYCON_L:
REPLY = JCL_REPLY02
elif controller_type == JOYCON_R:
REPLY = JCR_REPLY02
else:
REPLY = PRO_REPLY02
port_ctrl = 17
port_itr = 19
message_buffer = []
toggle_input_plugin(False)
bt = BlueZ(adapter_path="/org/bluez/hci0")
controller = Controller(bt, controller_type)
# Joy-Con Sockets
jc_ctrl = socket.socket(family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
jc_itr = socket.socket(family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
# Switch sockets
switch_itr = socket.socket(family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
switch_ctrl = socket.socket(family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
try:
switch_ctrl.bind((bt.address, port_ctrl))
switch_itr.bind((bt.address, port_itr))
# bt.set_alias("Joy-Con (L)")
bt.set_alias("Pro Controller")
bt.set_discoverable(True)
print("Waiting for Switch to connect...")
switch_itr.listen(1)
switch_ctrl.listen(1)
client_control, control_address = switch_ctrl.accept()
print("Got Switch Control Client Connection")
client_interrupt, interrupt_address = switch_itr.accept()
print("Got Switch Interrupt Client Connection")
bt.set_alias("Nintendo Switch")
print("Connecting to Joy-Con: ", jc_MAC)
jc_ctrl.bind((socket.BDADDR_ANY, port_ctrl))
jc_itr.bind((socket.BDADDR_ANY, port_itr))
jc_ctrl.listen(1)
jc_itr.listen(1)
jc_client_ctrl, _ = jc_ctrl.accept()
jc_client_itr, _ = jc_itr.accept()
print("Got connection.")
# Creating a non-blocking client interrupt connection
fcntl.fcntl(client_interrupt, fcntl.F_SETFL, os.O_NONBLOCK)
# Initial Input report from Joy-Con
jc_data = jc_client_itr.recv(350)
print("Got initial Joy-Con Empty Report")
# print_msg_controller(jc_data)
write_to_buffer(
message_buffer,
"Joy-Con Empty Report",
"comment")
write_to_buffer(message_buffer, jc_data, "controller")
print(message_buffer)
# Send the input report to the Switch a couple times
for i in range(3):
print("Sending input report", i)
client_interrupt.sendall(jc_data)
time.sleep(1)
# Get the Switch's reply and send it to the Joy-Con
reply = client_interrupt.recv(350)
# print_msg_switch(reply)
write_to_buffer(
message_buffer,
"Switch Input Report Reply",
"comment")
write_to_buffer(message_buffer, reply, "switch")
jc_client_itr.sendall(reply)
# Sending Switch the proxy's device info
if controller_type == JOYCON_R:
client_interrupt.sendall(REPLY)
elif controller_type == JOYCON_L:
client_interrupt.sendall(REPLY)
elif controller_type == PRO_CONTROLLER:
client_interrupt.sendall(REPLY)
# Waste some cycles here until we get the controllers info.
# We don't want to proxy the device's info to the Switch
# since it includes a MAC address.
print("Waiting on Joy-Con Device Info")
while True:
jc_data = jc_client_itr.recv(350)
if jc_data[1] == 0x21:
print("Got Device Info")
# print_msg_controller(jc_data)
print("Joy-Con Device Info Reply Length", len(jc_data))
write_to_buffer(
message_buffer,
"Joy-Con Device Info",
"comment")
write_to_buffer(message_buffer, jc_data, "controller")
break
# Main loop
print("Entering main proxy loop")
write_to_buffer(
message_buffer,
"Entering Main Loop",
"comment")
time_old = perf_counter()
timer_old = 0
timer_counter = 0
while True:
try:
reply = client_interrupt.recv(350)
# print_msg_switch(reply)
write_to_buffer(message_buffer, reply, "switch")
except BlockingIOError:
reply = None
if reply:
print("Sending to Controller")
jc_client_itr.sendall(reply)
jc_data = jc_client_itr.recv(350)
timer_new = int(jc_data[2])
if timer_new < timer_old:
timer_counter += timer_new - (timer_old - 255)
else:
timer_counter += timer_new - timer_old
timer_old = timer_new
# print_msg_controller(jc_data)
write_to_buffer(message_buffer, jc_data, "controller")
try:
client_interrupt.sendall(jc_data)
except BlockingIOError:
continue
time.sleep(1/2)
except KeyboardInterrupt:
print("Closing sockets")
# time_new = perf_counter()
# print(f"Total Delta: {(time_new - time_old) * 1000}")
# print(f"Timer Counter: {timer_counter}")
jc_ctrl.close()
jc_itr.close()
switch_itr.close()
switch_ctrl.close()
# Write the buffer
with open("messages.txt", "w") as f:
f.write("\n".join(message_buffer))
try:
sys.exit(1)
except SystemExit:
os._exit(1)
except OSError as e:
print("Closing sockets")
jc_ctrl.close()
jc_itr.close()
switch_itr.close()
switch_ctrl.close()
raise e
finally:
toggle_input_plugin(True)

13
scripts/switch_emu.py
View file

@ -3,7 +3,7 @@ Quick script to emulate a Switch connecting to a Joy-Con/Pro Controller.
Note: If you get an Invalid Exchange error when running this script, this means
that the Switch has paired to the controller, invalidating the original pairing
key we created. You'll need to remove the controller before continuing.
key we created. You'll need to re-pair the controller to this device.
"""
import socket
@ -115,7 +115,7 @@ def wait_for_reply(itr):
if __name__ == "__main__":
# Switch Controller Bluetooth MAC Address goes here
jc_MAC = "7C:BB:8A:FA:41:3D"
jc_MAC = "98:B6:E9:B0:05:E7"
port_ctrl = 17
port_itr = 19
@ -167,15 +167,6 @@ if __name__ == "__main__":
jc_itr.sendall(command)
wait_for_reply(jc_itr)
print("Here")
jc_itr.sendall(FLASH_PLAYER_LIGHTS)
count = 0
while count < 360:
data = jc_itr.recv(350)
print_msg_controller(data)
time.sleep(1/120)
count += 1
while True:
data = jc_itr.recv(350)
print_msg_controller(data)