from __future__ import with_statement
import dbus, os
+from fnmatch import fnmatch
from dbuscron.logger import Logger
log = Logger(__name__)
def dbus_to_str(value):
try:
- if isinstance(value, dbus.Byte):
+ if value is None:
+ return ''
+ elif isinstance(value, dbus.Byte):
result = str(int(value))
elif isinstance(value, dbus.ByteArray):
result = ','.join(str(ord(v)) for v in value)
log.error('convert exception', e)
raise e
-
def get_dbus_message_type(message):
result = message.__class__.__name__[0:-7]
for c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ':
__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):
- self.get_session_bus_address()
+ 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']
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:
return ','.join(rule)
def match(self, bus, message):
-
if self._bus not in (None, bus):
return False
for i, arg in enumerate(args_):
if i >= len(self._args):
break
- if self._args[i] not in (None, str(arg)):
- return False
+ a = dbus_to_str(arg)
+ if self._args[i] not in (None, a):
+ if not fnmatch(a, self._args[i]):
+ return False
return True