Mercurial > libervia-backend
view src/bridge/bridge_contructor.py @ 266:c4b84a2d2ad1
bridge: constructor and template improved, documentation added
author | Goffi <goffi@goffi.org> |
---|---|
date | Mon, 24 Jan 2011 17:47:45 +0100 |
parents | b5f1f3dc9ac6 |
children |
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/>. """ #consts NAME = u"bridge_constructor" VERSION="0.1.0" ABOUT = NAME+u""" v%s (c) Jérôme Poisson (aka Goffi) 2011 --- """+NAME+u""" Copyright (C) 2011 Jérôme Poisson (aka Goffi) This program comes with ABSOLUTELY NO WARRANTY; This is free software, and you are welcome to redistribute it under certain conditions. --- This script construct a SàT bridge using the given protocol """ MANAGED_PROTOCOLES=['dbus'] DEFAULT_PROTOCOLE='dbus' import sys import os from os import path from optparse import OptionParser from ConfigParser import SafeConfigParser as Parser from ConfigParser import NoOptionError import re class ParseError(Exception): #Used when the signature parsing is going wrong (invalid signature ?) pass class Constructor: def __init__(self, bridge_template, options): self.bridge_template = bridge_template self.options = options def getValues(self, name): """Return values of a function in a dict @param name: Name of the function to get @return: dict, each key has the config value or None if the value is not set""" function={} for option in ['type','category','sig_in','sig_out','doc']: try: value = self.bridge_template.get(name, option) except NoOptionError: value = None function[option] = value return function def getDefault(self, name): """Return default values of a function in a dict @param name: Name of the function to get @return: dict, each key is the integer param number (no key if no default value)""" default_dict={} def_re = re.compile(r"param_(\d+)_default") for option in self.bridge_template.options(name): if option == 'doc_return': default_dict['return'] = self.bridge_template.get(name, option) continue match = def_re.match(option) if match: try: idx = int(match.group(1)) except ValueError: raise ParseError("Invalid value [%s] for parameter number" % match.group(1)) default_dict[idx] = self.bridge_template.get(name, option) return default_dict def getArgumentsDoc(self, name): """Return documentation of arguments @param name: Name of the function to get @return: dict, each key is the integer param number (no key if no argument doc), value is a tuple (name, doc)""" doc_dict={} option_re = re.compile(r"doc_param_(\d+)") value_re = re.compile(r"^(\w+): (.*)$", re.MULTILINE | re.DOTALL) for option in self.bridge_template.options(name): match = option_re.match(option) if match: try: idx = int(match.group(1)) except ValueError: raise ParseError("Invalid value [%s] for parameter number" % match.group(1)) value_match = value_re.match(self.bridge_template.get(name, option)) if not value_match: raise ParseError("Invalid value for parameter doc [%i]" % idx) doc_dict[idx]=(value_match.group(1),value_match.group(2)) return doc_dict def getArguments(self, signature, name=None, default=None): """Return arguments to user given a signature @param signature: signature in the short form (using s,a,i,b etc) @return: list of arguments that correspond to a signature (e.g.: "sss" return "arg1, arg2, arg3")""" i=0 idx=0 attr_string="" while i<len(signature): if signature[i] not in ['b','y','n','i','x','q','u','t','d','s','a']: raise ParseError("Unmanaged attribute type [%c]" % signature[i]) attr_string += ("" if idx==0 else ", ") + ("%(name)s%(default)s" % { 'name':name[idx][0] if (name and name.has_key(idx)) else "arg_%i" % idx, 'default':"="+default[idx] if (default and default.has_key(idx)) else '' }) #give arg_1, arg2, etc or name1, name2=default, etc idx+=1 if signature[i] == 'a': i+=1 if signature[i]!='{' and signature[i]!='(': #FIXME: must manage tuples out of arrays i+=1 continue #we have a simple type for the array opening_car = signature[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(signature): raise ParseError("missing }") if signature[i] == opening_car: opening_count+=1 if signature[i] == closing_car: opening_count-=1 if opening_count == 0: break i+=1 return attr_string def generateCoreSide(self): """create the constructor in SàT core side (backend)""" raise NotImplementedError class DbusConstructor(Constructor): def __init__(self, bridge_template, options): Constructor.__init__(self, bridge_template, options) self.core_template="dbus_core_template.py" self.core_dest="DBus.py" def generateCoreSide(self): signals_part = [] methods_part = [] direct_calls = [] sections = self.bridge_template.sections() sections.sort() for section in sections: function = self.getValues(section) print ("Adding %s %s" % (section, function["type"])) default = self.getDefault(section) arg_doc = self.getArgumentsDoc(section) completion = { 'sig_in':function['sig_in'] or '', 'sig_out':function['sig_out'] or '', 'category':'REQ' if function['category'] == 'request' else 'COMM', 'name':section, 'args':self.getArguments(function['sig_in'], name=arg_doc, default=default, ) } if function["type"] == "signal": completion['body'] = "pass" if not self.options.debug else 'debug ("%s")' % section signals_part.append("""\ @dbus.service.signal(const_INT_PREFIX+const_%(category)s_SUFFIX, signature='%(sig_in)s') def %(name)s(self, %(args)s): %(body)s """ % completion) direct_calls.append("""\ def %(name)s(self, %(args)s): self.dbus_bridge.%(name)s(%(args)s) """ % completion) elif function["type"] == "method": completion['debug'] = "" if not self.options.debug else 'debug ("%s")\n%s' % (section,8*' ') completion['args_nodefault'] = self.getArguments(function['sig_in'], name=arg_doc) methods_part.append("""\ @dbus.service.method(const_INT_PREFIX+const_%(category)s_SUFFIX, in_signature='%(sig_in)s', out_signature='%(sig_out)s') def %(name)s(self, %(args)s): %(debug)sreturn self.cb["%(name)s"](%(args_nodefault)s) """ % completion) #at this point, signals_part, methods_part and direct_calls should be filled, #we just have to place them in the right part of the template core_bridge = [] try: with open(self.core_template) as core_template: for line in core_template: if line.startswith('##SIGNALS_PART##'): core_bridge.extend(signals_part) elif line.startswith('##METHODS_PART##'): core_bridge.extend(methods_part) elif line.startswith('##DIRECT_CALLS##'): core_bridge.extend(direct_calls) else: core_bridge.append(line.replace('\n','')) except IOError: print ("Can't open template file [%s]" % self.core_template) sys.exit(1) #now we write to final file if os.path.exists(self.core_dest) and not self.options.force: print ("The destination file [%s] already exists ! Use --force to overwrite it" % self.core_dest) try: with open(self.core_dest,'w') as dest_file: dest_file.write('\n'.join(core_bridge)) except IOError: print ("Can't open destination file [%s]" % self.dest_file) class ConstructorError(Exception): pass class ConstructorFactory: def create(self, bridge_template, options): if options.protocole=='dbus': return DbusConstructor(bridge_template, options) raise ConstructorError('Unknown constructor type') class BridgeConstructor: def __init__(self): self.options = None def check_options(self): """Check command line options""" _usage=""" %prog [options] %prog --help for options list """ parser = OptionParser(usage=_usage,version=ABOUT % VERSION) parser.add_option("-p", "--protocole", action="store", type="string", default=DEFAULT_PROTOCOLE, help="Generate bridge using PROTOCOLE (default: %s, possible values: [%s])" % (DEFAULT_PROTOCOLE, ", ".join(MANAGED_PROTOCOLES))) parser.add_option("-s", "--side", action="store", type="string", default="core", help="Which side of the bridge do you want to make ? (default: %default, possible values: [core, frontend])") parser.add_option("-t", "--template", action="store", type="string", default='bridge_template.ini', help="Use TEMPLATE to generate bridge (default: %default)") parser.add_option("-f", "--force", action="store_true", default=False, help=("Force overwritting of existing files")) parser.add_option("-d", "--debug", action="store_true", default=False, help=("Add debug information printing")) (self.options, args) = parser.parse_args() return args def go(self): self.check_options() self.template = Parser() try: self.template.readfp(open(self.options.template)) except IOError: print ("The template file doesn't exist or is not accessible") exit(1) constructor = ConstructorFactory().create(self.template, self.options) if self.options.side == "core": constructor.generateCoreSide() if __name__ == "__main__": bc = BridgeConstructor() bc.go()