Mercurial > libervia-backend
view src/bridge/bridge_constructor/dbus_core_template.py @ 291:7c79d4a8c9e6
plugins: fixed bad import names
author | Goffi <goffi@goffi.org> |
---|---|
date | Sun, 06 Feb 2011 23:49:27 +0100 |
parents | c25371424090 |
children | 15c8916317d0 |
line wrap: on
line source
#!/usr/bin/python #-*- coding: utf-8 -*- """ SAT: a jabber client Copyright (C) 2009, 2010, 2011 Jérôme Poisson (goffi@goffi.org) This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program 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 General Public License for more details. You should have received a copy of the GNU General Public License along with this program. If not, see <http://www.gnu.org/licenses/>. """ from bridge import Bridge import dbus import dbus.service import dbus.mainloop.glib from logging import debug, info const_INT_PREFIX = "org.goffi.SAT" #Interface prefix const_COMM_SUFFIX = ".communication" const_REQ_SUFFIX = ".request" class DbusObject(dbus.service.Object): def __init__(self, bus, path): dbus.service.Object.__init__(self, bus, path) debug("Init DbusObject...") self.cb={} def register(self, name, cb): self.cb[name]=cb ### signals ### ##SIGNALS_PART## ### methods ### ##METHODS_PART## def __attribute_string(self, in_sign): """Return arguments to user given a in_sign @param in_sign: in_sign in the short form (using s,a,i,b etc) @return: list of arguments that correspond to a in_sign (e.g.: "sss" return "arg1, arg2, arg3")""" i=0 idx=0 attr_string="" while i<len(in_sign): if in_sign[i] not in ['b','y','n','i','x','q','u','t','d','s','a']: raise ParseError("Unmanaged attribute type [%c]" % in_sign[i]) attr_string += ("" if idx==0 else ", ") + ("arg_%i" % idx) idx+=1 if in_sign[i] == 'a': i+=1 if in_sign[i]!='{' and in_sign[i]!='(': #FIXME: must manage tuples out of arrays i+=1 continue #we have a simple type for the array opening_car = in_sign[i] assert(opening_car in ['{','(']) closing_car = '}' if opening_car == '{' else ')' opening_count = 1 while (True): #we have a dict or a list of tuples i+=1 if i>=len(in_sign): raise ParseError("missing }") if in_sign[i] == opening_car: opening_count+=1 if in_sign[i] == closing_car: opening_count-=1 if opening_count == 0: break i+=1 return attr_string def addMethod(self, name, int_suffix, in_sign, out_sign): """Dynamically add a method to Dbus Bridge""" #FIXME: Better way ??? attributes = self.__attribute_string(in_sign) code = compile ('def '+name+' (self,'+attributes+'): return self.cb["'+name+'"]('+attributes+')', '<DBus bridge>','exec') exec (code) method = locals()[name] setattr(DbusObject, name, dbus.service.method( const_INT_PREFIX+int_suffix, in_signature=in_sign, out_signature=out_sign)(method)) function = getattr(self, name) func_table = self._dbus_class_table[self.__class__.__module__ + '.' + self.__class__.__name__][function._dbus_interface] func_table[function.__name__] = function #Needed for introspection def addSignal(self, name, int_suffix, signature): """Dynamically add a signal to Dbus Bridge""" #FIXME: Better way ??? attributes = self.__attribute_string(signature) code = compile ('def '+name+' (self,'+attributes+'): debug ("'+name+' signal")', '<DBus bridge>','exec') exec (code) signal = locals()[name] setattr(DbusObject, name, dbus.service.signal( const_INT_PREFIX+int_suffix, signature=signature)(signal)) function = getattr(self, name) func_table = self._dbus_class_table[self.__class__.__module__ + '.' + self.__class__.__name__][function._dbus_interface] func_table[function.__name__] = function #Needed for introspection class DBusBridge(Bridge): def __init__(self): dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) Bridge.__init__(self) info ("Init DBus...") self.session_bus = dbus.SessionBus() self.dbus_name = dbus.service.BusName(const_INT_PREFIX, self.session_bus) self.dbus_bridge = DbusObject(self.session_bus, '/org/goffi/SAT/bridge') ##DIRECT_CALLS## def register(self, name, callback): debug("registering DBus bridge method [%s]", name) self.dbus_bridge.register(name, callback) def addMethod(self, name, int_suffix, in_sign, out_sign, method): """Dynamically add a method to Dbus Bridge""" print ("Adding method [%s] to DBus bridge" % name) self.dbus_bridge.addMethod(name, int_suffix, in_sign, out_sign) self.register(name, method) def addSignal(self, name, int_suffix, signature): self.dbus_bridge.addSignal(name, int_suffix, signature) setattr(DBusBridge, name, getattr(self.dbus_bridge, name))