python-pyatspi 1.32.0-1ubuntu1 / usr / share / pyshared / pyatspi / accessible.py

'''
Creates functions at import time that are mixed into the 
Accessibility.Accessible base class to make it more Pythonic.

Based on public domain code originally posted at 
U{http://wwwx.cs.unc.edu/~parente/cgi-bin/RuntimeClassMixins}.

@var _ACCESSIBLE_CACHE: Pairs hash values for accessible objects to 
  L{_PropertyCache} bags. We do not store actual accessibles in the dictionary
  because that would +1 their ref counts and cause __del__ to never be called
  which is the method we rely on to properly invalidate cache entries.
@type _ACCESSIBLE_CACHE: dictionary
@var _CACHE_LEVEL: Current level of caching enabled. Checked dynamically by
  L{_AccessibleMixin}
@type _CACHE_LEVEL: integer

@author: Peter Parente
@organization: IBM Corporation
@copyright: Copyright (c) 2005, 2007 IBM Corporation
@license: LGPL

This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Library General Public
License as published by the Free Software Foundation; either
version 2 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
Library General Public License for more details.

You should have received a copy of the GNU Library General Public
License along with this library; if not, write to the
Free Software Foundation, Inc., 59 Temple Place - Suite 330,
Boston, MA 02111-1307, USA.

Portions of this code originally licensed and copyright (c) 2005, 2007
IBM Corporation under the BSD license, available at
U{http://www.opensource.org/licenses/bsd-license.php}
'''
import new
import types
import ORBit
import Accessibility
import constants
import utils
import registry
import weakref

_ACCESSIBLE_CACHE = weakref.WeakValueDictionary()
_ACCESSIBLE_USER_DATA = weakref.WeakValueDictionary()
_CACHE_LEVEL = None

class _PropertyCache:
  def wipe(self):
      self.__dict__ = {}

class _UserData:
  value = None

def getCacheLevel():
  '''
  Gets the current level of caching.
  
  @return: None indicating no caching is in effect. 
    L{constants.CACHE_INTERFACES} indicating all interface query results are
    cached. L{constants.CACHE_PROPERTIES} indicating all basic accessible
    properties are cached.
  @rtype: integer
  '''
  return _CACHE_LEVEL

def setCacheLevel(val):
  '''
  Sets the desired level of caching for all accessible objects created after
  this function is invoked. Immediately clears the current accessible cache.
  
  @param val: None indicating no caching is in effect. 
    L{constants.CACHE_INTERFACES} indicating all interface query results are
    cached. L{constants.CACHE_PROPERTIES} indicating all basic accessible
    properties are cached plus all interfaces.
  @type val: integer
  '''
  global _CACHE_LEVEL
  if _CACHE_LEVEL != val:
    # empty our accessible cache  
    _ACCESSIBLE_CACHE.clear()
    # need to register/unregister for listeners depending on caching level
    if val == constants.CACHE_PROPERTIES:
      r = registry.Registry()
      r.registerEventListener(_updateCache, *constants.CACHE_EVENTS)
    else:
      r = registry.Registry()
      r.deregisterEventListener(_updateCache, *constants.CACHE_EVENTS)
  _CACHE_LEVEL = val
  
def clearCache():
  '''Forces a clear of the entire cache.'''
  _ACCESSIBLE_CACHE.clear()
  
def printCache(template='%s'):
  '''
  Prints the contents of the cache.
  
  @param template: Format string to use when printing
  @type template: string
  '''
  print template % _ACCESSIBLE_CACHE

def _updateCache(event):
  '''
  Invalidates an entry in the cache when the hash value of a source of an event
  matches an entry in the cache.
  
  @param event: One of the L{constants.CACHE_EVENTS} event types
  @type event: L{event.Event}
  '''
  try:
    _ACCESSIBLE_CACHE[hash(event.source)].wipe()
  except KeyError:
    return

def _getAndCache(acc, value_name, get_method):
  '''
  If property caching is enabled, use the cached proprty, or get the 
  property and cache it. If property caching is disabled, simply get the 
  property.

  @param value_name: The name of the value, like 'role' or 'description'.
  @type value_name: string
  @param get_method: Method used to get the property, should not have any 
  arguments.
  @type get_method: callable
  @return: Value of property we are retrieving.
  @rtype: object
  '''
  if _CACHE_LEVEL != constants.CACHE_PROPERTIES:
    return get_method()
    
  cache = _ACCESSIBLE_CACHE
  h = hash(acc)

  try:
    pc = acc._property_cache
  except AttributeError:
    try:
      pc = cache[h]
    except KeyError:
      # no cached info for this accessible yet
      pc = _PropertyCache()
      cache[h] = pc
    acc._property_cache = pc  
    
  try:
    value = getattr(pc, value_name)
  except AttributeError:
    # no cached property of this type
    value = get_method()
    setattr(pc, value_name, value)
    
  return value
  

def _makeQuery(interface):
  '''
  Builds a function querying to a specific interface and returns it.
  
  @param interface: Class representing an AT-SPI interface
  @type interface: class
  @return: Function querying to the given interface
  @rtype: function
  '''
  def _inner(self):
    '''
    Queries an object for another interface.
  
    @return: An object with the desired interface
    @rtype: object
    @raise NotImplementedError: When the desired interface is not supported    
    '''
    iid = utils.getInterfaceIID(interface)
    try:
      i = self._icache[iid]
    except KeyError:
      # interface not cached
      caching = True
    except AttributeError:
      # determine if we're caching
      caching = _CACHE_LEVEL is not None
      if caching:
        # initialize the cache
        self._icache = {}
    else:
      # check if our cached result was an interface, or an indicator that the
      # interface is not supported
      if i is None:
        raise NotImplementedError
      else:
        return i

    try:
      # do the query remotely
      i = self.queryInterface(iid)
      if i is not None:
        i = i._narrow(interface)
    except Exception, e:
      raise LookupError(e)      
    if i is None:
      # cache that the interface is not supported
      if caching:
        self._icache[iid] = None
      raise NotImplementedError
    
    if caching:
      # cache the narrow'ed result, but only if we're caching for this object
      self._icache[iid] = i
    return i
  
  return _inner

def _makeExceptionHandler(func):
  '''
  Builds a function calling the one it wraps in try/except statements catching
  CORBA exceptions.
  
  @return: Function calling the method being wrapped
  @rtype: function
  '''
  def _inner(self, *args, **kwargs):
    try:
      # try calling the original func
      return func(self, *args, **kwargs)
    except ORBit.CORBA.NO_IMPLEMENT, e:
      # raise Python exception
      raise NotImplementedError(e)
    except ORBit.CORBA.Exception, e:
      # raise Python exception
      raise LookupError(e)
  return _inner

def _mixInterfaces(cls, interfaces):
  '''
  Add methods for querying to interfaces other than the base accessible to
  the given class.
  
  @param cls: Class to mix interface methods into
  @type cls: class
  @param interfaces: Classes representing AT-SPI interfaces
  @type interfaces: list of class
  '''
  # create functions in this module for all interfaces listed in constants
  for interface in interfaces:
    # build name of converter from the name of the interface
    name = 'query%s' % utils.getInterfaceName(interface)
    # build a function that queries to the given interface
    func = _makeQuery(interface)
    # build a new method that is a clone of the original function
    method = new.function(func.func_code, func.func_globals, name, 
                          func.func_defaults, func.func_closure)
    # add the method to the given class
    setattr(cls, name, method)

def _mixExceptions(cls):
  '''
  Wraps all methods and properties in a class with handlers for CORBA 
  exceptions.
  
  @param cls: Class to mix interface methods into
  @type cls: class
  '''
  # get a method type as a reference from a known method
  method_type = Accessibility.Accessible.getRole.__class__
  # loop over all names in the new class
  for name in cls.__dict__.keys():
    obj = cls.__dict__[name]
    # check if we're on a protected or private method
    if name.startswith('_'):
      continue
    # check if we're on a method
    elif isinstance(obj, method_type):
      # wrap the function in an exception handler
      method = _makeExceptionHandler(obj)
      # add the wrapped function to the class
      setattr(cls, name, method)
    # check if we're on a property
    elif isinstance(obj, property):
      # wrap the getters and setters
      if obj.fget:
        func = getattr(cls, obj.fget.__name__)
        getter = _makeExceptionHandler(func)
      else:
        getter = None
      if obj.fset:
        func = getattr(cls, obj.fset.__name__)
        setter = _makeExceptionHandler(func)
      else:
        setter = None
      setattr(cls, name, property(getter, setter))

def _mixClass(cls, new_cls, ignore=[]):
  '''
  Adds the methods in new_cls to cls. After mixing, all instances of cls will
  have the new methods. If there is a method name clash, the method already in
  cls will be prefixed with '_mix_' before the new method of the same name is 
  mixed in.
  
  @note: _ is not the prefix because if you wind up with __ in front of a 
  variable, it becomes private and mangled when an instance is created. 
  Difficult to invoke from the mixin class.

  @param cls: Existing class to mix features into
  @type cls: class
  @param new_cls: Class containing features to add
  @type new_cls: class
  @param ignore: Ignore these methods from the mixin
  @type ignore: iterable
  '''
  # loop over all names in the new class
  for name, func in new_cls.__dict__.items():
    if name in ignore:
      continue
    if isinstance(func, types.FunctionType):
      # build a new function that is a clone of the one from new_cls
      method = new.function(func.func_code, func.func_globals, name, 
                            func.func_defaults, func.func_closure)
      try:
        # check if a method of the same name already exists in the target
        old_method = getattr(cls, name)
      except AttributeError:
        pass
      else:
        # rename the old method so we can still call it if need be
        setattr(cls, '_mix_'+name, old_method)
      # add the clone to cls
      setattr(cls, name, method)
    elif isinstance(func, staticmethod):
      try:
        # check if a method of the same name already exists in the target
        old_method = getattr(cls, name)
      except AttributeError:
        pass
      else:
        # rename the old method so we can still call it if need be
        setattr(cls, '_mix_'+name, old_method)
      setattr(cls, name, func)
    elif isinstance(func, property):
      try:
        # check if a method of the same name already exists in the target
        old_prop = getattr(cls, name)
      except AttributeError:
        pass
      else:
        # IMPORTANT: We save the old property before overwriting it, even 
        # though we never end up calling the old prop from our mixin class.
        # If we don't save the old one, we seem to introduce a Python ref count
        # problem where the property get/set methods disappear before we can
        # use them at a later time. This is a minor waste of memory because
        # a property is a class object and we only overwrite a few of them.
        setattr(cls, '_mix_'+name, old_prop)
      setattr(cls, name, func)

class _AccessibleMixin(object):
  '''
  Defines methods to be added to the Accessibility.Accessible class. The
  features defined here will be added to the Accessible class at run time so
  that all instances of Accessible have them (i.e. there is no need to
  explicitly wrap an Accessible in this class or derive a new class from it.)
  
  @cvar SLOTTED_CLASSES: Mapping from raw Accessibility class to a new class
    having the slots defined by L{SLOTS}
  @type SLOTTED_CLASSES: dictionary
  @cvar SLOTS: All slots to create
  @type SLOTS: tuple
  '''
  SLOTTED_CLASSES = {}
  SLOTS = ('_icache', '_property_cache', '_user_data')
  
  def __new__(cls):
    '''
    Creates a new class mimicking the one requested, but with extra named 
    defined in __slots__. The _cache attribute is used internally for interface
    caching. The user_data field may be populated with whatever data structure
    a client wishes to use. Neither is set to a default value by default.
    
    Note that we can't simply mix __slots__ into this class because __slots__
    has an effect only at class creation time. 
    
    We also do not completely obliterate __slots__ to allow __dict__ to be
    instantiated as normal as reducing the initialization and memory overhead
    of the millions of accessible objects that are created is a good thing for
    many clients.
    
    @param cls: Accessibility object class
    @type cls: class
    @return: Instance of the new class
    @rtype: object
    '''
    try:
      # check if we've already created a new version of the class
      new_cls = _AccessibleMixin.SLOTTED_CLASSES[cls]
    except KeyError:
      # create the new class if not
      new_cls = type(cls.__name__, (cls,), 
                     {'__module__' : cls.__module__, 
                      '__slots__' : _AccessibleMixin.SLOTS})
      _AccessibleMixin.SLOTTED_CLASSES[cls] = new_cls
    obj = cls._mix___new__(new_cls)
    return obj
  
  def __del__(self):
    '''    
    Decrements the reference count on the accessible object when there are no
    Python references to this object. This provides automatic reference
    counting for AT-SPI objects. Also removes this object from the cache if
    we're caching properties. 
    '''
    try:
      self.unref()
    except Exception:
      pass
    
  def __iter__(self):
    '''
    Iterator that yields one accessible child per iteration. If an exception is
    encountered, None is yielded instead.
    
    @return: A child accessible
    @rtype: Accessibility.Accessible
    '''
    for i in xrange(self.childCount):
      try:
        yield self.getChildAtIndex(i)
      except LookupError:
        yield None
    
  def __str__(self):
    '''
    Gets a human readable representation of the accessible.
    
    @return: Role and name information for the accessible
    @rtype: string
    '''
    try:
      return '[%s | %s]' % (self.getRoleName(), self.name)
    except Exception:
      return '[DEAD]'
    
  def __nonzero__(self):
    '''
    @return: True, always
    @rtype: boolean
    '''
    return True
    
  def __getitem__(self, index):
    '''
    Thin wrapper around getChildAtIndex.
    
    @param index: Index of desired child
    @type index: integer
    @return: Accessible child
    @rtype: Accessibility.Accessible
    '''
    n = self.childCount
    if index >= n:
      raise IndexError
    elif index < -n:
      raise IndexError
    elif index < 0:
      index += n
    return self.getChildAtIndex(index)
  
  def __len__(self):
    '''
    Thin wrapper around childCount.
    
    @return: Number of child accessibles
    @rtype: integer
    '''
    return self.childCount
  
  def _get_user_data(self):
    '''
    Get user_data from global dictionay fo this accessible.

    @return: Any data the user assigned, or None.
    @rtype: object
    '''
    global _ACCESSIBLE_USER_DATA
    h = hash(self)
    
    try:
      ud = self._user_data
    except AttributeError:
      try:
        ud = _ACCESSIBLE_USER_DATA[h]
      except KeyError:
        # no cached info for this object yet
        ud = _UserData()
        _ACCESSIBLE_USER_DATA[h] = ud

    self._user_data = ud
    return ud.value

  def _set_user_data(self, value):
    '''
    Set arbitrary data to user_data.

    @param value: Value to set in user_data
    @type value: object
    '''
    global _ACCESSIBLE_USER_DATA
    h = hash(self)
    
    try:
      ud = self._user_data
    except AttributeError:
      try:
        ud = _ACCESSIBLE_USER_DATA[h]
      except KeyError:
        # no cached info for this object yet
        ud = _UserData()
        _ACCESSIBLE_USER_DATA[h] = ud

    self._user_data = ud
    ud.value = value

  user_data = property(_get_user_data, _set_user_data)

  def _get_name(self):
    '''
    Gets the name of the accessible from the cache if it is available, 
    otherwise, fetches it remotely.
    
    @return: Name of the accessible
    @rtype: string
    '''
    return _getAndCache(self, 'name', self._get_name)

  name = property(_get_name, Accessibility.Accessible._set_name)

  def _get_parent(self):
    '''
    Gets the parent of the accessible from the cache if it is available, 
    otherwise, fetches it remotely.
    
    @return: Parent of the accessible
    @rtype: Accessibility.Accessible
    '''
    return _getAndCache(self, 'parent', self._get_parent)

  parent = property(_get_parent)
  
  def getRoleName(self):
    '''
    Gets the unlocalized role name of the accessible from the cache if it is 
    available, otherwise, fetches it remotely.
    
    @return: Role name of the accessible
    @rtype: string
    '''
    return _getAndCache(self, 'rolename', self._mix_getRoleName)

  def getRole(self):
    '''
    Gets the role of the accessible from the cache if it is 
    available, otherwise, fetches it remotely.
    
    @return: Role of the accessible
    @rtype: Accessibility.Role
    '''
    return _getAndCache(self, 'role', self._mix_getRole)

  def _get_description(self):
    '''    
    Gets the description of the accessible from the cache if it is available,
    otherwise, fetches it remotely.
    
    @return: Description of the accessible
    @rtype: string
    '''
    return _getAndCache(self, 'description', self._get_description)
    
  description = property(_get_description, 
                         Accessibility.Accessible._set_description)
  
  def getIndexInParent(self):
    '''
    Gets the index of this accessible in its parent. Uses the implementation of
    this method provided by the Accessibility.Accessible object, but checks the
    bound of the value to ensure it is not outside the range of childCount 
    reported by this accessible's parent.
    
    @return: Index of this accessible in its parent
    @rtype: integer
    '''
    i = self._mix_getIndexInParent()
    try:
      # correct for out-of-bounds index reporting
      return min(self.parent.childCount-1, i)
    except AttributeError:
      # return sentinel if there is no parent
      return -1

  def getApplication(self):
    '''
    Gets the most-parent accessible (the application) of this
    accessible. Tries using the getApplication method introduced in
    AT-SPI 1.7.0 first before resorting to traversing parent links.
    
    @warning: Cycles involving more than the previously traversed accessible 
      are not detected by this code.
    @return: Application object
    @rtype: Accessibility.Application
    '''
    try:
      app = self._mix_getApplication()
    except AttributeError:
      app = None

    # Some toolkits (e.g., Java) do not support getApplication yet and
    # will return None as a result.
    #
    if app:
      return app

    # If we didn't find anything, traverse up the tree, making sure to
    # attempt to turn the thing we found into an Application object.
    #
    curr = self
    try:
      while curr.parent is not None and (not curr.parent == curr):
        curr = curr.parent
      curr.ref()
      return curr._narrow(Accessibility.Application)
    except:
      return None

class _RelationMixin(object):
  '''
  Defines methods to be added to the Relation class. At this time it only
  overrides L{_RelationMixin.getTarget} which by the IDL's standard is
  supposed to return CORBA.Objects but we expect LAccessibility.Accessible
  objects (see http://bugzilla.gnome.org/show_bug.cgi?id=435833). 
  This seems to be a problem especially with the Java implementation of CORBA.
  '''
  def getTarget(self, index):
    '''
    Overrides the regular getTarget to return Accessibility.Accessible
    objects.

    @return: The 'nth' target of this Relation.
    @rtype: Accessibility.Accessible
    '''
    target = self._mix_getTarget(index)
    if not isinstance(target, Accessibility.Accessible):
      target = target._narrow(Accessibility.Accessible)
    target.ref()
    return target

class _UnrefMixin(object):
  '''
  This mixin addresses the issue we have with unreferencing non-primitives.
  '''
  def __del__(self):
    '''
    Unrefence the instance when Python GCs it. Why do we need this twice?
    '''
    try:
      self.unref()
    except Exception:
      pass

# 1. mix the exception handlers into all queryable interfaces
map(_mixExceptions, constants.ALL_INTERFACES)
# 2. mix the exception handlers into other Accessibility objects
map(_mixExceptions, [Accessibility.StateSet])
# 3. mix the new functions
_mixClass(Accessibility.Accessible, _AccessibleMixin,
          ['_get_name', '_get_description', '_get_parent'])
# 4. mix queryInterface convenience methods
_mixInterfaces(Accessibility.Accessible, constants.ALL_INTERFACES)
# 5. mix Relation class
_mixClass(Accessibility.Relation, _RelationMixin)
# 6. mix in neccessary unrefs
map(lambda cls: _mixClass(cls, _UnrefMixin), 
    (Accessibility.StateSet,Accessibility.Relation))