"""The VFS importer allows importing Python modules from Panda3D's virtual
file system, through Python's standard import mechanism.
Calling the :func:`register()` function to register the import hooks should be
sufficient to enable this functionality.
"""
__all__ = ['register', 'sharedPackages',
'reloadSharedPackage', 'reloadSharedPackages']
from panda3d.core import Filename, VirtualFileSystem, VirtualFileMountSystem, OFileStream, copyStream
from direct.stdpy.file import open
import sys
import marshal
import imp
import types
#: The sharedPackages dictionary lists all of the "shared packages",
#: special Python packages that automatically span multiple directories
#: via magic in the VFSImporter. You can make a package "shared"
#: simply by adding its name into this dictionary (and then calling
#: reloadSharedPackages() if it's already been imported).
#:
#: When a package name is in this dictionary at import time, *all*
#: instances of the package are located along sys.path, and merged into
#: a single Python module with a __path__ setting that represents the
#: union. Thus, you can have a direct.showbase.foo in your own
#: application, and loading it won't shadow the system
#: direct.showbase.ShowBase which is in a different directory on disk.
sharedPackages = {}
vfs = VirtualFileSystem.getGlobalPtr()
compiledExtensions = [ 'pyc', 'pyo' ]
if not __debug__:
# In optimized mode, we prefer loading .pyo files over .pyc files.
# We implement that by reversing the extension names.
compiledExtensions = [ 'pyo', 'pyc' ]
[docs]class VFSImporter:
""" This class serves as a Python importer to support loading
Python .py and .pyc/.pyo files from Panda's Virtual File System,
which allows loading Python source files from mounted .mf files
(among other places). """
[docs] def __init__(self, path):
if isinstance(path, Filename):
self.dir_path = Filename(path)
else:
self.dir_path = Filename.fromOsSpecific(path)
[docs] def find_module(self, fullname, path = None):
if path is None:
dir_path = self.dir_path
else:
dir_path = path
#print >>sys.stderr, "find_module(%s), dir_path = %s" % (fullname, dir_path)
basename = fullname.split('.')[-1]
path = Filename(dir_path, basename)
# First, look for Python files.
filename = Filename(path)
filename.setExtension('py')
vfile = vfs.getFile(filename, True)
if vfile:
return VFSLoader(dir_path, vfile, filename,
desc=('.py', 'r', imp.PY_SOURCE))
# If there's no .py file, but there's a .pyc file, load that
# anyway.
for ext in compiledExtensions:
filename = Filename(path)
filename.setExtension(ext)
vfile = vfs.getFile(filename, True)
if vfile:
return VFSLoader(dir_path, vfile, filename,
desc=('.'+ext, 'rb', imp.PY_COMPILED))
# Look for a C/C++ extension module.
for desc in imp.get_suffixes():
if desc[2] != imp.C_EXTENSION:
continue
filename = Filename(path + desc[0])
vfile = vfs.getFile(filename, True)
if vfile:
return VFSLoader(dir_path, vfile, filename, desc=desc)
# Finally, consider a package, i.e. a directory containing
# __init__.py.
filename = Filename(path, '__init__.py')
vfile = vfs.getFile(filename, True)
if vfile:
return VFSLoader(dir_path, vfile, filename, packagePath=path,
desc=('.py', 'r', imp.PY_SOURCE))
for ext in compiledExtensions:
filename = Filename(path, '__init__.' + ext)
vfile = vfs.getFile(filename, True)
if vfile:
return VFSLoader(dir_path, vfile, filename, packagePath=path,
desc=('.'+ext, 'rb', imp.PY_COMPILED))
#print >>sys.stderr, "not found."
return None
[docs]class VFSLoader:
""" The second part of VFSImporter, this is created for a
particular .py file or directory. """
[docs] def __init__(self, dir_path, vfile, filename, desc, packagePath=None):
self.dir_path = dir_path
self.timestamp = None
if vfile:
self.timestamp = vfile.getTimestamp()
self.filename = filename
self.desc = desc
self.packagePath = packagePath
[docs] def load_module(self, fullname, loadingShared = False):
#print >>sys.stderr, "load_module(%s), dir_path = %s, filename = %s" % (fullname, self.dir_path, self.filename)
if self.desc[2] == imp.PY_FROZEN:
return self._import_frozen_module(fullname)
if self.desc[2] == imp.C_EXTENSION:
return self._import_extension_module(fullname)
# Check if this is a child of a shared package.
if not loadingShared and self.packagePath and '.' in fullname:
parentname = fullname.rsplit('.', 1)[0]
if parentname in sharedPackages:
# It is. That means it's a shared package too.
parent = sys.modules[parentname]
path = getattr(parent, '__path__', None)
importer = VFSSharedImporter()
sharedPackages[fullname] = True
loader = importer.find_module(fullname, path = path)
assert loader
return loader.load_module(fullname)
code = self._read_code()
if not code:
raise ImportError('No Python code in %s' % (fullname))
mod = sys.modules.setdefault(fullname, imp.new_module(fullname))
mod.__file__ = self.filename.toOsSpecific()
mod.__loader__ = self
if self.packagePath:
mod.__path__ = [self.packagePath.toOsSpecific()]
#print >> sys.stderr, "loaded %s, path = %s" % (fullname, mod.__path__)
exec(code, mod.__dict__)
return sys.modules[fullname]
[docs] def getdata(self, path):
path = Filename(self.dir_path, Filename.fromOsSpecific(path))
vfile = vfs.getFile(path)
if not vfile:
raise IOError("Could not find '%s'" % (path))
return vfile.readFile(True)
[docs] def is_package(self, fullname):
return bool(self.packagePath)
[docs] def get_code(self, fullname):
return self._read_code()
[docs] def get_source(self, fullname):
return self._read_source()
[docs] def get_filename(self, fullname):
return self.filename.toOsSpecific()
def _read_source(self):
""" Returns the Python source for this file, if it is
available, or None if it is not. May raise IOError. """
if self.desc[2] == imp.PY_COMPILED or \
self.desc[2] == imp.C_EXTENSION:
return None
filename = Filename(self.filename)
filename.setExtension('py')
filename.setText()
# Use the tokenize module to detect the encoding.
import tokenize
fh = open(self.filename, 'rb')
encoding, lines = tokenize.detect_encoding(fh.readline)
return (b''.join(lines) + fh.read()).decode(encoding)
def _import_extension_module(self, fullname):
""" Loads the binary shared object as a Python module, and
returns it. """
vfile = vfs.getFile(self.filename, False)
# We can only import an extension module if it already exists on
# disk. This means if it's a truly virtual file that has no
# on-disk equivalent, we have to write it to a temporary file
# first.
if hasattr(vfile, 'getMount') and \
isinstance(vfile.getMount(), VirtualFileMountSystem):
# It's a real file.
filename = self.filename
elif self.filename.exists():
# It's a virtual file, but it's shadowing a real file in
# the same directory. Assume they're the same, and load
# the real one.
filename = self.filename
else:
# It's a virtual file with no real-world existence. Dump
# it to disk. TODO: clean up this filename.
filename = Filename.temporary('', self.filename.getBasenameWoExtension(),
'.' + self.filename.getExtension(),
type = Filename.TDso)
filename.setExtension(self.filename.getExtension())
filename.setBinary()
sin = vfile.openReadFile(True)
sout = OFileStream()
if not filename.openWrite(sout):
raise IOError
if not copyStream(sin, sout):
raise IOError
vfile.closeReadFile(sin)
del sout
module = imp.load_module(fullname, None, filename.toOsSpecific(),
self.desc)
module.__file__ = self.filename.toOsSpecific()
return module
def _import_frozen_module(self, fullname):
""" Imports the frozen module without messing around with
searching any more. """
#print >>sys.stderr, "importing frozen %s" % (fullname)
module = imp.load_module(fullname, None, fullname,
('', '', imp.PY_FROZEN))
module.__path__ = []
return module
def _read_code(self):
""" Returns the Python compiled code object for this file, if
it is available, or None if it is not. May raise IOError,
ValueError, SyntaxError, or a number of other errors generated
by the low-level system. """
if self.desc[2] == imp.PY_COMPILED:
# It's a pyc file; just read it directly.
pycVfile = vfs.getFile(self.filename, False)
if pycVfile:
return self._loadPyc(pycVfile, None)
raise IOError('Could not read %s' % (self.filename))
elif self.desc[2] == imp.C_EXTENSION:
return None
# It's a .py file (or an __init__.py file; same thing). Read
# the .pyc file if it is available and current; otherwise read
# the .py file and compile it.
t_pyc = None
for ext in compiledExtensions:
pycFilename = Filename(self.filename)
pycFilename.setExtension(ext)
pycVfile = vfs.getFile(pycFilename, False)
if pycVfile:
t_pyc = pycVfile.getTimestamp()
break
code = None
if t_pyc and t_pyc >= self.timestamp:
try:
code = self._loadPyc(pycVfile, self.timestamp)
except ValueError:
code = None
if not code:
source = self._read_source()
filename = Filename(self.filename)
filename.setExtension('py')
code = self._compile(filename, source)
return code
def _loadPyc(self, vfile, timestamp):
""" Reads and returns the marshal data from a .pyc file.
Raises ValueError if there is a problem. """
code = None
data = vfile.readFile(True)
if data[:4] != imp.get_magic():
raise ValueError("Bad magic number in %s" % (vfile))
t = int.from_bytes(data[4:8], 'little')
data = data[12:]
if not timestamp or t == timestamp:
return marshal.loads(data)
else:
raise ValueError("Timestamp wrong on %s" % (vfile))
def _compile(self, filename, source):
""" Compiles the Python source code to a code object and
attempts to write it to an appropriate .pyc file. May raise
SyntaxError or other errors generated by the compiler. """
if source and source[-1] != '\n':
source = source + '\n'
code = compile(source, filename.toOsSpecific(), 'exec')
# try to cache the compiled code
pycFilename = Filename(filename)
pycFilename.setExtension(compiledExtensions[0])
try:
f = open(pycFilename.toOsSpecific(), 'wb')
except IOError:
pass
else:
f.write(imp.get_magic())
f.write((self.timestamp & 0xffffffff).to_bytes(4, 'little'))
f.write(b'\0\0\0\0')
f.write(marshal.dumps(code))
f.close()
return code
[docs]class VFSSharedImporter:
""" This is a special importer that is added onto the meta_path
list, so that it is called before sys.path is traversed. It uses
special logic to load one of the "shared" packages, by searching
the entire sys.path for all instances of this shared package, and
merging them. """
[docs] def __init__(self):
pass
[docs] def find_module(self, fullname, path = None, reload = False):
#print >>sys.stderr, "shared find_module(%s), path = %s" % (fullname, path)
if fullname not in sharedPackages:
# Not a shared package; fall back to normal import.
return None
if path is None:
path = sys.path
excludePaths = []
if reload:
# If reload is true, we are simply reloading the module,
# looking for new paths to add.
mod = sys.modules[fullname]
excludePaths = getattr(mod, '_vfs_shared_path', None)
if excludePaths is None:
# If there isn't a _vfs_shared_path symbol already,
# the module must have been loaded through
# conventional means. Try to guess which path it was
# found on.
d = self.getLoadedDirname(mod)
excludePaths = [d]
loaders = []
for dir in path:
if dir in excludePaths:
continue
importer = sys.path_importer_cache.get(dir, None)
if importer is None:
try:
importer = VFSImporter(dir)
except ImportError:
continue
sys.path_importer_cache[dir] = importer
try:
loader = importer.find_module(fullname)
if not loader:
continue
except ImportError:
continue
loaders.append(loader)
if not loaders:
return None
return VFSSharedLoader(loaders, reload = reload)
[docs] def getLoadedDirname(self, mod):
""" Returns the directory name that the indicated
conventionally-loaded module must have been loaded from. """
if not getattr(mod, '__file__', None):
return None
fullname = mod.__name__
dirname = Filename.fromOsSpecific(mod.__file__).getDirname()
parentname = None
basename = fullname
if '.' in fullname:
parentname, basename = fullname.rsplit('.', 1)
path = None
if parentname:
parent = sys.modules[parentname]
path = parent.__path__
if path is None:
path = sys.path
for dir in path:
pdir = str(Filename.fromOsSpecific(dir))
if pdir + '/' + basename == dirname:
# We found it!
return dir
# Couldn't figure it out.
return None
[docs]class VFSSharedLoader:
""" The second part of VFSSharedImporter, this imports a list of
packages and combines them. """
[docs] def __init__(self, loaders, reload):
self.loaders = loaders
self.reload = reload
[docs] def load_module(self, fullname):
#print >>sys.stderr, "shared load_module(%s), loaders = %s" % (fullname, map(lambda l: l.dir_path, self.loaders))
mod = None
message = None
path = []
vfs_shared_path = []
if self.reload:
mod = sys.modules[fullname]
path = mod.__path__ or []
if path == fullname:
# Work around Python bug setting __path__ of frozen modules.
path = []
vfs_shared_path = getattr(mod, '_vfs_shared_path', [])
for loader in self.loaders:
try:
mod = loader.load_module(fullname, loadingShared = True)
except ImportError:
etype, evalue, etraceback = sys.exc_info()
print("%s on %s: %s" % (etype.__name__, fullname, evalue))
if not message:
message = '%s: %s' % (fullname, evalue)
continue
for dir in getattr(mod, '__path__', []):
if dir not in path:
path.append(dir)
if mod is None:
# If all of them failed to load, raise ImportError.
raise ImportError(message)
# If at least one of them loaded successfully, return the
# union of loaded modules.
mod.__path__ = path
mod.__package__ = fullname
# Also set this special symbol, which records that this is a
# shared package, and also lists the paths we have already
# loaded.
mod._vfs_shared_path = vfs_shared_path + [l.dir_path for l in self.loaders]
return mod
_registered = False
[docs]def register():
""" Register the VFSImporter on the path_hooks, if it has not
already been registered, so that future Python import statements
will vector through here (and therefore will take advantage of
Panda's virtual file system). """
global _registered
if not _registered:
_registered = True
sys.path_hooks.insert(0, VFSImporter)
sys.meta_path.insert(0, VFSSharedImporter())
# Blow away the importer cache, so we'll come back through the
# VFSImporter for every folder in the future, even those
# folders that previously were loaded directly.
sys.path_importer_cache = {}
[docs]def reloadSharedPackage(mod):
""" Reloads the specific module as a shared package, adding any
new directories that might have appeared on the search path. """
fullname = mod.__name__
path = None
if '.' in fullname:
parentname = fullname.rsplit('.', 1)[0]
parent = sys.modules[parentname]
path = parent.__path__
importer = VFSSharedImporter()
loader = importer.find_module(fullname, path = path, reload = True)
if loader:
loader.load_module(fullname)
# Also force any child packages to become shared packages, if
# they aren't already.
for basename, child in list(mod.__dict__.items()):
if isinstance(child, types.ModuleType):
childname = child.__name__
if childname == fullname + '.' + basename and \
hasattr(child, '__path__') and \
childname not in sharedPackages:
sharedPackages[childname] = True
reloadSharedPackage(child)
[docs]def reloadSharedPackages():
""" Walks through the sharedPackages list, and forces a reload of
any modules on that list that have already been loaded. This
allows new directories to be added to the search path. """
#print >> sys.stderr, "reloadSharedPackages, path = %s, sharedPackages = %s" % (sys.path, sharedPackages.keys())
# Sort the list, just to make sure parent packages are reloaded
# before child packages are.
for fullname in sorted(sharedPackages.keys()):
mod = sys.modules.get(fullname, None)
if not mod:
continue
reloadSharedPackage(mod)
