| # |
| # DEPRECATED: implementation for ffi.verify() |
| # |
| import sys, os, binascii, shutil, io |
| from . import __version_verifier_modules__ |
| from . import ffiplatform |
| from .error import VerificationError |
| |
| if sys.version_info >= (3, 3): |
| import importlib.machinery |
| def _extension_suffixes(): |
| return importlib.machinery.EXTENSION_SUFFIXES[:] |
| else: |
| import imp |
| def _extension_suffixes(): |
| return [suffix for suffix, _, type in imp.get_suffixes() |
| if type == imp.C_EXTENSION] |
| |
| |
| if sys.version_info >= (3,): |
| NativeIO = io.StringIO |
| else: |
| class NativeIO(io.BytesIO): |
| def write(self, s): |
| if isinstance(s, unicode): |
| s = s.encode('ascii') |
| super(NativeIO, self).write(s) |
| |
| |
| class Verifier(object): |
| |
| def __init__(self, ffi, preamble, tmpdir=None, modulename=None, |
| ext_package=None, tag='', force_generic_engine=False, |
| source_extension='.c', flags=None, relative_to=None, **kwds): |
| if ffi._parser._uses_new_feature: |
| raise VerificationError( |
| "feature not supported with ffi.verify(), but only " |
| "with ffi.set_source(): %s" % (ffi._parser._uses_new_feature,)) |
| self.ffi = ffi |
| self.preamble = preamble |
| if not modulename: |
| flattened_kwds = ffiplatform.flatten(kwds) |
| vengine_class = _locate_engine_class(ffi, force_generic_engine) |
| self._vengine = vengine_class(self) |
| self._vengine.patch_extension_kwds(kwds) |
| self.flags = flags |
| self.kwds = self.make_relative_to(kwds, relative_to) |
| # |
| if modulename: |
| if tag: |
| raise TypeError("can't specify both 'modulename' and 'tag'") |
| else: |
| key = '\x00'.join(['%d.%d' % sys.version_info[:2], |
| __version_verifier_modules__, |
| preamble, flattened_kwds] + |
| ffi._cdefsources) |
| if sys.version_info >= (3,): |
| key = key.encode('utf-8') |
| k1 = hex(binascii.crc32(key[0::2]) & 0xffffffff) |
| k1 = k1.lstrip('0x').rstrip('L') |
| k2 = hex(binascii.crc32(key[1::2]) & 0xffffffff) |
| k2 = k2.lstrip('0').rstrip('L') |
| modulename = '_cffi_%s_%s%s%s' % (tag, self._vengine._class_key, |
| k1, k2) |
| suffix = _get_so_suffixes()[0] |
| self.tmpdir = tmpdir or _caller_dir_pycache() |
| self.sourcefilename = os.path.join(self.tmpdir, modulename + source_extension) |
| self.modulefilename = os.path.join(self.tmpdir, modulename + suffix) |
| self.ext_package = ext_package |
| self._has_source = False |
| self._has_module = False |
| |
| def write_source(self, file=None): |
| """Write the C source code. It is produced in 'self.sourcefilename', |
| which can be tweaked beforehand.""" |
| with self.ffi._lock: |
| if self._has_source and file is None: |
| raise VerificationError( |
| "source code already written") |
| self._write_source(file) |
| |
| def compile_module(self): |
| """Write the C source code (if not done already) and compile it. |
| This produces a dynamic link library in 'self.modulefilename'.""" |
| with self.ffi._lock: |
| if self._has_module: |
| raise VerificationError("module already compiled") |
| if not self._has_source: |
| self._write_source() |
| self._compile_module() |
| |
| def load_library(self): |
| """Get a C module from this Verifier instance. |
| Returns an instance of a FFILibrary class that behaves like the |
| objects returned by ffi.dlopen(), but that delegates all |
| operations to the C module. If necessary, the C code is written |
| and compiled first. |
| """ |
| with self.ffi._lock: |
| if not self._has_module: |
| self._locate_module() |
| if not self._has_module: |
| if not self._has_source: |
| self._write_source() |
| self._compile_module() |
| return self._load_library() |
| |
| def get_module_name(self): |
| basename = os.path.basename(self.modulefilename) |
| # kill both the .so extension and the other .'s, as introduced |
| # by Python 3: 'basename.cpython-33m.so' |
| basename = basename.split('.', 1)[0] |
| # and the _d added in Python 2 debug builds --- but try to be |
| # conservative and not kill a legitimate _d |
| if basename.endswith('_d') and hasattr(sys, 'gettotalrefcount'): |
| basename = basename[:-2] |
| return basename |
| |
| def get_extension(self): |
| ffiplatform._hack_at_distutils() # backward compatibility hack |
| if not self._has_source: |
| with self.ffi._lock: |
| if not self._has_source: |
| self._write_source() |
| sourcename = ffiplatform.maybe_relative_path(self.sourcefilename) |
| modname = self.get_module_name() |
| return ffiplatform.get_extension(sourcename, modname, **self.kwds) |
| |
| def generates_python_module(self): |
| return self._vengine._gen_python_module |
| |
| def make_relative_to(self, kwds, relative_to): |
| if relative_to and os.path.dirname(relative_to): |
| dirname = os.path.dirname(relative_to) |
| kwds = kwds.copy() |
| for key in ffiplatform.LIST_OF_FILE_NAMES: |
| if key in kwds: |
| lst = kwds[key] |
| if not isinstance(lst, (list, tuple)): |
| raise TypeError("keyword '%s' should be a list or tuple" |
| % (key,)) |
| lst = [os.path.join(dirname, fn) for fn in lst] |
| kwds[key] = lst |
| return kwds |
| |
| # ---------- |
| |
| def _locate_module(self): |
| if not os.path.isfile(self.modulefilename): |
| if self.ext_package: |
| try: |
| pkg = __import__(self.ext_package, None, None, ['__doc__']) |
| except ImportError: |
| return # cannot import the package itself, give up |
| # (e.g. it might be called differently before installation) |
| path = pkg.__path__ |
| else: |
| path = None |
| filename = self._vengine.find_module(self.get_module_name(), path, |
| _get_so_suffixes()) |
| if filename is None: |
| return |
| self.modulefilename = filename |
| self._vengine.collect_types() |
| self._has_module = True |
| |
| def _write_source_to(self, file): |
| self._vengine._f = file |
| try: |
| self._vengine.write_source_to_f() |
| finally: |
| del self._vengine._f |
| |
| def _write_source(self, file=None): |
| if file is not None: |
| self._write_source_to(file) |
| else: |
| # Write our source file to an in memory file. |
| f = NativeIO() |
| self._write_source_to(f) |
| source_data = f.getvalue() |
| |
| # Determine if this matches the current file |
| if os.path.exists(self.sourcefilename): |
| with open(self.sourcefilename, "r") as fp: |
| needs_written = not (fp.read() == source_data) |
| else: |
| needs_written = True |
| |
| # Actually write the file out if it doesn't match |
| if needs_written: |
| _ensure_dir(self.sourcefilename) |
| with open(self.sourcefilename, "w") as fp: |
| fp.write(source_data) |
| |
| # Set this flag |
| self._has_source = True |
| |
| def _compile_module(self): |
| # compile this C source |
| tmpdir = os.path.dirname(self.sourcefilename) |
| outputfilename = ffiplatform.compile(tmpdir, self.get_extension()) |
| try: |
| same = ffiplatform.samefile(outputfilename, self.modulefilename) |
| except OSError: |
| same = False |
| if not same: |
| _ensure_dir(self.modulefilename) |
| shutil.move(outputfilename, self.modulefilename) |
| self._has_module = True |
| |
| def _load_library(self): |
| assert self._has_module |
| if self.flags is not None: |
| return self._vengine.load_library(self.flags) |
| else: |
| return self._vengine.load_library() |
| |
| # ____________________________________________________________ |
| |
| _FORCE_GENERIC_ENGINE = False # for tests |
| |
| def _locate_engine_class(ffi, force_generic_engine): |
| if _FORCE_GENERIC_ENGINE: |
| force_generic_engine = True |
| if not force_generic_engine: |
| if '__pypy__' in sys.builtin_module_names: |
| force_generic_engine = True |
| else: |
| try: |
| import _cffi_backend |
| except ImportError: |
| _cffi_backend = '?' |
| if ffi._backend is not _cffi_backend: |
| force_generic_engine = True |
| if force_generic_engine: |
| from . import vengine_gen |
| return vengine_gen.VGenericEngine |
| else: |
| from . import vengine_cpy |
| return vengine_cpy.VCPythonEngine |
| |
| # ____________________________________________________________ |
| |
| _TMPDIR = None |
| |
| def _caller_dir_pycache(): |
| if _TMPDIR: |
| return _TMPDIR |
| result = os.environ.get('CFFI_TMPDIR') |
| if result: |
| return result |
| filename = sys._getframe(2).f_code.co_filename |
| return os.path.abspath(os.path.join(os.path.dirname(filename), |
| '__pycache__')) |
| |
| def set_tmpdir(dirname): |
| """Set the temporary directory to use instead of __pycache__.""" |
| global _TMPDIR |
| _TMPDIR = dirname |
| |
| def cleanup_tmpdir(tmpdir=None, keep_so=False): |
| """Clean up the temporary directory by removing all files in it |
| called `_cffi_*.{c,so}` as well as the `build` subdirectory.""" |
| tmpdir = tmpdir or _caller_dir_pycache() |
| try: |
| filelist = os.listdir(tmpdir) |
| except OSError: |
| return |
| if keep_so: |
| suffix = '.c' # only remove .c files |
| else: |
| suffix = _get_so_suffixes()[0].lower() |
| for fn in filelist: |
| if fn.lower().startswith('_cffi_') and ( |
| fn.lower().endswith(suffix) or fn.lower().endswith('.c')): |
| try: |
| os.unlink(os.path.join(tmpdir, fn)) |
| except OSError: |
| pass |
| clean_dir = [os.path.join(tmpdir, 'build')] |
| for dir in clean_dir: |
| try: |
| for fn in os.listdir(dir): |
| fn = os.path.join(dir, fn) |
| if os.path.isdir(fn): |
| clean_dir.append(fn) |
| else: |
| os.unlink(fn) |
| except OSError: |
| pass |
| |
| def _get_so_suffixes(): |
| suffixes = _extension_suffixes() |
| if not suffixes: |
| # bah, no C_EXTENSION available. Occurs on pypy without cpyext |
| if sys.platform == 'win32': |
| suffixes = [".pyd"] |
| else: |
| suffixes = [".so"] |
| |
| return suffixes |
| |
| def _ensure_dir(filename): |
| dirname = os.path.dirname(filename) |
| if dirname and not os.path.isdir(dirname): |
| os.makedirs(dirname) |
