#
##
## This file is part of pyFormex 2.0 (Mon Sep 14 12:29:05 CEST 2020)
## pyFormex is a tool for generating, manipulating and transforming 3D
## geometrical models by sequences of mathematical operations.
## Home page: http://pyformex.org
## Project page: http://savannah.nongnu.org/projects/pyformex/
## Copyright 2004-2020 (C) Benedict Verhegghe (benedict.verhegghe@ugent.be)
## Distributed under the GNU General Public License version 3 or later.
##
## This program is free software: you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation, either version 3 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program. If not, see http://www.gnu.org/licenses/.
##
"""Executing external commands.
This module provides some functions for executing external commands in
a subprocess. They are based on the standard Python :mod:`subprocess`
module, but provide some practical enhancements.
Contents:
- :class:`DoneProcess` is a class to hold information about a terminated
process.
- :func:`run` runs a command in a subprocess and waits for it to finish.
Returns a :class:`DoneProcess` about the outcome.
- :func:`start` starts a subprocess and does not wait for it.
Returns a :class:`subprocess.Popen` instance that can be used to communicate
with the process.
This module can be used independently from pyFormex.
"""
import subprocess
import shlex
import os
[docs]class DoneProcess(subprocess.CompletedProcess):
"""A class to return the outcome of a subprocess.
This is a subclass of :class:`subprocess.CompletedProcess` with some
extra attributes.
Attributes
----------
args: str or sequence of arguments
The args that were used to create the subprocess.
returncode: int
The exit code of the subprocess. Typical a 0 means it ran
succesfully. If the subprocess fails to start, the value
is 127. Other non-zero values can be returned by the child
process to flag some error condition.
stdout: str or bytes or None
The standard output captured from the child process.
stderr: str or bytes or None
The error output captured from the child process.
failed: bool
True if the child process failed to start, e.g. due to a
non-existing or non-loadable executable. The returncode
will be 127 in this case.
timedout: bool
True if the child process exited due to a timeout condition.
"""
def __init__(self, args, returncode, failed=False, timedout=False,
**kargs):
super().__init__(args, returncode, **kargs)
self.failed = failed
self.timedout = timedout
def __repr__(self):
"""Textual representation of the DoneProcess
This only shows the non-default values.
"""
s = super().__repr__()
i = s.rfind(')')
s = [ s[:i], s[i:] ]
if self.failed:
s.insert(-1, f", failed={self.failed!r}")
if self.timedout:
s.insert(-1, f", timedout={self.timedout!r}")
return ''.join(s)
def __str__(self):
"""User-friendly full tectual representation
Returns
-------
str
An extensive report about the last run command, including
output and error messages.
Notes
-----
This is mostly useful in interactive work, to find out
why a command failed.
"""
s = f"DoneProcess report\nargs: {self.args}\n"
if self.failed:
s += f"Command failed to run!\n"
s += f"returncode: {self.returncode}\n"
for atr in ['stdout', 'stderr']:
val = getattr(self, atr)
if val is not None:
s += f"{atr}:\n{val}"
if s[-1] != '\n':
s += '\n'
if self.timedout:
s += f"timedout: {self.timedout}\n"
return s
[docs]def run(args, *, capture_output=True, encoding='utf-8', **kargs):
"""Execute a command through the operating system.
Run an external command and wait for its termination.
This uses Python3's :func:`subprocess.run` with the following
enhancements:
- If a string is passed as `args` and `shell` is False, the string is
automatically tokenized into an args list.
- The output of the command is captured by default.
- The `encoding` is set to 'utf-8' by default, so that stdout
and stderr return as strings.
- stdin, stdout and stderr can be file names. They will be replaced
with the opened files (in mode 'r' for stdin, 'w' for the others).
- If `shell` is True and no executable is specified, it defaults
to the value of the 'SHELL' environment variable, or '/bin/sh/' if
that doesn't exist.
- All exceptions are captured by default and can be detected from
the return value.
Only the new and changed parameters are described here. Except for
the first (`args`), all parameters should be specified by keyword.
Parameters
----------
args: string or sequence of program arguments.
If a string is provided and `shell` is False (default), the string is
split into a sequence of program arguments using :func:`shlex.split`.
capture_output: bool
If True (default), both stdout and stderr are captured and available
from the returned :class:`DoneProcess`. Specifying any of `stdout` or
`stderr` values will however override the capture_output setting.
encoding: string or None
The captured stdout and stderr are returned as strings by default.
Specifying None will return them as bytes.
shell: bool
Default is False. If True, the command will be run in a new shell.
This uses more resources and may pose a security risk.
Unless you need some shell functionality (parameter expansion,
compound commands, pipes), the advised way to
execute a command is to use the default False.
executable: string
The full path name of the program to be executed.
This can be used to specify the real executable if the
program specified in `args` is not in your PATH, or,
when `shell` is True, if you want to use shell other than
the default: the value of the 'SHELL' environment variable,
or '/bin/sh/'.
timeout: int
If provided, this is the maximum number of seconds the process is
allowed to run. If a timeout condition occurs, the
subprocess.TimeoutExpired exception is caught and an attribute
timedout=True is set in the return value.
check: bool
Defauls is False. If True, an exception will be raised when
the command did not terminate with a 0 returncode or another
error condition occurred.
**kargs: keyword arguments
Any keyword arguments accepted by :func:`subprocess.run` (and
thus also :class:`subprocess.Popen`) can be specified.
See the Python documentation for :func:`subprocess.run` for full info.
Returns
-------
:class:`DoneProcess`
An object collecting all relevant info about the
finished subprocess. See :class:`DoneProcess`.
See Also
--------
start: start a subprocess but do not wait for its outcome
Examples
--------
>>> P = run("pwd")
>>> P.stdout.strip('\\n') == os.getcwd()
True
>>> P = run("echo 'This is stderr' > /dev/stderr", shell=True)
>>> P.stderr
'This is stderr\\n'
>>> P = run('false')
>>> P
DoneProcess(args=['false'], returncode=1, stdout='', stderr='')
>>> P = run('false', capture_output=False)
>>> P
DoneProcess(args=['false'], returncode=1)
>>> P.stdout is None
True
>>> P = run('False')
>>> P
DoneProcess(args=['False'], returncode=127, failed=True)
>>> P = run("sleep 5", timeout=1)
>>> P
DoneProcess(args=['sleep', '5'], returncode=-1, timedout=True)
>>> print(P)
DoneProcess report
args: ['sleep', '5']
returncode: -1
timedout: True
<BLANKLINE>
"""
shell = bool(kargs.get('shell', False))
if isinstance(args, str) and shell is False:
# Tokenize the command line
args = shlex.split(args)
if shell and 'executable' not in kargs:
kargs['executable'] = os.environ.get('SHELL', '/bin/sh')
if not 'stdout' in kargs and not 'stderr' in kargs:
kargs['capture_output'] = capture_output
if encoding:
kargs['encoding'] = encoding
check = kargs.get('check', False)
# TODO: NEED TO CAPTURE open ERRORS ?
for f in ['stdin', 'stdout', 'stderr']:
if f in kargs and isinstance(kargs[f], str):
if f[-1] == 'n':
mode = 'r'
else:
mode = 'w'
kargs[f] = open(kargs[f], mode)
try:
P = subprocess.run(args, **kargs)
P = DoneProcess(**P.__dict__)
except OSError:
if check:
raise
P = DoneProcess(args, 127, failed=True)
except subprocess.TimeoutExpired:
if check:
raise
P = DoneProcess(args, -1, timedout=True)
return P
[docs]def start(args, *, verbose=False, **kargs):
"""Start a subprocess for running an external command.
This creates a subprocess using Python3's :class:`subprocess.Popen`
running the command specified by `args`. This function has some
practical enhancements over subprocess.Popen:
- If the command is passed as a string and shell is False, the string is
automatically tokenized into an args list.
- The `verbose` option will print feedback.
- If `shell` = True is specified, and no `executable` is provided,
the shell will default to the value in the user's SHELL environment
variable, or '/bin/sh' it that isn't set
Note
----
This function will immediately return after creating the subprocess
and thus not wait for the command to terminate, nor return its
outcome. The user if fully responsible for handling the communication
with the process, its termination, and the capture of its outcome.
If you just want to wait for the command to finish, and capture
its output, use :func:`system` instead.
Parameters
----------
args: string or sequence of program arguments.
If a string is provided, it is the command as it should be entered
in a shell. If it is an args list, args[0] is the executable to run
and the remainder of args are the arguments to be passed to it.
If a string is provided and `shell` is False (default), the string is
split into a sequence of program arguments using :func:`shlex.split`.
verbose: bool
If True, the command is written to stdout.
shell: bool.
If True, the command will be run in a new shell.
This uses more resources, may cause problems with killing
the command and may pose a security risk.
Unless you need some shell functionality (parameter expansion,
compound commands, pipes), the advised way to
execute a command is to use the default False.
executable: string.
The full path name of the program to be executed.
This can be used to specify the real executable if the
program specified in `args` is not in your PATH, or,
when `shell` is True, if you want to use shell other than
the default: the value of the 'SHELL' environment variable,
or '/bin/sh/'.
**kargs: keyword arguments
Any other keyword arguments accepted by :class:`subprocess.Popen`.
See the Python documentation for :class:`subprocess.Popen` for full info.
Returns
-------
:class:`subprocess.Popen`
A :class:`subprocess.Popen` instance that can be used to communicate
with the started subprocess.
See Also
--------
run: start a subprocess and wait for its outcome
Examples
--------
>>> P = start("sleep 10")
>>> P.poll() is None
True
"""
shell = bool(kargs.get('shell', False))
if isinstance(args, str) and shell is False:
# Tokenize the command line
args = shlex.split(args)
if shell and 'executable' not in kargs:
kargs['executable'] = os.environ['SHELL']
return subprocess.Popen(args, **kargs)
if __name__ == "__main__":
print("This is process.py")
P = run("pwd", verbose=True, shell=True)
print(P)
P = run("ls -l", stdout="filelist.txt")
print(P)
### End
