from __future__ import division, absolute_import, print_function import os import sys import warnings __all__ = ['PackageLoader'] class PackageLoader(object): def __init__(self, verbose=False, infunc=False): """ Manages loading packages. """ if infunc: _level = 2 else: _level = 1 self.parent_frame = frame = sys._getframe(_level) self.parent_name = eval('__name__', frame.f_globals, frame.f_locals) parent_path = eval('__path__', frame.f_globals, frame.f_locals) if isinstance(parent_path, str): parent_path = [parent_path] self.parent_path = parent_path if '__all__' not in frame.f_locals: exec('__all__ = []', frame.f_globals, frame.f_locals) self.parent_export_names = eval('__all__', frame.f_globals, frame.f_locals) self.info_modules = {} self.imported_packages = [] self.verbose = None def _get_info_files(self, package_dir, parent_path, parent_package=None): """ Return list of (package name,info.py file) from parent_path subdirectories. """ from glob import glob files = glob(os.path.join(parent_path, package_dir, 'info.py')) for info_file in glob(os.path.join(parent_path, package_dir, 'info.pyc')): if info_file[:-1] not in files: files.append(info_file) info_files = [] for info_file in files: package_name = os.path.dirname(info_file[len(parent_path)+1:])\ .replace(os.sep, '.') if parent_package: package_name = parent_package + '.' + package_name info_files.append((package_name, info_file)) info_files.extend(self._get_info_files('*', os.path.dirname(info_file), package_name)) return info_files def _init_info_modules(self, packages=None): """Initialize info_modules = {<package_name>: <package info.py module>}. """ from numpy.compat import npy_load_module info_files = [] info_modules = self.info_modules if packages is None: for path in self.parent_path: info_files.extend(self._get_info_files('*', path)) else: for package_name in packages: package_dir = os.path.join(*package_name.split('.')) for path in self.parent_path: names_files = self._get_info_files(package_dir, path) if names_files: info_files.extend(names_files) break else: try: exec('import %s.info as info' % (package_name)) info_modules[package_name] = info except ImportError as msg: self.warn('No scipy-style subpackage %r found in %s. '\ 'Ignoring: %s'\ % (package_name, ':'.join(self.parent_path), msg)) for package_name, info_file in info_files: if package_name in info_modules: continue fullname = self.parent_name +'.'+ package_name if info_file[-1]=='c': filedescriptor = ('.pyc', 'rb', 2) else: filedescriptor = ('.py', 'U', 1) try: info_module = npy_load_module(fullname + '.info', info_file, filedescriptor) except Exception as msg: self.error(msg) info_module = None if info_module is None or getattr(info_module, 'ignore', False): info_modules.pop(package_name, None) else: self._init_info_modules(getattr(info_module, 'depends', [])) info_modules[package_name] = info_module return def _get_sorted_names(self): """ Return package names sorted in the order as they should be imported due to dependence relations between packages. """ depend_dict = {} for name, info_module in self.info_modules.items(): depend_dict[name] = getattr(info_module, 'depends', []) package_names = [] for name in list(depend_dict.keys()): if not depend_dict[name]: package_names.append(name) del depend_dict[name] while depend_dict: for name, lst in list(depend_dict.items()): new_lst = [n for n in lst if n in depend_dict] if not new_lst: package_names.append(name) del depend_dict[name] else: depend_dict[name] = new_lst return package_names def __call__(self,*packages, **options): """Load one or more packages into parent package top-level namespace. This function is intended to shorten the need to import many subpackages, say of scipy, constantly with statements such as import scipy.linalg, scipy.fftpack, scipy.etc... Instead, you can say: import scipy scipy.pkgload('linalg','fftpack',...) or scipy.pkgload() to load all of them in one call. If a name which doesn't exist in scipy's namespace is given, a warning is shown. Parameters ---------- *packages : arg-tuple the names (one or more strings) of all the modules one wishes to load into the top-level namespace. verbose= : integer verbosity level [default: -1]. verbose=-1 will suspend also warnings. force= : bool when True, force reloading loaded packages [default: False]. postpone= : bool when True, don't load packages [default: False] """ # 2014-10-29, 1.10 warnings.warn('pkgload and PackageLoader are obsolete ' 'and will be removed in a future version of numpy', DeprecationWarning, stacklevel=2) frame = self.parent_frame self.info_modules = {} if options.get('force', False): self.imported_packages = [] self.verbose = verbose = options.get('verbose', -1) postpone = options.get('postpone', None) self._init_info_modules(packages or None) self.log('Imports to %r namespace\n----------------------------'\ % self.parent_name) for package_name in self._get_sorted_names(): if package_name in self.imported_packages: continue info_module = self.info_modules[package_name] global_symbols = getattr(info_module, 'global_symbols', []) postpone_import = getattr(info_module, 'postpone_import', False) if (postpone and not global_symbols) \ or (postpone_import and postpone is not None): continue old_object = frame.f_locals.get(package_name, None) cmdstr = 'import '+package_name if self._execcmd(cmdstr): continue self.imported_packages.append(package_name) if verbose!=-1: new_object = frame.f_locals.get(package_name) if old_object is not None and old_object is not new_object: self.warn('Overwriting %s=%s (was %s)' \ % (package_name, self._obj2repr(new_object), self._obj2repr(old_object))) if '.' not in package_name: self.parent_export_names.append(package_name) for symbol in global_symbols: if symbol=='*': symbols = eval('getattr(%s,"__all__",None)'\ % (package_name), frame.f_globals, frame.f_locals) if symbols is None: symbols = eval('dir(%s)' % (package_name), frame.f_globals, frame.f_locals) symbols = [s for s in symbols if not s.startswith('_')] else: symbols = [symbol] if verbose!=-1: old_objects = {} for s in symbols: if s in frame.f_locals: old_objects[s] = frame.f_locals[s] cmdstr = 'from '+package_name+' import '+symbol if self._execcmd(cmdstr): continue if verbose!=-1: for s, old_object in old_objects.items(): new_object = frame.f_locals[s] if new_object is not old_object: self.warn('Overwriting %s=%s (was %s)' \ % (s, self._obj2repr(new_object), self._obj2repr(old_object))) if symbol=='*': self.parent_export_names.extend(symbols) else: self.parent_export_names.append(symbol) return def _execcmd(self, cmdstr): """ Execute command in parent_frame.""" frame = self.parent_frame try: exec (cmdstr, frame.f_globals, frame.f_locals) except Exception as msg: self.error('%s -> failed: %s' % (cmdstr, msg)) return True else: self.log('%s -> success' % (cmdstr)) return def _obj2repr(self, obj): """ Return repr(obj) with""" module = getattr(obj, '__module__', None) file = getattr(obj, '__file__', None) if module is not None: return repr(obj) + ' from ' + module if file is not None: return repr(obj) + ' from ' + file return repr(obj) def log(self, mess): if self.verbose>1: print(str(mess), file=sys.stderr) def warn(self, mess): if self.verbose>=0: print(str(mess), file=sys.stderr) def error(self, mess): if self.verbose!=-1: print(str(mess), file=sys.stderr) def _get_doc_title(self, info_module): """ Get the title from a package info.py file. """ title = getattr(info_module, '__doc_title__', None) if title is not None: return title title = getattr(info_module, '__doc__', None) if title is not None: title = title.lstrip().split('\n', 1)[0] return title return '* Not Available *' def _format_titles(self,titles,colsep='---'): display_window_width = 70 # How to determine the correct value in runtime?? lengths = [len(name)-name.find('.')-1 for (name, title) in titles]+[0] max_length = max(lengths) lines = [] for (name, title) in titles: name = name[name.find('.')+1:] w = max_length - len(name) words = title.split() line = '%s%s %s' % (name, w*' ', colsep) tab = len(line) * ' ' while words: word = words.pop(0) if len(line)+len(word)>display_window_width: lines.append(line) line = tab line += ' ' + word else: lines.append(line) return '\n'.join(lines) def get_pkgdocs(self): """ Return documentation summary of subpackages. """ import sys self.info_modules = {} self._init_info_modules(None) titles = [] symbols = [] for package_name, info_module in self.info_modules.items(): global_symbols = getattr(info_module, 'global_symbols', []) fullname = self.parent_name +'.'+ package_name note = '' if fullname not in sys.modules: note = ' [*]' titles.append((fullname, self._get_doc_title(info_module) + note)) if global_symbols: symbols.append((package_name, ', '.join(global_symbols))) retstr = self._format_titles(titles) +\ '\n [*] - using a package requires explicit import (see pkgload)' if symbols: retstr += """\n\nGlobal symbols from subpackages"""\ """\n-------------------------------\n""" +\ self._format_titles(symbols, '-->') return retstr class PackageLoaderDebug(PackageLoader): def _execcmd(self, cmdstr): """ Execute command in parent_frame.""" frame = self.parent_frame print('Executing', repr(cmdstr), '...', end=' ') sys.stdout.flush() exec (cmdstr, frame.f_globals, frame.f_locals) print('ok') sys.stdout.flush() return if int(os.environ.get('NUMPY_IMPORT_DEBUG', '0')): PackageLoader = PackageLoaderDebug