<s> from . stream import Stream <EOL> from . wrappers import StreamIOThreadWrapper <EOL> from . . compat import str <EOL> from . . exceptions import StreamError <EOL> from . . packages import pbs as sh <EOL> import os <EOL> import time <EOL> import tempfile <EOL> class StreamProcessIO ( StreamIOThreadWrapper ) : <EOL> def __init__ ( self , session , process , ** kwargs ) : <EOL> self . process = process <EOL> StreamIOThreadWrapper . __init__ ( self , session , <EOL> process . stdout , <EOL> ** kwargs ) <EOL> def close ( self ) : <EOL> try : <EOL> self . process . kill ( ) <EOL> except Exception : <EOL> pass <EOL> StreamIOThreadWrapper . close ( self ) <EOL> class StreamProcess ( Stream ) : <EOL> def __init__ ( self , session , params = None , timeout = <NUM_LIT> ) : <EOL> Stream . __init__ ( self , session ) <EOL> if not params : <EOL> params = { } <EOL> self . params = params <EOL> self . errorlog = self . session . options . get ( "<STR_LIT>" ) <EOL> self . timeout = timeout <EOL> def open ( self ) : <EOL> cmd = self . _check_cmd ( ) <EOL> params = self . params . copy ( ) <EOL> params [ "<STR_LIT>" ] = True <EOL> if self . errorlog : <EOL> tmpfile = tempfile . NamedTemporaryFile ( prefix = "<STR_LIT>" , <EOL> suffix = "<STR_LIT>" , delete = False ) <EOL> params [ "<STR_LIT>" ] = tmpfile <EOL> else : <EOL> params [ "<STR_LIT>" ] = open ( os . devnull , "<STR_LIT:wb>" ) <EOL> with params [ "<STR_LIT>" ] : <EOL> stream = cmd ( ** params ) <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> process_alive = stream . process . returncode is None <EOL> if not process_alive : <EOL> if self . errorlog : <EOL> raise StreamError ( ( "<STR_LIT>" <EOL> "<STR_LIT>" ) . format ( tmpfile . name ) ) <EOL> else : <EOL> raise StreamError ( "<STR_LIT>" ) <EOL> return StreamProcessIO ( self . session , stream . process , <EOL> timeout = self . timeout ) <EOL> def _check_cmd ( self ) : <EOL> try : <EOL> cmd = sh . create_command ( self . cmd ) <EOL> except sh . CommandNotFound as err : <EOL> raise StreamError ( "<STR_LIT>" . format ( err ) ) <EOL> return cmd <EOL> def cmdline ( self ) : <EOL> cmd = self . _check_cmd ( ) <EOL> return str ( cmd . bake ( ** self . params ) ) <EOL> @ classmethod <EOL> def is_usable ( cls , cmd ) : <EOL> try : <EOL> cmd = sh . create_command ( cmd ) <EOL> except sh . CommandNotFound : <EOL> return False <EOL> return True <EOL> __all__ = [ "<STR_LIT>" ] </s>
