_db.py

Go to the documentation of this file.

 
import os
import sys
import logging
import atexit
 
 
class ConfDB2(object):
    def __init__(self):
        import shelve
        self._dbs = {}
        pathvar = ('DYLD_LIBRARY_PATH'
                   if sys.platform == 'darwin' else 'LD_LIBRARY_PATH')
        for path in os.getenv(pathvar, '').split(os.pathsep):
            if not os.path.isdir(path):
                continue
            dbfiles = [
                os.path.join(path, f) for f in os.listdir(path)
                if f.endswith('.confdb2')
                and os.path.isfile(os.path.join(path, f))
            ]
            dbfiles.sort()
            for db in [shelve.open(f, 'r') for f in dbfiles]:
                for key in db:
                    if key not in self._dbs:
                        self._dbs[key] = db
 
    def __getitem__(self, key):
        return self._dbs[key][key]
 
    def __contains__(self, key):
        return key in self._dbs
 
    def __iter__(self):
        return iter(self._dbs)
 
 
# allow overriding the low level DB access for testing
if 'GAUDICONFIG2_DB' in os.environ:
    exec('from {} import {} as _DB'.format(
        *os.environ['GAUDICONFIG2_DB'].rsplit('.', 1)))
else:  # pragma no cover
    _DB = ConfDB2()
 
 
# Workaround for a bug in Python (2) clean up
# see https://stackoverflow.com/q/2180946
def _fix_clean_up():  # pragma no cover
    global _DB
    del _DB
 
 
atexit.register(_fix_clean_up)
 
if sys.version_info >= (3, ):  # pragma no cover
    maketrans = str.maketrans
else:  # pragma no cover
    from string import maketrans
_TRANS_TABLE = maketrans("<>&*,: ().", "__rp__s___")
 
 
def _normalize_cpp_type_name(name):
    '''
    Translate a C++ type name (with templates etc.) to Python identifier.
    '''
    return name.replace(', ', ',').translate(_TRANS_TABLE)
 
 
def split_namespace(typename):
    '''
    Split a C++ qualified namespace in the tuple (top_namespace, rest) starting
    searching the separator from pos.
 
    >>> split_namespace('std::chrono::time_point')
    ('std', 'chrono::time_point')
    '''
    # find next namespace separator skipping template arguments
    tpos = typename.find('<')
    pos = typename.find('::')
    # '<' can appear only in a class name, if we find a '::'
    # earlier than a '<', then we got a namespace, else a class
    if pos > 0 and (tpos < 0 or pos < tpos):
        head = typename[:pos]
        tail = typename[pos + 2:]
    else:
        head = None
        tail = typename
    return (head, tail)
 
 
class ConfigurablesDB(object):
    '''
    Helper to expose Configurables classes (from Configurables database) as
    a tree of subpackages, each mapped to a namespace.
    '''
 
    def __init__(self, modulename, root=None):
        '''
        @param modulename: name of the module
        @param root: name of the root modules (that pointing to the root C++
                     namespace), None is ewuivalent to pass modulename as root
        '''
        self._log = logging.getLogger(modulename)
        self._log.debug('initializing module %r (with root %r)', modulename,
                        root)
 
        self.__name__ = modulename
        self.__loader__ = None
        self._root = root or modulename
        assert ((not root) or modulename.startswith(root + '.')), \
            'modulename should be (indirect submodule of root)'
 
        self._namespace = modulename[len(root) + 1:].replace(
            '.', '::') if root else ''
        self._log.debug('%r mapping namespace %r', modulename, self._namespace
                        or '::')
 
        self._namespaces, self._classes = self._getEntries()
        self._alt_names = {}
        for cname in self._classes:
            alt_name = _normalize_cpp_type_name(cname)
            if alt_name != cname:
                self._alt_names[alt_name] = cname
            if ' ' in cname:  # allow matching of 'T<A, B>' a well as 'T<A,B>'
                self._alt_names[cname.replace(' ', '')] = cname
        self.__all__ = list(
            self._namespaces.union(self._classes).union(self._alt_names))
 
        sys.modules[modulename] = self
 
        for submodule in self._namespaces:
            setattr(
                self, submodule,
                ConfigurablesDB('.'.join([self.__name__, submodule]),
                                self._root))
 
    def _getEntries(self):
        '''
        Extract from the Configurables DB the namespaces and classes names in
        the namespace this instance represents.
        '''
        self._log.debug('getting list of entries under %r', self._namespace)
        prefix = self._namespace + '::' if self._namespace else ''
        prefix_len = len(prefix)
 
        namespaces = set()
        classes = set()
        for name in _DB:
            if name.startswith(prefix):
                head, tail = split_namespace(name[prefix_len:])
                if head:
                    namespaces.add(head)
                else:
                    classes.add(tail)
 
        self._log.debug('found %d namespaces and %d classes', len(namespaces),
                        len(classes))
        return (namespaces, classes)
 
    def __getattr__(self, name):
        '''
        Helper to instantiate on demand Configurable classes.
        '''
        if name in self._classes:
            fullname = ('::'.join([self._namespace, name])
                        if self._namespace else name)
            from ._configurables import makeConfigurableClass
            self._log.debug('generating %r (%s)', name, fullname)
            entry = makeConfigurableClass(
                name,
                __module__=self.__name__,
                __qualname__=name,
                __cpp_type__=fullname,
                **_DB[fullname])
        elif name.replace(' ', '') in self._alt_names:
            entry = getattr(self, self._alt_names[name.replace(' ', '')])
        elif name == "__spec__":  # pragma no cover
            import importlib
            entry = importlib.machinery.ModuleSpec(
                name=self.__package__,
                loader=self.__loader__,
            )
        elif name == "__package__":  # pragma no cover
            entry = self.__name__
        else:
            raise AttributeError('module {!r} has no attribute {!r}'.format(
                self.__name__, name))
        setattr(self, name, entry)
        return entry
 
    def getByType(self, typename):
        '''
        Return a configurable from the fully qualified type name (relative to
        the current namespace).any
        '''
        head, tail = split_namespace(typename)
        if head:
            return getattr(self, head).getByType(tail)
        else:
            return getattr(self, tail)