Source code for honeybee_plus.radiance.command.pyRad.pyRadLib.pyrad_proc

# -*- coding: utf-8 -*-
''' pyrad_proc.py - Process and pipeline management for Python Radiance scripts
2016 - Georg Mischler

Use as:
    from pyradlib.pyrad_proc import PIPE, Error, ProcMixin

For a single-file installation, include the contents of this file
at the same place (minus the __future__ import below).
'''
from __future__ import division, print_function, unicode_literals

import sys
import subprocess
PIPE = subprocess.PIPE


[docs]class Error(Exception): pass
[docs]class ProcMixin(): '''Process and pipeline management for Python Radiance scripts '''
[docs] def raise_on_error(self, actstr, e): try: self._strtypes except AttributeError: self.__configure_subprocess() if hasattr(e, 'strerror'): eb = e.strerror elif isinstance(e, self._strtypes): eb = e else: eb = e if isinstance(eb, type(b'')): estr = eb.decode(encoding='utf-8', errors='ignore') else: estr = eb raise Error('Unable to %s - %s' % (actstr, estr))
def __configure_subprocess(self): '''Prevent subprocess module failure in frozen scripts on Windows. Prevent console windows from popping up when not console based. Make sure we use the version-specific string types. ''' # On Windows, sys.stdxxx may not be available when: # - built as *.exe with "pyinstaller --noconsole" # - invoked via CreateProcess() and stream not redirected try: sys.__stdin__.fileno() self._stdin = sys.stdin except BaseException: self._stdin = PIPE try: sys.__stdout__.fileno() self._stdout = sys.stdout except BaseException: self._stdout = PIPE try: sys.__stderr__.fileno() self._stderr = sys.stderr # keep subprocesses from opening their own console. except BaseException: self._stderr = PIPE if hasattr(subprocess, 'STARTUPINFO'): si = subprocess.STARTUPINFO() si.dwFlags |= subprocess.STARTF_USESHOWWINDOW self._pipeargs = {'startupinfo': si} else: self._pipeargs = {} # type names vary between Py2.7 and 3.x self._strtypes = (type(b''), type(u''))
[docs] def qjoin(self, sl): '''Join a list with quotes around each element containing whitespace. We only use this to display command lines on sys.stderr, the actual Popen() calls are made with the original list. ''' def _q(s): if ' ' in s or '\t' in s or ';' in s: return "'" + s + "'" return s return ' '.join([_q(s) for s in sl])
def __parse_args(self, _in, out): try: self._strtypes except AttributeError: self.__configure_subprocess() instr = '' if _in == PIPE: stdin = _in elif isinstance(_in, self._strtypes): if not getattr(self, 'donothing', None): stdin = open(_in, 'rb') else: stdin = None instr = ' < "%s"' % _in elif hasattr(_in, 'read'): stdin = _in instr = ' < "%s"' % _in.name else: stdin = self._stdin outstr = '' if out == PIPE: stdout = out elif isinstance(out, self._strtypes): if not getattr(self, 'donothing', None): stdout = open(out, 'wb') else: stdout = None outstr = ' > "%s"' % out elif hasattr(out, 'write'): stdout = out outstr = ' > "%s"' % out.name else: stdout = self._stdout return stdin, stdout, instr, outstr
[docs] def call_one(self, cmdl, actstr, _in=None, out=None, universal_newlines=False): '''Create a single subprocess, possibly with an incoming and outgoing pipe at each end. - cmdl A list of strings, leading with the name of the executable (without the *.exe suffix), followed by individual arguments. - actstr A text string of the form "do something". Used in verbose mode as "### do someting\n### [command line]" Used in error messages as "Scriptname: Unable to do something". - _in / out What to do with the input and output pipes of the process: * a filename as string Open file and use for reading/writing. * a file like object Use for reading/writing * PIPE Pipe will be available in returned object for reading/writing. * None (default) System stdin/stdout will be used if available If _in or out is a PIPE, the caller should call p.wait() on the returned Popen instance after writing to and closing it. ''' stdin, stdout, instr, outstr = self.__parse_args(_in, out) if getattr(self, 'verbose', None): sys.stderr.write('### %s \n' % actstr) sys.stderr.write(self.qjoin(cmdl) + instr + outstr + '\n') if not getattr(self, 'donothing', None): try: p = subprocess.Popen( cmdl, stdin=stdin, stdout=stdout, stderr=self._stderr, universal_newlines=universal_newlines, **self._pipeargs) except Exception as e: self.raise_on_error(actstr, e) if stdin != PIPE and stdout != PIPE: # caller needs to wait after reading or writing (else deadlock) res = p.wait() if res != 0: self.raise_on_error( actstr, 'Nonzero exit (%d) from command [%s].' % (res, self.qjoin(cmdl) + instr + outstr + '\n')) return p
[docs] def call_two(self, cmdl_1, cmdl_2, actstr_1, actstr_2, _in=None, out=None, universal_newlines=False): '''Create two processes, chained via a pipe, possibly with an incoming and outgoing pipe at each end. Returns a tuple of two Popen instances. Arguments are equivalent to call_one(), with _in and out applying to the ends of the chain. If _in or out is PIPE, the caller should call p.wait() on both returned popen instances after writing to and closing the first on . ''' stdin, stdout, instr, outstr = self.__parse_args(_in, out) if getattr(self, 'verbose', None): sys.stderr.write('### %s \n' % actstr_1) sys.stderr.write('### %s \n' % actstr_2) sys.stderr.write(self.qjoin(cmdl_1) + instr + ' | ') if not getattr(self, 'donothing', None): try: p1 = subprocess.Popen(cmdl_1, stdin=stdin, stdout=PIPE, stderr=self._stderr, **self._pipeargs) except Exception as e: self.raise_on_error(actstr_1, e) if getattr(self, 'verbose', None): sys.stderr.write(self.qjoin(cmdl_2) + outstr + '\n') if not getattr(self, 'donothing', None): try: p2 = subprocess.Popen( cmdl_2, stdin=p1.stdout, stdout=stdout, stderr=self._stderr, universal_newlines=universal_newlines, **self._pipeargs) p1.stdout.close() except Exception as e: self.raise_on_error(actstr_2, e) if stdin != PIPE and stdout != PIPE: # caller needs to wait after reading or writing (else deadlock) res = p1.wait() if res != 0: self.raise_on_error(actstr_1, 'Nonzero exit (%d) from command [%s].' % (res, self.qjoin(cmdl_1))) res = p2.wait() if res != 0: self.raise_on_error(actstr_2, 'Nonzero exit (%d) from command [%s].' % (res, self.qjoin(cmdl_2))) return p1, p2
[docs] def call_many(self, cmdlines, actstr, _in=None, out=None, universal_newlines=False): '''Create a series of N processes, chained via pipes, possibly with an incoming and outgoing pipe at each end. Returns a tuple of N subprocess.Popen instances. Depending on the values of _in and out, the first and last may be available to write to or read from respectively. Most arguments are equivalent to call_one(), with - cmdlines a list of N command argument lists - _in / out applying to the ends of the chain. If _in or out is PIPE, the caller should call p.wait() on all returned Popen instances after writing to and closing the first one. ''' if len(cmdlines) == 1: # other than direct call_one(), this returns a one-item tuple! return (self.call_one(cmdlines[0], actstr, _in=_in, out=out, universal_newlines=universal_newlines),) stdin, stdout, instr, outstr = self.__parse_args(_in, out) procs = [] if getattr(self, 'verbose', None): sys.stderr.write('### %s \n' % actstr) sys.stderr.write(self.qjoin(cmdlines[0]) + instr + ' | ') if not getattr(self, 'donothing', None): try: prevproc = subprocess.Popen( cmdlines[0], stdin=stdin, stdout=PIPE, stderr=self._stderr, **self._pipeargs) procs.append(prevproc) except Exception as e: self.raise_on_error(actstr, e) for cmdl in cmdlines[1:-1]: if getattr(self, 'verbose', None): sys.stderr.write(self.qjoin(cmdl) + ' | ') if not getattr(self, 'donothing', None): try: nextproc = subprocess.Popen( cmdl, stdin=prevproc.stdout, stdout=PIPE, stderr=self._stderr, **self._pipeargs) procs.append(nextproc) prevproc.stdout.close() prevproc = nextproc except Exception as e: self.raise_on_error(actstr, e) if getattr(self, 'verbose', None): sys.stderr.write(self.qjoin(cmdlines[-1]) + outstr + '\n') if not getattr(self, 'donothing', None): try: lastproc = subprocess.Popen( cmdlines[-1], stdin=prevproc.stdout, stdout=stdout, stderr=self._stderr, universal_newlines=universal_newlines, **self._pipeargs) prevproc.stdout.close() procs.append(lastproc) prevproc.stdout.close() except Exception as e: self.raise_on_error(actstr, e) if stdin != PIPE and stdout != PIPE: # caller needs to wait after reading or writing (else deadlock) for proc, cmdl in zip(procs, cmdlines): res = proc.wait() if res != 0: self.raise_on_error(actstr, 'Nonzero exit (%d) from command [%s].' % (res, self.qjoin(cmdl))) return procs
# end of proc_mixin.py