Mercurial > hg > periprocess
changeset 0:d28351df9a8a
initial commit from http://k0s.org/hg/config/file/tip/python/example/process.py
author | Jeff Hammel <k0scist@gmail.com> |
---|---|
date | Wed, 30 Nov 2016 12:40:25 -0800 |
parents | |
children | a1b13d024582 |
files | INSTALL.py README.txt periprocess/__init__.py periprocess/process.py setup.py tests/test_periprocess.py tests/testall.py |
diffstat | 7 files changed, 468 insertions(+), 0 deletions(-) [+] |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/INSTALL.py Wed Nov 30 12:40:25 2016 -0800 @@ -0,0 +1,67 @@ +#!/usr/bin/env python + +""" +installation script for periprocess +subprocess front-end that keeps buffers of stdout/stderr +""" + +import os +import sys +import urllib2 +import subprocess +try: + from subprocess import check_call as call +except: + from subprocess import call + +REPO='http://k0s.org/hg/periprocess' +DEST='periprocess' # name of the virtualenv +VIRTUALENV='https://raw.github.com/pypa/virtualenv/develop/virtualenv.py' + +def which(binary, path=os.environ['PATH']): + dirs = path.split(os.pathsep) + for dir in dirs: + if os.path.isfile(os.path.join(dir, fileName)): + return os.path.join(dir, fileName) + if os.path.isfile(os.path.join(dir, fileName + ".exe")): + return os.path.join(dir, fileName + ".exe") + +def main(args=sys.argv[1:]): + + # create a virtualenv + virtualenv = which('virtualenv') or which('virtualenv.py') + if virtualenv: + call([virtualenv, DEST]) + else: + process = subproces.Popen([sys.executable, '-', DEST], stdin=subprocess.PIPE) + process.communicate(stdin=urllib2.urlopen(VIRTUALENV).read()) + + # create a src directory + src = os.path.join(DEST, 'src') + os.mkdir(src) + + # clone the repository + call(['hg', 'clone', REPO], cwd=src) + + # find the virtualenv python + python = None + for path in (('bin', 'python'), ('Scripts', 'python.exe')): + _python = os.path.join(DEST, *path) + if os.path.exists(_python) + python = _python + break + else: + raise Exception("Python binary not found in %s" % DEST) + + # find the clone + filename = REPO.rstrip('/') + filename = filename.split('/')[-1] + clone = os.path.join(src, filename) + assert os.path.exists(clone), "Clone directory not found in %s" % src + + # ensure setup.py exists + assert os.path.exists(os.path.join(clone, 'setup.py')), 'setup.py not found in %s' % clone + + # install the package in develop mode + call([python 'setup.py', 'develop'], cwd=clone) +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/README.txt Wed Nov 30 12:40:25 2016 -0800 @@ -0,0 +1,11 @@ +periprocess +=========== + +subprocess front-end that keeps buffers of stdout/stderr + +---- + +Jeff Hammel + +http://k0s.org/hg/periprocess +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/periprocess/__init__.py Wed Nov 30 12:40:25 2016 -0800 @@ -0,0 +1,2 @@ +# +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/periprocess/process.py Wed Nov 30 12:40:25 2016 -0800 @@ -0,0 +1,274 @@ +#!/usr/bin/env python + +""" +multiprocessing/subprocess front-end +""" + +# imports +import argparse +import atexit +import os +import signal +import subprocess # http://bugs.python.org/issue1731717 +import sys +import time +import tempfile + + +# globals +__all__ = ['Process'] +string = (str, unicode) +PIDS = set() + +# ensure subprocesses gets killed on exit +def killall(): + """kill all subprocess PIDs""" + global PIDS + for pid in PIDS.copy(): + try: + os.kill(pid, 9) # SIGKILL + PIDS.discard(pid) + except: + sys.stderr.write("Unable to kill PID {}\n".format(pid)) +atexit.register(killall) + + +signals = (signal.SIGHUP, signal.SIGINT, signal.SIGQUIT, signal.SIGSEGV, signal.SIGTERM) # signals to handle +fatal = set([signal.SIGINT, signal.SIGSEGV, signal.SIGKILL, signal.SIGTERM]) +# ensure subprocesses get killed on signals +def sighandler(signum, frame): + """https://docs.python.org/2/library/signal.html""" + sys.stderr.write('Signal handler called with signal {}\n; terminating subprocesses: {}'.format(signum, + ', '.join([str(pid) for pid in sorted(PIDS)]))) + killall() + if signum in fatal: + print ("Caught signal {}; exiting".format(signum)) + sys.exit() +for signum in signals: + try: + signal.signal(signum, sighandler) + except RuntimeError as e: + print ('[{}] {}'.format(signum, e)) + raise + +class Process(subprocess.Popen): + """why would you name a subprocess object Popen?""" + + # http://docs.python.org/2/library/subprocess.html#popen-constructor + defaults = {'bufsize': 1, # line buffered + 'store_output': True, # store stdout + } + + def __init__(self, command, **kwargs): + + # get verbosity + self.verbose = kwargs.pop('verbose', False) + + # setup arguments + self.command = command + _kwargs = self.defaults.copy() + _kwargs.update(kwargs) + + + # on unix, ``shell={True|False}`` should always come from the + # type of command (string or list) + if not subprocess.mswindows: + _kwargs['shell'] = isinstance(command, string) + + # output buffer + self.location = 0 + self.output_buffer = tempfile.SpooledTemporaryFile() + self.output = '' if _kwargs.pop('store_output') else None + _kwargs['stdout'] = self.output_buffer + + # ensure child in process group + # see : + # - http://pymotw.com/2/subprocess/#process-groups-sessions + # - http://ptspts.blogspot.com/2012/11/how-to-start-and-kill-unix-process-tree.html + _kwargs['preexec_fn'] = os.setpgrp + + # runtime + self.start = time.time() + self.end = None + + if self.verbose: + # print useful info + print ("Running `{}`; started: {}".format(str(self), self.start)) + + # launch subprocess + try: + subprocess.Popen.__init__(self, command, **_kwargs) + PIDS.add(self.pid) + if self.verbose: + # print the PID + print ("PID: {}".format(self.pid)) + except: + # print the command + print ("Failure to run:") + print (self.command) + + # reraise the hard way: + # http://www.ianbicking.org/blog/2007/09/re-raising-exceptions.html + exc = sys.exc_info() + raise exc[0], exc[1], exc[2] + + + def _finalize(self, process_output): + """internal function to finalize""" + + # read final output + if process_output is not None: + self.read(process_output) + + # reset output buffer location + self.output_buffer.seek(0) + + # set end time + self.end = time.time() + + # remove PID from list + PIDS.discard(self.pid) + + def poll(self, process_output=None): + + if process_output is not None: + self.read(process_output) # read from output buffer + poll = subprocess.Popen.poll(self) + if poll is not None: + self._finalize(process_output) + return poll + + def wait(self, maxtime=None, sleep=1., process_output=None): + """ + maxtime -- timeout in seconds + sleep -- number of seconds to sleep between polling + """ + while self.poll(process_output) is None: + + # check for timeout + curr_time = time.time() + run_time = self.runtime() + if maxtime is not None and run_time > maxtime: + self.kill() + self._finalize(process_output) + return + + # naptime + if sleep: + time.sleep(sleep) + + # finalize + self._finalize(process_output) + + return self.returncode # set by ``.poll()`` + + def read(self, process_output=None): + """read from the output buffer""" + + self.output_buffer.seek(self.location) + read = self.output_buffer.read() + if self.output is not None: + self.output += read + if process_output: + process_output(read) + self.location += len(read) + return read + + def commandline(self): + """returns string of command line""" + + if isinstance(self.command, string): + return self.command + return subprocess.list2cmdline(self.command) + + __str__ = commandline + + def runtime(self): + """returns time spent running or total runtime if completed""" + + if self.end is None: + return time.time() - self.start + return self.end - self.start + + +def main(args=sys.argv[1:]): + """CLI""" + + description = """demonstration of how to do things with subprocess""" + + # available programs + progs = {'yes': ["yes"], + 'ping': ['ping', 'google.com']} + + # parse command line + parser = argparse.ArgumentParser(description=description) + parser.add_argument("-t", "--time", dest="time", + type=float, default=4., + help="seconds to run for (or <= 0 for forever)") + parser.add_argument("-s", "--sleep", dest="sleep", + type=float, default=1., + help="sleep this number of seconds between polling") + parser.add_argument("-p", "--prog", dest='program', + choices=progs.keys(), default='ping', + help="subprocess to run") + parser.add_argument("--list-programs", dest='list_programs', + action='store_true', default=False, + help="list available programs") + parser.add_argument("--wait", dest='wait', + action='store_true', default=False, + help="run with ``.wait()`` and a callback") + parser.add_argument("--callback", dest='callback', + action='store_true', default=False, + help="run with polling and a callback") + options = parser.parse_args(args) + + # list programs + if options.list_programs: + for key in sorted(progs.keys()): + print ('{}: {}'.format(key, subprocess.list2cmdline(progs[key]))) + sys.exit(0) + + # select program + prog = progs[options.program] + + # start process + proc = Process(prog) + + # demo function for processing output + def output_processor(output): + print ('[{}]:\n{}\n{}'.format(proc.runtime(), + output.upper(), + '-==-'*10)) + if options.callback: + process_output = output_processor + else: + process_output = None + + if options.wait: + # wait for being done + proc.wait(maxtime=options.time, sleep=options.sleep, process_output=output_processor) + else: + # start the main subprocess loop + while proc.poll(process_output) is None: + + if options.time > 0 and proc.runtime() > options.time: + proc.kill() + + if options.sleep: + time.sleep(options.sleep) + + if process_output is None: + # process the output with ``.read()`` call + read = proc.read() + output_processor(read) + + # correctness tests + assert proc.end is not None + + # print summary + output = proc.output + n_lines = len(output.splitlines()) + print ("{}: {} lines, ran for {} seconds".format(subprocess.list2cmdline(prog), n_lines, proc.runtime())) + +if __name__ == '__main__': + main()
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/setup.py Wed Nov 30 12:40:25 2016 -0800 @@ -0,0 +1,45 @@ +""" +setup packaging script for periprocess +""" + +import os + +version = "0.0" +dependencies = [] + +# allow use of setuptools/distribute or distutils +kw = {} +try: + from setuptools import setup + kw['entry_points'] = """ + [console_scripts] + periprocess = periprocess.main:main + periprocess-template = periprocess.template:main +""" + kw['install_requires'] = dependencies +except ImportError: + from distutils.core import setup + kw['requires'] = dependencies + +try: + here = os.path.dirname(os.path.abspath(__file__)) + description = file(os.path.join(here, 'README.txt')).read() +except IOError: + description = '' + + +setup(name='periprocess', + version=version, + description="subprocess front-end that keeps buffers of stdout/stderr", + long_description=description, + classifiers=[], # Get strings from http://www.python.org/pypi?%3Aaction=list_classifiers + author='Jeff Hammel', + author_email='k0scist@gmail.com', + url='http://k0s.org/hg/periprocess', + license='MPL', + packages=['periprocess'], + include_package_data=True, + zip_safe=False, + **kw + ) +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/tests/test_periprocess.py Wed Nov 30 12:40:25 2016 -0800 @@ -0,0 +1,29 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +""" +unit tests for periprocess +""" + +import os +import sys +import tempfile +import unittest + +# globals +here = os.path.dirname(os.path.abspath(__file__)) + +class periprocessUnitTest(unittest.TestCase): + + def test_periprocess(self): + tf = tempfile.mktemp() + try: + # pass + pass + finally: + if os.path.exists(tf): + os.remove(tf) + +if __name__ == '__main__': + unittest.main() +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/tests/testall.py Wed Nov 30 12:40:25 2016 -0800 @@ -0,0 +1,40 @@ +#!/usr/bin/env python + +""" +run all unit tests +""" + +import os +import sys +import unittest + +here = os.path.dirname(os.path.abspath(__file__)) + +def main(args=sys.argv[1:]): + + results = unittest.TestResult() + suite = unittest.TestLoader().discover(here, 'test_*.py') + suite.run(results) + n_errors = len(results.errors) + n_failures = len(results.failures) + print ("Run {} tests ({} failures; {} errors)".format(results.testsRun, + n_failures, + n_errors)) + if results.wasSuccessful(): + print ("Success") + sys.exit(0) + else: + # print failures and errors + for label, item in (('FAIL', results.failures), + ('ERROR', results.errors)): + if item: + print ("\n{}::\n".format(label)) + for index, (i, message) in enumerate(item): + print ('{}) {}:'.format(index, str(i))) + print (message) + sys.exit(1) + +if __name__ == '__main__': + main() + +