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

Added stick input and loops to macros

Browse source
This commit is contained in:
Brikwerk 2020-06-08 21:23:23 -07:00
commit db4b550166
10 changed files with 384 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

67
demo.py
View file

@ -1,8 +1,7 @@
import time
from random import randint
from nxbt import Nxbt
from nxbt import ControllerTypes
import nxbt
MACRO = """
B 0.1s
@ -13,25 +12,44 @@ 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
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
0.1s
"""
@ -48,25 +66,26 @@ if __name__ == "__main__":
# Loop over all Bluetooth adapters and create
# Switch Pro Controllers
nxbt = Nxbt()
adapters = nxbt.get_available_adapters()
# adapters = ["/org/bluez/hci0"]
nx = nxbt.Nxbt()
adapters = nx.get_available_adapters()
controller_idxs = []
for i in range(0, len(adapters)):
index = nxbt.create_controller(
ControllerTypes.PRO_CONTROLLER,
index = nx.create_controller(
nxbt.PRO_CONTROLLER,
adapters[i],
colour_body=random_colour(),
colour_buttons=random_colour())
controller_idxs.append(index)
# Run a macro on the last controller
nxbt.macro(controller_idxs[-1], MACRO, block=False)
# 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 nxbt.state.keys():
state = nxbt.state[key]
for key in nx.state.keys():
state = nx.state[key]
if not state["errors"]:
print(state)
else:

4
nxbt/__init__.py
View file

@ -1,8 +1,10 @@
from .controller import ControllerServer
from .controller import ControllerTypes
from .controller import ControllerProtocol
from .controller import SwitchReportParser
from .controller import SwitchResponses
from .controller import Controller
from .bluez import *
from .nxbt import Nxbt
from .nxbt import JOYCON_L
from .nxbt import JOYCON_R
from .nxbt import PRO_CONTROLLER

121
nxbt/cli.py
View file

@ -1,5 +1,124 @@
import argparse
import time
from random import randint
from .web import start_web_app
from .nxbt import Nxbt, PRO_CONTROLLER
parser = argparse.ArgumentParser()
parser.add_argument('command', default=False, choices=['start', 'demo'],
help="Specifies the Nxbt command to run")
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
A 0.1s
3.0s
20.0s
B 0.1s
LOOP 100
B 0.1s
0.1s
"""
def random_colour():
return [
randint(0, 255),
randint(0, 255),
randint(0, 255),
]
def demo():
"""Loops over all available Bluetooth adapters
and creates controllers on each. The last available adapter
is used to run a macro.
"""
nx = Nxbt()
adapters = nx.get_available_adapters()
controller_idxs = []
for i in range(0, len(adapters)):
index = nx.create_controller(
PRO_CONTROLLER,
adapters[i],
colour_body=random_colour(),
colour_buttons=random_colour())
controller_idxs.append(index)
# 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"])
def main():
start_web_app()
if args.command == 'start':
start_web_app()
elif args.command == 'demo':
demo()

161
nxbt/controller/input.py
View file

@ -3,12 +3,34 @@ from time import perf_counter
class InputParser():
# Left Stick calibration values
LEFT_STICK_CALIBRATION = {
"center_x": 2159,
"center_y": 1916,
# Zeroed Min/Max X and Y
"min_x": -1466,
"max_x": 1517,
"min_y": -1583,
"max_y": 1465,
}
# Right Stick calibration values
RIGHT_STICK_CALIBRATION = {
"center_x": 2070,
"center_y": 2013,
# Zeroed Min/Max X and Y
"min_x": -1522,
"max_x": 1414,
"min_y": -1531,
"max_y": 1510,
}
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
@ -16,8 +38,10 @@ class InputParser():
# 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
@ -47,8 +71,7 @@ class InputParser():
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 = self.parse_macro(macro[0])
self.current_macro_id = macro[1]
# Check if we can load the next set of commands
@ -62,6 +85,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)
# Check if we're done inputting the current command
@ -78,6 +102,57 @@ class InputParser():
return controller_input
def parse_macro(self, macro):
parsed = macro.split("\n")
parsed = list(filter(lambda s: not s.strip() == "", parsed))
parsed = self.parse_loops(parsed)
return parsed
def parse_loops(self, macro):
parsed = []
i = 0
while i < len(macro):
line = macro[i]
if line.startswith("LOOP"):
loop_count = int(line.split(" ")[1])
loop_buffer = []
# Detect delimiter and record
if macro[i+1].startswith("\t"):
loop_delimiter = "\t"
elif macro[i+1].startswith(" "):
loop_delimiter = " "
else:
loop_delimiter = " "
# Gather looping commands
for j in range(i+1, len(macro)):
loop_line = macro[j]
if loop_line.startswith(loop_delimiter):
# Replace the first instance of the delimiter
loop_line = loop_line.replace(loop_delimiter, "", 1)
loop_buffer.append(loop_line)
# Set the new position if we either encounter the end
# of the loop or we reach the end of the macro
else:
i = j - 1
break
if j+1 >= len(macro):
i = j
# Recursively gather other loops if present
if any(s.startswith("LOOP") for s in loop_buffer):
loop_buffer = self.parse_loops(loop_buffer)
# Multiply out the loop and concatenate
parsed = parsed + (loop_buffer * loop_count)
else:
parsed.append(line)
i += 1
return parsed
def parse_macro_input(self, macro_input):
# Checking if this is a wait macro command
@ -89,6 +164,9 @@ class InputParser():
upper = ['0'] * 8
shared = ['0'] * 8
lower = ['0'] * 8
# Analog stick byte placeholders
stick_left = None
stick_right = None
for i in range(0, len(macro_input)-1):
button = macro_input[i]
# Upper Byte
@ -114,9 +192,9 @@ class InputParser():
shared[7] = '1'
elif button == "+":
shared[6] = '1'
elif button == "R_ANALOG_DOWN":
elif button == "R_STICK_PRESS":
shared[5] = '1'
elif button == "L_ANALOG_DOWN":
elif button == "L_STICK_PRESS":
shared[4] = '1'
elif button == "HOME":
shared[3] = '1'
@ -141,9 +219,84 @@ class InputParser():
elif button == "ZL":
lower[0] = '1'
# Analog Stick Positions
elif button.startswith("L_STICK@"):
stick_left = self.parse_macro_stick_position(button)
elif button.startswith("R_STICK@"):
stick_right = self.parse_macro_stick_position(button)
# 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)
if stick_left:
self.protocol.set_left_stick_inputs(stick_left)
if stick_right:
self.protocol.set_right_stick_inputs(stick_right)
def parse_macro_stick_position(self, stick_pos):
stick_type = stick_pos.split("@")[0]
positions = stick_pos.split("@")[1]
if len(positions) < 8:
return None
# Converting macro to proper ratios
sign_x = positions[0]
ratio_x = int(positions[1:4]) / 100
if sign_x == "-":
ratio_x = ratio_x * -1
sign_y = positions[4]
ratio_y = int(positions[5:8]) / 100
if sign_y == "-":
ratio_y = ratio_y * -1
calibrated_position = self.stick_ratio_to_calibrated_position(
ratio_x, ratio_y, stick_type)
return calibrated_position
def stick_ratio_to_calibrated_position(self, ratio_x, ratio_y, stick_type):
# Using the appropriate calibration values for the stick type
if stick_type == "L_STICK":
cal = self.LEFT_STICK_CALIBRATION
else:
cal = self.RIGHT_STICK_CALIBRATION
# Converting ratios to uint16 values
if ratio_x < 0:
data_x_converted = (
abs(ratio_x) * cal["min_x"] + cal["center_x"])
else:
data_x_converted = (
abs(ratio_x) * cal["max_x"] + cal["center_x"])
data_x_converted = int(round(data_x_converted))
if ratio_y < 0:
data_y_converted = (
abs(ratio_y) * cal["min_y"] + cal["center_y"])
else:
data_y_converted = (
abs(ratio_y) * cal["max_y"] + cal["center_y"])
data_y_converted = int(round(data_y_converted))
# Converting the two X/Y uint16 values to 3 uint8 Little Endian values
# using bitshifting techniques
converted_values = [
# Get the last two hex digits
data_x_converted & 0xFF,
# Combine the last digit of the Y uint16 and the first digit
# of the X uint16
((data_y_converted & 0xF) << 4) + (data_x_converted >> 8),
# Get the first two digits of the Y uint16
data_y_converted >> 4]
return converted_values
def reassign_protocol(self, protocol):
self.protocol = protocol

36
nxbt/controller/protocol.py
View file

@ -3,7 +3,7 @@ import random
from time import perf_counter
from .controller import ControllerTypes
from .utils import replace_subarray
from .utils import replace_subarray, format_msg_controller
class SwitchResponses(Enum):
@ -97,19 +97,19 @@ class ControllerProtocol():
# Disable left stick if we have a right Joy-Con
if self.controller_type == ControllerTypes.JOYCON_R:
self.left_stick_status = [0x00] * 3
self.left_stick_centre = [0x00] * 3
else:
# Center values which are also reported under
# SPI Stick calibration reads
self.left_stick_status = [0x6F, 0xC8, 0x77]
self.left_stick_centre = [0x6F, 0xC8, 0x77]
# Disable right stick if we have a left Joy-Con
if self.controller_type == ControllerTypes.JOYCON_L:
self.right_stick_status = [0x00] * 3
self.right_stick_centre = [0x00] * 3
else:
# Center values which are also reported under
# SPI Stick calibration reads
self.right_stick_status = [0x16, 0xD8, 0x7D]
self.right_stick_centre = [0x16, 0xD8, 0x7D]
self.vibrator_report = random.choice(self.VIBRATOR_BYTES)
@ -130,6 +130,7 @@ class ControllerProtocol():
def get_report(self):
report = bytes(self.report)
print(format_msg_controller(report))
# Clear report
self.set_empty_report()
return report
@ -266,22 +267,35 @@ class ControllerProtocol():
self.report[5] = self.button_status[1]
self.report[6] = self.button_status[2]
self.report[7] = self.left_stick_status[0]
self.report[8] = self.left_stick_status[1]
self.report[9] = self.left_stick_status[2]
self.report[7] = self.left_stick_centre[0]
self.report[8] = self.left_stick_centre[1]
self.report[9] = self.left_stick_centre[2]
self.report[10] = self.right_stick_status[0]
self.report[11] = self.right_stick_status[1]
self.report[12] = self.right_stick_status[2]
self.report[10] = self.right_stick_centre[0]
self.report[11] = self.right_stick_centre[1]
self.report[12] = self.right_stick_centre[2]
self.report[13] = self.vibrator_report
def set_button_inputs(self, upper, shared, lower):
print(upper, shared, lower)
self.report[4] = upper
self.report[5] = shared
self.report[6] = lower
def set_left_stick_inputs(self, left):
self.report[7] = left[0]
self.report[8] = left[1]
self.report[9] = left[2]
def set_right_stick_inputs(self, right):
self.report[10] = right[0]
self.report[11] = right[1]
self.report[12] = right[2]
def set_device_info(self):
# ACK Reply

6
nxbt/controller/server.py
View file

@ -104,7 +104,6 @@ class ControllerServer():
if self.task_queue:
try:
msg = self.task_queue.get_nowait()
print(msg)
if msg:
self.input.buffer_macro(
msg["macro"], msg["macro_id"])
@ -115,8 +114,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:
# print(format_msg_controller(msg))
try:
itr.sendall(msg)
@ -168,6 +167,7 @@ class ControllerServer():
self.bt.address,
colour_body=self.colour_body,
colour_buttons=self.colour_buttons)
self.input.reassign_protocol(self.protocol)
if self.lock:
self.lock.acquire()

9
nxbt/nxbt.py
View file

@ -8,10 +8,16 @@ import os
import dbus
from .controller import ControllerServer
from .controller import ControllerTypes
from .bluez import find_objects, toggle_input_plugin
from .bluez import SERVICE_NAME, ADAPTER_INTERFACE
JOYCON_L = ControllerTypes.JOYCON_L
JOYCON_R = ControllerTypes.JOYCON_R
PRO_CONTROLLER = ControllerTypes.PRO_CONTROLLER
class NxbtCommands(Enum):
CREATE_CONTROLLER = 0
@ -147,7 +153,7 @@ class Nxbt():
# Block until the controller is ready
# This needs to be done to prevent race conditions
# on DBus resources.
# on Bluetooth resources.
if type(controller_index) == int:
while True:
if controller_index in self.manager_state.keys():
@ -157,7 +163,6 @@ class Nxbt():
break
finally:
self.__controller_lock.release()
pass
return controller_index

8
scripts/proxy.py
View file

@ -15,7 +15,7 @@ from time import perf_counter
from nxbt import toggle_input_plugin
from nxbt import BlueZ
from nxbt import Controller
from nxbt import ControllerTypes
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'
@ -95,10 +95,10 @@ if __name__ == "__main__":
# Switch Controller Bluetooth MAC Address goes here
jc_MAC = "7C:BB:8A:FA:41:3D"
# Specify the type of controller here
controller_type = ControllerTypes.PRO_CONTROLLER
if controller_type == ControllerTypes.JOYCON_L:
controller_type = PRO_CONTROLLER
if controller_type == JOYCON_L:
REPLY = JCL_REPLY02
elif controller_type == ControllerTypes.JOYCON_R:
elif controller_type == JOYCON_R:
REPLY = JCR_REPLY02
else:
REPLY = PRO_REPLY02

31
scripts/sticks.py
View file

@ -24,10 +24,10 @@ data_right[5] = (stick_cal_right[8] << 4) | (stick_cal_right[7] >> 4)
# Left Stick Decode
left_center_x = data_left[2]
left_center_y = data_left[3]
left_x_min = left_center_x - data_left[0]
left_x_max = left_center_x + data_left[4]
left_y_min = left_center_y - data_left[1]
left_y_max = left_center_y + data_left[5]
left_x_min = (left_center_x - data_left[0]) - left_center_x
left_x_max = (left_center_x + data_left[4]) - left_center_x
left_y_min = (left_center_y - data_left[1]) - left_center_y
left_y_max = (left_center_y + data_left[5]) - left_center_y
print("Left Stick Values:")
print("~~~~~~~~~~~~~~~~~~")
@ -38,10 +38,10 @@ print("Left Y Min/Max: ", left_y_min, "", left_y_max)
# Right Stick Decode
right_center_x = data_right[0]
right_center_y = data_right[1]
right_x_min = right_center_x - data_right[2]
right_x_max = right_center_x + data_right[4]
right_y_min = right_center_y - data_right[3]
right_y_max = right_center_y + data_right[5]
right_x_min = (right_center_x - data_right[2]) - right_center_x
right_x_max = (right_center_x + data_right[4]) - right_center_x
right_y_min = (right_center_y - data_right[3]) - right_center_y
right_y_max = (right_center_y + data_right[5]) - right_center_y
print("\nRight Stick Values:")
print("~~~~~~~~~~~~~~~~~~~")
@ -63,9 +63,20 @@ print("Relative X/Y Values", ratio_x, ratio_y)
print("\nExample Left Stick Ratio to Data Conversion:")
print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~")
data_x_converted = (abs(ratio_x) * (left_x_min - left_center_x) + left_center_x)
if ratio_x < 0:
data_x_converted = (
abs(ratio_x) * (left_x_min - left_center_x) + left_center_x)
else:
data_x_converted = (
abs(ratio_x) * (left_x_max - left_center_x) + left_center_x)
data_x_converted = int(round(data_x_converted))
data_y_converted = (abs(ratio_y) * (left_y_min - left_center_y) + left_center_y)
if ratio_y < 0:
data_y_converted = (
abs(ratio_y) * (left_y_min - left_center_y) + left_center_y)
else:
data_y_converted = (
abs(ratio_y) * (left_y_max - left_center_y) + left_center_y)
data_y_converted = int(round(data_y_converted))
print("X/Y Converted Values:", data_x_converted, data_y_converted)

2
setup.cfg
View file

@ -43,5 +43,5 @@ filterwarnings =
error
[flake8]
max-line-length = 80
max-line-length = 100
exclude = .git, .eggs, __pycache__, tests/, docs/, build/, dist/