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

Merge pull request #31 from Brikwerk/v12_fix

Browse source 
SwitchOS v12 and v13 fixes
This commit is contained in:
Reece Walsh 2021-10-03 12:18:23 -07:00 • committed by GitHub
commit 27451e61b4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
25 changed files with 1358 additions and 225 deletions
Download patch file Download diff file Expand all files Collapse all files

6
.gitignore vendored
View file

@ -140,4 +140,8 @@ cython_debug/
# Project Specific Excludes
.vscode
secrets.txt
messages.txt
messages.txt
Vagrantfile
.vagrant
cert.pem
key.pem

14
README.md
View file

@ -50,13 +50,9 @@ sudo pip3 install nxbt
**Please Note:** NXBT needs root privileges to toggle the BlueZ Input plugin. If you're not comfortable running this program as root, you can disable the Input plugin manually, and install NXBT as a regular user.
### MacOS
### Windows and macOS
Coming Soon (time permitting)
### Windows
Under Investigation
See the installation guide [here.](docs/Windows-and-macOS-Installation.md)
## Getting Started
@ -295,7 +291,7 @@ controller_index = nx.create_controller(
reconnect_address=nx.get_switch_addresses())
```
**Stopping or Clearning Macros**
**Stopping or Clearing Macros**
```python
# Stops/deletes a single macro from a specified controller
nx.stop_macro(controller_index, macro_id)
@ -309,6 +305,10 @@ nx.clear_all_macros()
## Troubleshooting
### My controller disconnects after exiting the "Change Grip/Order" Menu
This can occasionally occur due to timing sensitivites when transitioning off of the "Change Grip/Order" menu. To avoid disconnections when exiting this menu, please only press A (or B) a single time and wait until the menu has fully exited. If a disconnect still occurs, you should be able to reconnect your controller and use NXBT as normal.
### "No Available Adapters"
This means that NXBT wasn't able to find a suitable Bluetooth adapter to use for Nintendo Switch controller emulation. Only one controller can be emulated per adapter on the system, so if you've got one Bluetooth adapter available, you'll only be able to emulate one Nintendo Switch controller. The general causes (and solutions) to the above error follows:

106
docs/Windows-and-macOS-Installation.md Normal file
View file

@ -0,0 +1,106 @@
# Windows and macOS Installation
To support the necessary Bluetooth APIs leveraged within NXBT, installation within a Virtual Machine (VM) is necessary. To install on Windows or macOS using a VM, please follow the instructions below.
## Prerequisites
Before continuing, please ensure you have the following:
- A **USB** Bluetooth Adapter
- Internal Bluetooth adapters are incompatible (generally) with the process that allows a VM to use external resources.
- VirtualBox v6 or above
- If you don't have this, you can install VirtualBox [here](https://www.virtualbox.org/wiki/Downloads)
- VirtualBox Extension Pack
- The Extension Pack should be available to download on the [same page as VirtualBox.](https://www.virtualbox.org/wiki/Downloads)
- Vagrant
- Available to download [here](https://www.vagrantup.com/downloads)
- Python 3
Additionally, please ensure that VBoxManage (a CLI that ships with VirtualBox) is available on your system path. Eg: A help message should be displayed if `VBoxManage` is entered into Terminal (macOS) or Command Prompt (Windows). If you don't see a help message, please add VirtualBox's installation directory to your system path.
## Installation
1. Clone the NXBT repo to a location of your choosing:
```bash
git clone https://github.com/Brikwerk/nxbt
```
2. Navigate inside the cloned directory and run the Vagrant setup tool.
```bash
cd nxbt
python3 vagrant_setup.py
```
3. Follow the tool's directions and choose the USB Bluetooth adapter you would like to use with NXBT. Additionally, you'll be able to choose between intalling NXBT from PyPi or from the cloned repository. Installing NXBT from the cloned repository allows for use of development version (as well as editing NXBT itself)
4. Once the Vagrant setup tool is finished, you should see a file called `Vagrantfile` located in the same directory as the setup tool. You should now be able to boot the VM with the following command:
```bash
vagrant up
```
5. After the VM has fully completed its setup, you can SSH into the terminal. Please note that your terminal's current working directory must be in the same directory at the Vagrantfile you generated earlier.
```bash
# SSHing into the VM
vagrant ssh
```
6. Unplug the USB Bluetooth adapter from your machine and plug it back in. The allows for VirtualBox to properly claim and forward to the USB into the Vagrant VM.
7. Inside the VirtualBox, check that your Bluetooth Adapter is available with `lsusb`:
```bash
> lsusb
# Something like the following will be printed
# if your USB Bluetooth adapter is available:
Bus 001 Device 001: ID 1d6b:0002 Linux Foundation 2.0 root hub
Bus 002 Device 002: ID 0a5c:21e9 Broadcom Corp. BCM20702A0 Bluetooth 4.0
Bus 002 Device 001: ID 1d6b:0001 Linux Foundation 1.1 root hub
```
Next, use `bluetoothctl` to test if Bluetooth is functional with the adapter:
```bash
> sudo bluetoothctl
# bluetoothctl should print something like below
# if the adapter is functional
Agent registered
[CHG] Controller XX:XX:XX:XX:XX:XX Pairable: yes
# You can additionally run the `show` command in
# bluetoothctl to list your adapters stats as a final check
> [bluetooth]# show
Controller XX:XX:XX:XX:XX:XX (public)
Name: ubuntu2010.localdomain
...
```
If you're not able to see your adapter within the VM or the adapter isn't functional, please refer to the troubleshooting section below.
8. If the above checks pass, NXBT should be functional within your VM. You can run NXBT commands as normal while SSHed into the VM:
```bash
# Eg:
sudo nxbt test
```
9. Finally, Vagrant exposes the following other commands to halt the VM and completely destroy it:
```bash
# Stop the VM but don't destroy it
vagrant halt
# Completely destroy the VM
vagrant destroy
```
## Troubleshooting
### My USB Bluetooth adapter won't show up inside the VM
First, halt your VM (`vagrant halt`) and unplug your adapter. Next, restart the VM (`vagrant up`) and SSH into it (`vagrant ssh`). Plug the Bluetooth adapter in and check if it's listed with `lsusb`. If the adapter still isn't listed, unplug the adapter again and manually add a USB passthrough with the VirtualBox application. Instructions on this can be found [here](https://help.ubuntu.com/community/VirtualBox/USB) under the "For persistent device connection to VM" section.
### My adapter appears but Bluetooth isn't functional
Typically, restarting the VM resolves the issue. Make sure you unplug the adapter and plug it back in when the VM has fully booted (AKA when it's possible to SSH into it).

348
nxbt/bluez.py
View file

@ -3,6 +3,9 @@ import re
import os
import time
import logging
from shutil import which
import random
from pathlib import Path
import dbus
@ -90,69 +93,179 @@ 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.
def toggle_clean_bluez(toggle):
"""Enables or disables all BlueZ plugins,
BlueZ compatibility mode, and removes all extraneous
SDP Services offered.
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)
:param toggle: A boolean element indicating if BlueZ
should be cleaned (True) or not (False)
:type toggle: boolean
:raises PermissionError: If the user is not root
:raises Exception: If the units can't be reloaded
:raises Exception: If sdptool, hciconfig, or hcitool are not available.
"""
service_path = "/lib/systemd/system/bluetooth.service"
service = None
with open(service_path, "r") as f:
service = f.read()
override_dir = Path("/run/systemd/system/bluetooth.service.d")
override_path = override_dir / "nxbt.conf"
# 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)
if toggle:
if override_path.is_file():
# Override exist, no need to restart bluetooth
return
with open(service_path) as f:
for line in f:
if line.startswith("ExecStart="):
exec_start = line.strip() + " --compat --noplugin=*"
break
else:
# If input is already disabled
if "--noplugin=input" in line:
return
# If not, add the flag
lines[i] = line + " --noplugin=input"
raise Exception("systemd service file doesn't have a ExecStart line")
service = "\n".join(lines)
with open(service_path, "w") as f:
f.write(service)
override = f"[Service]\nExecStart=\n{exec_start}"
override_dir.mkdir(parents=True, exist_ok=True)
with override_path.open("w") as f:
f.write(override)
else:
try:
os.remove(override_path)
except FileNotFoundError:
# Override doesn't exist, no need to restart bluetooth
return
# 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)
_run_command(["systemctl", "daemon-reload"])
# 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)
_run_command(["systemctl", "restart", "bluetooth"])
# Kill a bit of time here to ensure all services have restarted
time.sleep(0.5)
def find_devices_by_alias(alias):
def clean_sdp_records():
"""Cleans all SDP Records from BlueZ with sdptool
:raises Exception: On CLI error or sdptool missing
"""
# TODO: sdptool is deprecated in BlueZ 5. This should ideally
# use the DBus API, however, bugs seemingly exist with the
# UnregisterProfile interface.
# Check if sdptool is available for use
if which("sdptool") is None:
raise Exception("sdptool is not available on this system." +
"If you can, please install this tool, as " +
"it is required for proper functionality.")
# Enable Read/Write to the SDP server. This is a remedy for a
# compatibility mode bug introduced in later versions of BlueZ 5
_run_command(["chmod", "777", "/var/run/sdp"])
# Identify/List all SDP services available with sdptool
result = _run_command(['sdptool', 'browse', 'local']).stdout.decode('utf-8')
if result is None or len(result.split('\n\n')) < 1:
return
# Record all service record handles
exceptions = ["PnP Information"]
service_rec_handles = []
for rec in result.split('\n\n'):
# Skip if exception is in record
exception_found = False
for exception in exceptions:
if exception in rec:
exception_found = True
break
if exception_found:
continue
# Read lines and add Record Handles to the list
for line in rec.split('\n'):
if "Service RecHandle" in line:
service_rec_handles.append(line.split(" ")[2])
# Delete all found service records
if len(service_rec_handles) > 0:
for record_handle in service_rec_handles:
_run_command(['sdptool', 'del', record_handle])
def _run_command(command):
"""Runs a specified command on the shell of the system.
If the command is run unsuccessfully, an error is raised.
The command must be in the form of an array with each term
individually listed. Eg: ["which", "bash"]
:param command: A list of command terms
:type command: list
:raises Exception: On command failure or error
"""
result = subprocess.run(
command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
cmd_err = result.stderr.decode("utf-8").replace("\n", "")
if cmd_err != "":
raise Exception(cmd_err)
return result
def get_random_controller_mac():
"""Generates a random Switch-compliant MAC address
"""
def seg():
random_number = random.randint(0,255)
hex_number = str(hex(random_number))
hex_number = hex_number[2:].upper()
return str(hex_number)
return f"7C:BB:8A:{seg()}:{seg()}:{seg()}"
def replace_mac_addresses(adapter_paths, addresses):
"""Replaces a list of adapter's Bluetooth MAC addresses
with Switch-compliant Controller MAC addresses. If the
addresses argument is specified, the adapter path's
MAC addresses will be reset to respective (index-wise)
address in the list.
:param adapter_paths: A list of Bluetooth adapter paths
:type adapter_paths: list
:param addresses: A list of Bluetooth MAC addresses,
defaults to False
:type addresses: bool, optional
"""
if which("hcitool") is None:
raise Exception("hcitool is not available on this system." +
"If you can, please install this tool, as " +
"it is required for proper functionality.")
if which("hciconfig") is None:
raise Exception("hciconfig is not available on this system." +
"If you can, please install this tool, as " +
"it is required for proper functionality.")
if addresses:
assert len(addresses) == len(adapter_paths)
for i in range(len(adapter_paths)):
adapter_id = adapter_paths[i].split('/')[-1]
mac = addresses[i].split(':')
cmds = ['hcitool', '-i', adapter_id, 'cmd', '0x3f', '0x001',
f'0x{mac[5]}',f'0x{mac[4]}',f'0x{mac[3]}',f'0x{mac[2]}',
f'0x{mac[1]}',f'0x{mac[0]}']
_run_command(cmds)
_run_command(['hciconfig', adapter_id, 'reset'])
def find_devices_by_alias(alias, return_path=False, created_bus=None):
"""Finds the Bluetooth addresses of devices
that have a specified Bluetooth alias. Aliases
are converted to uppercase before comparison
@ -164,7 +277,10 @@ def find_devices_by_alias(alias):
:rtype: string or None
"""
bus = dbus.SystemBus()
if created_bus is not None:
bus = created_bus
else:
bus = dbus.SystemBus()
# Find all connected/paired/discovered devices
devices = find_objects(
bus,
@ -172,6 +288,7 @@ def find_devices_by_alias(alias):
DEVICE_INTERFACE)
addresses = []
matching_paths = []
for path in devices:
# Get the device's address and paired status
device_props = dbus.Interface(
@ -187,9 +304,59 @@ def find_devices_by_alias(alias):
# Check for an address match
if device_alias.upper() == alias.upper():
addresses.append(device_addr)
matching_paths.append(path)
bus.close()
return addresses
# Close the dbus connection if we created one
if created_bus is None:
bus.close()
if return_path:
return addresses, matching_paths
else:
return addresses
def disconnect_devices_by_alias(alias, created_bus=None):
"""Disconnects all devices matching an alias.
:param alias: The device's alias
:type alias: string
"""
if created_bus is not None:
bus = created_bus
else:
bus = dbus.SystemBus()
# Find all connected/paired/discovered devices
devices = find_objects(
bus,
SERVICE_NAME,
DEVICE_INTERFACE)
addresses = []
matching_paths = []
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()
# Check for an alias match
if device_alias.upper() == alias.upper():
device = dbus.Interface(
bus.get_object(SERVICE_NAME, path),
DEVICE_INTERFACE)
try:
device.Disconnect()
except Exception as e:
print(e)
# Close the dbus connection if we created one
if created_bus is None:
bus.close()
class BlueZ():
@ -246,6 +413,44 @@ class BlueZ():
return self.device.Get(ADAPTER_INTERFACE, "Address").upper()
def set_address(self, mac):
"""Sets the Bluetooth MAC address of the Bluetooth adapter.
The hciconfig CLI is required for setting the address.
For changes to apply, the Bluetooth interface needs to be
restarted.
:param mac: A Bluetooth MAC address in
the form of "XX:XX:XX:XX:XX:XX
:type mac: str
:raises PermissionError: On run as non-root user
:raises Exception: On CLI errors
"""
if which("hcitool") is None:
raise Exception("hcitool is not available on this system." +
"If you can, please install this tool, as " +
"it is required for proper functionality.")
# Reverse MAC (element position-wise) for use with hcitool
mac = mac.split(":")
cmds = ['hcitool', '-i', self.device_id, 'cmd', '0x3f', '0x001',
f'0x{mac[5]}',f'0x{mac[4]}',f'0x{mac[3]}',f'0x{mac[2]}',
f'0x{mac[1]}',f'0x{mac[0]}']
_run_command(cmds)
_run_command(['hciconfig', self.device_id, 'reset'])
def set_class(self, device_class):
if which("hciconfig") is None:
raise Exception("hciconfig is not available on this system." +
"If you can, please install this tool, as " +
"it is required for proper functionality.")
_run_command(['hciconfig', self.device_id, 'class', device_class])
def reset_adapter(self):
if which("hciconfig") is None:
raise Exception("hciconfig is not available on this system." +
"If you can, please install this tool, as " +
"it is required for proper functionality.")
_run_command(['hciconfig', self.device_id, 'reset'])
@property
def name(self):
"""Gets the name of the Bluetooth adapter.
@ -470,7 +675,16 @@ class BlueZ():
:type opts: dict
"""
self.profile_manager.RegisterProfile(profile_path, uuid, opts)
return self.profile_manager.RegisterProfile(profile_path, uuid, opts)
def unregister_profile(self, profile):
"""Unregisters a given SDP record from the BlueZ SDP server.
:param profile: A SDP record profile object
:type profile: Profile
"""
self.profile_manager.UnregisterProfile(profile)
def reset(self):
"""Restarts the Bluetooth Service
@ -563,6 +777,8 @@ class BlueZ():
devices = self.get_discovered_devices()
# Start discovering new devices and loop
self.set_powered(True)
self.set_pairable(True)
self.adapter.StartDiscovery()
try:
for i in range(0, timeout):
@ -577,6 +793,7 @@ class BlueZ():
callback(devices)
finally:
self.adapter.StopDiscovery()
time.sleep(1)
# Filter out paired devices or devices that don't
# match a specified alias.
@ -665,3 +882,38 @@ class BlueZ():
return path
return None
def find_connected_devices(self, alias_filter=False):
"""Finds the D-Bus path to a device that contains the
specified address.
:param address: The Bluetooth MAC address
:type address: string
:return: The path to the D-Bus object or None
:rtype: string or None
"""
devices = find_objects(
self.bus,
SERVICE_NAME,
DEVICE_INTERFACE)
conn_devices = []
for path in devices:
# Get the device's connection status
device_props = dbus.Interface(
self.bus.get_object(SERVICE_NAME, path),
"org.freedesktop.DBus.Properties")
device_conn_status = device_props.Get(
DEVICE_INTERFACE,
"Connected")
device_alias = device_props.Get(
DEVICE_INTERFACE,
"Alias").upper()
if device_conn_status:
if alias_filter and device_alias == alias_filter.upper():
conn_devices.append(path)
else:
conn_devices.append(path)
return conn_devices

132
nxbt/cli.py
View file

@ -2,6 +2,7 @@ import argparse
from random import randint
from time import sleep
import os
import traceback
from .nxbt import Nxbt, PRO_CONTROLLER
from .bluez import find_devices_by_alias
@ -10,7 +11,7 @@ from .tui import InputTUI
parser = argparse.ArgumentParser()
parser.add_argument('command', default=False, choices=[
'webapp', 'demo', 'macro', 'tui', 'addresses'
'webapp', 'demo', 'macro', 'tui', 'remote_tui', 'addresses', 'test'
],
help="""Specifies the nxbt command to run:
webapp - Runs web server and allows for controller/macro
@ -19,9 +20,12 @@ parser.add_argument('command', default=False, choices=[
is on the main menu's Change Grip/Order menu before running).
macro - Allows for input of a specified macro from the command line
(with the argument -s) or from a file (with the argument -f).
input - Opens a TUI that allows for direct input from the keyboard
to the Switch. addresses - Lists the Bluetooth MAC addresses for
all previously connected Nintendo Switches""")
tui/remote_tui - Opens a TUI that allows for direct input from the keyboard
to the Switch.
addresses - Lists the Bluetooth MAC addresses for
all previously connected Nintendo Switches.
test - Runs through a series of tests to ensure NXBT is working and
compatible with your system.""")
parser.add_argument('-c', '--commands', required=False, default=False,
help="""Used in conjunction with the macro command. Specifies a
macro string or a file location to load a macro string from.""")
@ -42,26 +46,34 @@ parser.add_argument('-i', '--ip', required=False, default="0.0.0.0", type=str,
help="""Specifies the IP to run the webapp at. Defaults to 0.0.0.0""")
parser.add_argument('-p', '--port', required=False, default=8000, type=int,
help="""Specifies the port to run the webapp at. Defaults to 8000""")
parser.add_argument('--usessl', required=False, default=False, action='store_true',
help="""Enables or disables SSL use in the webapp""")
parser.add_argument('--certpath', required=False, default=None, type=str,
help="""Specifies the folder location for SSL certificates used
in the webapp. Certificates in this folder should be in the form of
a 'cert.pem' and 'key.pem' pair.""")
args = parser.parse_args()
MACRO = """
B 0.1s
0.1s
0.5s
B 0.1s
0.1s
0.5s
B 0.1s
0.1s
0.5s
B 0.1s
1.5s
DPAD_RIGHT 0.075s
0.075s
A 0.1s
1.5s
DPAD_DOWN 1.0s
LOOP 12
DPAD_DOWN 0.075s
0.075s
A 0.1s
0.25s
DPAD_DOWN 0.95s
DPAD_DOWN 0.93s
A 0.1s
0.25s
L_STICK_PRESS 0.1s
@ -138,6 +150,9 @@ def demo():
nx = Nxbt(debug=args.debug, log_to_file=args.logfile)
adapters = nx.get_available_adapters()
if len(adapters) < 1:
raise OSError("Unable to detect any Bluetooth adapters.")
controller_idxs = []
for i in range(0, len(adapters)):
index = nx.create_controller(
@ -148,8 +163,94 @@ def demo():
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)
print("Running Demo...")
macro_id = nx.macro(controller_idxs[-1], MACRO, block=False)
while macro_id not in nx.state[controller_idxs[-1]]["finished_macros"]:
state = nx.state[controller_idxs[-1]]
if state['state'] == 'crashed':
print("An error occurred while running the demo:")
print(state['errors'])
exit(1)
sleep(1.0)
print("Finished!")
def test():
"""Tests NXBT functionality"""
# Init
print("[1] Attempting to initialize NXBT...")
nx = None
try:
nx = Nxbt(debug=args.debug, log_to_file=args.logfile)
except Exception as e:
print("Failed to initialize:")
print(traceback.format_exc())
exit(1)
print("Successfully initialized NXBT.\n")
# Adapter Check
print("[2] Checking for Bluetooth adapter availability...")
adapters = None
try:
adapters = nx.get_available_adapters()
except Exception as e:
print("Failed to check for adapters:")
print(traceback.format_exc())
exit(1)
if len(adapters) < 1:
print("Unable to detect any Bluetooth adapters.")
print("Please ensure you system has Bluetooth capability.")
exit(1)
print(f"{len(adapters)} Bluetooth adapter(s) available.")
print("Adapters:", adapters, "\n")
# Creating a controller
print("[3] Please turn on your Switch and navigate to the 'Change Grip/Order menu.'")
input("Press Enter to continue...")
print("Creating a controller with the first Bluetooth adapter...")
cindex = None
try:
cindex = nx.create_controller(
PRO_CONTROLLER,
adapters[0],
colour_body=random_colour(),
colour_buttons=random_colour())
except Exception as e:
print("Failed to create a controller:")
print(traceback.format_exc())
exit(1)
print("Successfully created a controller.\n")
# Controller connection check
print("[4] Waiting for controller to connect with the Switch...")
timeout = 120
print(f"Connection timeout is {timeout} seconds for this test script.")
elapsed = 0
while nx.state[cindex]['state'] != 'connected':
if elapsed >= timeout:
print("Timeout reached, exiting...")
exit(1)
elif nx.state[cindex]['state'] == 'crashed':
print("An error occurred while connecting:")
print(nx.state[cindex]['errors'])
exit(1)
elapsed += 1
sleep(1)
print("Successfully connected.\n")
# Exit the Change Grip/Order Menu
print("[5] Attempting to exit the 'Change Grip/Order Menu'...")
nx.macro(cindex, "B 0.1s\n0.1s")
sleep(5)
if nx.state[cindex]['state'] != 'connected':
print("Controller disconnected after leaving the menu.")
print("Exiting...")
exit(1)
print("Controller successfully exited the menu.\n")
print("All tests passed.")
def macro():
@ -218,7 +319,8 @@ def main():
if args.command == 'webapp':
from .web import start_web_app
start_web_app(ip=args.ip, port=args.port)
start_web_app(ip=args.ip, port=args.port,
usessl=args.usessl, cert_path=args.certpath)
elif args.command == 'demo':
demo()
elif args.command == 'macro':
@ -227,5 +329,11 @@ def main():
reconnect_target = get_reconnect_target()
tui = InputTUI(reconnect_target=reconnect_target)
tui.start()
elif args.command == 'remote_tui':
reconnect_target = get_reconnect_target()
tui = InputTUI(reconnect_target=reconnect_target, force_remote=True)
tui.start()
elif args.command == 'addresses':
list_switch_addresses()
elif args.command == 'test':
test()

6
nxbt/controller/controller.py
View file

@ -1,5 +1,6 @@
from enum import Enum
import os
import logging
import dbus
@ -27,6 +28,7 @@ class Controller():
def __init__(self, bluetooth, controller_type):
self.bt = bluetooth
self.logger = logging.getLogger('nxbt')
if controller_type not in self.ALIASES.keys():
raise ValueError("Unknown controller type specified")
@ -63,5 +65,5 @@ class Controller():
# catch the error and continue
try:
self.bt.register_profile(self.SDP_RECORD_PATH, self.SDP_UUID, opts)
except dbus.exceptions.DBusException:
pass
except dbus.exceptions.DBusException as e:
self.logger.debug(e)

24
nxbt/controller/input.py
View file

@ -150,6 +150,30 @@ class InputParser():
def set_controller_input(self, controller_input):
self.controller_input = controller_input
def commands_queued(self):
check = dumps(self.controller_input) != dumps(DIRECT_INPUT_IDLE_PACKET)
check = check or self.macro_buffer
check = check or self.current_macro
check = check or self.current_macro_commands
return check
def active_input_queued(self):
"""Checks if an active command input is queued. An active command
is a depressed button or tilted stick.
:return: True (on an active button) or False (no active buttons)
:rtype: bool
"""
if (self.current_macro_commands is not None):
if len(self.current_macro_commands) < 2:
return False
else:
return True
elif dumps(self.controller_input) != dumps(DIRECT_INPUT_IDLE_PACKET):
return True
else:
return False
def set_protocol_input(self, state=None):

2
nxbt/controller/protocol.py
View file

@ -242,7 +242,7 @@ class ControllerProtocol():
# Get how many ticks have passed in hex with overflow at 255
# Joy-Con uses 4.96ms as the timer tick rate
elapsed_ticks = int(delta_t // 4.96)
elapsed_ticks = int(delta_t * 4)
self.timer = (self.timer + elapsed_ticks) & 0xFF
self.report[2] = self.timer

316
nxbt/controller/server.py
View file

@ -5,9 +5,12 @@ import time
import queue
import logging
import traceback
import atexit
from threading import Thread
import statistics as stat
from .controller import Controller, ControllerTypes
from ..bluez import BlueZ
from ..bluez import BlueZ, find_devices_by_alias
from .protocol import ControllerProtocol
from .input import InputParser
from .utils import format_msg_controller, format_msg_switch
@ -23,6 +26,8 @@ class ControllerServer():
# Cache logging level to increase performance on checks
self.logger_level = self.logger.level
atexit.register(self._on_exit)
if state:
self.state = state
else:
@ -56,7 +61,12 @@ class ControllerServer():
self.input = InputParser(self.protocol)
self.slow_input_frequency = False
# Debug timekeeping storage array
self.times = []
# Initial reconnection overload protection
self.tick = 1
self.cached_msg = ''
def run(self, reconnect_address=None):
"""Runs the mainloop of the controller server.
@ -78,14 +88,18 @@ class ControllerServer():
self.controller.setup()
if reconnect_address:
itr, ctrl = self.reconnect(reconnect_address)
try:
itr, ctrl = self.reconnect(reconnect_address)
except OSError:
itr, ctrl = self.connect()
else:
itr, ctrl = self.connect()
finally:
if self.lock:
self.lock.release()
self.switch_address = itr.getsockname()[0]
self.switch_address = itr.getpeername()[0]
self.state["last_connection"] = self.switch_address
self.state["state"] = "connected"
@ -94,21 +108,25 @@ class ControllerServer():
except KeyboardInterrupt:
pass
except Exception:
self.state["state"] = "crashed"
self.state["errors"] = traceback.format_exc()
return self.state
try:
self.state["state"] = "crashed"
self.state["errors"] = traceback.format_exc()
return self.state
except Exception as e:
self.logger.debug("Error during graceful shutdown:")
self.logger.debug(traceback.format_exc())
def mainloop(self, itr, ctrl):
# Mainloop
duration_start = time.perf_counter()
while True:
# Start timing the command processing
# Start timing command processing
timer_start = time.perf_counter()
# Attempt to get output from Switch
try:
reply = itr.recv(50)
if self.logger_level <= logging.DEBUG and len(reply) > 40:
if len(reply) > 40:
self.logger.debug(format_msg_switch(reply))
except BlockingIOError:
reply = None
@ -142,7 +160,16 @@ class ControllerServer():
self.logger.debug(format_msg_controller(msg))
try:
itr.sendall(msg)
# Cache the last packet to prevent overloading the switch
# with packets on the "Change Grip/Order" menu.
if msg[3:] != self.cached_msg:
itr.sendall(msg)
self.cached_msg = msg[3:]
# Send a blank packet every so often to keep the Switch
# from disconnecting from the controller.
elif self.tick >= 132:
itr.sendall(msg)
self.tick = 0
except BlockingIOError:
continue
except OSError as e:
@ -150,27 +177,24 @@ class ControllerServer():
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)
duration_end = time.perf_counter()
duration_elapsed = duration_end - duration_start
duration_start = duration_end
sleep_time = 1/132 - duration_elapsed
if sleep_time >= 0:
time.sleep(sleep_time)
self.tick += 1
if self.slow_input_frequency:
# Check if we can switch out of slow frequency input
if self.input.exited_grip_order_menu:
self.slow_input_frequency = False
if self.logger_level <= logging.DEBUG:
self.times.append(duration_elapsed)
if len(self.times) > 100:
self.times.pop()
mean_time = stat.mean(self.times)
self.logger.debug(
f"Tick: {self.tick}, Mean Time: {str(1/mean_time)}")
if elapsed_time < 1/15:
time.sleep(1/15 - elapsed_time)
else:
# Respond at 120Hz for Pro Controller
# or 60Hz for Joy-Cons.
# Sleep timers are compensated with the elapsed command
# processing time.
if self.controller_type == ControllerTypes.PRO_CONTROLLER:
if elapsed_time < 1/120:
time.sleep(1/120 - elapsed_time)
else:
if elapsed_time < 1/60:
time.sleep(1/60 - elapsed_time)
def save_connection(self, error, state=None):
@ -183,17 +207,57 @@ 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()
try:
itr, ctrl = self.reconnect(self.switch_address)
received_first_message = False
while True:
# Attempt to get output from Switch
try:
reply = itr.recv(50)
if self.logger_level <= logging.DEBUG and len(reply) > 40:
self.logger.debug(format_msg_switch(reply))
except BlockingIOError:
reply = None
if reply:
received_first_message = True
self.protocol.process_commands(reply)
msg = self.protocol.get_report()
if self.logger_level <= logging.DEBUG and reply:
self.logger.debug(format_msg_controller(msg))
try:
itr.sendall(msg)
except BlockingIOError:
continue
# 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
# Pairing cycle responds optimally on a 15Hz loop
if not received_first_message:
time.sleep(1)
else:
time.sleep(1/15)
self.state["state"] = "connected"
return itr, ctrl
finally:
if self.lock:
self.lock.release()
except OSError:
self.reconnect_counter += 1
self.logger.exception(error)
self.logger.debug(error)
time.sleep(0.5)
# If we can't reconnect, transition to attempting
@ -201,6 +265,9 @@ class ControllerServer():
self.logger.debug("Connecting to any Switch")
self.reconnect_counter = 0
# Reinitialize initial communication overload protections
self.tick = 1
# Reinitialize the protocol
self.protocol = ControllerProtocol(
self.controller_type,
@ -233,81 +300,143 @@ class ControllerServer():
return itr, ctrl
def connection_reset_watchdog(self):
connected_devices = []
connected_devices_count = {}
while self._crw_running:
paths = self.bt.find_connected_devices(alias_filter="Nintendo Switch")
# Keep track of Switches that connect
if len(paths) > 0:
connected_devices = list(set(connected_devices + paths))
# Increment a counter if a Switch connected and disconnected
disconnected = list(set(connected_devices) - set(paths))
if len(disconnected) > 0:
for path in disconnected:
if path not in connected_devices_count.keys():
connected_devices_count[path] = 1
else:
connected_devices_count[path] += 1
connected_devices = list(set(connected_devices) - set(disconnected))
# Delete Switches that connect/disconnect twice.
# This behaviour is characteristic of connection issues and is corrected
# by removing the Switch's connection to the system.
if len(connected_devices_count.keys()) > 0:
for key in connected_devices_count.keys():
if connected_devices_count[key] >= 2:
self.logger.debug(
"A Nintendo Switch disconnected. Resetting Connection...")
self.logger.debug(f"Removing {str(key)}")
self.bt.remove_device(key)
connected_devices_count[key] = 0
time.sleep(0.1)
def connect(self):
"""Configures as a specified controller, pairs with a Nintendo Switch,
and creates/accepts sockets for communication with the Switch.
"""
self.state["state"] = "connecting"
# Creating control and interrupt sockets
s_ctrl = socket.socket(
family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
s_itr = socket.socket(
family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
# Setting up HID interrupt/control sockets
try:
s_ctrl.bind((self.bt.address, 17))
s_itr.bind((self.bt.address, 19))
except OSError:
s_ctrl.bind((socket.BDADDR_ANY, 17))
s_itr.bind((socket.BDADDR_ANY, 19))
s_itr.listen(1)
s_ctrl.listen(1)
self.bt.set_discoverable(True)
ctrl, ctrl_address = s_ctrl.accept()
itr, itr_address = s_itr.accept()
# 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)
# 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)
# Mainloop
# The controller server will continue attempting to connect
# to any Nintendo Switch until the connection procedure fully
# succeeds. This prevents situations where the Switch will
# disconnect during a connection.
while True:
# Attempt to get output from Switch
try:
reply = itr.recv(50)
if self.logger_level <= logging.DEBUG and len(reply) > 40:
self.logger.debug(format_msg_switch(reply))
except BlockingIOError:
reply = None
self.state["state"] = "connecting"
self.protocol.process_commands(reply)
msg = self.protocol.get_report()
# Creating control and interrupt sockets
s_ctrl = socket.socket(
family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
s_itr = socket.socket(
family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
if self.logger_level <= logging.DEBUG and reply:
self.logger.debug(format_msg_controller(msg))
# Setting up HID interrupt/control sockets
try:
s_ctrl.bind((self.bt.address, 17))
s_itr.bind((self.bt.address, 19))
except OSError:
s_ctrl.bind((socket.BDADDR_ANY, 17))
s_itr.bind((socket.BDADDR_ANY, 19))
try:
s_itr.listen(1)
s_ctrl.listen(1)
self.bt.set_discoverable(True)
# WARNING:
# A device's class must be set **AFTER** discoverability
# is set. If it is set before or in a similar timeframe,
# the class will be reset to the default value.
self.bt.set_class("0x02508")
self._crw_running = True
crw = Thread(target = self.connection_reset_watchdog)
crw.start()
itr, itr_address = s_itr.accept()
ctrl, ctrl_address = s_ctrl.accept()
self._crw_running = False
# 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)
except BlockingIOError:
continue
# 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):
# 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)
# Mainloop
received_first_message = False
while True:
# Attempt to get output from Switch
try:
reply = itr.recv(50)
if self.logger_level <= logging.DEBUG and len(reply) > 40:
self.logger.debug(format_msg_switch(reply))
except BlockingIOError:
reply = None
if reply:
received_first_message = True
self.protocol.process_commands(reply)
msg = self.protocol.get_report()
if self.logger_level <= logging.DEBUG and reply:
self.logger.debug(format_msg_controller(msg))
try:
itr.sendall(msg)
except BlockingIOError:
continue
# 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
# Pairing cycle responds optimally on a 15Hz loop
if not received_first_message:
time.sleep(1)
else:
time.sleep(1/15)
break
except OSError as e:
self.logger.debug(e)
# Switch responds to packets slower during pairing
# Pairing cycle responds optimally on a 15Hz loop
time.sleep(1/15)
self.slow_input_frequency = True
self.input.exited_grip_order_menu = False
return itr, ctrl
@ -377,3 +506,6 @@ class ControllerServer():
fcntl.fcntl(itr, fcntl.F_SETFL, os.O_NONBLOCK)
return itr, ctrl
def _on_exit(self):
self.bt.reset_address()

32
nxbt/nxbt.py
View file

@ -12,7 +12,8 @@ import dbus
from .controller import ControllerServer
from .controller import ControllerTypes
from .bluez import find_objects, toggle_input_plugin
from .bluez import BlueZ, find_objects, toggle_clean_bluez
from .bluez import replace_mac_addresses
from .bluez import find_devices_by_alias
from .bluez import SERVICE_NAME, ADAPTER_INTERFACE
from .logging import create_logger
@ -177,10 +178,7 @@ class Nxbt():
# Disable the BlueZ input plugin so we can use the
# HID control/interrupt Bluetooth ports
try:
toggle_input_plugin(False)
except PermissionError:
pass
toggle_clean_bluez(True)
# Exit handler
atexit.register(self._on_exit)
@ -209,11 +207,8 @@ class Nxbt():
self.resource_manager.shutdown()
# Re-enable the BlueZ input plugin, if we have permission
try:
toggle_input_plugin(True)
except PermissionError:
pass
# Re-enable the BlueZ plugins, if we have permission
toggle_clean_bluez(False)
def _command_manager(self, task_queue, state):
"""Used as the main multiprocessing Process that is launched
@ -601,6 +596,13 @@ class Nxbt():
"""
if controller_index not in self.manager_state.keys():
if controller_index in self._controller_adapter_lookup.keys():
# Attempt to free any adapters claimed by a crashed controller
try:
adapter_path = self._controller_adapter_lookup.pop(controller_index, None)
self._adapters_in_use.pop(adapter_path, None)
except Exception:
pass
raise ValueError("Specified controller does not exist")
self._controller_lock.acquire()
@ -641,6 +643,7 @@ class Nxbt():
bus = dbus.SystemBus()
adapters = find_objects(bus, SERVICE_NAME, ADAPTER_INTERFACE)
bus.close()
return adapters
@ -737,6 +740,11 @@ class _ControllerManager():
controller_state["finished_macros"] = []
controller_state["errors"] = False
controller_state["direct_input"] = json.loads(json.dumps(DIRECT_INPUT_PACKET))
controller_state["colour_body"] = colour_body
controller_state["colour_buttons"] = colour_buttons
controller_state["type"] = str(controller_type)
controller_state["adapter_path"] = adapter_path
controller_state["last_connection"] = None
self._controller_queues[index] = controller_queue
@ -776,8 +784,8 @@ class _ControllerManager():
})
def remove_controller(self, index):
self._children[index].kill()
self._children[index].terminate()
self.state.pop(index, None)
def shutdown(self):

30
nxbt/tui.py
View file

@ -264,13 +264,28 @@ class InputTUI():
"9": "ZR",
}
def __init__(self, reconnect_target=None, debug=False, logfile=False):
def __init__(self, reconnect_target=None, debug=False, logfile=False, force_remote=False):
self.reconnect_target = reconnect_target
self.term = Terminal()
self.remote_connection = self.detect_remote_connection()
if force_remote:
self.remote_connection = True
else:
self.remote_connection = self.detect_remote_connection()
self.controller = ControllerTUI(self.term)
# Check if direct connection will fail
if not self.remote_connection:
try:
from pynput import keyboard
except ImportError as e:
print("Unable to import pynput for direct input.")
print("If you're accessing NXBT over a remote shell, ", end="")
print("please use the 'remote_tui' option instead of 'tui'.")
print("The original pynput import is displayed below:\n")
print(e)
exit(1)
self.debug = debug
self.logfile = logfile
@ -391,7 +406,7 @@ class InputTUI():
if len(term._keyboard_buf) > 1:
term._keyboard_buf = deque([term._keyboard_buf.pop()])
inp = term.inkey(1/60)
inp = term.inkey(1/66)
pressed_key = None
if inp.is_sequence:
@ -426,9 +441,6 @@ class InputTUI():
self.check_for_disconnect(term)
def direct_input_loop(self, term):
# pynput must be imported here since earlier imports
# will cause errors on remote connections
from pynput import keyboard
self.controller.toggle_auto_keypress_deactivation(False)
@ -610,6 +622,12 @@ class InputTUI():
print(term.bold_black_on_red(term.center(state.title())))
print(term.bold_black_on_red(term.center("")))
if state == 'crashed':
time.sleep(3)
term.clear()
errors = self.nx.state[self.controller_index]["errors"]
raise ConnectionError(errors)
while True:
inp = term.inkey(1/30)
if inp == chr(113):

55
nxbt/web/app.py
View file

@ -1,7 +1,10 @@
import json
import os
from threading import RLock
import time
from socket import gethostname
from .cert import generate_cert
from ..nxbt import Nxbt, PRO_CONTROLLER
from flask import Flask, render_template, request
from flask_socketio import SocketIO, emit
@ -88,6 +91,7 @@ def on_create_controller():
@sio.on('input')
def handle_input(message):
# print("Webapp Input", time.perf_counter())
message = json.loads(message)
index = message[0]
input_packet = message[1]
@ -102,8 +106,55 @@ def handle_macro(message):
nxbt.macro(index, macro)
def start_web_app(ip='0.0.0.0', port=8000):
eventlet.wsgi.server(eventlet.listen((ip, port)), app)
def start_web_app(ip='0.0.0.0', port=8000, usessl=False, cert_path=None):
if usessl:
if cert_path is None:
# Store certs in the package directory
cert_path = os.path.join(
os.path.dirname(__file__), "cert.pem"
)
key_path = os.path.join(
os.path.dirname(__file__), "key.pem"
)
else:
# If specified, store certs at the user's preferred location
cert_path = os.path.join(
cert_path, "cert.pem"
)
key_path = os.path.join(
cert_path, "key.pem"
)
if not os.path.isfile(cert_path) or not os.path.isfile(key_path):
print(
"\n"
"-----------------------------------------\n"
"---------------->WARNING<----------------\n"
"The NXBT webapp is being run with self-\n"
"signed SSL certificates for use on your\n"
"local network.\n"
"\n"
"These certificates ARE NOT safe for\n"
"production use. Please generate valid\n"
"SSL certificates if you plan on using the\n"
"NXBT webapp anywhere other than your own\n"
"network.\n"
"-----------------------------------------\n"
"\n"
"The above warning will only be shown once\n"
"on certificate generation."
"\n"
)
print("Generating certificates...")
cert, key = generate_cert(gethostname())
with open(cert_path, "wb") as f:
f.write(cert)
with open(key_path, "wb") as f:
f.write(key)
eventlet.wsgi.server(eventlet.wrap_ssl(eventlet.listen((ip, port)),
certfile=cert_path, keyfile=key_path), app)
else:
eventlet.wsgi.server(eventlet.listen((ip, port)), app)
if __name__ == "__main__":

88
nxbt/web/cert.py Normal file
View file

@ -0,0 +1,88 @@
# Copyright 2018 Simon Davy
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# WARNING: the code in the gist generates self-signed certs, for the purposes of testing in development.
# Do not use these certs in production, or You Will Have A Bad Time.
#
# Caveat emptor
#
from datetime import datetime, timedelta
import ipaddress
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
def generate_cert(hostname, ip_addresses=None, key=None):
"""Generates self signed certificate for a hostname, and optional IP addresses."""
# Generate our key
if key is None:
key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend(),
)
name = x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, hostname)
])
# best practice seem to be to include the hostname in the SAN, which *SHOULD* mean COMMON_NAME is ignored.
alt_names = [x509.DNSName(hostname)]
# allow addressing by IP, for when you don't have real DNS (common in most testing scenarios
if ip_addresses:
for addr in ip_addresses:
# openssl wants DNSnames for ips...
alt_names.append(x509.DNSName(addr))
# ... whereas golang's crypto/tls is stricter, and needs IPAddresses
# note: older versions of cryptography do not understand ip_address objects
alt_names.append(x509.IPAddress(ipaddress.ip_address(addr)))
san = x509.SubjectAlternativeName(alt_names)
# path_len=0 means this cert can only sign itself, not other certs.
basic_contraints = x509.BasicConstraints(ca=True, path_length=0)
now = datetime.utcnow()
cert = (
x509.CertificateBuilder()
.subject_name(name)
.issuer_name(name)
.public_key(key.public_key())
.serial_number(1000)
.not_valid_before(now - timedelta(days=10*365))
.not_valid_after(now - timedelta(days=9*365))
.add_extension(basic_contraints, False)
.add_extension(san, False)
.sign(key, hashes.SHA256(), default_backend())
)
cert_pem = cert.public_bytes(encoding=serialization.Encoding.PEM)
key_pem = key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption(),
)
return cert_pem, key_pem

54
nxbt/web/static/js/main.js
View file

@ -394,6 +394,11 @@ function recreateProController() {
socket.emit('create_pro_controller');
}
function restartController() {
shutdownController();
setTimeout(recreateProController, 2000);
}
function checkForLoad() {
if (STATE[NXBT_CONTROLLER_INDEX]) {
controller_state = STATE[NXBT_CONTROLLER_INDEX].state
@ -463,6 +468,22 @@ function changeInput(evt) {
}
}
function changeFrequency(evt) {
let newFrequency = evt.target.value;
if (newFrequency === "RAF") {
useRAF = true;
} else {
newFrequency = Number(newFrequency);
if (!isNaN(newFrequency)) {
useRAF = false;
frequency = (1/newFrequency) * 1000;
} else {
console.log("New frequency is not a number");
}
}
}
const LOADER_ANIMATION_FRAMES = [0,1,2,3,3,2,1,0];
let loaderFrame = 1;
let highlightedBlock = false;
@ -548,6 +569,7 @@ function updateGamepadDisplay() {
let timeOld = false;
let frequency = (1/120) * 1000;
let useRAF = true;
function eventLoop() {
// Update x/y ratio for the sticks based on
// pressed buttons if we're using a keyboard
@ -601,21 +623,23 @@ function eventLoop() {
updateGamepadDisplay()
// if (!timeOld) {
// timeOld = performance.now();
// }
// timeNew = performance.now();
// delta = timeNew - timeOld;
// diff = delta - frequency;
// if (diff > 0) {
// setTimeout(eventLoop, frequency - diff);
// } else {
// setTimeout(eventLoop, frequency);
// }
// timeOld = timeNew;
requestAnimationFrame(eventLoop);
if (useRAF) {
requestAnimationFrame(eventLoop);
} else {
if (!timeOld) {
timeOld = performance.now();
}
timeNew = performance.now();
delta = timeNew - timeOld;
diff = delta - frequency;
if (diff > 0) {
setTimeout(eventLoop, frequency - diff);
} else {
setTimeout(eventLoop, frequency);
}
timeOld = timeNew;
}
}
function sendMacro() {

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

@ -11,7 +11,7 @@
<header class="top-bar">
<div class="top-bar-wrapper">
<div>
<h1>NXBT</h1><div class="mono">v0.1.3</div>
<h1>NXBT</h1><div class="mono">v0.1.4</div>
</div>
<div id="status-indicator" class="hidden surface-lighter">
<div id="status-indicator-light" class="indicator-yellow"></div>
@ -70,7 +70,7 @@
</div>
</section>
<section id="controller-config" class="hidden surface">
<section id="controller-config" class="surface hidden">
<div id="controller-underlay-container">
<div class="controller-display hidden">
<div class="control-indicator" id="jl_"></div>
@ -120,6 +120,19 @@
a button on the gamepad before it shows up on the
input device list.
</p>
<h2 class="surface-lighter">Input Sampling Frequency</h2>
<select id="input-frequency" name="Select an Input Sampling Frequency" onchange="changeFrequency(event)">
<option value="RAF">Monitor Refresh Rate</option>
<option value="240">240Hz</option>
<option value="120">120Hz</option>
<option value="60">60Hz</option>
</select>
<p>
Note: "Monitor Refresh Rate" is a more stable but
(potentially) slower frequency. This loop uses the
"requestAnimationFrame()" function, rather than
"setTimeout()".
</p>
<h2 class="surface-lighter">Control Mapping</h2>
<div id="keyboard-map" class="surface-lighter">
<table>
@ -184,6 +197,7 @@
<h2 class="surface-lighter">Controller Commands</h2>
<button onclick="shutdownController()">Shutdown Controller</button>
<button onclick="recreateProController()">Recreate Controller</button>
<button onclick="restartController()">Restart Controller</button>
<h2 class="surface-lighter">Controller Macro</h2>
<p>

6
scripts/crash_switch.py
View file

@ -44,7 +44,7 @@ import os
import time
import fcntl
from nxbt import toggle_input_plugin
from nxbt import toggle_clean_bluez
from nxbt import BlueZ
from nxbt import Controller
from nxbt import PRO_CONTROLLER
@ -142,7 +142,7 @@ if __name__ == "__main__":
port_ctrl = 17
port_itr = 19
toggle_input_plugin(False)
toggle_clean_bluez(False)
bt = BlueZ(adapter_path="/org/bluez/hci0")
controller = Controller(bt, PRO_CONTROLLER)
@ -225,4 +225,4 @@ if __name__ == "__main__":
raise e
finally:
toggle_input_plugin(True)
toggle_clean_bluez(True)

107
scripts/demo_loop.py Normal file
View file

@ -0,0 +1,107 @@
from random import randint
from time import sleep
from nxbt import Nxbt, PRO_CONTROLLER
MACRO = """
B 0.1s
0.5s
B 0.1s
0.5s
B 0.1s
0.5s
B 0.1s
1.5s
DPAD_RIGHT 0.075s
0.075s
A 0.1s
1.5s
LOOP 12
DPAD_DOWN 0.075s
0.075s
A 0.1s
0.25s
LOOP 3
DPAD_DOWN 0.1s
0.1s
DPAD_DOWN 0.54s
0.1s
A 0.1s
0.25s
L_STICK_PRESS 0.1s
1.0s
L_STICK@-100+000 1.0s
L_STICK@+000+100 1.0s
L_STICK@+100+000 1.0s
L_STICK@+000-100 1.0s
B 0.1s
0.25s
R_STICK_PRESS 0.1s
1.0s
R_STICK@-100+000 1.0s
R_STICK@+000+100 1.0s
R_STICK@+100+000 1.0s
R_STICK@+000-100 1.0s
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.75s
A 0.1s
5.0s
"""
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(debug=False)
adapters = nx.get_available_adapters()
if len(adapters) < 1:
raise OSError("Unable to detect any Bluetooth 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
for i in range(100):
print(f"Running Demo: Iteration {i}")
macro_id = nx.macro(controller_idxs[-1], MACRO, block=False)
while macro_id not in nx.state[controller_idxs[-1]]["finished_macros"]:
state = nx.state[controller_idxs[-1]]
if state['state'] == 'crashed':
print("An error occurred while running the demo:")
print(state['errors'])
exit(1)
sleep(1.0)
print("Finished!")
if __name__ == "__main__":
demo()

47
scripts/proxy.py
View file

@ -35,7 +35,7 @@ import time
import fcntl
from time import perf_counter
from nxbt import toggle_input_plugin
from nxbt import toggle_clean_bluez
from nxbt import BlueZ
from nxbt import Controller
from nxbt import JOYCON_L, JOYCON_R, PRO_CONTROLLER
@ -116,7 +116,8 @@ def write_to_buffer(buffer, message, message_type):
if __name__ == "__main__":
# Switch Controller Bluetooth MAC Address goes here
jc_MAC = "98:B6:E9:B0:05:E7"
# jc_MAC = "98:B6:E9:B0:05:E7"
jc_MAC = "7C:BB:8A:FA:41:3D"
# Specify the type of controller here
controller_type = PRO_CONTROLLER
if controller_type == JOYCON_L:
@ -130,7 +131,7 @@ if __name__ == "__main__":
port_itr = 19
message_buffer = []
toggle_input_plugin(False)
toggle_clean_bluez(True)
bt = BlueZ(adapter_path="/org/bluez/hci0")
controller = Controller(bt, controller_type)
@ -150,6 +151,12 @@ if __name__ == "__main__":
switch_ctrl = socket.socket(family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
switch_test1 = socket.socket(family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
switch_test3 = socket.socket(family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
try:
# Remove the device before we try to re-pair
@ -179,6 +186,23 @@ if __name__ == "__main__":
jc_itr.connect((jc_MAC, port_itr))
print("Got connection.")
# print("Bruteforcing sockets")
# for i in range(1,99999,2):
# try:
# test_socket = socket.socket(family=socket.AF_BLUETOOTH,
# type=socket.SOCK_SEQPACKET,
# proto=socket.BTPROTO_L2CAP)
# test_socket.settimeout(3)
# # print("Connecting to", i)
# test_socket.connect((jc_MAC, i))
# print("!!!!!! Got connection to", i)
# except Exception as e:
# # print(str(e))
# pass
# switch_test1.bind((bt.address, 1))
# switch_test3.bind((bt.address, 3))
switch_ctrl.bind((bt.address, port_ctrl))
switch_itr.bind((bt.address, port_itr))
@ -285,9 +309,9 @@ if __name__ == "__main__":
except KeyboardInterrupt:
print("Closing sockets")
time_new = perf_counter()
print(f"Total Delta: {(time_new - time_old) * 1000}")
print(f"Timer Counter: {timer_counter}")
# 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()
@ -299,11 +323,6 @@ if __name__ == "__main__":
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")
@ -313,7 +332,11 @@ if __name__ == "__main__":
switch_itr.close()
switch_ctrl.close()
# Write the buffer
with open("messages.txt", "w") as f:
f.write("\n".join(message_buffer))
raise e
finally:
toggle_input_plugin(True)
toggle_clean_bluez(False)

6
scripts/reconnect_proxy.py
View file

@ -16,7 +16,7 @@ import time
import fcntl
from time import perf_counter
from nxbt import toggle_input_plugin
from nxbt import toggle_clean_bluez
from nxbt import BlueZ
from nxbt import Controller
from nxbt import JOYCON_L, JOYCON_R, PRO_CONTROLLER
@ -112,7 +112,7 @@ if __name__ == "__main__":
port_itr = 19
message_buffer = []
toggle_input_plugin(False)
toggle_clean_bluez(True)
bt = BlueZ(adapter_path="/org/bluez/hci0")
controller = Controller(bt, controller_type)
@ -288,4 +288,4 @@ if __name__ == "__main__":
raise e
finally:
toggle_input_plugin(True)
toggle_clean_bluez(False)

6
scripts/reconnect_test.py
View file

@ -26,7 +26,7 @@ import os
import time
import fcntl
from nxbt import toggle_input_plugin
from nxbt import toggle_clean_bluez
from nxbt import BlueZ
from nxbt import Controller
from nxbt import PRO_CONTROLLER
@ -125,7 +125,7 @@ if __name__ == "__main__":
port_ctrl = 17
port_itr = 19
toggle_input_plugin(False)
toggle_clean_bluez(True)
bt = BlueZ(adapter_path="/org/bluez/hci0")
controller = Controller(bt, PRO_CONTROLLER)
@ -199,4 +199,4 @@ if __name__ == "__main__":
raise e
finally:
toggle_input_plugin(True)
toggle_clean_bluez(False)

6
scripts/switch_emu.py
View file

@ -11,7 +11,7 @@ import sys
import os
import time
from nxbt import toggle_input_plugin
from nxbt import toggle_clean_bluez
from nxbt import BlueZ
REQUEST_INFO = b'\xA2\x01\x02\x00\x00\x00\x00\x00\x00\x00\x00\x02\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\x00'
@ -126,7 +126,7 @@ if __name__ == "__main__":
jc_itr = socket.socket(family=socket.AF_BLUETOOTH,
type=socket.SOCK_SEQPACKET,
proto=socket.BTPROTO_L2CAP)
toggle_input_plugin(False)
toggle_clean_bluez(True)
bt = BlueZ(adapter_path="/org/bluez/hci0")
try:
@ -194,4 +194,4 @@ if __name__ == "__main__":
raise e
finally:
toggle_input_plugin(True)
toggle_clean_bluez(False)

2
setup.cfg
View file

@ -1,6 +1,6 @@
[metadata]
name = nxbt
version = 0.1.3
version = 0.1.4
author = Reece Walsh
author-email = reece@brikwerk.com
project_urls =

1
setup.py
View file

@ -12,6 +12,7 @@ setup(
"blessed==1.17.10",
"pynput==1.7.1",
"psutil==5.6.6",
"cryptography==3.0",
],
extra_require={
"dev": [

32
template_vagrantfile Normal file
View file

@ -0,0 +1,32 @@
# -*- mode: ruby -*-
# vi: set ft=ruby :
Vagrant.configure("2") do |config|
config.vm.box = "generic/ubuntu2010"
# Publically forwarded ports.
# The below ports are accessible to all machines on the same network.
# To limit access to the local network, add "host_ip".
# Eg: config.vm.network "forwarded_port", guest: 80, host: 8080, host_ip: "127.0.0.1"
config.vm.network "forwarded_port", guest: 8000, host: 8000 # Web App
config.vm.network "forwarded_port", guest: 9000, host: 9000 # Remote server
# Explicitly create a shared folder of the Vagrantfile directory at /vagrant in the VM
config.vm.synced_folder ".", "/vagrant"
# Install NXBT
config.vm.provision "shell", inline: <<-SHELL
apt-get update
apt-get install -y python3-pip bluez bluetooth pkg-config build-essential libdbus-glib-1-dev libgirepository1.0-dev
{{SHELL_CONFIG}}
SHELL
# Enable USB Controller on VirtualBox
config.vm.provider "virtualbox" do |vb|
vb.memory = "2048"
vb.cpus = 2
vb.customize ["modifyvm", :id, "--usb", "on"]
vb.customize ["modifyvm", :id, "--usbehci", "on"]
{{USB_FILTER}}
end
end

139
vagrant_setup.py Normal file
View file

@ -0,0 +1,139 @@
import re
from shutil import which
import subprocess
import os
def find_line_items(identifier, input_string):
pattern = re.compile(fr"(?<={re.escape(identifier)}: ).*.")
matches = pattern.findall(input_string)
matches = list(map(str.strip, matches))
return matches
def get_usb_devices():
usb_string = subprocess.check_output(['VBoxManage', 'list', 'usbhost'])
usb_string = usb_string.decode("utf-8").replace('r', '')
usb_devices = usb_string.split("\n\n")
devices = []
for device in usb_devices:
productid = find_line_items("ProductId", device)
vendorid = find_line_items("VendorId", device)
manufacturer = find_line_items("Manufacturer", device)
product = find_line_items("Product", device)
if (len(productid) < 1 or len(vendorid) < 1 or
len(manufacturer) < 1 or len(product) < 1):
continue
productid = productid[0]
vendorid = vendorid[0]
manufacturer = manufacturer[0]
product = product[0]
if len(productid) != 13 or len(vendorid) != 13:
continue
devices.append({
'product': product,
'manufacturer': manufacturer,
'productid': productid[8:12],
'vendorid': vendorid[8:12]
})
return devices
def is_cli(cli_string):
return which(cli_string) is not None
def check_cli(name, cli_string, msg=None):
print(name, end="")
if is_cli(cli_string):
print(" [OK]")
else:
print(" [ERROR]")
print(f" -> {name} wasn't found on your system")
if msg is not None:
print(msg)
exit(1)
GH_SHELL_CONFIG = """cd /vagrant
pip3 install -e ."""
PYPI_SHELL_CONFIG = """pip3 install nxbt"""
if __name__ == "__main__":
print("Checking for the required utilities...")
vg_msg = (" Please ensure that Vagrant is installed and available on\n"
" your system path.")
check_cli("Vagrant", "vagrant", msg=vg_msg)
vb_msg = (" VBoxManage (part of the VirtualBox CLI) wasn't found\n"
" on your system path. Please ensure that VirtualBox is\n"
" installed and VBoxManage is on your system path.")
check_cli("VirtualBox", "VBoxManageaa", msg=vb_msg)
print("")
print("---")
print("Welcome to the nxbt-vagrant setup.")
print("As part of the first step in this process, you will "
"select the USB Bluetooth adapter that will be used with NXBT.")
print("Please ensure that your adapter is plugged into this computer.")
print("---")
input("Press the enter key to continue.")
print("")
print("USB Devices:")
print("---")
devices = get_usb_devices()
for i, device in enumerate(devices):
print(f"{i:3}. {device['product']} ({device['manufacturer']})")
print()
# Choose a USB Bluetooth adapter
invalid_choice = True
while invalid_choice:
usb_choice = input(
f"Please choose your Bluetooth USB Adapter from the above list [0-{len(devices)-1}]: ")
if usb_choice.isdigit() and int(usb_choice) < len(devices):
invalid_choice = False
else:
print(f"Invalid choice. Please choose a number from 0 to {len(devices)-1}.")
adapter_info = devices[int(usb_choice)]
print("")
# Choose how to install NXBT (PyPi or Github)
invalid_choice = True
while invalid_choice:
install_choice = input(
"Would you like to install NXBT from (1) PyPi or (2) install from local files? (1/2) ")
if install_choice in ['1', '2']:
invalid_choice = False
else:
print("Invalid choice. Please choose PyPi (1) or Github clone/install (2)")
print("")
print("Configuring...")
with open("template_vagrantfile", "r") as f:
vagrantfile = f.read()
vb_usb_filter = f"""vb.customize ["usbfilter", "add", "0",
"--target", :id,
"--name", "{adapter_info['product']} ({adapter_info['manufacturer']})",
"--product", "{adapter_info['product']}",
"--manufacturer", "{adapter_info['manufacturer']}",
"--productid", "{adapter_info['productid']}",
"--vendorid", "{adapter_info['vendorid']}",]"""
vagrantfile = vagrantfile.replace("{{USB_FILTER}}", vb_usb_filter)
if install_choice == '1':
vagrantfile = vagrantfile.replace("{{SHELL_CONFIG}}", PYPI_SHELL_CONFIG)
else:
vagrantfile = vagrantfile.replace("{{SHELL_CONFIG}}", GH_SHELL_CONFIG)
with open("Vagrantfile", "w") as f:
f.write(vagrantfile)
print("Done!")
print("")
print("You can now start the NXBT Vagrant Box with 'vagrant up'.")
print("After booting up, the Vagrant Box can be access with 'vagrant ssh'.")