import os
import sys
__all__ = ['PackageLoader']
class PackageLoader:
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>}.
"""
import imp
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, 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 = imp.load_module(fullname+'.info',
open(info_file,filedescriptor[1]),
info_file,
filedescriptor)
except Exception,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 depend_dict.keys():
if not depend_dict[name]:
package_names.append(name)
del depend_dict[name]
while depend_dict:
for name, lst in 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]
"""
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 = filter(lambda s:not s.startswith('_'),symbols)
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,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 >> sys.stderr, str(mess)
def warn(self,mess):
if self.verbose>=0:
print >> sys.stderr, str(mess)
def error(self,mess):
if self.verbose!=-1:
print >> sys.stderr, str(mess)
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',`cmdstr`,'...',
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
|