Added docstrings and adjust reconnect arguments
This commit is contained in:
Brikwerk
parent
3545613a92
commit
47b6ada12d
1 changed files with 267 additions and 50 deletions
317
nxbt/nxbt.py
317
nxbt/nxbt.py
|
|
@ -20,6 +20,8 @@ PRO_CONTROLLER = ControllerTypes.PRO_CONTROLLER
|
|||
|
||||
|
||||
class Buttons():
|
||||
"""The button object containing the button string constants.
|
||||
"""
|
||||
|
||||
Y = 'Y'
|
||||
X = 'X'
|
||||
|
|
@ -46,12 +48,17 @@ class Buttons():
|
|||
|
||||
|
||||
class Sticks():
|
||||
"""The sticks object containing the joystick string constants.
|
||||
"""
|
||||
|
||||
RIGHT_STICK = "R_STICK"
|
||||
LEFT_STICK = "L_STICK"
|
||||
|
||||
|
||||
class NxbtCommands(Enum):
|
||||
"""An enumeration containing the nxbt message
|
||||
commands.
|
||||
"""
|
||||
|
||||
CREATE_CONTROLLER = 0
|
||||
INPUT_MACRO = 1
|
||||
|
|
@ -63,6 +70,17 @@ class NxbtCommands(Enum):
|
|||
|
||||
|
||||
class Nxbt():
|
||||
"""The nxbt object implements the core multiprocessing logic
|
||||
and message passing API that acts as the central of the application.
|
||||
Upon creation, a multiprocessing Process is spun off to act at the
|
||||
manager for all emulated Nintendo Switch controllers. Messages
|
||||
are passed into a queue which is consumed and acted upon by the
|
||||
__command_manager__.
|
||||
|
||||
All function calls that interact or control the emulated controllers
|
||||
are simply message constructors that submit to the central task_queue.
|
||||
This allows for thread-safe control of emulated controllers.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
|
||||
|
|
@ -92,7 +110,7 @@ class Nxbt():
|
|||
toggle_input_plugin(False)
|
||||
|
||||
# Exit handler
|
||||
atexit.register(self.on_exit)
|
||||
atexit.register(self._on_exit)
|
||||
|
||||
# Starting the nxbt worker process
|
||||
self.controllers = Process(
|
||||
|
|
@ -104,7 +122,12 @@ class Nxbt():
|
|||
self.controllers.daemon = False
|
||||
self.controllers.start()
|
||||
|
||||
def on_exit(self):
|
||||
def _on_exit(self):
|
||||
"""The exit handler function used with the atexit module.
|
||||
This function attempts to gracefully exit by terminating
|
||||
all spun up multiprocessing Processes. This is done to
|
||||
ensure no zombie processes linger after exit.
|
||||
"""
|
||||
|
||||
# Need to explicitly kill the controllers process
|
||||
# since it isn't daemonized.
|
||||
|
|
@ -117,8 +140,20 @@ class Nxbt():
|
|||
toggle_input_plugin(True)
|
||||
|
||||
def __command_manager__(self, task_queue, state):
|
||||
"""Used as the main multiprocessing Process that is launched
|
||||
on startup to handle the message passing and instantiation of
|
||||
the controllers. Messages are pulled out of a Queue and passed
|
||||
as appropriately phrased function calls to the ControllerManager.
|
||||
|
||||
cm = ControllerManager(state, self.__bluetooth_lock__)
|
||||
:param task_queue: A multiprocessing Queue used as the source
|
||||
of messages
|
||||
:type task_queue: multiprocessing.Queue
|
||||
:param state: A dict used to store the shared state of the
|
||||
emulated controllers.
|
||||
:type state: multiprocessing.Manager().dict
|
||||
"""
|
||||
|
||||
cm = __ControllerManager__(state, self.__bluetooth_lock__)
|
||||
# Ensure a SystemExit exception is raised on SIGTERM
|
||||
# so that we can gracefully shutdown.
|
||||
signal.signal(signal.SIGTERM, lambda sigterm_handler: quit())
|
||||
|
|
@ -160,6 +195,29 @@ class Nxbt():
|
|||
quit()
|
||||
|
||||
def macro(self, controller_index, macro, block=True):
|
||||
"""Used to input a given macro on a specified controller.
|
||||
This is done by creating and passing an INPUT_MACRO
|
||||
message into the task queue with the given macro.
|
||||
|
||||
If block is set to True, this function waits until the
|
||||
macro_id (generated on the submission of the macro)
|
||||
shows up under the "finished_macros" list communicated
|
||||
under the controllers shared state.
|
||||
|
||||
:param controller_index: The index of a given controller
|
||||
:type controller_index: int
|
||||
:param macro: The series of button presses and timings
|
||||
to be passed to the controller
|
||||
:type macro: string
|
||||
:param block: A boolean variable indicating whether or not
|
||||
to block until the macro completes, defaults to True
|
||||
:type block: bool, optional
|
||||
:raises ValueError: If the controller_index does not exist
|
||||
:return: The generated ID of the passed macro. This ID
|
||||
will show up under the "finished_macros" list communicated
|
||||
in the controllers shared state.
|
||||
:rtype: str
|
||||
"""
|
||||
|
||||
if controller_index not in self.manager_state.keys():
|
||||
raise ValueError("Specified controller does not exist")
|
||||
|
|
@ -185,38 +243,67 @@ class Nxbt():
|
|||
|
||||
return macro_id
|
||||
|
||||
def press_buttons(self, controller_index, buttons, up=0.1, down=0.1, block=True):
|
||||
def press_buttons(self, controller_index, buttons, down=0.1, up=0.1, block=True):
|
||||
"""Used to press a given set of buttons on the controller for a
|
||||
specified up and down duration. This is done by inputting a macro
|
||||
configured with the specified button presses and timings.
|
||||
|
||||
if controller_index not in self.manager_state.keys():
|
||||
raise ValueError("Specified controller does not exist")
|
||||
:param controller_index: The index of a given controller
|
||||
:type controller_index: int
|
||||
:param buttons: A list of nxbt.Buttons
|
||||
:type buttons: list
|
||||
:param down: How long to hold the buttons down for
|
||||
in seconds, defaults to 0.1
|
||||
:type down: float, optional
|
||||
:param up: How long to release the button for
|
||||
in seconds, defaults to 0.1
|
||||
:type up: float, optional
|
||||
:param block: A boolean variable indicating whether or not
|
||||
to block until the macro completes, defaults to True
|
||||
:type block: bool, optional
|
||||
:return: The generated ID of the passed macro. This ID
|
||||
will show up under the "finished_macros" list communicated
|
||||
in the controllers shared state.
|
||||
:rtype: str
|
||||
"""
|
||||
|
||||
macro_buttons = " ".join(buttons)
|
||||
macro_times = f"{up}s \n{down}s"
|
||||
macro_times = f"{down}s \n{up}s"
|
||||
macro = macro_buttons + " " + macro_times
|
||||
|
||||
# Get a unique ID to identify the button press
|
||||
# so we can check when the controller is done inputting it
|
||||
macro_id = os.urandom(24).hex()
|
||||
self.task_queue.put({
|
||||
"command": NxbtCommands.INPUT_MACRO,
|
||||
"arguments": {
|
||||
"controller_index": controller_index,
|
||||
"macro": macro,
|
||||
"macro_id": macro_id,
|
||||
}
|
||||
})
|
||||
|
||||
if block:
|
||||
while True:
|
||||
finished = (self.manager_state
|
||||
[controller_index]["finished_macros"])
|
||||
if macro_id in finished:
|
||||
break
|
||||
macro_id = self.macro(controller_index, macro, block=block)
|
||||
|
||||
return macro_id
|
||||
|
||||
def tilt_stick(self, controller_index, stick, x, y,
|
||||
tilted=0.1, released=0.1, block=True):
|
||||
"""Used to tilt a given stick on the controller for a
|
||||
specified tilted and released duration. This is done by
|
||||
inputting a macro configured with the specified stick tilts
|
||||
and timings.
|
||||
|
||||
:param controller_index: The index of a given controller
|
||||
:type controller_index: int
|
||||
:param stick: The right or left nxbt.Stick
|
||||
:type stick: nxbt.Stick
|
||||
:param x: The positive or negative X-Axis of the stick on
|
||||
a 0 to 100 scale
|
||||
:type x: int
|
||||
:param y: The positive or negative Y-Axis of the stick on
|
||||
a 0 to 100 scale
|
||||
:type y: int
|
||||
:param tilted: The time the stick should remain tilted
|
||||
for, defaults to 0.1
|
||||
:type tilted: float, optional
|
||||
:param released: The time the stick should remain
|
||||
released for, defaults to 0.1
|
||||
:type released: float, optional
|
||||
:type block: bool, optional
|
||||
:return: The generated ID of the passed macro. This ID
|
||||
will show up under the "finished_macros" list communicated
|
||||
in the controllers shared state.
|
||||
:rtype: str
|
||||
"""
|
||||
|
||||
if controller_index not in self.manager_state.keys():
|
||||
raise ValueError("Specified controller does not exist")
|
||||
|
|
@ -233,28 +320,25 @@ class Nxbt():
|
|||
|
||||
macro = f'{stick}@{x_parsed}{y_parsed} {tilted}s\n{released}s'
|
||||
|
||||
# Get a unique ID to identify the button press
|
||||
# so we can check when the controller is done inputting it
|
||||
macro_id = os.urandom(24).hex()
|
||||
self.task_queue.put({
|
||||
"command": NxbtCommands.INPUT_MACRO,
|
||||
"arguments": {
|
||||
"controller_index": controller_index,
|
||||
"macro": macro,
|
||||
"macro_id": macro_id,
|
||||
}
|
||||
})
|
||||
|
||||
if block:
|
||||
while True:
|
||||
finished = (self.manager_state
|
||||
[controller_index]["finished_macros"])
|
||||
if macro_id in finished:
|
||||
break
|
||||
macro_id = self.macro(controller_index, macro, block=block)
|
||||
|
||||
return macro_id
|
||||
|
||||
def stop_macro(self, controller_index, macro_id, block=True):
|
||||
"""Used to stop a given macro by its macro ID. After
|
||||
the macro has been stopped, its macro ID will show up
|
||||
as a finished macro in the respective controllers
|
||||
"finished_macros" list communicated in its state.
|
||||
|
||||
:param controller_index: The index of a given controller
|
||||
:type controller_index: int
|
||||
:param macro_id: The ID of a given macro (queued or running)
|
||||
:type macro_id: str
|
||||
:param block: A boolean variable indicating whether or not
|
||||
to block until the macro is stopped, defaults to True
|
||||
:type block: bool, optional
|
||||
:raises ValueError: If the controller_index does not exist
|
||||
"""
|
||||
|
||||
if controller_index not in self.manager_state.keys():
|
||||
raise ValueError("Specified controller does not exist")
|
||||
|
|
@ -275,6 +359,13 @@ class Nxbt():
|
|||
break
|
||||
|
||||
def clear_macros(self, controller_index):
|
||||
"""Clears all running and queued macros on a specified
|
||||
controller.
|
||||
|
||||
:param controller_index: The index of a given controller
|
||||
:type controller_index: int
|
||||
:raises ValueError: If the controller_index does not exist
|
||||
"""
|
||||
|
||||
if controller_index not in self.manager_state.keys():
|
||||
raise ValueError("Specified controller does not exist")
|
||||
|
|
@ -287,19 +378,65 @@ class Nxbt():
|
|||
})
|
||||
|
||||
def clear_all_macros(self):
|
||||
"""Clears all running and queued macros on all
|
||||
controllers.
|
||||
"""
|
||||
|
||||
for controller in self.manager_state.keys():
|
||||
self.clear_macros(controller)
|
||||
|
||||
def create_controller(self, controller_type, adapter_path,
|
||||
def create_controller(self, controller_type, adapter_path=None,
|
||||
colour_body=None, colour_buttons=None,
|
||||
reconnect_address=None):
|
||||
"""Used to create a Nintendo Switch controller of a
|
||||
given type and colour on an (optionally) specified
|
||||
bluetooth adapter.
|
||||
|
||||
if adapter_path not in self.get_available_adapters():
|
||||
raise ValueError("Specified adapter is unavailable")
|
||||
If no Bluetooth adapter is specified, the first available
|
||||
adapter is used.
|
||||
|
||||
if adapter_path in self.__adapters_in_use__.keys():
|
||||
raise ValueError("Specified adapter in use")
|
||||
If the reconnect_address is specified, the controller
|
||||
will attempt to reconnect to the Switch, rather than
|
||||
simply letting any Switch connect to it. To ensure
|
||||
that the reconnect succeeds, the Switch must be on
|
||||
and *not* on the Change Grip/Order menu.
|
||||
|
||||
:param controller_type: The type of controller to create
|
||||
:type controller_type: ControllerTypes
|
||||
:param adapter_path: The DBus path to a given Bluetooth
|
||||
adapter, defaults to None
|
||||
:type adapter_path: str, optional
|
||||
:param colour_body: The body colour of the controller
|
||||
represented by a hexadecimal colour value (a list of
|
||||
three ints (0-255)), defaults to None
|
||||
:type colour_body: list, optional
|
||||
:param colour_buttons: The button colour of the controller
|
||||
represented by a hexadecimal colour value (a list of
|
||||
three ints (0-255)), defaults to None
|
||||
:type colour_buttons: list, optional
|
||||
:param reconnect_address: A previously connected to
|
||||
Switch's Bluetooth MAC address, defaults to None
|
||||
:type reconnect_address: str or list, optional
|
||||
:raises ValueError: If specified adapter is unavailable
|
||||
:raises ValueError: If specified adapter is in use
|
||||
:return: The index of the created controller
|
||||
:rtype: int
|
||||
"""
|
||||
if adapter_path:
|
||||
if adapter_path not in self.get_available_adapters():
|
||||
raise ValueError("Specified adapter is unavailable")
|
||||
|
||||
if adapter_path in self.__adapters_in_use__.keys():
|
||||
raise ValueError("Specified adapter in use")
|
||||
else:
|
||||
# Get all adapters we can use
|
||||
usable_adapters = list(
|
||||
set(self.get_available_adapters()) - set(self.__adapters_in_use__))
|
||||
if len(usable_adapters) > 0:
|
||||
# Use the first available adapter
|
||||
adapter_path = usable_adapters[0]
|
||||
else:
|
||||
raise ValueError("No adapters available")
|
||||
|
||||
controller_index = None
|
||||
try:
|
||||
|
|
@ -328,7 +465,8 @@ class Nxbt():
|
|||
if controller_index in self.manager_state.keys():
|
||||
state = self.manager_state[controller_index]
|
||||
if (state["state"] == "connecting" or
|
||||
state["state"] == "reconnecting"):
|
||||
state["state"] == "reconnecting" or
|
||||
state["state"] == "crashed"):
|
||||
break
|
||||
finally:
|
||||
self.__controller_lock__.release()
|
||||
|
|
@ -336,6 +474,12 @@ class Nxbt():
|
|||
return controller_index
|
||||
|
||||
def remove_controller(self, controller_index):
|
||||
"""Terminates and removes a given controller.
|
||||
|
||||
:param controller_index: The index of a given controller
|
||||
:type controller_index: int
|
||||
:raises ValueError: If controller does not exist
|
||||
"""
|
||||
|
||||
if controller_index not in self.manager_state.keys():
|
||||
raise ValueError("Specified controller does not exist")
|
||||
|
|
@ -355,11 +499,26 @@ class Nxbt():
|
|||
})
|
||||
|
||||
def wait_for_connection(self, controller_index):
|
||||
"""Blocks until a given controller is connected
|
||||
to a Nintendo Switch.
|
||||
|
||||
:param controller_index: The index of a given controller
|
||||
:type controller_index: int
|
||||
"""
|
||||
|
||||
while not self.state[controller_index]["state"] == "connected":
|
||||
if self.state[controller_index]["state"] == "crashed":
|
||||
raise OSError("The watched controller has crashe with error",
|
||||
self.state[controller_index]["errors"])
|
||||
pass
|
||||
|
||||
def get_available_adapters(self):
|
||||
"""Gets the DBus paths of all available Bluetooth
|
||||
adapters.
|
||||
|
||||
:return: A list of available adapter paths
|
||||
:rtype: list
|
||||
"""
|
||||
|
||||
bus = dbus.SystemBus()
|
||||
adapters = find_objects(bus, SERVICE_NAME, ADAPTER_INTERFACE)
|
||||
|
|
@ -367,16 +526,52 @@ class Nxbt():
|
|||
return adapters
|
||||
|
||||
def get_switch_addresses(self):
|
||||
"""Gets the Bluetooth MAC addresses of all
|
||||
previously connected Nintendo Switchs
|
||||
|
||||
:return: A list of Bluetooth MAC addresses
|
||||
:rtype: list
|
||||
"""
|
||||
|
||||
return (find_devices_by_alias("Nintendo Switch"))
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""The state of all created and running controllers.
|
||||
This state is read-only and is represented as a dict.
|
||||
The state dict's structure follows:
|
||||
|
||||
{
|
||||
"controller_index"
|
||||
{
|
||||
"state":
|
||||
"initializing" or
|
||||
"connecting" or
|
||||
"reconnecting" or
|
||||
"crashed"
|
||||
"finished_macros":
|
||||
A list of UUIDs
|
||||
"errors":
|
||||
A string with the crash error
|
||||
}
|
||||
}
|
||||
|
||||
:return: The state dict
|
||||
:rtype: dict
|
||||
"""
|
||||
|
||||
return self.manager_state
|
||||
|
||||
|
||||
class ControllerManager():
|
||||
class __ControllerManager__():
|
||||
"""Used as the manager for all controllers. Each controller is
|
||||
a daemon multiprocessing Process that the ControllerManager
|
||||
object creates and manages.
|
||||
|
||||
The ControllerManager object submits messages to the respective
|
||||
queues of each controller process for tasks such as macro submission
|
||||
or macro clearing/stopping.
|
||||
"""
|
||||
|
||||
def __init__(self, state, lock):
|
||||
|
||||
|
|
@ -389,6 +584,28 @@ class ControllerManager():
|
|||
def create_controller(self, index, controller_type, adapter_path,
|
||||
colour_body=None, colour_buttons=None,
|
||||
reconnect_address=None):
|
||||
"""Instantiates a given controller as a multiprocessing
|
||||
Process with a shared state dict and a task queue.
|
||||
|
||||
Configuration options are available in the form of
|
||||
controller colours.
|
||||
|
||||
:param index: The index of the controller
|
||||
:type index: int
|
||||
:param controller_type: The type of Nintendo Switch controller
|
||||
:type controller_type: ControllerTypes
|
||||
:param adapter_path: The DBus path to the Bluetooth adapter
|
||||
:type adapter_path: str
|
||||
:param colour_body: A list of three ints representing the hex
|
||||
colour of the controller, defaults to None
|
||||
:type colour_body: list, optional
|
||||
:param colour_buttons: A list of three ints representing the
|
||||
hex colour of the controller, defaults to None
|
||||
:type colour_buttons: list, optional
|
||||
:param reconnect_address: The address of a Nintendo Switch
|
||||
to reconnect to, defaults to None
|
||||
:type reconnect_address: str, optional
|
||||
"""
|
||||
|
||||
controller_queue = Queue()
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue