summaryrefslogtreecommitdiffstats
path: root/python/dbus.py
diff options
context:
space:
mode:
Diffstat (limited to 'python/dbus.py')
-rw-r--r--python/dbus.py267
1 files changed, 0 insertions, 267 deletions
diff --git a/python/dbus.py b/python/dbus.py
deleted file mode 100644
index c7ab5dd3..00000000
--- a/python/dbus.py
+++ /dev/null
@@ -1,267 +0,0 @@
-
-"""Module for high-level communication over the FreeDesktop.org Bus (DBus)
-
-DBus allows you to share and access remote objects between processes
-running on the desktop, and also to access system services (such as
-the print spool).
-
-To use DBus, first get a Bus object, which provides a connection to one
-of a few standard dbus-daemon instances that might be running. From the
-Bus you can get a RemoteService. A service is provided by an application or
-process connected to the Bus, and represents a set of objects. Once you
-have a RemoteService you can get a RemoteObject that implements a specific interface
-(an interface is just a standard group of member functions). Then you can call
-those member functions directly.
-
-You can think of a complete method call as looking something like:
-
-Bus:SESSION -> Service:org.gnome.Evolution -> Object:/org/gnome/Evolution/Inbox -> Interface: org.gnome.Evolution.MailFolder -> Method: Forward('message1', 'seth@gnome.org')
-
-This communicates over the SESSION Bus to the org.gnome.Evolution process to call the
-Forward method of the /org/gnome/Evolution/Inbox object (which provides the
-org.gnome.Evolution.MailFolder interface) with two string arguments.
-
-For example, the dbus-daemon itself provides a service and some objects:
-
-# Get a connection to the desktop-wide SESSION bus
-bus = dbus.Bus(dbus.Bus.TYPE_SESSION)
-
-# Get the service provided by the dbus-daemon named org.freedesktop.DBus
-dbus_service = bus.get_service('org.freedesktop.DBus')
-
-# Get a reference to the desktop bus' standard object, denoted
-# by the path /org/freedesktop/DBus. The object /org/freedesktop/DBus
-# implements the 'org.freedesktop.DBus' interface
-dbus_object = dbus_service.get_object('/org/freedesktop/DBus',
- 'org.freedesktop.DBus')
-
-# One of the member functions in the org.freedesktop.DBus interface
-# is ListServices(), which provides a list of all the other services
-# registered on this bus. Call it, and print the list.
-print(dbus_object.ListServices())
-"""
-
-import dbus_bindings
-
-class Bus:
- """A connection to a DBus daemon.
-
- One of three possible standard buses, the SESSION, SYSTEM,
- or ACTIVATION bus
- """
- TYPE_SESSION = dbus_bindings.BUS_SESSION
- TYPE_SYSTEM = dbus_bindings.BUS_SYSTEM
- TYPE_ACTIVATION = dbus_bindings.BUS_ACTIVATION
-
- def __init__(self, bus_type=TYPE_SESSION, glib_mainloop=True):
- self._connection = dbus_bindings.bus_get(bus_type)
-
- self._connection.add_filter(self._signal_func)
- self._match_rule_to_receivers = { }
- if (glib_mainloop):
- self._connection.setup_with_g_main()
-
- def get_service(self, service_name="org.freedesktop.Broadcast"):
- """Get one of the RemoteServices connected to this Bus. service_name
- is just a string of the form 'com.widgetcorp.MyService'
- """
- return RemoteService(self._connection, service_name)
-
- def add_signal_receiver(self, receiver, interface=None, service=None, path=None):
- match_rule = self._get_match_rule(interface, service, path)
-
- if (not self._match_rule_to_receivers.has_key(match_rule)):
- self._match_rule_to_receivers[match_rule] = [ ]
- self._match_rule_to_receivers[match_rule].append(receiver)
-
- dbus_bindings.bus_add_match(self._connection, match_rule)
-
- def get_connection(self):
- """Get the dbus_bindings.Connection object associated with this Bus"""
- return self._connection
-
- def _get_match_rule(self, interface, service, path):
-## if (interface):
-## match_rule = match_rule + ",interface='%s'" % (interface)
-## if (service):
-## match_rule = match_rule + ",service='%s'" % (service)
-## if (path):
-## match_rule = match_rule + ",path='%s'" % (path)
- # FIXME: use the service here too!!!
- return "type='signal',interface='%s',path='%s'" % (interface, path)
-
- def _signal_func(self, connection, message):
- if (message.get_type() != dbus_bindings.MESSAGE_TYPE_SIGNAL):
- return
-
- interface = message.get_interface()
- service = message.get_sender()
- path = message.get_path()
- member = message.get_member()
-
- match_rule = self._get_match_rule(interface, service, path)
-
- if (self._match_rule_to_receivers.has_key(match_rule)):
- receivers = self._match_rule_to_receivers[match_rule]
- args = [interface, member, service, path]
- for receiver in receivers:
- receiver(*args)
-
-
-class RemoteObject:
- """A remote Object.
-
- A RemoteObject is provided by a RemoteService on a particular Bus. RemoteObjects
- have member functions, and can be called like normal Python objects.
- """
- def __init__(self, connection, service_name, object_path, interface):
- self._connection = connection
- self._service_name = service_name
- self._object_path = object_path
- self._interface = interface
-
- def __getattr__(self, member):
- if member == '__call__':
- return object.__call__
- else:
- return RemoteMethod(self._connection, self._service_name,
- self._object_path, self._interface, member)
-
-
-class RemoteMethod:
- """A remote Method.
-
- Typically a member of a RemoteObject. Calls to the
- method produce messages that travel over the Bus and are routed
- to a specific Service.
- """
- def __init__(self, connection, service_name, object_path, interface, method_name):
- self._connection = connection
- self._service_name = service_name
- self._object_path = object_path
- self._interface = interface
- self._method_name = method_name
-
- def __call__(self, *args):
- message = dbus_bindings.MethodCall(self._object_path, self._interface, self._method_name)
- message.set_destination(self._service_name)
-
- # Add the arguments to the function
- iter = message.get_iter()
- for arg in args:
- iter.append(arg)
-
- reply_message = self._connection.send_with_reply_and_block(message, 5000)
-
- args_tuple = reply_message.get_args_list()
- if len(args_tuple) == 0:
- return
- elif len(args_tuple) == 1:
- return args_tuple[0]
- else:
- return args_tuple
-
-class Service:
- """A base class for exporting your own Services across the Bus
-
- Just inherit from Service, providing the name of your service
- (e.g. org.designfu.SampleService).
- """
- def __init__(self, service_name, bus=None):
- self._service_name = service_name
-
- if bus == None:
- # Get the default bus
- self._bus = Bus()
- else:
- self._bus = bus
-
- dbus_bindings.bus_acquire_service(self._bus.get_connection(), service_name)
-
- def get_bus(self):
- """Get the Bus this Service is on"""
- return self._bus
-
- def get_service_name(self):
- """Get the name of this service"""
- return self._service_name
-
-class Object:
- """A base class for exporting your own Objects across the Bus.
-
- Just inherit from Object and provide a list of methods to share
- across the Bus. These will appear as member functions of your
- ServiceObject.
- """
- def __init__(self, object_path, methods_to_share, service):
- self._object_path = object_path
- self._service = service
- self._bus = service.get_bus()
- self._connection = self._bus.get_connection()
-
- self._method_name_to_method = self._build_method_dictionary(methods_to_share)
-
- self._connection.register_object_path(object_path, self._unregister_cb, self._message_cb)
-
- def broadcast_signal(self, interface, signal_name):
- message = dbus_bindings.Signal(self._object_path, interface, signal_name)
- #FIXME: need to set_sender, but it always disconnects when we do this
- #message.set_sender(self._service.get_service_name())
- self._connection.send(message)
-
- def _unregister_cb(self, connection):
- print ("Unregister")
-
- def _message_cb(self, connection, message):
- target_method_name = message.get_member()
- target_method = self._method_name_to_method[target_method_name]
- args = message.get_args_list()
-
- try:
- retval = target_method(*args)
- except Exception, e:
- if e.__module__ == '__main__':
- # FIXME: is it right to use .__name__ here?
- error_name = e.__class__.__name__
- else:
- error_name = e.__module__ + '.' + str(e.__class__.__name__)
- error_contents = str(e)
- reply = dbus_bindings.Error(message, error_name, error_contents)
- else:
- reply = dbus_bindings.MethodReturn(message)
- if retval != None:
- iter = reply.get_iter()
- iter.append(retval)
-
- self._connection.send(reply)
-
- def _build_method_dictionary(self, methods):
- method_dict = {}
- for method in methods:
- if method_dict.has_key(method.__name__):
- print ('WARNING: registering DBus Object methods, already have a method named %s' % (method.__name__))
- method_dict[method.__name__] = method
- return method_dict
-
-class RemoteService:
- """A remote service providing objects.
-
- A service is typically a process or application that provides
- remote objects, but can also be the broadcast service that
- receives signals from all applications on the Bus.
- """
-
- def __init__(self, connection, service_name):
- self._connection = connection
- self._service_name = service_name
-
- def get_object(self, object_path, interface):
- """Get an object provided by this Service that implements a
- particular interface. object_path is a string of the form
- '/com/widgetcorp/MyService/MyObject1'. interface looks a lot
- like a service_name (they're often the same) and is of the form,
- 'com.widgetcorp.MyInterface', and mostly just defines the
- set of member functions that will be present in the object.
- """
- return RemoteObject(self._connection, self._service_name, object_path, interface)
-