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

Added button functionality and info

Browse source
This commit is contained in:
Brikwerk 2020-05-31 16:30:52 -07:00
commit 21f7f42cc9
14 changed files with 666 additions and 182 deletions
Download patch file Download diff file Expand all files Collapse all files

50
demo.py
View file

@ -3,13 +3,57 @@ import time
from nxbt import Nxbt
from nxbt import ControllerTypes
MACRO = """
B 0.1s
0.1s
B 0.1s
0.1s
B 0.1s
0.1s
B 0.1s
1.5s
DPAD_LEFT 0.1s
0.1s
DPAD_LEFT 0.1s
0.1s
DPAD_LEFT 0.1s
0.1s
DPAD_LEFT 0.1s
0.1s
DPAD_LEFT 0.1s
0.1s
DPAD_RIGHT 0.075s
0.075s
DPAD_RIGHT 0.075s
0.075s
DPAD_RIGHT 0.075s
0.075s
A 0.1s
1.5s
A 0.1s
"""
if __name__ == "__main__":
nxbt = Nxbt()
adapters = nxbt.get_available_adapters()
index = nxbt.create_controller(
ControllerTypes.PRO_CONTROLLER, "/org/bluez/hci0")
ControllerTypes.PRO_CONTROLLER,
adapters[0],
colour_body=[0xFF, 0x7B, 0x83],
colour_buttons=[0xFF, 0xF0, 0x78])
index2 = nxbt.create_controller(
ControllerTypes.PRO_CONTROLLER,
adapters[1],
colour_body=[0xFF, 0xFF, 0xFF],
colour_buttons=[0xFF, 0xF0, 0x78])
nxbt.macro(index2, MACRO, block=False)
while True:
time.sleep(1)
print(nxbt.get_state()[0])
state = nxbt.state[0]
if not state["errors"]:
print(state["finished_macros"])
else:
print(state["errors"])
break

93
docs/Analog Stick and Button Input.md Normal file
View file

@ -0,0 +1,93 @@
# Analog Stick and Button Input Information
**Disclaimer:** The info within this document is sourced from the Switch reverse engineering
effort at [DekuNukem's Repository](https://github.com/dekuNukem/Nintendo_Switch_Reverse_Engineering).
The below sections contain info on the formulation and derivation of data
pertaining to the Nintendo Switch's controllers. The section on the analog
sticks contains info on encoding/decoding stick X/Y data, deadzones,
maximum range, etc. The button info section contains info on how each
button's state is communicated.
## Analog Stick Information
Information on a controller's analog sticks is stored in three primary
locations (user calibration excluded):
| Obtained From | Byte # | Data Type | Info |
| --- | --- | --- | --- | --- |
| Standard Input Report | 6-11 | 2 uint16 | Contains X/Y Data of Analog Sticks<sup>1</sup>
| SPI Flash Read (Offset 0x6080) | 13-30 | 12 uint16 LE | Dead Zone, Range ratio |
| SPI Flash Read (Offset 0x603D) | 7-24 | 12 uint16 LE | X/Y Min/Max and Centers |
<sup>1</sup> This data is relative, meaning that stick calibration data
*must* be used to encode/decode X and Y positions.
## Decoding a Stick's Position
**Note:** The following configuration values are used within Nxbt.
First, we use the data obtained from the 0x603D SPI flash read to
derive the right/left stick calibration parameters.
Sample data output by Nxbt:
```
Payload: 0xA1 0x21 0x2B 0x90 0x00 0x00 0x00 0x74 0x58 0x75 0x4B 0x68 0x7C 0x90
0 1 2 3 4 5 6 7 8 9 10 11 12 13
Subcommand: 0x90 0x10 0x3D 0x60 0x00 0x00 0x19 0xBA 0xF5 0x62 0x6F 0xC8 0x77 0xED
14 15 16 17 18 19 20 21 22 23 24 25 26 27
0x95 0x5B 0x16 0xD8 0x7D 0xF2 0xB5 0x5F 0x86 0x65 0x5E 0xFF 0x82 0x82
28 29 30 31 32 33 34 35
0x82 0x0F 0x0F 0x0F 0x00 0x00 0x00 0x00
```
Which gives us:
```
Left Stick: 0xBA 0xF5 0x62 0x6F 0xC8 0x77 0xED 0x95 0x5B
Right Stick: 0x16 0xD8 0x7D 0xF2 0xB5 0x5F 0x86 0x65 0x5E
```
Using the following equations, we can decode these values into meaningful ones.
Each stick's data is treated as an array of byte values for the equations.
```
# The nine stick bytes are labelled stick_cal[0] - stick_cal[8] here
uint16_t data[6]
data[0] = (stick_cal[1] << 8) & 0xF00 | stick_cal[0];
data[1] = (stick_cal[2] << 4) | (stick_cal[1] >> 4);
data[2] = (stick_cal[4] << 8) & 0xF00 | stick_cal[3];
data[3] = (stick_cal[5] << 4) | (stick_cal[4] >> 4);
data[4] = (stick_cal[7] << 8) & 0xF00 | stick_cal[6];
data[5] = (stick_cal[8] << 4) | (stick_cal[7] >> 4);
# These values used as such in, for example, a right stick
uint16_t rstick_center_x = data[0];
uint16_t rstick_center_y = data[1];
uint16_t rstick_x_min = rstick_center_x - data[2];
uint16_t rstick_x_max = rstick_center_x + data[4];
uint16_t rstick_y_min = rstick_center_y - data[3];
uint16_t rstick_y_max = rstick_center_y + data[5];
```
Resulting in the following values for the sticks:
```
Left Stick
~~~~~~~~~~
Center X =
Center Y =
X Min =
X Max =
Y Min =
Y Max =
Right Stick
~~~~~~~~~~~
Center X =
Center Y =
X Min =
X Max =
Y Min =
Y Max =
```

5
docs/Example Pairing Session.md
View file

@ -748,12 +748,13 @@ Subcommand: 0x90 0x10 0x3D 0x60 0x00 0x00 0x19 0xD5 0xA4 0x43 0xA4 0xE7 0x87 0x2
| 1 | Subcommand reply |
| 5-2 | SPI memory address (Little Endian) |
| 6 | Read length |
| 7-25 | Stick factory calibration |
| 7-24 | Stick factory calibration |
| 25 | Spacer byte (always 0xFF) |
| 26-31 | Controller colours |
#### Stick Factory Calibration
Byte 7-15 comprises the left stick calibration data. Byte 16-25 comprises the
Byte 7-15 comprises the left stick calibration data. Byte 16-24 comprises the
right stick data. These 9 bytes are used to generate 6 uint16 values which are
used to calculate the stick dead zones and min/max x/y values.

4
docs/Miscellaneous Notes.md
View file

@ -7,7 +7,9 @@ this project.
- Controller SDP record. They all share the same one (generally), so a only a single record is needed to emulate all three controllers
- The Bluetooth alias "Joy-Con (L)", "Joy-Con (R)", or "Pro Controller"
- The Bluetooth Gamepad HID Class
**Note:** Setting the device's major and minor class is *not* required
to get the Switch to connect. Only the alias and SDP record are required.
### Weird Tibit:

132
nxbt/bluez.py
View file

@ -89,19 +89,75 @@ def find_objects(bus, service_name, interface_name):
return paths
def toggle_input_plugin(toggle):
"""Enables or disables the BlueZ input plugin. Requires
root user to be run. The units and Bluetooth service will
not be restarted if the input plugin already matches
the toggle.
:param toggle: A boolean element indicating if the plugin
is enabled (True) or disabled (False)
:type toggle: boolean
:raises PermissionError: If the user is not root
:raises Exception: If the units can't be reloaded
"""
if os.geteuid() != 0:
raise PermissionError("The input plugin must be toggled as root")
service_path = "/lib/systemd/system/bluetooth.service"
service = None
with open(service_path, "r") as f:
service = f.read()
# Find the bluetooth service execution line
lines = service.split("\n")
for i in range(0, len(lines)):
line = lines[i]
if line.startswith("ExecStart="):
# If we want to ensure the plugin is enabled
if toggle:
# If input is already enabled
if "--noplugin=input" not in line:
return
lines[i] = re.sub(" --noplugin=input", "", line)
else:
# If input is already disabled
if "--noplugin=input" in line:
return
# If not, add the flag
lines[i] = line + " --noplugin=input"
service = "\n".join(lines)
with open(service_path, "w") as f:
f.write(service)
# Reload units
result = subprocess.run(
["systemctl", "daemon-reload"],
stderr=subprocess.PIPE)
cmd_err = result.stderr.decode("utf-8").replace("\n", "")
if cmd_err != "":
raise Exception(cmd_err)
# Reload the bluetooth service with input disabled
result = subprocess.run(
["systemctl", "restart", "bluetooth"],
stderr=subprocess.PIPE)
cmd_err = result.stderr.decode("utf-8").replace("\n", "")
if cmd_err != "":
raise Exception(cmd_err)
class BlueZ():
"""Exposes the BlueZ D-Bus API as a Python object.
"""
def __init__(self, device_id="hci0"):
def __init__(self, adapter_path="/org/bluez/hci0"):
self.bus = dbus.SystemBus()
# Try to find the default adapter (hci0) or a user specified adapter
self.device_path = find_object_path(
self.bus,
SERVICE_NAME,
ADAPTER_INTERFACE,
object_name=device_id)
self.device_path = adapter_path
# If we weren't able to find an adapter with the specified ID,
# try to find any usable Bluetooth adapter
@ -123,10 +179,7 @@ class BlueZ():
self.device_path),
"org.freedesktop.DBus.Properties")
if device_id:
self.device_id = device_id
else:
self.device_id = self.device_path.split("/")[-1]
self.device_id = self.device_path.split("/")[-1]
# Load the ProfileManager interface
self.profile_manager = dbus.Interface(self.bus.get_object(
@ -400,61 +453,6 @@ class BlueZ():
BLUEZ_OBJECT_PATH),
PROFILEMANAGER_INTERFACE)
def toggle_input_plugin(self, toggle):
"""Enables or disables the BlueZ input plugin. Requires
root user to be run. The units and Bluetooth service will
not be restarted if the input plugin already matches
the toggle.
:param toggle: A boolean element indicating if the plugin
is enabled (True) or disabled (False)
:type toggle: boolean
:raises PermissionError: If the user is not root
:raises Exception: If the units can't be reloaded
"""
if os.geteuid() != 0:
raise PermissionError("The input plugin must be toggled as root")
service_path = "/lib/systemd/system/bluetooth.service"
service = None
with open(service_path, "r") as f:
service = f.read()
# Find the bluetooth service execution line
lines = service.split("\n")
for i in range(0, len(lines)):
line = lines[i]
if line.startswith("ExecStart="):
# If we want to ensure the plugin is enabled
if toggle:
# If input is already enabled
if "--noplugin=input" not in line:
return
lines[i] = re.sub(" --noplugin=input", "", line)
else:
# If input is already disabled
if "--noplugin=input" in line:
return
# If not, add the flag
lines[i] = line + " --noplugin=input"
service = "\n".join(lines)
with open(service_path, "w") as f:
f.write(service)
# Reload units
result = subprocess.run(
["systemctl", "daemon-reload"],
stderr=subprocess.PIPE)
cmd_err = result.stderr.decode("utf-8").replace("\n", "")
if cmd_err != "":
raise Exception(cmd_err)
# Reload the bluetooth service with input disabled
self.reset()
def get_discovered_devices(self):
"""Gets a dict of all discovered (or previously discovered
and connected) devices. The key is the device's dbus object

2
nxbt/controller/controller.py
View file

@ -66,4 +66,4 @@ class Controller():
except dbus.exceptions.DBusException:
pass
self.bt.set_device_class(self.GAMEPAD_CLASS)
# self.bt.set_device_class(self.GAMEPAD_CLASS)

150
nxbt/controller/input.py
View file

@ -1,4 +1,150 @@
from time import perf_counter
class InputParser():
def __init__():
print("")
def __init__(self, protocol):
self.protocol = protocol
# Buffers a list of unparsed macros
self.macro_buffer = []
# Keeps track of the entire current
# list of macro commands.
self.current_macro = None
self.current_macro_id = None
# Keeps track of the macro commands being
# input over a period of time.
self.current_macro_commands = None
# The time length of the current macro
self.macro_timer_length = 0
# The start time for the current macro commands
self.macro_timer_start = 0
self.controller_input = None
def buffer_macro(self, macro, macro_id):
# Doesn't have any info
if len(macro) < 4:
return
self.macro_buffer.append([macro, macro_id])
def set_controller_input(self, controller_input):
self.controller_input = controller_input
def set_protocol_input(self, state=None):
if self.controller_input:
self.parse_controller_input(self.controller_input)
self.controller_input = None
elif (self.macro_buffer or self.current_macro or
self.current_macro_commands):
# Check if we can start on a new macro.
if not self.current_macro and self.macro_buffer:
# Preprocess command lines of current macro
macro = self.macro_buffer.pop(0)
self.current_macro = macro[0].strip("\n")
self.current_macro = self.current_macro.split("\n")
self.current_macro_id = macro[1]
# Check if we can load the next set of commands
if not self.current_macro_commands and self.current_macro:
self.current_macro_commands = (
self.current_macro.pop(0).strip(" ").split(" "))
# Timing metadata extraction
timer_length = self.current_macro_commands[-1]
timer_length = timer_length[0:len(timer_length)-1]
self.macro_timer_length = float(timer_length)
self.macro_timer_start = perf_counter()
self.parse_macro_input(self.current_macro_commands)
# Check if we're done inputting the current command
time_delta = perf_counter() - self.macro_timer_start
if time_delta > self.macro_timer_length:
self.current_macro_commands = None
# Check if we're done the current macro
if not self.current_macro and state:
finished = state["finished_macros"]
finished.append(self.current_macro_id)
state["finished_macros"] = finished
def parse_controller_input(self, controller_input):
return controller_input
def parse_macro_input(self, macro_input):
# Checking if this is a wait macro command
if len(macro_input) < 2:
print("waiting")
return
# Arrays representing the 3 button bytes in the
# standard input report as binary.
upper = ['0'] * 8
shared = ['0'] * 8
lower = ['0'] * 8
for i in range(0, len(macro_input)-1):
button = macro_input[i]
# Upper Byte
if button == "Y":
upper[7] = '1'
elif button == "X":
upper[6] = '1'
elif button == "B":
upper[5] = '1'
elif button == "A":
upper[4] = '1'
elif button == "SR":
upper[3] = '1'
elif button == "SL":
upper[2] = '1'
elif button == "R":
upper[1] = '1'
elif button == "ZR":
upper[0] = '1'
# Shared byte
elif button == "-":
shared[7] = '1'
elif button == "+":
shared[6] = '1'
elif button == "R_ANALOG_DOWN":
shared[5] = '1'
elif button == "L_ANALOG_DOWN":
shared[4] = '1'
elif button == "HOME":
shared[3] = '1'
elif button == "CAPTURE":
shared[2] = '1'
# Lower byte
elif button == "DPAD_DOWN":
lower[7] = '1'
elif button == "DPAD_UP":
lower[6] = '1'
elif button == "DPAD_RIGHT":
lower[5] = '1'
elif button == "DPAD_LEFT":
lower[4] = '1'
elif button == "SR":
lower[3] = '1'
elif button == "SL":
lower[2] = '1'
elif button == "L":
lower[1] = '1'
elif button == "ZL":
lower[0] = '1'
# 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)

40
nxbt/controller/protocol.py
View file

@ -42,7 +42,8 @@ class ControllerProtocol():
}
VIBRATOR_BYTES = [0xA0, 0xB0, 0xC0, 0x90]
def __init__(self, controller_type, bt_address, report_size=50):
def __init__(self, controller_type, bt_address, report_size=50,
colour_body=None, colour_buttons=None):
"""Initializes the protocol for the controller.
:param controller_type: The type of controller (Joy-Con (L),
@ -52,6 +53,12 @@ class ControllerProtocol():
:type bt_address: string
:param report_size: The size of the protocol report, defaults to 50
:type report_size: int, optional
:param colour_body: Sets the body colour of the controller, defaults
to None
:type colour_body: list of bytes, optional
:param colour_buttons: Sets the colour of the controller buttons,
defaults to None
:type colour_buttons: list of bytes, optional
:raises ValueError: On unknown controller type
"""
@ -107,8 +114,14 @@ class ControllerProtocol():
# Controller colours
# Body Colour
self.colour_body = [0x82] * 3
self.colour_buttons = [0x0F] * 3
if not colour_body:
self.colour_body = [0x82] * 3
else:
self.colour_body = colour_body
if not colour_buttons:
self.colour_buttons = [0x0F] * 3
else:
self.colour_buttons = colour_buttons
def get_report(self):
@ -167,9 +180,11 @@ class ControllerProtocol():
# Bad Packet handling statements
elif message.response == SwitchResponses.UNKNOWN_SUBCOMMAND:
# Currently set so that the controller ignores any unknown
# subcommands. This is better than sending a NACK response
# since we'd just get stuck in an infinite loop arguing
# with the Switch.
self.set_full_input_report()
# self.set_subcommand_reply()
# self.set_unknown_subcommand(message.subcommand_id)
elif message.response == SwitchResponses.NO_DATA:
self.set_full_input_report()
@ -257,6 +272,14 @@ class ControllerProtocol():
self.report[13] = self.vibrator_report
def set_button_inputs(self, upper, shared, lower):
self.report[4] = upper
self.report[5] = shared
self.report[6] = lower
print(self.report)
def set_device_info(self):
# ACK Reply
@ -436,13 +459,16 @@ class ControllerProtocol():
else:
replace_subarray(self.report, 30, 9, value=0xFF)
# Spacer byte
self.report[39] = 0xFF
# Body colour
replace_subarray(
self.report, 39, 3,
self.report, 40, 3,
replace_arr=self.colour_body)
# Buttons colour
replace_subarray(
self.report, 42, 3,
self.report, 43, 3,
replace_arr=self.colour_buttons)
# Six-Axis motion sensor factor calibration

182
nxbt/controller/server.py
View file

@ -2,26 +2,41 @@ import socket
import fcntl
import os
import time
import traceback
import queue
from .controller import Controller, ControllerTypes
from ..bluez import BlueZ
from .protocol import ControllerProtocol
from .input import InputParser
from .utils import format_msg_controller, format_msg_switch
class ControllerServer():
def __init__(self, controller_type, bt_device_id="hci0"):
def __init__(self, controller_type, adapter_path="/org/bluez/hci0",
lock=None, colour_body=None, colour_buttons=None):
self.controller_type = controller_type
self.colour_body = colour_body
self.colour_buttons = colour_buttons
if lock:
self.lock = lock
self.reconnect_counter = 0
# Intializing Bluetooth
self.bt = BlueZ(device_id=bt_device_id)
self.bt = BlueZ(adapter_path=adapter_path)
self.controller = Controller(self.bt, self.controller_type)
self.protocol = ControllerProtocol(
self.controller_type,
self.bt.address)
self.bt.address,
colour_body=self.colour_body,
colour_buttons=self.colour_buttons)
self.input = InputParser(self.protocol)
def run(self, reconnect_address=None, state=None, task_queue=None):
"""Runs the mainloop of the controller server.
@ -30,50 +45,108 @@ class ControllerServer():
previously connected to Nintendo Switch, defaults to None
:type reconnect_address: string, optional
"""
print(reconnect_address, state, task_queue)
if state:
state["state"] = "initializing"
self.controller.setup()
if reconnect_address:
itr, s_itr, ctrl, s_ctrl = self.reconnect(
reconnect_address, state=state)
else:
itr, s_itr, ctrl, s_ctrl = self.connect(state=state)
if state:
state["state"] = "connected"
# Mainloop
while True:
# Attempt to get output from Switch
try:
# If we have a lock, prevent other controllers
# from initializing at the same time and saturating
# the DBus
if self.lock:
self.lock.acquire()
try:
reply = itr.recv(50)
if len(reply) > 40:
print(format_msg_switch(reply))
except BlockingIOError:
reply = None
self.controller.setup()
self.protocol.process_commands(reply)
msg = self.protocol.get_report()
if reconnect_address:
itr, ctrl = self.reconnect(reconnect_address, state=state)
else:
itr, ctrl = self.connect(state=state)
except Exception:
if self.lock:
self.lock.release()
if reply:
print(format_msg_controller(msg))
self.switch_address = itr.getsockname()[0]
try:
itr.sendall(msg)
except BlockingIOError:
continue
if state:
state["state"] = "connected"
# Respond at 120Hz for Pro Controller
# or 60Hz for Joy-Cons
if self.controller_type == ControllerTypes.PRO_CONTROLLER:
time.sleep(1/120)
# Mainloop
while True:
# Attempt to get output from Switch
try:
reply = itr.recv(50)
if len(reply) > 40:
print(format_msg_switch(reply))
except BlockingIOError:
reply = None
# Getting any inputs from the task queue
if task_queue:
try:
msg = task_queue.get_nowait()
print(msg)
if msg:
self.input.buffer_macro(
msg["macro"], msg["macro_id"])
except queue.Empty:
pass
self.protocol.process_commands(reply)
self.input.set_protocol_input(state=state)
msg = self.protocol.get_report()
if reply:
print(format_msg_controller(msg))
try:
itr.sendall(msg)
except BlockingIOError:
continue
except OSError as e:
# Attempt to reconnect to the Switch
if self.reconnect_counter < 2:
try:
print("Attempting to reconnect")
# Reinitialize the protocol
self.protocol = ControllerProtocol(
self.controller_type,
self.bt.address,
colour_body=self.colour_body,
colour_buttons=self.colour_buttons)
itr, ctrl = self.reconnect(self.switch_address,
state=state)
except OSError:
self.reconnect_counter += 1
print(e)
time.sleep(0.5)
continue
# If we can't reconnect, transition to attempting
# to connect to any Switch.
else:
print("Connecting")
# Reinitialize the protocol
self.protocol = ControllerProtocol(
self.controller_type,
self.bt.address,
colour_body=self.colour_body,
colour_buttons=self.colour_buttons)
itr, ctrl = self.connect(state=state)
self.switch_address = itr.getsockname()[0]
# Respond at 120Hz for Pro Controller
# or 60Hz for Joy-Cons
if self.controller_type == ControllerTypes.PRO_CONTROLLER:
time.sleep(1/120)
else:
time.sleep(1/60)
except Exception as e:
if state:
state["state"] = "crashed"
state["errors"] = traceback.format_exc()
else:
time.sleep(1/60)
raise e
def connect(self, state=None):
"""Configures as a specified controller, pairs with a Nintendo Switch,
@ -132,8 +205,8 @@ class ControllerServer():
self.protocol.process_commands(reply)
msg = self.protocol.get_report()
# if reply:
# print(format_msg_controller(msg))
if reply:
print(format_msg_controller(msg))
try:
itr.sendall(msg)
@ -148,7 +221,7 @@ class ControllerServer():
# Pairing cycle responds optimally on a 15Hz loop
time.sleep(1/15)
return itr, s_itr, ctrl, s_ctrl
return itr, ctrl
def reconnect(self, reconnect_address, state=None):
"""Attempts to reconnect with a Switch at the given address.
@ -160,31 +233,30 @@ class ControllerServer():
if state:
state["state"] = "reconnecting"
device_path = self.bt.find_device_by_address(reconnect_address)
if not device_path:
raise ValueError(
"No device Switch found with MAC address " + reconnect_address)
# Creating control and interrupt sockets
s_ctrl = socket.socket(
ctrl = socket.socket(
family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
s_itr = socket.socket(
itr = socket.socket(
family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
# Setting up HID interrupt/control sockets
s_ctrl.bind((self.bt.address, 17))
s_itr.bind((self.bt.address, 19))
ctrl.connect((reconnect_address, 17))
itr.connect((reconnect_address, 19))
s_itr.listen(1)
s_ctrl.listen(1)
fcntl.fcntl(itr, fcntl.F_SETFL, os.O_NONBLOCK)
self.bt.connect_device(device_path)
# Send an empty input report to the Switch to prompt a reply
self.protocol.process_commands(None)
msg = self.protocol.get_report()
itr.sendall(msg)
ctrl, ctrl_address = s_ctrl.accept()
itr, itr_address = s_itr.accept()
# Setting interrupt connection as non-blocking
# In this case, non-blocking means it throws a "BlockingIOError"
# for sending and receiving, instead of blocking
fcntl.fcntl(itr, fcntl.F_SETFL, os.O_NONBLOCK)
print("Here")
return itr, ctrl

16
nxbt/controller/utils.py
View file

@ -1,4 +1,20 @@
def replace_subarray(arr, start, num_elms, value=0, replace_arr=None):
"""Replaces a subsection within an array with another
set of values.
:param arr: The array to replace values within
:type arr: list
:param start: The starting index for replacement
:type start: int
:param num_elms: The number of elements to be replaced
:type num_elms: int
:param value: The value to replace elements within the
subarray with, defaults to 0
:type value: any, optional
:param replace_arr: A subarray to insert within
the passed array, defaults to None
:type replace_arr: list, optional
"""
if replace_arr:
arr[start:start + num_elms] = replace_arr

108
nxbt/nxbt.py
View file

@ -3,16 +3,19 @@ from multiprocessing import Queue, Manager
import queue
from enum import Enum
import atexit
import os
import dbus
from .controller import ControllerServer
from .bluez import find_objects, SERVICE_NAME, ADAPTER_INTERFACE
from .bluez import find_objects, toggle_input_plugin
from .bluez import SERVICE_NAME, ADAPTER_INTERFACE
class NxbtCommands(Enum):
CREATE_CONTROLLER = 0
INPUT_MACRO = 1
class Nxbt():
@ -22,12 +25,16 @@ class Nxbt():
# Main queue for nbxt tasks
self.task_queue = Queue()
# Sychronizes bluetooth actions
self.__bluetooth_lock = Lock()
# Creates/manages shared resources
self.resource_manager = Manager()
# Shared dictionary for viewing overall nxbt state.
# Should only be read by threads and wrote to by
# the main nxbt multiprocessing process.
self.state = self.resource_manager.dict()
self.manager_state = self.resource_manager.dict()
self.manager_state_lock = Lock()
# Shared, controller management properties.
# The controller lock is used to sychronize use.
@ -35,13 +42,17 @@ class Nxbt():
self.__controller_counter = 0
self.__adapters_in_use = []
# Disable the BlueZ input plugin so we can use the
# HID control/interrupt Bluetooth ports
toggle_input_plugin(False)
# Exit handler
atexit.register(self.on_exit)
# Starting the nxbt worker process
self.controllers = Process(
target=self.__command_manager,
args=((self.task_queue), (self.state)))
args=((self.task_queue), (self.manager_state)))
# Disabling daemonization since we need to spawn
# other controller processes, however, this means
# we need to cleanup on exit.
@ -55,9 +66,12 @@ class Nxbt():
if hasattr(self, "controllers") and self.controllers.is_alive():
self.controllers.terminate()
# Re-enable the BlueZ input plugin
toggle_input_plugin(True)
def __command_manager(self, task_queue, state):
cm = ControllerManager(state)
cm = ControllerManager(state, self.__bluetooth_lock)
while True:
try:
@ -70,13 +84,43 @@ class Nxbt():
cm.create_controller(
msg["arguments"]["controller_index"],
msg["arguments"]["controller_type"],
msg["arguments"]["adapter_path"])
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"])
def send_input(self, msg):
def macro(self, controller_index, macro, block=True):
self.task_queue.put(msg)
if controller_index not in self.manager_state.keys():
raise ValueError("Specified controller does not exist")
def create_controller(self, controller_type, adapter_path):
# Get a unique ID to identify the macro
# 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 create_controller(self, controller_type, adapter_path, block=True,
colour_body=None, colour_buttons=None):
if adapter_path not in self.get_available_adapters():
raise ValueError("Specified adapter is unavailable")
@ -93,11 +137,21 @@ class Nxbt():
"controller_index": self.__controller_counter,
"controller_type": controller_type,
"adapter_path": adapter_path,
"colour_body": colour_body,
"colour_buttons": colour_buttons,
}
})
controller_index = self.__controller_counter
self.__controller_counter += 1
self.__adapters_in_use.append(adapter_path)
if type(controller_index) == int:
while True:
if controller_index in self.manager_state.keys():
state = self.manager_state[controller_index]
if (state["state"] == "connecting" or
state["state"] == "reconnecting"):
break
finally:
self.__controller_lock.release()
pass
@ -111,21 +165,23 @@ class Nxbt():
return adapters
def get_state(self):
@property
def state(self):
return self.state
return self.manager_state
class ControllerManager():
def __init__(self, state):
def __init__(self, state, lock):
self.state = state
self.lock = lock
self.controller_resources = Manager()
self.controller_states = []
self.controller_queues = []
self.__controller_queues = {}
def create_controller(self, index, controller_type, adapter_path):
def create_controller(self, index, controller_type, adapter_path,
colour_body=None, colour_buttons=None):
controller_queue = Queue()
@ -134,12 +190,28 @@ class ControllerManager():
controller_state["finished_macros"] = []
controller_state["errors"] = False
self.state[index] = controller_state
# Get the last parameter of the path, AKA the ID
device_id = adapter_path.split("/")[-1]
self.__controller_queues[index] = controller_queue
server = ControllerServer(controller_type, bt_device_id=device_id)
self.state[index] = controller_state
server = ControllerServer(controller_type,
adapter_path=adapter_path,
lock=self.lock,
colour_body=colour_body,
colour_buttons=colour_buttons)
controller = Process(target=server.run, args=(
None, controller_state, controller_queue))
controller.daemon = True
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({
"type": "macro",
"macro": macro,
"macro_id": macro_id
})

2
nxbt/web/templates/index.html
View file

@ -57,7 +57,7 @@
"DPAD_LEFT": gp["buttons"][14]["pressed"],
"DPAD_RIGHT": gp["buttons"][15]["pressed"],
"HOME": gp["buttons"][16]["pressed"],
"SCREENSHOT": gp["buttons"][17]["pressed"],
"CAPTURE": gp["buttons"][17]["pressed"],
}
}

20
scripts/proxy.py
View file

@ -12,6 +12,7 @@ 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 ControllerTypes
@ -92,15 +93,22 @@ def write_to_buffer(buffer, message, message_type):
if __name__ == "__main__":
# Switch Controller Bluetooth MAC Address goes here
jc_MAC = "XX:XX:XX:XX:XX:XX"
jc_MAC = "7C:BB:8A:FA:41:3D"
# Specify the type of controller here
controller_type = ControllerTypes.JOYCON_L
controller_type = ControllerTypes.PRO_CONTROLLER
if controller_type == ControllerTypes.JOYCON_L:
REPLY = JCL_REPLY02
elif controller_type == ControllerTypes.JOYCON_R:
REPLY = JCR_REPLY02
else:
REPLY = PRO_REPLY02
port_ctrl = 17
port_itr = 19
message_buffer = []
bt = BlueZ()
bt.toggle_input_plugin(False)
toggle_input_plugin(False)
controller = Controller(bt, controller_type)
@ -196,11 +204,11 @@ if __name__ == "__main__":
# Sending Switch the proxy's device info
if controller_type == ControllerTypes.JOYCON_R:
client_interrupt.sendall(JCR_REPLY02)
client_interrupt.sendall(REPLY)
elif controller_type == ControllerTypes.JOYCON_L:
client_interrupt.sendall(JCL_REPLY02)
client_interrupt.sendall(REPLY)
elif controller_type == ControllerTypes.PRO_CONTROLLER:
client_interrupt.sendall(PRO_REPLY02)
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

44
test.py
View file

@ -1,23 +1,29 @@
import time
from ctypes import c_uint16
from nxbt import ControllerTypes
from nxbt import ControllerProtocol
# Left Stick Calibration
stick_cal = [0xBA, 0xF5, 0x62, 0x6F, 0xC8, 0x77, 0xED, 0x95, 0x5B]
# Right Stick Calibration
stick_cal = [0x16, 0xD8, 0x7D, 0xF2, 0xB5, 0x5F, 0x86, 0x65, 0x5E]
data = [0] * 6
# The nine stick bytes are labelled stick_cal[0] - stick_cal[8] here
data[0] = (stick_cal[1] << 8) & 0xF00 | stick_cal[0]
data[1] = (stick_cal[2] << 4) | (stick_cal[1] >> 4)
data[2] = (stick_cal[4] << 8) & 0xF00 | stick_cal[3]
data[3] = (stick_cal[5] << 4) | (stick_cal[4] >> 4)
data[4] = (stick_cal[7] << 8) & 0xF00 | stick_cal[6]
data[5] = (stick_cal[8] << 4) | (stick_cal[7] >> 4)
INPUT_REPORT = b'\xa2\x01\x0E\x00\x00\x00\x00\x00\x00\x00\x00\x02\x80\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
# These values used as such in, for example, a right stick
center_x = data[0]
center_y = data[1]
x_min = c_uint16(center_x - data[2])
x_max = c_uint16(center_x + data[4])
y_min = c_uint16(center_y - data[3])
y_max = c_uint16(center_y + data[5])
center_x = c_uint16(data[0])
center_y = c_uint16(data[1])
protocol = ControllerProtocol(
ControllerTypes.JOYCON_L,
"AA:AA:AA:AA:AA:AA")
protocol.process_commands(None)
print(hex(protocol.get_report()[2]))
time.sleep(1)
protocol.process_commands(None)
print(hex(protocol.get_report()[2]))
protocol.process_commands(INPUT_REPORT)
print(hex(protocol.get_report()[2]))
time.sleep(1)
protocol.process_commands(None)
print(hex(protocol.get_report()[2]))
protocol.process_commands(None)
print(hex(protocol.get_report()[2]))
print("Center X and Y", center_x, center_y)
print("X Min/Max", x_min, x_max)
print("Y Min/Max", y_min, y_max)