<s> """<STR_LIT>""" <EOL> import locale <EOL> import logging <EOL> import random <EOL> import subprocess <EOL> import time <EOL> import warnings <EOL> from contextlib import contextmanager <EOL> from functools import wraps <EOL> from . import keys as keyboard <EOL> from . version import get_version <EOL> __version__ = get_version ( ) <EOL> logger = logging . getLogger ( __name__ ) <EOL> VALID_VERSIONS = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> WRONG_VERSION_MESSAGE = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) . format ( ver = VALID_VERSIONS [ <NUM_LIT:0> ] ) <EOL> try : <EOL> output = subprocess . check_output ( [ '<STR_LIT>' , '<STR_LIT>' ] , stderr = subprocess . STDOUT ) <EOL> output = output . decode ( locale . getdefaultlocale ( ) [ <NUM_LIT:1> ] ) <EOL> version_checks = ( ver in output for ver in VALID_VERSIONS ) <EOL> if not any ( version_checks ) : <EOL> warnings . warn ( WRONG_VERSION_MESSAGE , Warning ) <EOL> except Exception as exc : <EOL> warnings . warn ( WRONG_VERSION_MESSAGE , Warning ) <EOL> class TmuxError ( subprocess . CalledProcessError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class ConnectionFailedError ( TmuxError ) : <EOL> """<STR_LIT>""" <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' <EOL> class SessionNotFoundError ( TmuxError ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . session = kwargs . pop ( '<STR_LIT>' , None ) <EOL> super ( SessionNotFoundError , self ) . __init__ ( * args , ** kwargs ) <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' . format ( repr ( self . session ) ) <EOL> def send_keys ( session , keys , literal = True ) : <EOL> """<STR_LIT>""" <EOL> args = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> if literal : <EOL> args . append ( '<STR_LIT>' ) <EOL> args . append ( "<STR_LIT>" . format ( session ) ) <EOL> args . append ( keys ) <EOL> cmd = '<STR_LIT:U+0020>' . join ( args ) <EOL> logger . debug ( '<STR_LIT>' , cmd ) <EOL> try : <EOL> subprocess . check_output ( args , stderr = subprocess . STDOUT ) <EOL> except subprocess . CalledProcessError as exc : <EOL> output = exc . output . decode ( locale . getdefaultlocale ( ) [ <NUM_LIT:1> ] ) <EOL> if '<STR_LIT>' in output : <EOL> raise SessionNotFoundError ( exc . returncode , cmd , exc . output , <EOL> session = session ) <EOL> elif '<STR_LIT>' in output : <EOL> raise ConnectionFailedError ( exc . returncode , cmd , exc . output ) <EOL> else : <EOL> raise <EOL> def prompt ( func , input_func = None ) : <EOL> """<STR_LIT>""" <EOL> if input_func is None : <EOL> try : <EOL> input_func = raw_input <EOL> except NameError : <EOL> input_func = input <EOL> @ wraps ( func ) <EOL> def wrapper ( * args , ** kwargs ) : <EOL> self = args [ <NUM_LIT:0> ] <EOL> keys = args [ <NUM_LIT:1> ] if len ( args ) > <NUM_LIT:1> else None <EOL> if not args [ <NUM_LIT:0> ] . auto_advancing : <EOL> if keys is not None : <EOL> msg = "<STR_LIT>" . format ( <EOL> keys = repr ( keys ) , <EOL> session = self . session , <EOL> ) <EOL> else : <EOL> msg = "<STR_LIT>" . format ( <EOL> session = self . session <EOL> ) <EOL> input_func ( msg ) <EOL> return func ( * args , ** kwargs ) <EOL> return wrapper <EOL> class Session ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , session , enable_auto_advance = False , <EOL> teletype_delay = None ) : <EOL> self . session = session <EOL> self . auto_advancing = enable_auto_advance <EOL> self . teletype_delay = teletype_delay <EOL> def send_keys ( self , keys , literal = True ) : <EOL> """<STR_LIT>""" <EOL> send_keys ( self . session , keys , literal = literal ) <EOL> @ prompt <EOL> def teletype ( self , keys , delay = None ) : <EOL> """<STR_LIT>""" <EOL> if delay is None : <EOL> delay = ( self . teletype_delay if self . teletype_delay is not None <EOL> else <NUM_LIT> ) <EOL> delay_variation = delay / <NUM_LIT:10> <EOL> with self . auto_advance ( ) : <EOL> logger . info ( '<STR_LIT>' , self . session , repr ( keys ) ) <EOL> for key in keys : <EOL> self . send_keys ( key ) <EOL> current_delay = random . randint ( delay - delay_variation , <EOL> delay + delay_variation ) <EOL> time . sleep ( current_delay / <NUM_LIT> ) <EOL> @ prompt <EOL> def enter ( self , keys = None , teletype = True , after = keyboard . enter ) : <EOL> """<STR_LIT>""" <EOL> if keys : <EOL> if teletype : <EOL> with self . auto_advance ( ) : <EOL> self . teletype ( keys ) <EOL> else : <EOL> self . send_keys ( keys ) <EOL> if after : <EOL> with self . auto_advance ( ) : <EOL> self . send_keys ( after , literal = False ) <EOL> @ contextmanager <EOL> def auto_advance ( self ) : <EOL> """<STR_LIT>""" <EOL> initial_auto_state = self . auto_advancing <EOL> self . auto_advancing = True <EOL> yield <EOL> self . auto_advancing = initial_auto_state <EOL> __all__ = [ '<STR_LIT>' , '<STR_LIT>' ] </s>
