Files
PyGame-Snake/controls/Gamepad/Gamepad.py
2025-03-21 15:13:03 +01:00

618 lines
24 KiB
Python
Executable File

#!/usr/bin/env python
# coding: utf-8
"""
This module is designed to read inputs from a gamepad or joystick.
See Controllers.py the names which can be used with specific gamepad types.
For basic use see the following examples:
AsyncExample.py - Updates read in the background.
EventExample.py - Updates passed to callback functions.
PollingExample.py - Reading updates one at a time.
AsyncAndEventExample.py - Mixing callbacks and background updates.
"""
import os
import sys
import struct
import time
import threading
import inspect
def available(joystickNumber = 0):
"""Check if a joystick is connected and ready to use."""
joystickPath = '/dev/input/js' + str(joystickNumber)
return os.path.exists(joystickPath)
class Gamepad:
EVENT_CODE_BUTTON = 0x01
EVENT_CODE_AXIS = 0x02
EVENT_CODE_INIT_BUTTON = 0x80 | EVENT_CODE_BUTTON
EVENT_CODE_INIT_AXIS = 0x80 | EVENT_CODE_AXIS
MIN_AXIS = -32767.0
MAX_AXIS = +32767.0
EVENT_BUTTON = 'BUTTON'
EVENT_AXIS = 'AXIS'
fullName = 'Generic (numbers only)'
class UpdateThread(threading.Thread):
"""Thread used to continually run the updateState function on a Gamepad in the background
One of these is created by the Gamepad startBackgroundUpdates function and closed by stopBackgroundUpdates"""
def __init__(self, gamepad):
threading.Thread.__init__(self)
if isinstance(gamepad, Gamepad):
self.gamepad = gamepad
else:
raise ValueError('Gamepad update thread was not created with a valid Gamepad object')
self.running = True
def run(self):
try:
while self.running:
self.gamepad.updateState()
self.gamepad = None
except:
self.running = False
self.gamepad = None
raise
def __init__(self, joystickNumber = 0):
self.joystickNumber = str(joystickNumber)
self.joystickPath = '/dev/input/js' + self.joystickNumber
retryCount = 5
while True:
try:
self.joystickFile = open(self.joystickPath, 'rb')
break
except IOError as e:
retryCount -= 1
if retryCount > 0:
time.sleep(0.5)
else:
raise IOError('Could not open gamepad %s: %s' % (self.joystickNumber, str(e)))
self.eventSize = struct.calcsize('IhBB')
self.pressedMap = {}
self.wasPressedMap = {}
self.wasReleasedMap = {}
self.axisMap = {}
self.buttonNames = {}
self.buttonIndex = {}
self.axisNames = {}
self.axisIndex = {}
self.lastTimestamp = 0
self.updateThread = None
self.connected = True
self.pressedEventMap = {}
self.releasedEventMap = {}
self.changedEventMap = {}
self.movedEventMap = {}
def __del__(self):
try:
self.joystickFile.close()
except AttributeError:
pass
def _setupReverseMaps(self):
for index in self.buttonNames:
self.buttonIndex[self.buttonNames[index]] = index
for index in self.axisNames:
self.axisIndex[self.axisNames[index]] = index
def _getNextEventRaw(self):
"""Returns the next raw event from the gamepad.
The return format is:
timestamp (ms), value, event type code, axis / button number
Throws an IOError if the gamepad is disconnected"""
if self.connected:
try:
rawEvent = self.joystickFile.read(self.eventSize)
except IOError as e:
self.connected = False
raise IOError('Gamepad %s disconnected: %s' % (self.joystickNumber, str(e)))
if rawEvent is None:
self.connected = False
raise IOError('Gamepad %s disconnected' % self.joystickNumber)
else:
return struct.unpack('IhBB', rawEvent)
else:
raise IOError('Gamepad has been disconnected')
def _rawEventToDescription(self, event):
"""Decodes the raw event from getNextEventRaw into a formatted string."""
timestamp, value, eventType, index = event
if eventType == Gamepad.EVENT_CODE_BUTTON:
if index in self.buttonNames:
button = self.buttonNames[index]
else:
button = str(index)
if value == 0:
return '%010u: Button %s released' % (timestamp, button)
elif value == 1:
return '%010u: button %s pressed' % (timestamp, button)
else:
return '%010u: button %s state %i' % (timestamp, button, value)
elif eventType == Gamepad.EVENT_CODE_AXIS:
if index in self.axisNames:
axis = self.axisNames[index]
else:
axis = str(index)
position = value / Gamepad.MAX_AXIS
return '%010u: Axis %s at %+06.1f %%' % (timestamp, axis, position * 100)
elif eventType == Gamepad.EVENT_CODE_INIT_BUTTON:
if index in self.buttonNames:
button = self.buttonNames[index]
else:
button = str(index)
if value == 0:
return '%010u: Button %s initially released' % (timestamp, button)
elif value == 1:
return '%010u: button %s initially pressed' % (timestamp, button)
else:
return '%010u: button %s initially state %i' % (timestamp, button, value)
elif eventType == Gamepad.EVENT_CODE_INIT_AXIS:
if index in self.axisNames:
axis = self.axisNames[index]
else:
axis = str(index)
position = value / Gamepad.MAX_AXIS
return '%010u: Axis %s initially at %+06.1f %%' % (timestamp, axis, position * 100)
else:
return '%010u: Unknown event %u, Index %u, Value %i' % (timestamp, eventType, index, value)
def getNextEvent(self, skipInit = True):
"""Returns the next event from the gamepad.
The return format is:
event name, entity name, value
For button events the event name is BUTTON and value is either True or False.
For axis events the event name is AXIS and value is between -1.0 and +1.0.
Names are string based when found in the button / axis decode map.
When not available the raw index is returned as an integer instead.
After each call the internal state used by getPressed and getAxis is updated.
Throws an IOError if the gamepad is disconnected"""
self.lastTimestamp, value, eventType, index = self._getNextEventRaw()
skip = False
eventName = None
entityName = None
finalValue = None
if eventType == Gamepad.EVENT_CODE_BUTTON:
eventName = Gamepad.EVENT_BUTTON
if index in self.buttonNames:
entityName = self.buttonNames[index]
else:
entityName = index
if value == 0:
finalValue = False
self.wasReleasedMap[index] = True
for callback in self.releasedEventMap[index]:
callback()
else:
finalValue = True
self.wasPressedMap[index] = True
for callback in self.pressedEventMap[index]:
callback()
self.pressedMap[index] = finalValue
for callback in self.changedEventMap[index]:
callback(finalValue)
elif eventType == Gamepad.EVENT_CODE_AXIS:
eventName = Gamepad.EVENT_AXIS
if index in self.axisNames:
entityName = self.axisNames[index]
else:
entityName = index
finalValue = value / Gamepad.MAX_AXIS
self.axisMap[index] = finalValue
for callback in self.movedEventMap[index]:
callback(finalValue)
elif eventType == Gamepad.EVENT_CODE_INIT_BUTTON:
eventName = Gamepad.EVENT_BUTTON
if index in self.buttonNames:
entityName = self.buttonNames[index]
else:
entityName = index
if value == 0:
finalValue = False
else:
finalValue = True
self.pressedMap[index] = finalValue
self.wasPressedMap[index] = False
self.wasReleasedMap[index] = False
self.pressedEventMap[index] = []
self.releasedEventMap[index] = []
self.changedEventMap[index] = []
skip = skipInit
elif eventType == Gamepad.EVENT_CODE_INIT_AXIS:
eventName = Gamepad.EVENT_AXIS
if index in self.axisNames:
entityName = self.axisNames[index]
else:
entityName = index
finalValue = value / Gamepad.MAX_AXIS
self.axisMap[index] = finalValue
self.movedEventMap[index] = []
skip = skipInit
else:
skip = True
if skip:
return self.getNextEvent()
else:
return eventName, entityName, finalValue
def updateState(self):
"""Updates the internal button and axis states with the next pending event.
This call waits for a new event if there are not any waiting to be processed."""
self.lastTimestamp, value, eventType, index = self._getNextEventRaw()
if eventType == Gamepad.EVENT_CODE_BUTTON:
if value == 0:
finalValue = False
self.wasReleasedMap[index] = True
for callback in self.releasedEventMap[index]:
callback()
else:
finalValue = True
self.wasPressedMap[index] = True
for callback in self.pressedEventMap[index]:
callback()
self.pressedMap[index] = finalValue
for callback in self.changedEventMap[index]:
callback(finalValue)
elif eventType == Gamepad.EVENT_CODE_AXIS:
finalValue = value / Gamepad.MAX_AXIS
self.axisMap[index] = finalValue
for callback in self.movedEventMap[index]:
callback(finalValue)
elif eventType == Gamepad.EVENT_CODE_INIT_BUTTON:
if value == 0:
finalValue = False
else:
finalValue = True
self.pressedMap[index] = finalValue
self.wasPressedMap[index] = False
self.wasReleasedMap[index] = False
self.pressedEventMap[index] = []
self.releasedEventMap[index] = []
self.changedEventMap[index] = []
elif eventType == Gamepad.EVENT_CODE_INIT_AXIS:
finalValue = value / Gamepad.MAX_AXIS
self.axisMap[index] = finalValue
self.movedEventMap[index] = []
def startBackgroundUpdates(self, waitForReady = True):
"""Starts a background thread which keeps the gamepad state updated automatically.
This allows for asynchronous gamepad updates and event callback code.
Do not use with getNextEvent"""
if self.updateThread is not None:
if self.updateThread.running:
raise RuntimeError('Called startBackgroundUpdates when the update thread is already running')
self.updateThread = Gamepad.UpdateThread(self)
self.updateThread.start()
if waitForReady:
while not self.isReady() and self.connected:
time.sleep(1.0)
def stopBackgroundUpdates(self):
"""Stops the background thread which keeps the gamepad state updated automatically.
This may be called even if the background thread was never started.
The thread will stop on the next event after this call was made."""
if self.updateThread is not None:
self.updateThread.running = False
def isReady(self):
"""Used with updateState to indicate that the gamepad is now ready for use.
This is usually after the first button press or stick movement."""
return len(self.axisMap) + len(self.pressedMap) > 1
def waitReady(self):
"""Convenience function which waits until the isReady call is True."""
self.updateState()
while not self.isReady() and self.connected:
time.sleep(1.0)
self.updateState()
def isPressed(self, buttonName):
"""Returns the last observed state of a gamepad button specified by name or index.
True if pressed, False if not pressed.
Status is updated by getNextEvent calls.
Throws ValueError if the button name or index cannot be found."""
try:
if buttonName in self.buttonIndex:
buttonIndex = self.buttonIndex[buttonName]
else:
buttonIndex = int(buttonName)
return self.pressedMap[buttonIndex]
except KeyError:
raise ValueError('Button %i was not found' % buttonIndex)
except ValueError:
raise ValueError('Button name %s was not found' % buttonName)
def beenPressed(self, buttonName):
"""Returns True if the button specified by name or index has been pressed since the last beenPressed call.
Used in conjunction with updateState.
Throws ValueError if the button name or index cannot be found."""
try:
if buttonName in self.buttonIndex:
buttonIndex = self.buttonIndex[buttonName]
else:
buttonIndex = int(buttonName)
if self.wasPressedMap[buttonIndex]:
self.wasPressedMap[buttonIndex] = False
return True
else:
return False
except KeyError:
raise ValueError('Button %i was not found' % buttonIndex)
except ValueError:
raise ValueError('Button name %s was not found' % buttonName)
def beenReleased(self, buttonName):
"""Returns True if the button specified by name or index has been released since the last beenReleased call.
Used in conjunction with updateState.
Throws ValueError if the button name or index cannot be found."""
try:
if buttonName in self.buttonIndex:
buttonIndex = self.buttonIndex[buttonName]
else:
buttonIndex = int(buttonName)
if self.wasReleasedMap[buttonIndex]:
self.wasReleasedMap[buttonIndex] = False
return True
else:
return False
except KeyError:
raise ValueError('Button %i was not found' % buttonIndex)
except ValueError:
raise ValueError('Button name %s was not found' % buttonName)
def axis(self, axisName):
"""Returns the last observed state of a gamepad axis specified by name or index.
Throws a ValueError if the axis index is unavailable.
Status is updated by getNextEvent calls.
Throws ValueError if the button name or index cannot be found."""
try:
if axisName in self.axisIndex:
axisIndex = self.axisIndex[axisName]
else:
axisIndex = int(axisName)
return self.axisMap[axisIndex]
except KeyError:
raise ValueError('Axis %i was not found' % axisIndex)
except ValueError:
raise ValueError('Axis name %s was not found' % axisName)
def availableButtonNames(self):
"""Returns a list of available button names for this gamepad.
An empty list means that no button mapping has been provided."""
return self.buttonIndex.keys()
def availableAxisNames(self):
"""Returns a list of available axis names for this gamepad.
An empty list means that no axis mapping has been provided."""
return self.axisIndex.keys()
def isConnected(self):
"""Returns True until reading from the device fails."""
return self.connected
def addButtonPressedHandler(self, buttonName, callback):
"""Adds a callback for when a specific button specified by name or index is pressed.
This callback gets no parameters passed."""
try:
if buttonName in self.buttonIndex:
buttonIndex = self.buttonIndex[buttonName]
else:
buttonIndex = int(buttonName)
if callback not in self.pressedEventMap[buttonIndex]:
self.pressedEventMap[buttonIndex].append(callback)
except KeyError:
raise ValueError('Button %i was not found' % buttonIndex)
except ValueError:
raise ValueError('Button name %s was not found' % buttonName)
def removeButtonPressedHandler(self, buttonName, callback):
"""Removes a callback for when a specific button specified by name or index is pressed."""
try:
if buttonName in self.buttonIndex:
buttonIndex = self.buttonIndex[buttonName]
else:
buttonIndex = int(buttonName)
if callback in self.pressedEventMap[buttonIndex]:
self.pressedEventMap[buttonIndex].remove(callback)
except KeyError:
raise ValueError('Button %i was not found' % buttonIndex)
except ValueError:
raise ValueError('Button name %s was not found' % buttonName)
def addButtonReleasedHandler(self, buttonName, callback):
"""Adds a callback for when a specific button specified by name or index is released.
This callback gets no parameters passed."""
try:
if buttonName in self.buttonIndex:
buttonIndex = self.buttonIndex[buttonName]
else:
buttonIndex = int(buttonName)
if callback not in self.releasedEventMap[buttonIndex]:
self.releasedEventMap[buttonIndex].append(callback)
except KeyError:
raise ValueError('Button %i was not found' % buttonIndex)
except ValueError:
raise ValueError('Button name %s was not found' % buttonName)
def removeButtonReleasedHandler(self, buttonName, callback):
"""Removes a callback for when a specific button specified by name or index is released."""
try:
if buttonName in self.buttonIndex:
buttonIndex = self.buttonIndex[buttonName]
else:
buttonIndex = int(buttonName)
if callback in self.releasedEventMap[buttonIndex]:
self.releasedEventMap[buttonIndex].remove(callback)
except KeyError:
raise ValueError('Button %i was not found' % buttonIndex)
except ValueError:
raise ValueError('Button name %s was not found' % buttonName)
def addButtonChangedHandler(self, buttonName, callback):
"""Adds a callback for when a specific button specified by name or index changes.
This callback gets a boolean for the button pressed state."""
try:
if buttonName in self.buttonIndex:
buttonIndex = self.buttonIndex[buttonName]
else:
buttonIndex = int(buttonName)
if callback not in self.changedEventMap[buttonIndex]:
self.changedEventMap[buttonIndex].append(callback)
except KeyError:
raise ValueError('Button %i was not found' % buttonIndex)
except ValueError:
raise ValueError('Button name %s was not found' % buttonName)
def removeButtonChangedHandler(self, buttonName, callback):
"""Removes a callback for when a specific button specified by name or index changes."""
try:
if buttonName in self.buttonIndex:
buttonIndex = self.buttonIndex[buttonName]
else:
buttonIndex = int(buttonName)
if callback in self.changedEventMap[buttonIndex]:
self.changedEventMap[buttonIndex].remove(callback)
except KeyError:
raise ValueError('Button %i was not found' % buttonIndex)
except ValueError:
raise ValueError('Button name %s was not found' % buttonName)
def addAxisMovedHandler(self, axisName, callback):
"""Adds a callback for when a specific axis specified by name or index changes.
This callback gets the updated position of the axis."""
try:
if axisName in self.axisIndex:
axisIndex = self.axisIndex[axisName]
else:
axisIndex = int(axisName)
if callback not in self.movedEventMap[axisIndex]:
self.movedEventMap[axisIndex].append(callback)
except KeyError:
raise ValueError('Button %i was not found' % axisIndex)
except ValueError:
raise ValueError('Button name %s was not found' % axisName)
def removeAxisMovedHandler(self, axisName, callback):
"""Removes a callback for when a specific axis specified by name or index changes."""
try:
if axisName in self.axisIndex:
axisIndex = self.axisIndex[axisName]
else:
axisIndex = int(axisName)
if callback in self.movedEventMap[axisIndex]:
self.movedEventMap[axisIndex].remove(callback)
except KeyError:
raise ValueError('Button %i was not found' % axisIndex)
except ValueError:
raise ValueError('Button name %s was not found' % axisName)
def removeAllEventHandlers(self):
"""Removes all event handlers from all axes and buttons."""
for index in self.pressedEventMap.keys():
self.pressedEventMap[index] = []
self.releasedEventMap[index] = []
self.changedEventMap[index] = []
self.movedEventMap[index] = []
def disconnect(self):
"""Cleanly disconnect and remove any threads and event handlers."""
self.connected = False
self.removeAllEventHandlers()
self.stopBackgroundUpdates()
del self.joystickFile
###########################
# Import gamepad mappings #
###########################
scriptDir = os.path.dirname(os.path.realpath(__file__))
controllerScript = os.path.join(scriptDir, "Controllers.py")
exec(open(controllerScript).read())
# Generate a list of available gamepad types
moduleDict = globals()
classList = [moduleDict[a] for a in moduleDict.keys() if inspect.isclass(moduleDict[a])]
controllerDict = {}
deviceNames = []
for gamepad in classList:
controllerDict[gamepad.__name__.upper()] = gamepad
deviceNames.append(gamepad.__name__)
deviceNames.sort()
##################################################################
# When this script is run it provides testing code for a gamepad #
##################################################################
if __name__ == "__main__":
# Python 2/3 compatibility
try:
input = raw_input
except NameError:
pass
# ANSI colour code sequences
GREEN = '\033[0;32m'
CYAN = '\033[0;36m'
BLUE = '\033[1;34m'
RESET = '\033[0m'
# Ask for the gamepad to use
print('Gamepad axis and button events...')
print('Press CTRL+C to exit')
print('')
print('Available device names:')
formatString = ' ' + GREEN + '%s' + RESET + ' - ' + CYAN + '%s' + RESET
for device in deviceNames:
print(formatString % (device, controllerDict[device.upper()].fullName))
print('')
print('What device name are you using (leave blank if not in the list)')
device = input('? ' + GREEN).strip().upper()
print(RESET)
# Wait for a connection
if not available():
print('Please connect your gamepad...')
while not available():
time.sleep(1.0)
print('Gamepad connected')
# Pick the correct class
if device in controllerDict:
print(controllerDict[device].fullName)
gamepad = controllerDict[device]()
elif device == '':
print('Unspecified gamepad')
print('')
gamepad = Gamepad()
else:
print('Unknown gamepad')
print('')
sys.exit()
# Display the event messages as they arrive
while True:
eventType, index, value = gamepad.getNextEvent()
print(BLUE + eventType + RESET + ',\t ' +
GREEN + str(index) + RESET + ',\t' +
CYAN + str(value) + RESET)
time.sleep(1.0)