-
-import dbus
+from __future__ import with_statement
+
+import dbus, os
+
+from dbuscron.logger import Logger
+log = Logger(__name__)
+
+def dbus_to_str(value):
+ try:
+ if isinstance(value, dbus.Byte):
+ result = str(int(value))
+ elif isinstance(value, dbus.ByteArray):
+ result = ','.join(str(ord(v)) for v in value)
+ elif isinstance(value, dbus.Array):
+ result = ','.join(dbus_to_str(v) for v in value)
+ elif isinstance(value, dbus.Dictionary):
+ result = ','.join('%s:%s' % (k, dbus_to_str(v)) for k, v in value.iteritems())
+ elif isinstance(value, dbus.String):
+ result = value.encode('utf-8')
+ else:
+ result = str(value)
+ return result
+ except Exception, e:
+ log.error('convert exception', e)
+ raise e
def get_dbus_message_type(message):
- return message.__class__.__name__.lower()[0:-7]
+ result = message.__class__.__name__[0:-7]
+ for c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ':
+ result = result.replace(c, '_'+c.lower())
+ return result.strip('_')
class DbusBus(object):
__bus = None
__system_bus = None
__session_bus = None
- def __new__(cls):
+ def __new__(cls, *args, **kw):
if not cls.__bus:
cls.__bus = super(DbusBus, cls).__new__(cls)
return cls.__bus
- def __init__(self):
+ def __init__(self, session_bus_address=None):
+ if session_bus_address:
+ self.set_session_bus_address(session_bus_address)
+ else:
+ self.get_session_bus_address()
+
from dbus.mainloop.glib import DBusGMainLoop
DBusGMainLoop(set_as_default=True)
+ def set_session_bus_address(self, value):
+ os.environ['DBUS_SESSION_BUS_ADDRESS'] = str(value)
+
+ def get_session_bus_address(self):
+ try:
+ return os.environ['DBUS_SESSION_BUS_ADDRESS']
+ except KeyError:
+ with open('/tmp/session_bus_address.user', 'rb') as f:
+ session_bus_address = f.readline().strip().split('=', 1).pop().strip("'\"")
+ os.environ['DBUS_SESSION_BUS_ADDRESS'] = session_bus_address
+ log('session bus address aquired', session_bus_address)
+ return session_bus_address
+
+ session_bus_address = property(get_session_bus_address, set_session_bus_address)
+
@property
def system(self):
if not self.__system_bus:
def register(self):
rule = str(self)
if rule:
- self._bus.add_match_string(str(self))
+ self._bus.add_match_string(rule)
+
+ def unregister(self):
+ rule = str(self)
+ if rule:
+ self._bus.remove_match_string(rule)
+
+ def __del__(self):
+ self.unregister()
def __str__(self):
rule = []
if self._destination not in (None, message.get_destination()):
return False
- args_ = message.get_args_list()
- for i, arg in enumerate(args_):
- if i >= len(self._args):
- break
- if self._args[i] not in (None, arg):
- return False
+ if self._args is not None:
+ args_ = message.get_args_list()
+ for i, arg in enumerate(args_):
+ if i >= len(self._args):
+ break
+ if self._args[i] not in (None, str(arg)):
+ return False
return True