app_name = _os.environ["ATSPI_TEST_APP_NAME"]
if app_name:
- self._appcache = TestApplicationCache(self, self._bus, app_name)
+ self._app_cache = TestApplicationCache(self, self._bus, app_name)
self.dev = _TestDeviceEventController()
else:
- self._appcache = ApplicationCache(self, self._bus)
+ self._app_cache = ApplicationCache(self, self._bus)
self.dev = DeviceEventController(self._bus)
self._event_listeners = {}
self._children_changed_type = _EventType("object:children-changed")
self._children_changed_listeners = {}
- self.queue = Queue()
self.clients = {}
+ self.deviceClients = {}
def __call__(self):
"""
"""
return self
+ @property
+ def cache (self):
+ """
+ This is the accessible application cache through which
+ all accessible objects are accessed.
+ """
+ return self._app_cache
+
def start(self, async=False, gil=True):
"""
Enter the main loop to start receiving and dispatching events.
@return: Desktop reference
@rtype: Accessibility.Desktop
"""
- return _Desktop(self._appcache)
+ return _Desktop(self._app_cache)
+
+ # -------------------------------------------------------------------------------
def _callClients(self, register, event):
for client in register.keys():
if new_type.is_subtype(type):
registered.append(new_type.name)
+ if registered == []:
+ del(register[client])
+
def _deregisterFake(self, type, register, client, *names):
"""
Deregisters a client from a register of clients
for name in names:
remove_type = _EventType(name)
- for i in range(0, len(registered) - 1):
- type_name = registered[i]
+ copy = registered[:]
+ for i in range(0, len(copy)):
+ type_name = copy[i]
registered_type = _EventType(type_name)
if remove_type.is_subtype(registered_type):
if registered == []:
del(register[client])
+ # -------------------------------------------------------------------------------
+
def registerEventListener(self, client, *names):
"""
Registers a new client callback for the given event names. Supports
for name in names:
new_type = _EventType(name)
registered.append((new_type.name,
- _event_type_to_signal_reciever(self._bus, self._appcache, client, new_type)))
+ _event_type_to_signal_reciever(self._bus, self._app_cache, client, new_type)))
self._registerFake(self._name_type, self._name_listeners, client, *names)
self._registerFake(self._description_type, self._description_listeners, client, *names)
for name in names:
remove_type = _EventType(name)
-
- for i in range(0, len(registered) - 1):
- (type_name, signal_match) = registered[i]
+ copy = registered[:]
+ for i in range (0, len(copy)):
+ type_name, signal_match = copy[i]
registered_type = _EventType(type_name)
if remove_type.is_subtype(registered_type):
return missing
+ # -------------------------------------------------------------------------------
+
def registerKeystrokeListener(self,
client,
key_set=[],
"""
try:
# see if we already have an observer for this client
- ob = self.clients[client]
+ ob = self.deviceClients[client]
except KeyError:
# create a new device observer for this client
ob = KeyboardDeviceEventListener(self, synchronous, preemptive, global_)
# store the observer to client mapping, and the inverse
- self.clients[ob] = client
- self.clients[client] = ob
+ self.deviceClients[ob] = client
+ self.deviceClients[client] = ob
if mask is None:
# None means all modifier combinations
mask = utils.allModifiers()
@raise KeyError: When the client isn't already registered for events
"""
# see if we already have an observer for this client
- ob = self.clients[client]
+ ob = self.deviceClients[client]
if mask is None:
# None means all modifier combinations
mask = utils.allModifiers()
# register for new keystrokes on the observer
ob.unregister(self.dev, key_set, mask, kind)
- def generateKeyboardEvent(self, keycode, keysym, kind):
- """
- Generates a keyboard event. One of the keycode or the keysym parameters
- should be specified and the other should be None. The kind parameter is
- required and should be one of the KEY_PRESS, KEY_RELEASE, KEY_PRESSRELEASE,
- KEY_SYM, or KEY_STRING.
-
- @param keycode: Hardware keycode or None
- @type keycode: integer
- @param keysym: Symbolic key string or None
- @type keysym: string
- @param kind: Kind of event to synthesize
- @type kind: integer
- """
- if keysym is None:
- self.dev.generateKeyboardEvent(keycode, '', kind)
- else:
- self.dev.generateKeyboardEvent(None, keysym, kind)
-
- def generateMouseEvent(self, x, y, name):
- """
- Generates a mouse event at the given absolute x and y coordinate. The kind
- of event generated is specified by the name. For example, MOUSE_B1P
- (button 1 press), MOUSE_REL (relative motion), MOUSE_B3D (butten 3
- double-click).
-
- @param x: Horizontal coordinate, usually left-hand oriented
- @type x: integer
- @param y: Vertical coordinate, usually left-hand oriented
- @type y: integer
- @param name: Name of the event to generate
- @type name: string
- """
- self.dev.generateMouseEvent(x, y, name)
-
def handleDeviceEvent(self, event, ob):
"""
Dispatches L{event.DeviceEvent}s to registered clients. Clients are called
"""
try:
# try to get the client registered for this event type
- client = self.clients[ob]
+ client = self.deviceClients[ob]
except KeyError:
# client may have unregistered recently, ignore event
return False
except Exception:
# print the exception, but don't let it stop notification
traceback.print_exc()
-
- def handleEvent(self, event):
+
+ # -------------------------------------------------------------------------------
+
+ def pumpQueuedEvents (self):
+ """
+ No Longer needed all application events are asyncronous.
"""
- Handles an AT-SPI event by either queuing it for later dispatch when the
- L{Registry.async} flag is set, or dispatching it immediately.
+ pass
- @param event: AT-SPI event
- @type event: L{event.Event}
+ def flushEvents (self):
"""
- if self.async:
- # queue for now
- self.queue.put_nowait(event)
- else:
- # dispatch immediately
- self._dispatchEvent(event)
+ No Longer needed all application events are asyncronous.
+ """
+ pass
+
+ # -------------------------------------------------------------------------------
- def _dispatchEvent(self, event):
+ def generateKeyboardEvent(self, keycode, keysym, kind):
"""
- Dispatches L{event.Event}s to registered clients. Clients are called in
- the order they were registered for the given AT-SPI event. If any client
- returns True, callbacks cease for the event for clients of this registry
- instance. Clients of other registry instances and clients in other processes
- are unaffected.
+ Generates a keyboard event. One of the keycode or the keysym parameters
+ should be specified and the other should be None. The kind parameter is
+ required and should be one of the KEY_PRESS, KEY_RELEASE, KEY_PRESSRELEASE,
+ KEY_SYM, or KEY_STRING.
- @param event: AT-SPI event
- @type event: L{event.Event}
+ @param keycode: Hardware keycode or None
+ @type keycode: integer
+ @param keysym: Symbolic key string or None
+ @type keysym: string
+ @param kind: Kind of event to synthesize
+ @type kind: integer
"""
- et = event.type
- try:
- # try to get the client registered for this event type
- clients = self.clients[et.name]
- except KeyError:
- try:
- # we may not have registered for the complete subtree of events
- # if our tree does not list all of a certain type (e.g.
- # object:state-changed:*); try again with klass and major only
- if et.detail is not None:
- # Strip the 'detail' field.
- clients = self.clients['%s:%s:%s' % (et.klass, et.major, et.minor)]
- elif et.minor is not None:
- # The event could possibly be object:state-changed:*.
- clients = self.clients['%s:%s' % (et.klass, et.major)]
- except KeyError:
- # client may have unregistered recently, ignore event
- return
- # make the call to each client
- consume = False
- for client in clients:
- try:
- consume = client(event) or False
- except Exception:
- # print the exception, but don't let it stop notification
- traceback.print_exc()
- if consume or event.consume:
- # don't allow further processing if a client returns True
- break
-
- def flushEvents(self):
- """
- Flushes the event queue by destroying it and recreating it.
- """
- self.queue = Queue()
-
- def pumpQueuedEvents(self, num=-1):
- """
- Provides asynch processing of events in the queue by executeing them with
- _dispatchEvent() (as is done immediately when synch processing).
- This method would normally be called from a main loop or idle function.
-
- @param num: Number of events to pump. If number is negative it pumps
- the entire queue. Default is -1.
- @type num: integer
- @return: True if queue is not empty after events were pumped.
- @rtype: boolean
+ if keysym is None:
+ self.dev.generateKeyboardEvent(keycode, '', kind)
+ else:
+ self.dev.generateKeyboardEvent(None, keysym, kind)
+
+ def generateMouseEvent(self, x, y, name):
+ """
+ Generates a mouse event at the given absolute x and y coordinate. The kind
+ of event generated is specified by the name. For example, MOUSE_B1P
+ (button 1 press), MOUSE_REL (relative motion), MOUSE_B3D (butten 3
+ double-click).
+
+ @param x: Horizontal coordinate, usually left-hand oriented
+ @type x: integer
+ @param y: Vertical coordinate, usually left-hand oriented
+ @type y: integer
+ @param name: Name of the event to generate
+ @type name: string
"""
- if num < 0:
- # Dequeue as many events as currently in the queue.
- num = self.queue.qsize()
- for i in xrange(num):
- try:
- # get next waiting event
- event = self.queue.get_nowait()
- except Queue.Empty:
- break
- self._dispatchEvent(event)
-
- return not self.queue.empty()
+ self.dev.generateMouseEvent(_dbus.Int32(x), _dbus.Int32(y), name)
projects / platform / core / uifw / at-spi2-atk.git / blobdiff