diff --git a/demo.py b/demo.py
index 9c9cf2f..06abe7c 100644
--- a/demo.py
+++ b/demo.py
@@ -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
diff --git a/docs/Analog Stick and Button Input.md b/docs/Analog Stick and Button Input.md
new file mode 100644
index 0000000..c7e6d0b
--- /dev/null
+++ b/docs/Analog Stick and Button Input.md
@@ -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 Sticks1
+| 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 |
+
+1 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 =
+```
diff --git a/docs/Example Pairing Session.md b/docs/Example Pairing Session.md
index ebd8c5a..c4c1d3c 100644
--- a/docs/Example Pairing Session.md
+++ b/docs/Example Pairing Session.md
@@ -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.
diff --git a/docs/Miscellaneous Notes.md b/docs/Miscellaneous Notes.md
index 66583e1..2bd3bfe 100644
--- a/docs/Miscellaneous Notes.md
+++ b/docs/Miscellaneous Notes.md
@@ -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:
diff --git a/nxbt/bluez.py b/nxbt/bluez.py
index c41b468..6bb33e8 100644
--- a/nxbt/bluez.py
+++ b/nxbt/bluez.py
@@ -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
diff --git a/nxbt/controller/controller.py b/nxbt/controller/controller.py
index 8270059..7f8cd14 100644
--- a/nxbt/controller/controller.py
+++ b/nxbt/controller/controller.py
@@ -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)
diff --git a/nxbt/controller/input.py b/nxbt/controller/input.py
index ed5f5d6..a7d2d89 100644
--- a/nxbt/controller/input.py
+++ b/nxbt/controller/input.py
@@ -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)
diff --git a/nxbt/controller/protocol.py b/nxbt/controller/protocol.py
index 7ae1559..332a3f3 100644
--- a/nxbt/controller/protocol.py
+++ b/nxbt/controller/protocol.py
@@ -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
diff --git a/nxbt/controller/server.py b/nxbt/controller/server.py
index ef23d2c..cf213d3 100644
--- a/nxbt/controller/server.py
+++ b/nxbt/controller/server.py
@@ -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
diff --git a/nxbt/controller/utils.py b/nxbt/controller/utils.py
index 5216961..5387cd0 100644
--- a/nxbt/controller/utils.py
+++ b/nxbt/controller/utils.py
@@ -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
diff --git a/nxbt/nxbt.py b/nxbt/nxbt.py
index 4d6d693..4e9f4e2 100644
--- a/nxbt/nxbt.py
+++ b/nxbt/nxbt.py
@@ -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
+ })
diff --git a/nxbt/web/templates/index.html b/nxbt/web/templates/index.html
index b3b5317..f8fa295 100644
--- a/nxbt/web/templates/index.html
+++ b/nxbt/web/templates/index.html
@@ -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"],
}
}
diff --git a/scripts/proxy.py b/scripts/proxy.py
index 587a26c..e95b18c 100644
--- a/scripts/proxy.py
+++ b/scripts/proxy.py
@@ -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
diff --git a/test.py b/test.py
index d943c66..d320a88 100644
--- a/test.py
+++ b/test.py
@@ -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)