def process_train_task(task):
(dataset_path, models_dir, config) = task
config_str = serialize_config(config)
model_path = os.path.join(models_dir, 'model.{}.pkz'.format(config_str))
env = os.environ.copy()
env['TREECAT_PROFILE'] = '1'
env['TREECAT_THREADS'] = '1'
check_call_env([
sys.executable,
'-O',
'-m',
'treecat.validate',
'train_task',
dataset_path,
model_path,
config_str,
], env)
python类executable()的实例源码
def get_command_line():
'''
Returns prefix of command line used for spawning a child process
'''
if process.current_process()._identity==() and is_forking(sys.argv):
raise RuntimeError('''
Attempt to start a new process before the current process
has finished its bootstrapping phase.
This probably means that you are on Windows and you have
forgotten to use the proper idiom in the main module:
if __name__ == '__main__':
freeze_support()
...
The "freeze_support()" line can be omitted if the program
is not going to be frozen to produce a Windows executable.''')
if getattr(sys, 'frozen', False):
return [sys.executable, '--multiprocessing-fork']
else:
prog = 'from multiprocessing.forking import main; main()'
return [_python_exe, '-c', prog, '--multiprocessing-fork']
def _findLib_ldconfig(name):
# XXX assuming GLIBC's ldconfig (with option -p)
expr = r'/[^\(\)\s]*lib%s\.[^\(\)\s]*' % re.escape(name)
f = os.popen('LC_ALL=C LANG=C /sbin/ldconfig -p 2>/dev/null')
try:
data = f.read()
finally:
f.close()
res = re.search(expr, data)
if not res:
# Hm, this works only for libs needed by the python executable.
cmd = 'ldd %s 2>/dev/null' % sys.executable
f = os.popen(cmd)
try:
data = f.read()
finally:
f.close()
res = re.search(expr, data)
if not res:
return None
return res.group(0)
def filemetadata(filename):
# type: (str) -> Optional[FileMeta]
p_filename = which(filename)
if p_filename is None:
return None
filename = p_filename
s = os.stat(filename)
if filename != sys.executable:
result = run_executable(filename, ['--version'])
versionstring = result.stdout
else:
# filename is the Python interpreter itself
versionstring = bytestr(sys.version)
return FileMeta(filename, s.st_size, s.st_mtime, filesha(filename), versionstring)
# ----------------------------------------------------------------------
def run_setup_commands(project, logger, commands) :
reports_dir = project.expand_path("$dir_reports/distutils")
if not os.path.exists(reports_dir) :
os.mkdir(reports_dir)
setup_script = project.expand_path("$dir_dist/setup.py")
for command in commands :
logger.debug("Executing distutils command %s", command)
output_file_path = os.path.join(reports_dir, command.replace("/", ""))
with open(output_file_path, "w") as output_file :
commandexec = [sys.executable, setup_script]
commandexec.extend(command.split())
working_dir = project.expand_path("$dir_dist")
process = subprocess.Popen(commandexec, cwd=working_dir, stdout=output_file, stderr=output_file, shell=False)
return_code = process.wait()
if return_code != 0 :
raise BuildFailedException("Error while executing setup command %s, see %s for details" % (command, output_file_path))
def make_virtualenv(args):
with cleanup_dir(tempfile.mkdtemp()) as tempdir:
venv, python, pip = dirs(tempdir)
print_call(
sys.executable, '-m', 'virtualenv', venv,
'-p', args.python, '--never-download',
)
def pip_install(*argv):
print_call(pip, 'install', '-i', args.index_url, *argv)
# Latest pip installs python3.5 wheels
pip_install('pip', 'setuptools', '--upgrade')
pip_install('-r', 'requirements-minimal.txt')
pip_install('-r', 'requirements-dev-minimal.txt')
reexec(
python, __file__.rstrip('c'),
'--tempdir', tempdir,
# Pass along existing args
'--index-url', args.index_url,
'--exec-count', str(args.exec_count),
'--exec-limit', str(args.exec_limit),
reason='to use the virtualenv python',
)
def test_exit_crash():
# For each Widget subclass, run a simple python script that creates an
# instance and then shuts down. The intent is to check for segmentation
# faults when each script exits.
tmp = tempfile.mktemp(".py")
path = os.path.dirname(pg.__file__)
initArgs = {
'CheckTable': "[]",
'ProgressDialog': '"msg"',
'VerticalLabel': '"msg"',
}
for name in dir(pg):
obj = getattr(pg, name)
if not isinstance(obj, type) or not issubclass(obj, pg.QtGui.QWidget):
continue
print(name)
argstr = initArgs.get(name, "")
open(tmp, 'w').write(code.format(path=path, classname=name, args=argstr))
proc = subprocess.Popen([sys.executable, tmp])
assert proc.wait() == 0
os.remove(tmp)
def test_exit_crash():
# For each Widget subclass, run a simple python script that creates an
# instance and then shuts down. The intent is to check for segmentation
# faults when each script exits.
tmp = tempfile.mktemp(".py")
path = os.path.dirname(pg.__file__)
initArgs = {
'CheckTable': "[]",
'ProgressDialog': '"msg"',
'VerticalLabel': '"msg"',
}
for name in dir(pg):
obj = getattr(pg, name)
if not isinstance(obj, type) or not issubclass(obj, pg.QtGui.QWidget):
continue
print(name)
argstr = initArgs.get(name, "")
open(tmp, 'w').write(code.format(path=path, classname=name, args=argstr))
proc = subprocess.Popen([sys.executable, tmp])
assert proc.wait() == 0
os.remove(tmp)
def test_colors(tmpdir):
"""Test colors.
:param tmpdir: pytest fixture.
"""
script = dedent("""\
import logging
from sphinxcontrib.versioning.setup_logging import setup_logging
setup_logging(verbose=False, colors=True)
logger = logging.getLogger("{logger}")
logger.critical("Critical")
logger.error("Error")
logger.warning("Warning")
logger.info("Info")
logger.debug("Debug") # Not printed since verbose = False.
""").format(logger=ColorFormatter.SPECIAL_SCOPE + '.sample')
tmpdir.join('script.py').write(script)
output = pytest.run(tmpdir, [sys.executable, 'script.py'])
assert '\033[31m=> Critical\033[39m\n' in output
assert '\033[31m=> Error\033[39m\n' in output
assert '\033[33m=> Warning\033[39m\n' in output
assert '\033[36m=> Info\033[39m' in output
assert 'Debug' not in output
def fix_script(path):
"""Replace #!python with #!/path/to/python
Return True if file was changed."""
# XXX RECORD hashes will need to be updated
if os.path.isfile(path):
script = open(path, 'rb')
try:
firstline = script.readline()
if not firstline.startswith(binary('#!python')):
return False
exename = sys.executable.encode(sys.getfilesystemencoding())
firstline = binary('#!') + exename + binary(os.linesep)
rest = script.read()
finally:
script.close()
script = open(path, 'wb')
try:
script.write(firstline)
script.write(rest)
finally:
script.close()
return True
def _build_one(self, req):
"""Build one wheel."""
base_args = [
sys.executable, '-c',
"import setuptools;__file__=%r;"\
"exec(compile(open(__file__).read().replace('\\r\\n', '\\n'), __file__, 'exec'))" % req.setup_py] + \
list(self.global_options)
logger.notify('Running setup.py bdist_wheel for %s' % req.name)
logger.notify('Destination directory: %s' % self.wheel_dir)
wheel_args = base_args + ['bdist_wheel', '-d', self.wheel_dir] + self.build_options
try:
call_subprocess(wheel_args, cwd=req.source_dir, show_stdout=False)
return True
except:
logger.error('Failed building wheel for %s' % req.name)
return False
def restart_with_reloader(self):
"""Spawn a new Python interpreter with the same arguments as this one,
but running the reloader thread.
"""
while 1:
_log('info', ' * Restarting with %s' % self.name)
args = [sys.executable] + sys.argv
new_environ = os.environ.copy()
new_environ['WERKZEUG_RUN_MAIN'] = 'true'
# a weird bug on windows. sometimes unicode strings end up in the
# environment and subprocess.call does not like this, encode them
# to latin1 and continue.
if os.name == 'nt' and PY2:
for key, value in iteritems(new_environ):
if isinstance(value, text_type):
new_environ[key] = value.encode('iso-8859-1')
exit_code = subprocess.call(args, env=new_environ,
close_fds=False)
if exit_code != 3:
return exit_code
def fix_jython_executable(executable, options):
if sys.platform.startswith('java') and is_sh(executable):
# Workaround for Jython is not needed on Linux systems.
import java
if java.lang.System.getProperty("os.name") == "Linux":
return executable
# Workaround Jython's sys.executable being a .sh (an invalid
# shebang line interpreter)
if options:
# Can't apply the workaround, leave it broken
log.warn(
"WARNING: Unable to adapt shebang line for Jython,"
" the following script is NOT executable\n"
" see http://bugs.jython.org/issue1112 for"
" more information.")
else:
return '/usr/bin/env %s' % executable
return executable
def _install_build_reqs(self, reqs, prefix):
# Local import to avoid circular import (wheel <-> req_install)
from pip._internal.req.req_install import InstallRequirement
from pip._internal.index import FormatControl
# Ignore the --no-binary option when installing the build system, so
# we don't recurse trying to build a self-hosting build system.
finder = copy.copy(self.finder)
finder.format_control = FormatControl(set(), set())
urls = [finder.find_requirement(InstallRequirement.from_line(r),
upgrade=False).url
for r in reqs]
args = [sys.executable, '-m', 'pip', 'install', '--ignore-installed',
'--prefix', prefix] + list(urls)
with open_spinner("Installing build dependencies") as spinner:
call_subprocess(args, show_stdout=False, spinner=spinner)
def get_install_args(self, global_options, record_filename, root, prefix):
install_args = [sys.executable, "-u"]
install_args.append('-c')
install_args.append(SETUPTOOLS_SHIM % self.setup_py)
install_args += list(global_options) + \
['install', '--record', record_filename]
install_args += ['--single-version-externally-managed']
if root is not None:
install_args += ['--root', root]
if prefix is not None:
install_args += ['--prefix', prefix]
if self.pycompile:
install_args += ["--compile"]
else:
install_args += ["--no-compile"]
if running_under_virtualenv():
py_ver_str = 'python' + sysconfig.get_python_version()
install_args += ['--install-headers',
os.path.join(sys.prefix, 'include', 'site',
py_ver_str, self.name)]
return install_args
def pyrun(src):
"""Run python code 'src' in a separate interpreter.
Return subprocess exit code.
"""
if PY3:
src = bytes(src, 'ascii')
with tempfile.NamedTemporaryFile(suffix='.py', delete=False) as f:
f.write(src)
f.flush()
test_files.append(f.name)
code = subprocess.call(
[sys.executable, f.name],
stdout=None, stderr=None,
# creationflags=subprocess.CREATE_NEW_PROCESS_GROUP
)
return code
def run_as_admin(extra_args=None):
"""If required, rerun the script and request admin privileges.
:params iterable extra_args:
Additional arguments to pass to the script in case it has to be
restarted.
"""
if not am_admin():
try:
print('Restarting and requesting admin privileges.')
args = sys.argv
if extra_args:
args = args + extra_args
exe, args = sys.executable, ' '.join(args)
if sys.version_info[0] == 2:
exe = _decode(exe)
ctypes.windll.shell32.ShellExecuteW(None, 'Runas', exe, args, None, 1)
sys.exit()
except Exception as e:
print(e)
msg = ('Unable to elevate privileges. You need to rerun the script'
' with Administrator privileges yourself. E.g. try pressing'
' the Windows key + x, then select "Command Prompt (Admin)"'
' and run the script in that console.')
sys.exit(msg)
def testOutputStreams(self):
output_spec = {
'mode': 'http',
'method': 'PUT',
'url': 'http://localhost:%d' % _socket_port
}
fd = os.open(_pipepath, os.O_RDONLY | os.O_NONBLOCK)
adapters = {
fd: make_stream_push_adapter(output_spec)
}
cmd = [sys.executable, _oscript, _pipepath]
try:
with captureOutput() as stdpipes:
run_process(cmd, adapters)
except Exception:
print('Stdout/stderr from exception: ')
print(stdpipes)
raise
self.assertEqual(stdpipes, ['start\ndone\n', ''])
self.assertEqual(len(_req_chunks), 1)
self.assertEqual(_req_chunks[0], (9, 'a message'))
def testInputStreams(self):
input_spec = {
'mode': 'http',
'method': 'GET',
'url': 'http://mockedhost'
}
@httmock.urlmatch(netloc='^mockedhost$', method='GET')
def mock_fetch(url, request):
return 'hello\nworld'
adapters = {
_pipepath: make_stream_fetch_adapter(input_spec)
}
cmd = [sys.executable, _iscript, _pipepath]
try:
with captureOutput() as stdpipes, httmock.HTTMock(mock_fetch):
run_process(cmd, input_pipes=adapters)
except Exception:
print('Stdout/stderr from exception: ')
print(stdpipes)
raise
self.assertEqual(stdpipes, ['olleh\ndlrow\n', ''])
def configure(conf):
"""
Detect the python interpreter
"""
v = conf.env
v['PYTHON'] = Options.options.python or os.environ.get('PYTHON', sys.executable)
if Options.options.pythondir:
v['PYTHONDIR'] = Options.options.pythondir
if Options.options.pythonarchdir:
v['PYTHONARCHDIR'] = Options.options.pythonarchdir
conf.find_program('python', var='PYTHON')
v['PYFLAGS'] = ''
v['PYFLAGS_OPT'] = '-O'
v['PYC'] = getattr(Options.options, 'pyc', 1)
v['PYO'] = getattr(Options.options, 'pyo', 1)
try:
v.PYTAG = conf.cmd_and_log(conf.env.PYTHON + ['-c', "import imp;print(imp.get_tag())"]).strip()
except Errors.WafError:
pass
def configure(conf):
"""
Detect the python interpreter
"""
v = conf.env
v['PYTHON'] = Options.options.python or os.environ.get('PYTHON', sys.executable)
if Options.options.pythondir:
v['PYTHONDIR'] = Options.options.pythondir
if Options.options.pythonarchdir:
v['PYTHONARCHDIR'] = Options.options.pythonarchdir
conf.find_program('python', var='PYTHON')
v['PYFLAGS'] = ''
v['PYFLAGS_OPT'] = '-O'
v['PYC'] = getattr(Options.options, 'pyc', 1)
v['PYO'] = getattr(Options.options, 'pyo', 1)
try:
v.PYTAG = conf.cmd_and_log(conf.env.PYTHON + ['-c', "import imp;print(imp.get_tag())"]).strip()
except Errors.WafError:
pass
def configure(conf):
"""
Detect the python interpreter
"""
v = conf.env
v['PYTHON'] = Options.options.python or os.environ.get('PYTHON', sys.executable)
if Options.options.pythondir:
v['PYTHONDIR'] = Options.options.pythondir
if Options.options.pythonarchdir:
v['PYTHONARCHDIR'] = Options.options.pythonarchdir
conf.find_program('python', var='PYTHON')
v['PYFLAGS'] = ''
v['PYFLAGS_OPT'] = '-O'
v['PYC'] = getattr(Options.options, 'pyc', 1)
v['PYO'] = getattr(Options.options, 'pyo', 1)
try:
v.PYTAG = conf.cmd_and_log(conf.env.PYTHON + ['-c', "import imp;print(imp.get_tag())"]).strip()
except Errors.WafError:
pass
def __init__(self, qt_toolkit='pyside'):
"""
:param qt_toolkit: Toolkit used to generate files: ``'pyside'`` or ``'pyqt'``
"""
if qt_toolkit.lower() == 'pyside':
if sys.platform.startswith('win'):
PYPATH = os.path.dirname(sys.executable)
PYSIDEPATH = os.path.join(get_python_lib(), 'PySide')
self.PYUIC = os.path.join(PYPATH, "Scripts/pyside-uic")
self.PYRCC = os.path.join(PYSIDEPATH, "pyside-rcc")
else:
self.PYUIC = "pyside-uic"
self.PYRCC = "pyside-rcc"
else: # pyqt
if sys.platform.startswith('win'):
PYQTPATH = os.path.join(get_python_lib(), 'PyQt4')
self.PYUIC = os.path.join(PYQTPATH, "pyuic4")
self.PYRCC = os.path.join(PYQTPATH, "pyrcc4.exe")
else:
self.PYUIC = "pyuic4"
self.PYRCC = "pyrcc4"
def restart_with_reloader(self):
"""Spawn a new Python interpreter with the same arguments as this one,
but running the reloader thread.
"""
while 1:
_log('info', ' * Restarting with %s' % self.name)
args = [sys.executable] + sys.argv
new_environ = os.environ.copy()
new_environ['WERKZEUG_RUN_MAIN'] = 'true'
# a weird bug on windows. sometimes unicode strings end up in the
# environment and subprocess.call does not like this, encode them
# to latin1 and continue.
if os.name == 'nt' and PY2:
for key, value in iteritems(new_environ):
if isinstance(value, text_type):
new_environ[key] = value.encode('iso-8859-1')
exit_code = subprocess.call(args, env=new_environ,
close_fds=False)
if exit_code != 3:
return exit_code
def _python_cmd(*args):
"""
Return True if the command succeeded.
"""
args = (sys.executable,) + args
return subprocess.call(args) == 0
def check(request):
return {
'version': '{major}.{minor}.{micro}'.format(
major=sys.version_info.major,
minor=sys.version_info.minor,
micro=sys.version_info.micro
),
'info': sys.version,
'executable': sys.executable,
'platform': sys.platform
}
def _init_non_posix(vars):
"""Initialize the module as appropriate for NT"""
# set basic install directories
vars['LIBDEST'] = get_path('stdlib')
vars['BINLIBDEST'] = get_path('platstdlib')
vars['INCLUDEPY'] = get_path('include')
vars['SO'] = '.pyd'
vars['EXE'] = '.exe'
vars['VERSION'] = _PY_VERSION_SHORT_NO_DOT
vars['BINDIR'] = os.path.dirname(_safe_realpath(sys.executable))
#
# public APIs
#
def get_executable():
# The __PYVENV_LAUNCHER__ dance is apparently no longer needed, as
# changes to the stub launcher mean that sys.executable always points
# to the stub on macOS
# if sys.platform == 'darwin' and ('__PYVENV_LAUNCHER__'
# in os.environ):
# result = os.environ['__PYVENV_LAUNCHER__']
# else:
# result = sys.executable
# return result
result = os.path.normcase(sys.executable)
if not isinstance(result, text_type):
result = fsdecode(result)
return result
def _copy_dist_from_dir(link_path, location):
"""Copy distribution files in `link_path` to `location`.
Invoked when user requests to install a local directory. E.g.:
pip install .
pip install ~/dev/git-repos/python-prompt-toolkit
"""
# Note: This is currently VERY SLOW if you have a lot of data in the
# directory, because it copies everything with `shutil.copytree`.
# What it should really do is build an sdist and install that.
# See https://github.com/pypa/pip/issues/2195
if os.path.isdir(location):
rmtree(location)
# build an sdist
setup_py = 'setup.py'
sdist_args = [sys.executable]
sdist_args.append('-c')
sdist_args.append(SETUPTOOLS_SHIM % setup_py)
sdist_args.append('sdist')
sdist_args += ['--dist-dir', location]
logger.info('Running setup.py sdist for %s', link_path)
with indent_log():
call_subprocess(sdist_args, cwd=link_path, show_stdout=False)
# unpack sdist into `location`
sdist = os.path.join(location, os.listdir(location)[0])
logger.info('Unpacking sdist %s into %s', sdist, location)
unpack_file(sdist, location, content_type=None, link=None)
def get_prog():
try:
if os.path.basename(sys.argv[0]) in ('__main__.py', '-c'):
return "%s -m pip" % sys.executable
except (AttributeError, TypeError, IndexError):
pass
return 'pip'
# Retry every half second for up to 3 seconds
