Source code for psychopy.iohub.client.keyboard

# -*- coding: utf-8 -*-
# Part of the PsychoPy library
# Copyright (C) 2012-2020 iSolver Software Solutions (C) 2021 Open Science Tools Ltd.
# Distributed under the terms of the GNU General Public License (GPL).

from collections import deque
import time
from ..client import ioHubDeviceView, ioEvent, DeviceRPC
from ..devices import DeviceEvent, Computer
from ..util import win32MessagePump
from ..devices.keyboard import KeyboardInputEvent
from ..constants import EventConstants, KeyboardConstants

#pylint: disable=protected-access

getTime = Computer.getTime
kb_cls_attr_names = KeyboardInputEvent.CLASS_ATTRIBUTE_NAMES
kb_mod_codes2labels = KeyboardConstants._modifierCodes2Labels

class KeyboardEvent(ioEvent):
    """
    Base class for KeyboardPress and KeyboardRelease events.

    Note that keyboard events can be compared using a single character
    basestring. For example::

        kb_evts = keyboard.getKeys(['a','b','c'])
        for event in kb_evts:
            if event.key == 'b' or event.char == 'b':
                return True
        return False

    can be written as:

        return 'b' in keyboard.getKeys()
    """
    _attrib_index = dict()
    _attrib_index['key'] = kb_cls_attr_names.index('key')
    _attrib_index['char'] = kb_cls_attr_names.index('char')
    _attrib_index['modifiers'] = kb_cls_attr_names.index('modifiers')

    def __init__(self, ioe_array):
        super(KeyboardEvent, self).__init__(ioe_array)
        for aname, aindex, in list(self._attrib_index.items()):
            setattr(self, '_%s'%aname, ioe_array[aindex])
        self._modifiers = kb_mod_codes2labels(self._modifiers)

    @property
    def key(self):
        return self._key
        
    @property
    def char(self):
        """The unicode value of the keyboard event, if available. This field is
        only populated when the keyboard event results in a character that
        could be printable.

        :return: unicode, '' if no char value is available for the event.

        """
        return self._char

    @property
    def modifiers(self):
        """
        A list of any modifier keys that were pressed when this keyboard event
        occurred.  Each element of the list contains a keyboard modifier string
        constant. Possible values are:

        * 'lctrl', 'rctrl'
        * 'lshift', 'rshift'
        * 'lalt', 'ralt' (labelled as 'option' keys on Apple Keyboards)
        * 'lcmd', 'rcmd' (map to the 'windows' key(s) on Windows keyboards)
        * 'menu'
        * 'capslock'
        * 'numlock'
        * 'function' (OS X only)
        * 'modhelp' (OS X only)

        If no modifiers were active when the event occurred, an empty list is
        returned.

        :return: tuple
        """
        return self._modifiers

    def __str__(self):
        pstr = ioEvent.__str__(self)
        return '{}, key: {} char: {}, modifiers: {}'.format(pstr, self.key,
                self.char, self.modifiers)

    def __eq__(self, v):
        if isinstance(v, KeyboardEvent):
            return self.key == v.key
        return self.key == v

    def __ne__(self, v):
        return not self.__eq__(v)


[docs]class KeyboardPress(KeyboardEvent): """An iohub Keyboard device key press event.""" def __init__(self, ioe_array): super(KeyboardPress, self).__init__(ioe_array)
[docs]class KeyboardRelease(KeyboardEvent): """An iohub Keyboard device key release event.""" _attrib_index = dict(KeyboardEvent._attrib_index) _attrib_index['duration'] = kb_cls_attr_names.index('duration') _attrib_index['press_event_id'] = kb_cls_attr_names.index('press_event_id') def __init__(self, ioe_array): super(KeyboardRelease, self).__init__(ioe_array) #self._duration = ioe_array[self._attrib_index['duration']] #self._press_event_id = ioe_array[self._attrib_index['press_event_id']] @property def duration(self): """The duration (in seconds) of the key press. This is calculated by subtracting the current event.time from the associated keypress.time. If no matching keypress event was reported prior to this event, then 0.0 is returned. This can happen, for example, when the key was pressed prior to psychopy starting to monitor the device. This condition can also happen when keyboard.reset() method is called between the press and release event times. :return: float """ return self._duration @property def pressEventID(self): """The event.id of the associated press event. The key press id is 0 if no associated KeyboardPress event was found. See the duration property documentation for details on when this can occur. :return: unsigned int """ return self._press_event_id def __str__(self): return '%s, duration: %.3f press_event_id: %d' % ( KeyboardEvent.__str__(self), self.duration, self.pressEventID)
[docs]class Keyboard(ioHubDeviceView): """The Keyboard device provides access to KeyboardPress and KeyboardRelease events as well as the current keyboard state. Examples: A. Print all keyboard events received for 5 seconds:: from psychopy.iohub import launchHubServer from psychopy.core import getTime # Start the ioHub process. 'io' can now be used during the # experiment to access iohub devices and read iohub device events. io = launchHubServer() keyboard = io.devices.keyboard # Check for and print any Keyboard events received for 5 seconds. stime = getTime() while getTime()-stime < 5.0: for e in keyboard.getEvents(): print(e) # Stop the ioHub Server io.quit() B. Wait for a keyboard press event (max of 5 seconds):: from psychopy.iohub import launchHubServer from psychopy.core import getTime # Start the ioHub process. 'io' can now be used during the # experiment to access iohub devices and read iohub device events. io = launchHubServer() keyboard = io.devices.keyboard # Wait for a key keypress event ( max wait of 5 seconds ) presses = keyboard.waitForPresses(maxWait=5.0) print(presses) # Stop the ioHub Server io.quit() """ KEY_PRESS = EventConstants.KEYBOARD_PRESS KEY_RELEASE = EventConstants.KEYBOARD_RELEASE _type2class = {KEY_PRESS: KeyboardPress, KEY_RELEASE: KeyboardRelease} def __init__(self, ioclient, dev_cls_name, dev_config): super(Keyboard, self).__init__(ioclient, 'client.Keyboard', dev_cls_name, dev_config) self.clock = Computer.global_clock self._events = dict() self._reporting = self.isReportingEvents() self._pressed_keys = {} self._device_config = dev_config self._event_buffer_length = dev_config.get('event_buffer_length') self._clearEventsRPC = DeviceRPC(self.hubClient._sendToHubServer, self.device_class, 'clearEvents') def _clearLocalEvents(self, event_type=None): for etype, elist in list(self._events.items()): if event_type is None: elist.clear() elif event_type == etype: elist.clear()
[docs] def _syncDeviceState(self): """An optimized iohub server request that receives all device state and event information in one response. :return: None """ kb_state = self.getCurrentDeviceState() events = {int(k):v for k,v in list(kb_state.get('events').items())} pressed_keys = {int(k):v for k,v in list(kb_state.get('pressed_keys',{}).items())} self._reporting = kb_state.get('reporting_events') self._pressed_keys.clear() akeyix = KeyboardEvent._attrib_index['key'] iotimeix = DeviceEvent.EVENT_HUB_TIME_INDEX for _, (key_array, _) in pressed_keys.items(): self._pressed_keys[key_array[akeyix]] = key_array[iotimeix] for etype, event_arrays in events.items(): ddeque = deque(maxlen=self._event_buffer_length) evts = [self._type2class[etype](e) for e in event_arrays] self._events.setdefault(etype, ddeque).extend(evts)
@property def state(self): """ Returns all currently pressed keys as a dictionary of key : time values. The key is taken from the originating press event .key field. The time value is time of the key press event. Note that any pressed, or active, modifier keys are included in the return value. :return: dict """ self._syncDeviceState() self._pressed_keys = {keys: vals for keys, vals in self._pressed_keys.items()} return self._pressed_keys @property def reporting(self): """Specifies if the keyboard device is reporting / recording events. * True: keyboard events are being reported. * False: keyboard events are not being reported. By default, the Keyboard starts reporting events automatically when the ioHub process is started and continues to do so until the process is stopped. This property can be used to read or set the device reporting state:: # Read the reporting state of the keyboard. is_reporting_keyboard_event = keyboard.reporting # Stop the keyboard from reporting any new events. keyboard.reporting = False """ return self._reporting @reporting.setter def reporting(self, r): """Sets the state of keyboard event reporting / recording.""" self._reporting = self.enableEventReporting(r) return self._reporting def clearEvents(self, event_type=None, filter_id=None): self._clearLocalEvents(event_type) return self._clearEventsRPC(event_type=event_type, filter_id=filter_id)
[docs] def getKeys(self, keys=None, chars=None, ignoreKeys=None, mods=None, duration=None, etype=None, clear=True): """ Return a list of any KeyboardPress or KeyboardRelease events that have occurred since the last time either: * this method was called with the kwarg clear=True (default) * the keyboard.clear() method was called. Other than the 'clear' kwarg, any kwargs that are not None or an empty list are used to filter the possible events that can be returned. If multiple filter criteria are provided, only events that match **all** specified criteria are returned. If no KeyboardEvent's are found that match the filtering criteria, an empty tuple is returned. Returned events are sorted by time. :param keys: Include events where .key in keys. :param chars: Include events where .char in chars. :param ignoreKeys: Ignore events where .key in ignoreKeys. :param mods: Include events where .modifiers include >=1 mods element. :param duration: Include KeyboardRelease events where .duration > duration or .duration < -(duration). :param etype: Include events that match etype of Keyboard.KEY_PRESS or Keyboard.KEY_RELEASE. :param clear: True (default) = clear returned events from event buffer, False = leave the keyboard event buffer unchanged. :return: tuple of KeyboardEvent instances, or () """ self._syncDeviceState() ecount = 0 for elist in list(self._events.values()): ecount += len(elist) if ecount == 0: return [] def filterEvent(e): r1 = (keys is None or e.key in keys) and ( ignoreKeys is None or e.key not in ignoreKeys) r2 = (chars is None or e.char in chars) r3 = True if duration is not None: r3 = (duration // abs(duration) * e.duration) >= duration r4 = True if mods: r4 = len([m for m in mods if m in e.modifiers]) > 0 return r1 and r2 and r3 and r4 press_evt = [] release_evt = [] if etype is None or etype == Keyboard.KEY_PRESS: press_evt = [ e for e in self._events.get( Keyboard.KEY_PRESS, []) if filterEvent(e)] if etype is None or etype == Keyboard.KEY_RELEASE: release_evt = [ e for e in self._events.get( Keyboard.KEY_RELEASE, []) if filterEvent(e)] return_events = sorted(press_evt + release_evt, key=lambda x: x.time) if clear is True: for e in return_events: self._events[e._type].remove(e) return return_events
[docs] def getPresses(self, keys=None, chars=None, ignoreKeys=None, mods=None, clear=True): """See the getKeys() method documentation. This method is identical, but only returns KeyboardPress events. """ return self.getKeys( keys=keys, chars=chars, ignoreKeys=ignoreKeys, mods=mods, duration=None, etype=self.KEY_PRESS, clear=clear )
[docs] def getReleases(self, keys=None, chars=None, ignoreKeys=None, mods=None, duration=None, clear=True): """See the getKeys() method documentation. This method is identical, but only returns KeyboardRelease events. """ return self.getKeys( keys=keys, chars=chars, ignoreKeys=ignoreKeys, mods=mods, duration=duration, etype=self.KEY_RELEASE, clear=clear )
[docs] def waitForKeys(self, maxWait=None, keys=None, chars=None, mods=None, duration=None, etype=None, clear=True, checkInterval=0.002): """Blocks experiment execution until at least one matching KeyboardEvent occurs, or until maxWait seconds has passed since the method was called. Keyboard events are filtered the same way as in the getKeys() method. As soon as at least one matching KeyboardEvent occurs prior to maxWait, the matching events are returned as a tuple. Returned events are sorted by time. :param maxWait: Maximum seconds method waits for >=1 matching event. If <=0.0, method functions the same as getKeys(). If None, the methods blocks indefinitely. :param keys: Include events where .key in keys. :param chars: Include events where .char in chars. :param mods: Include events where .modifiers include >=1 mods element. :param duration: Include KeyboardRelease events where .duration > duration or .duration < -(duration). :param etype: Include events that match etype of Keyboard.KEY_PRESS or Keyboard.KEY_RELEASE. :param clear: True (default) = clear returned events from event buffer, False = leave the keyboard event buffer unchanged. :param checkInterval: The time between geyKeys() calls while waiting. The method sleeps between geyKeys() calls, up until checkInterval*2.0 sec prior to the maxWait. After that time, keyboard events are constantly checked until the method times out. :return: tuple of KeyboardEvent instances, or () """ start_time = getTime() if maxWait is None: maxWait = 60000.0 timeout = start_time + maxWait key = [] def pumpKeys(): key = self.getKeys( keys=keys, chars=chars, mods=mods, duration=duration, etype=etype, clear=clear ) if key: return key win32MessagePump() return key # Don't wait if maxWait is <= 0 if maxWait <= 0: key = pumpKeys() return key while getTime() < (timeout - checkInterval * 2): # Pump events on pyglet windows if they exist ltime = getTime() key = pumpKeys() if key: return key sleep_dur = max(checkInterval - (getTime() - ltime), 0.0001) time.sleep(sleep_dur) while getTime() < timeout: key = pumpKeys() if key: return key return key
[docs] def waitForPresses(self, maxWait=None, keys=None, chars=None, mods=None, duration=None, clear=True, checkInterval=0.002): """See the waitForKeys() method documentation. This method is identical, but only returns KeyboardPress events. """ return self.waitForKeys( maxWait=maxWait, keys=keys, chars=chars, mods=mods, duration=duration, etype=Keyboard.KEY_PRESS, # used this instead of `self.KEY_PRESS` clear=clear, checkInterval=checkInterval )
[docs] def waitForReleases(self, maxWait=None, keys=None, chars=None, mods=None, duration=None, clear=True, checkInterval=0.002): """See the waitForKeys() method documentation. This method is identical, but only returns KeyboardRelease events. """ return self.waitForKeys( maxWait=maxWait, keys=keys, chars=chars, mods=mods, duration=duration, etype=Keyboard.KEY_RELEASE, # used this instead of `self.KEY_RLEASE` clear=clear, checkInterval=checkInterval )

Back to top