signature:
def submit ( self , command , blocksize , tasks_per_node , job_name = "parsl.auto" ) :
    """Submits the command onto an Local Resource Manager job of blocksize parallel elements .
 example file with the complex case of multiple submits per job :
 Universe = vanilla
 output = out . $ ( Cluster ) . $ ( Process )
 error = err . $ ( Cluster ) . $ ( Process )
 log = log . $ ( Cluster )
 leave _ in _ queue = true
 executable = test . sh
 queue 5
 executable = foo
 queue 1
 $ condor _ submit test . sub
 Submitting job ( s ) . . . . .
 5 job ( s ) submitted to cluster 118907.
 1 job ( s ) submitted to cluster 118908.
 Parameters
 command : str
 Command to execute
 blocksize : int
 Number of blocks to request .
 job _ name : str
 Job name prefix .
 tasks _ per _ node : int
 command invocations to be launched per node
 Returns
 None or str
 None if at capacity and cannot provision more ; otherwise the identifier for the job ."""

implementation:
logger . debug ( "Attempting to launch with blocksize: {}" . format ( blocksize ) )
if self . provisioned_blocks >= self . max_blocks :
    template = "Provider {} is currently using {} blocks while max_blocks is {}; no blocks will be added"
    logger . warn ( template . format ( self . label , self . provisioned_blocks , self . max_blocks ) )
    return None
# Note : Fix this later to avoid confusing behavior .
# We should always allocate blocks in integer counts of node _ granularity
blocksize = max ( self . nodes_per_block , blocksize )
job_name = "parsl.{0}.{1}" . format ( job_name , time . time ( ) )
script_path = "{0}/{1}.submit" . format ( self . script_dir , job_name )
script_path = os . path . abspath ( script_path )
userscript_path = "{0}/{1}.script" . format ( self . script_dir , job_name )
userscript_path = os . path . abspath ( userscript_path )
self . environment [ "JOBNAME" ] = "'{}'" . format ( job_name )
job_config = { }
job_config [ "job_name" ] = job_name
job_config [ "submit_script_dir" ] = self . channel . script_dir
job_config [ "project" ] = self . project
job_config [ "nodes" ] = self . nodes_per_block
job_config [ "scheduler_options" ] = self . scheduler_options
job_config [ "worker_init" ] = self . worker_init
job_config [ "user_script" ] = command
job_config [ "tasks_per_node" ] = tasks_per_node
job_config [ "requirements" ] = self . requirements
job_config [ "environment" ] = ' ' . join ( [ '{}={}' . format ( key , value ) for key , value in self . environment . items ( ) ] )
# Move the user script
# This is where the command should be wrapped by the launchers .
wrapped_command = self . launcher ( command , tasks_per_node , self . nodes_per_block )
with open ( userscript_path , 'w' ) as f :
    f . write ( job_config [ "worker_init" ] + '\n' + wrapped_command )
user_script_path = self . channel . push_file ( userscript_path , self . channel . script_dir )
the_input_files = [ user_script_path ] + self . transfer_input_files
job_config [ "input_files" ] = ',' . join ( the_input_files )
job_config [ "job_script" ] = os . path . basename ( user_script_path )
# Construct and move the submit script
self . _write_submit_script ( template_string , script_path , job_name , job_config )
channel_script_path = self . channel . push_file ( script_path , self . channel . script_dir )
cmd = "condor_submit {0}" . format ( channel_script_path )
retcode , stdout , stderr = super ( ) . execute_wait ( cmd , 30 )
logger . debug ( "Retcode:%s STDOUT:%s STDERR:%s" , retcode , stdout . strip ( ) , stderr . strip ( ) )
job_id = [ ]
if retcode == 0 :
    for line in stdout . split ( '\n' ) :
        if re . match ( '^[0-9]' , line ) is not None :
            cluster = line . split ( " " ) [ 5 ]
            # We know the first job id ( " process " in condor terms ) within a
            # cluster is 0 and we know the total number of jobs from
            # condor _ submit , so we use some list comprehensions to expand
            # the condor _ submit output into job IDs
            # e . g . , [ ' 118907.0 ' , ' 118907.1 ' , ' 118907.2 ' , ' 118907.3 ' , ' 118907.4 ' , ' 118908.0 ' ]
            processes = [ str ( x ) for x in range ( 0 , int ( line [ 0 ] ) ) ]
            job_id += [ cluster + process for process in processes ]
    self . _add_resource ( job_id )
return job_id [ 0 ]


signature:
def compose_containerized_launch_cmd ( self , filepath , engine_dir , container_image ) :
    """Reads the json contents from filepath and uses that to compose the engine launch command .
 Notes : Add this to the ipengine launch for debug logs :
 - - log - to - file - - debug
 Args :
 filepath ( str ) : Path to the engine file
 engine _ dir ( str ) : CWD for the engines .
 container _ image ( str ) : The container to be used to launch workers"""

implementation:
self . engine_file = os . path . expanduser ( filepath )
uid = str ( uuid . uuid4 ( ) )
engine_json = None
try :
    with open ( self . engine_file , 'r' ) as f :
        engine_json = f . read ( )
except OSError as e :
    logger . error ( "Could not open engine_json : " , self . engine_file )
    raise e
return """mkdir -p {0}
cd {0}
cat <<EOF > ipengine.{uid}.json
{1}
EOF

DOCKER_ID=$(docker create --network host {2} ipengine --file=/tmp/ipengine.{uid}.json) {debug_option}
docker cp ipengine.{uid}.json $DOCKER_ID:/tmp/ipengine.{uid}.json

# Copy current dir to the working directory
DOCKER_CWD=$(docker image inspect --format='{{{{.Config.WorkingDir}}}}' {2})
docker cp -a . $DOCKER_ID:$DOCKER_CWD
docker start $DOCKER_ID

at_exit() {{
  echo "Caught SIGTERM/SIGINT signal!"
  docker stop $DOCKER_ID
}}

trap at_exit SIGTERM SIGINT
sleep infinity
""" . format ( engine_dir , engine_json , container_image , debug_option = self . debug_option , uid = uid )


signature:
def scale_out ( self , blocks = 1 ) :
    """Scales out the number of active workers by 1.
 This method is notImplemented for threads and will raise the error if called .
 Parameters :
 blocks : int
 Number of blocks to be provisioned ."""

implementation:
r = [ ]
for i in range ( blocks ) :
    if self . provider :
        block = self . provider . submit ( self . launch_cmd , 1 , self . workers_per_node )
        logger . debug ( "Launched block {}:{}" . format ( i , block ) )
        if not block :
            raise ( ScalingFailed ( self . provider . label , "Attempts to provision nodes via provider has failed" ) )
        self . engines . extend ( [ block ] )
        r . extend ( [ block ] )
else :
    logger . error ( "No execution provider available" )
    r = None
return r


signature:
def scale_in ( self , blocks ) :
    """Scale in the number of active blocks by the specified number ."""

implementation:
status = dict ( zip ( self . engines , self . provider . status ( self . engines ) ) )
# This works for blocks = 0
to_kill = [ engine for engine in status if status [ engine ] == "RUNNING" ] [ : blocks ]
if self . provider :
    r = self . provider . cancel ( to_kill )
else :
    logger . error ( "No execution provider available" )
    r = None
return r


signature:
def status ( self ) :
    """Returns the status of the executor via probing the execution providers ."""

implementation:
if self . provider :
    status = self . provider . status ( self . engines )
else :
    status = [ ]
return status


signature:
def shutdown ( self , hub = True , targets = 'all' , block = False ) :
    """Shutdown the executor , including all workers and controllers .
 The interface documentation for IPP is ` here < http : / / ipyparallel . readthedocs . io / en / latest / api / ipyparallel . html # ipyparallel . Client . shutdown > ` _
 Kwargs :
 - hub ( Bool ) : Whether the hub should be shutdown , Default : True ,
 - targets ( list of ints | ' all ' ) : List of engine id ' s to kill , Default : ' all '
 - block ( Bool ) : To block for confirmations or not
 Raises :
 NotImplementedError"""

implementation:
if self . controller :
    logger . debug ( "IPP:Shutdown sequence: Attempting controller kill" )
    self . controller . close ( )
# We do not actually do executor . shutdown because
# this blocks even when requested to not block , killing the
# controller is more effective although impolite .
# x = self . executor . shutdown ( targets = targets ,
# hub = hub ,
# block = block )
logger . debug ( "Done with executor shutdown" )
return True


signature:
def parent_callback ( self , executor_fu ) :
    """Callback from a parent future to update the AppFuture .
 Used internally by AppFuture , and should not be called by code using AppFuture .
 Args :
 - executor _ fu ( Future ) : Future returned by the executor along with callback .
 This may not be the current parent future , as the parent future may have
 already been updated to point to a retrying execution , and in that case ,
 this is logged .
 In the case that a new parent has been attached , we must immediately discard
 this result no matter what it contains ( although it might be interesting
 to log if it was successful . . . )
 Returns :
 - None
 Updates the super ( ) with the result ( ) or exception ( )"""

implementation:
with self . _update_lock :
    if not executor_fu . done ( ) :
        raise ValueError ( "done callback called, despite future not reporting itself as done" )
    # this is for consistency checking
    if executor_fu != self . parent :
        if executor_fu . exception ( ) is None and not isinstance ( executor_fu . result ( ) , RemoteExceptionWrapper ) : # . . . then we completed with a value , not an exception or wrapped exception ,
        # but we ' ve got an updated executor future .
        # This is bad - for example , we ' ve started a retry even though we have a result
            raise ValueError ( "internal consistency error: AppFuture done callback called without an exception, but parent has been changed since then" )
    try :
        res = executor_fu . result ( )
        if isinstance ( res , RemoteExceptionWrapper ) :
            res . reraise ( )
        super ( ) . set_result ( executor_fu . result ( ) )
    except Exception as e :
        if executor_fu . retries_left > 0 : # ignore this exception , because assume some later
        # parent executor , started external to this class ,
        # will provide the answer
            pass
        else :
            super ( ) . set_exception ( e )


signature:
def update_parent ( self , fut ) :
    """Add a callback to the parent to update the state .
 This handles the case where the user has called result on the AppFuture
 before the parent exists ."""

implementation:
self . parent = fut
try :
    fut . add_done_callback ( self . parent_callback )
except Exception as e :
    logger . error ( "add_done_callback got an exception {} which will be ignored" . format ( e ) )


signature:
def parent_callback ( self , parent_fu ) :
    """Callback from executor future to update the parent .
 Args :
 - parent _ fu ( Future ) : Future returned by the executor along with callback
 Returns :
 - None
 Updates the super ( ) with the result ( ) or exception ( )"""

implementation:
if parent_fu . done ( ) is True :
    e = parent_fu . _exception
    if e :
        super ( ) . set_exception ( e )
    else :
        super ( ) . set_result ( self . file_obj )
return


signature:
def remote_side_bash_executor ( func , * args , ** kwargs ) :
    """Execute the bash app type function and return the command line string .
 This string is reformatted with the * args , and * * kwargs
 from call time ."""

implementation:
import os
import time
import subprocess
import logging
import parsl . app . errors as pe
logging . basicConfig ( filename = '/tmp/bashexec.{0}.log' . format ( time . time ( ) ) , level = logging . DEBUG )
# start _ t = time . time ( )
func_name = func . __name__
partial_cmdline = None
# Try to run the func to compose the commandline
try : # Execute the func to get the commandline
    partial_cmdline = func ( * args , ** kwargs )
    # Reformat the commandline with current args and kwargs
    executable = partial_cmdline . format ( * args , ** kwargs )
except AttributeError as e :
    if partial_cmdline is not None :
        raise pe . AppBadFormatting ( "App formatting failed for app '{}' with AttributeError: {}" . format ( func_name , e ) )
    else :
        raise pe . BashAppNoReturn ( "Bash app '{}' did not return a value, or returned none - with this exception: {}" . format ( func_name , e ) , None )
except IndexError as e :
    raise pe . AppBadFormatting ( "App formatting failed for app '{}' with IndexError: {}" . format ( func_name , e ) )
except Exception as e :
    logging . error ( "Caught exception during formatting of app '{}': {}" . format ( func_name , e ) )
    raise e
logging . debug ( "Executable: %s" , executable )
# Updating stdout , stderr if values passed at call time .
def open_std_fd ( fdname ) : # fdname is ' stdout ' or ' stderr '
    stdfspec = kwargs . get ( fdname )
    # spec is str name or tuple ( name , mode )
    if stdfspec is None :
        return None
    elif isinstance ( stdfspec , str ) :
        fname = stdfspec
        mode = 'a+'
    elif isinstance ( stdfspec , tuple ) :
        if len ( stdfspec ) != 2 :
            raise pe . BadStdStreamFile ( "std descriptor %s has incorrect tuple length %s" % ( fdname , len ( stdfspec ) ) , TypeError ( 'Bad Tuple Length' ) )
        fname , mode = stdfspec
    else :
        raise pe . BadStdStreamFile ( "std descriptor %s has unexpected type %s" % ( fdname , str ( type ( stdfspec ) ) ) , TypeError ( 'Bad Tuple Type' ) )
    try :
        fd = open ( fname , mode )
    except Exception as e :
        raise pe . BadStdStreamFile ( fname , e )
    return fd
std_out = open_std_fd ( 'stdout' )
std_err = open_std_fd ( 'stderr' )
timeout = kwargs . get ( 'walltime' )
returncode = None
try :
    proc = subprocess . Popen ( executable , stdout = std_out , stderr = std_err , shell = True , executable = '/bin/bash' )
    proc . wait ( timeout = timeout )
    returncode = proc . returncode
except subprocess . TimeoutExpired : # print ( " Timeout " )
    raise pe . AppTimeout ( "[{}] App exceeded walltime: {}" . format ( func_name , timeout ) )
except Exception as e : # print ( " Caught exception : " , e )
    raise pe . AppException ( "[{}] App caught exception: {}" . format ( func_name , proc . returncode ) , e )
if returncode != 0 :
    raise pe . AppFailure ( "[{}] App failed with exit code: {}" . format ( func_name , proc . returncode ) , proc . returncode )
# TODO : Add support for globs here
missing = [ ]
for outputfile in kwargs . get ( 'outputs' , [ ] ) :
    fpath = outputfile
    if type ( outputfile ) != str :
        fpath = outputfile . filepath
    if not os . path . exists ( fpath ) :
        missing . extend ( [ outputfile ] )
if missing :
    raise pe . MissingOutputs ( "[{}] Missing outputs" . format ( func_name ) , missing )
# exec _ duration = time . time ( ) - start _ t
return returncode


signature:
def submit ( self , command , blocksize , tasks_per_node , job_name = "parsl.auto" ) :
    '''The submit method takes the command string to be executed upon
 instantiation of a resource most often to start a pilot .
 Args :
 - command ( str ) : The bash command string to be executed .
 - blocksize ( int ) : Blocksize to be requested
 - tasks _ per _ node ( int ) : command invocations to be launched per node
 KWargs :
 - job _ name ( str ) : Human friendly name to be assigned to the job request
 Returns :
 - A job identifier , this could be an integer , string etc
 Raises :
 - ExecutionProviderException or its subclasses'''

implementation:
wrapped_cmd = self . launcher ( command , tasks_per_node , 1 )
instance , name = self . create_instance ( command = wrapped_cmd )
self . provisioned_blocks += 1
self . resources [ name ] = { "job_id" : name , "status" : translate_table [ instance [ 'status' ] ] }
return name


signature:
def status ( self , job_ids ) :
    '''Get the status of a list of jobs identified by the job identifiers
 returned from the submit request .
 Args :
 - job _ ids ( list ) : A list of job identifiers
 Returns :
 - A list of status from [ ' PENDING ' , ' RUNNING ' , ' CANCELLED ' , ' COMPLETED ' ,
 ' FAILED ' , ' TIMEOUT ' ] corresponding to each job _ id in the job _ ids list .
 Raises :
 - ExecutionProviderException or its subclasses'''

implementation:
statuses = [ ]
for job_id in job_ids :
    instance = self . client . instances ( ) . get ( instance = job_id , project = self . project_id , zone = self . zone ) . execute ( )
    self . resources [ job_id ] [ 'status' ] = translate_table [ instance [ 'status' ] ]
    statuses . append ( translate_table [ instance [ 'status' ] ] )
return statuses


signature:
def cancel ( self , job_ids ) :
    '''Cancels the resources identified by the job _ ids provided by the user .
 Args :
 - job _ ids ( list ) : A list of job identifiers
 Returns :
 - A list of status from cancelling the job which can be True , False
 Raises :
 - ExecutionProviderException or its subclasses'''

implementation:
statuses = [ ]
for job_id in job_ids :
    try :
        self . delete_instance ( job_id )
        statuses . append ( True )
        self . provisioned_blocks -= 1
    except Exception :
        statuses . append ( False )
return statuses


signature:
def _queue_management_worker ( self ) :
    """Listen to the queue for task status messages and handle them .
 Depending on the message , tasks will be updated with results , exceptions ,
 or updates . It expects the following messages :
 . . code : : python
 " task _ id " : < task _ id >
 " result " : serialized result object , if task succeeded
 . . . more tags could be added later
 " task _ id " : < task _ id >
 " exception " : serialized exception object , on failure
 We do not support these yet , but they could be added easily .
 . . code : : python
 " task _ id " : < task _ id >
 " cpu _ stat " : < >
 " mem _ stat " : < >
 " io _ stat " : < >
 " started " : tstamp
 The ` None ` message is a die request ."""

implementation:
while True :
    logger . debug ( "[MTHREAD] Management thread active" )
    try :
        msg = self . incoming_q . get ( block = True , timeout = 1 )
    except queue . Empty : # Timed out .
        pass
    except IOError as e :
        logger . debug ( "[MTHREAD] Caught broken queue with exception code {}: {}" . format ( e . errno , e ) )
        return
    except Exception as e :
        logger . debug ( "[MTHREAD] Caught unknown exception: {}" . format ( e ) )
    else :
        if msg is None :
            logger . debug ( "[MTHREAD] Got None" )
            return
        else :
            logger . debug ( "[MTHREAD] Received message: {}" . format ( msg ) )
            task_fut = self . tasks [ msg [ 'task_id' ] ]
            if 'result' in msg :
                result , _ = deserialize_object ( msg [ 'result' ] )
                task_fut . set_result ( result )
            elif 'exception' in msg :
                exception , _ = deserialize_object ( msg [ 'exception' ] )
                task_fut . set_exception ( exception )
    if not self . is_alive :
        break


signature:
def shutdown ( self ) :
    """Shutdown method , to kill the threads and workers ."""

implementation:
self . is_alive = False
logging . debug ( "Waking management thread" )
self . incoming_q . put ( None )
# Wake up the thread
self . _queue_management_thread . join ( )
# Force join
logging . debug ( "Exiting thread" )
self . worker . join ( )
return True


signature:
def submit ( self , func , * args , ** kwargs ) :
    """Submits work to the the outgoing _ q .
 The outgoing _ q is an external process listens on this
 queue for new work . This method is simply pass through and behaves like a
 submit call as described here ` Python docs : < https : / / docs . python . org / 3 / library / concurrent . futures . html # concurrent . futures . ThreadPoolExecutor > ` _
 Args :
 - func ( callable ) : Callable function
 - * args ( list ) : List of arbitrary positional arguments .
 Kwargs :
 - * * kwargs ( dict ) : A dictionary of arbitrary keyword args for func .
 Returns :
 Future"""

implementation:
task_id = uuid . uuid4 ( )
logger . debug ( "Pushing function {} to queue with args {}" . format ( func , args ) )
self . tasks [ task_id ] = Future ( )
fn_buf = pack_apply_message ( func , args , kwargs , buffer_threshold = 1024 * 1024 , item_threshold = 1024 )
msg = { "task_id" : task_id , "buffer" : fn_buf }
# Post task to the the outgoing queue
self . outgoing_q . put ( msg )
# Return the future
return self . tasks [ task_id ]


signature:
def filepath ( self ) :
    """Return the resolved filepath on the side where it is called from .
 The appropriate filepath will be returned when called from within
 an app running remotely as well as regular python on the client side .
 Args :
 - self
 Returns :
 - filepath ( string )"""

implementation:
if hasattr ( self , 'local_path' ) :
    return self . local_path
if self . scheme in [ 'ftp' , 'http' , 'https' , 'globus' ] :
    return self . filename
elif self . scheme in [ 'file' ] :
    return self . path
else :
    raise Exception ( 'Cannot return filepath for unknown scheme {}' . format ( self . scheme ) )


signature:
def execute_wait ( self , cmd , walltime = None , envs = { } ) :
    '''Synchronously execute a commandline string on the shell .
 Args :
 - cmd ( string ) : Commandline string to execute
 - walltime ( int ) : walltime in seconds , this is not really used now .
 Kwargs :
 - envs ( dict ) : Dictionary of env variables . This will be used
 to override the envs set at channel initialization .
 Returns :
 - retcode : Return code from the execution , - 1 on fail
 - stdout : stdout string
 - stderr : stderr string
 Raises :
 None .'''

implementation:
retcode = - 1
stdout = None
stderr = None
current_env = copy . deepcopy ( self . _envs )
current_env . update ( envs )
try :
    proc = subprocess . Popen ( cmd , stdout = subprocess . PIPE , stderr = subprocess . PIPE , cwd = self . userhome , env = current_env , shell = True )
    proc . wait ( timeout = walltime )
    stdout = proc . stdout . read ( )
    stderr = proc . stderr . read ( )
    retcode = proc . returncode
except Exception as e :
    print ( "Caught exception: {0}" . format ( e ) )
    logger . warn ( "Execution of command [%s] failed due to \n %s " , cmd , e )
    # Set retcode to non - zero so that this can be handled in the provider .
    if retcode == 0 :
        retcode = - 1
    return ( retcode , None , None )
return ( retcode , stdout . decode ( "utf-8" ) , stderr . decode ( "utf-8" ) )


signature:
def push_file ( self , source , dest_dir ) :
    '''If the source files dirpath is the same as dest _ dir , a copy
 is not necessary , and nothing is done . Else a copy is made .
 Args :
 - source ( string ) : Path to the source file
 - dest _ dir ( string ) : Path to the directory to which the files is to be copied
 Returns :
 - destination _ path ( String ) : Absolute path of the destination file
 Raises :
 - FileCopyException : If file copy failed .'''

implementation:
local_dest = dest_dir + '/' + os . path . basename ( source )
# Only attempt to copy if the target dir and source dir are different
if os . path . dirname ( source ) != dest_dir :
    try :
        shutil . copyfile ( source , local_dest )
        os . chmod ( local_dest , 0o777 )
    except OSError as e :
        raise FileCopyException ( e , self . hostname )
return local_dest


signature:
def App ( apptype , data_flow_kernel = None , walltime = 60 , cache = False , executors = 'all' ) :
    """The App decorator function .
 Args :
 - apptype ( string ) : Apptype can be bash | python
 Kwargs :
 - data _ flow _ kernel ( DataFlowKernel ) : The : class : ` ~ parsl . dataflow . dflow . DataFlowKernel ` responsible for
 managing this app . This can be omitted only
 after calling : meth : ` parsl . dataflow . dflow . DataFlowKernelLoader . load ` .
 - walltime ( int ) : Walltime for app in seconds ,
 default = 60
 - executors ( str | list ) : Labels of the executors that this app can execute over . Default is ' all ' .
 - cache ( Bool ) : Enable caching of the app call
 default = False
 Returns :
 A PythonApp or BashApp object , which when called runs the apps through the executor ."""

implementation:
from parsl . app . python import PythonApp
from parsl . app . bash import BashApp
logger . warning ( "The 'App' decorator will be deprecated in Parsl 0.8. Please use 'python_app' or 'bash_app' instead." )
if apptype == 'python' :
    app_class = PythonApp
elif apptype == 'bash' :
    app_class = BashApp
else :
    raise InvalidAppTypeError ( "Invalid apptype requested {}; must be 'python' or 'bash'" . format ( apptype ) )
def wrapper ( f ) :
    return app_class ( f , data_flow_kernel = data_flow_kernel , walltime = walltime , cache = cache , executors = executors )
return wrapper


signature:
def python_app ( function = None , data_flow_kernel = None , walltime = 60 , cache = False , executors = 'all' ) :
    """Decorator function for making python apps .
 Parameters
 function : function
 Do not pass this keyword argument directly . This is needed in order to allow for omitted parenthesis ,
 for example , ` @ python _ app ` if using all defaults or ` @ python _ app ( walltime = 120 ) ` . If the
 decorator is used alone , function will be the actual function being decorated , whereas if it
 is called with arguments , function will be None . Default is None .
 data _ flow _ kernel : DataFlowKernel
 The : class : ` ~ parsl . dataflow . dflow . DataFlowKernel ` responsible for managing this app . This can
 be omitted only after calling : meth : ` parsl . dataflow . dflow . DataFlowKernelLoader . load ` . Default is None .
 walltime : int
 Walltime for app in seconds . Default is 60.
 executors : string or list
 Labels of the executors that this app can execute over . Default is ' all ' .
 cache : bool
 Enable caching of the app call . Default is False ."""

implementation:
from parsl . app . python import PythonApp
def decorator ( func ) :
    def wrapper ( f ) :
        return PythonApp ( f , data_flow_kernel = data_flow_kernel , walltime = walltime , cache = cache , executors = executors )
    return wrapper ( func )
if function is not None :
    return decorator ( function )
return decorator


signature:
def bash_app ( function = None , data_flow_kernel = None , walltime = 60 , cache = False , executors = 'all' ) :
    """Decorator function for making bash apps .
 Parameters
 function : function
 Do not pass this keyword argument directly . This is needed in order to allow for omitted parenthesis ,
 for example , ` @ bash _ app ` if using all defaults or ` @ bash _ app ( walltime = 120 ) ` . If the
 decorator is used alone , function will be the actual function being decorated , whereas if it
 is called with arguments , function will be None . Default is None .
 data _ flow _ kernel : DataFlowKernel
 The : class : ` ~ parsl . dataflow . dflow . DataFlowKernel ` responsible for managing this app . This can
 be omitted only after calling : meth : ` parsl . dataflow . dflow . DataFlowKernelLoader . load ` . Default is None .
 walltime : int
 Walltime for app in seconds . Default is 60.
 executors : string or list
 Labels of the executors that this app can execute over . Default is ' all ' .
 cache : bool
 Enable caching of the app call . Default is False ."""

implementation:
from parsl . app . bash import BashApp
def decorator ( func ) :
    def wrapper ( f ) :
        return BashApp ( f , data_flow_kernel = data_flow_kernel , walltime = walltime , cache = cache , executors = executors )
    return wrapper ( func )
if function is not None :
    return decorator ( function )
return decorator


signature:
def make_rundir ( path ) :
    """When a path has not been specified , make the run directory .
 Creates a rundir with the following hierarchy :
 . / runinfo < - Home of all run directories
 | - - - - 000
 | - - - - 001 < - Directories for each run
 | - - - - NNN
 Kwargs :
 - path ( str ) : String path to a specific run dir
 Default : None ."""

implementation:
try :
    if not os . path . exists ( path ) :
        os . makedirs ( path )
    prev_rundirs = glob ( os . path . join ( path , "[0-9]*" ) )
    current_rundir = os . path . join ( path , '000' )
    if prev_rundirs : # Since we globbed on files named as 0-9
        x = sorted ( [ int ( os . path . basename ( x ) ) for x in prev_rundirs ] ) [ - 1 ]
        current_rundir = os . path . join ( path , '{0:03}' . format ( x + 1 ) )
    os . makedirs ( current_rundir )
    logger . debug ( "Parsl run initializing in rundir: {0}" . format ( current_rundir ) )
    return os . path . abspath ( current_rundir )
except Exception as e :
    logger . error ( "Failed to create a run directory" )
    logger . error ( "Error: {0}" . format ( e ) )
    raise


signature:
def put ( self , task_id , buffer ) :
    """TODO : docstring"""

implementation:
task_id_bytes = task_id . to_bytes ( 4 , "little" )
message = [ b"" , task_id_bytes ] + buffer
self . zmq_socket . send_multipart ( message )
logger . debug ( "Sent task {}" . format ( task_id ) )


signature:
def monitor ( pid , task_id , monitoring_hub_url , run_id , sleep_dur = 10 ) :
    """Internal
 Monitors the Parsl task ' s resources by pointing psutil to the task ' s pid and watching it and its children ."""

implementation:
import psutil
radio = UDPRadio ( monitoring_hub_url , source_id = task_id )
# these values are simple to log . Other information is available in special formats such as memory below .
simple = [ "cpu_num" , 'cpu_percent' , 'create_time' , 'cwd' , 'exe' , 'memory_percent' , 'nice' , 'name' , 'num_threads' , 'pid' , 'ppid' , 'status' , 'username' ]
# values that can be summed up to see total resources used by task process and its children
summable_values = [ 'cpu_percent' , 'memory_percent' , 'num_threads' ]
pm = psutil . Process ( pid )
pm . cpu_percent ( )
first_msg = True
while True :
    try :
        d = { "psutil_process_" + str ( k ) : v for k , v in pm . as_dict ( ) . items ( ) if k in simple }
        d [ "run_id" ] = run_id
        d [ "task_id" ] = task_id
        d [ 'resource_monitoring_interval' ] = sleep_dur
        d [ 'first_msg' ] = first_msg
        d [ 'timestamp' ] = datetime . datetime . now ( )
        children = pm . children ( recursive = True )
        d [ "psutil_cpu_count" ] = psutil . cpu_count ( )
        d [ 'psutil_process_memory_virtual' ] = pm . memory_info ( ) . vms
        d [ 'psutil_process_memory_resident' ] = pm . memory_info ( ) . rss
        d [ 'psutil_process_time_user' ] = pm . cpu_times ( ) . user
        d [ 'psutil_process_time_system' ] = pm . cpu_times ( ) . system
        d [ 'psutil_process_children_count' ] = len ( children )
        try :
            d [ 'psutil_process_disk_write' ] = pm . io_counters ( ) . write_bytes
            d [ 'psutil_process_disk_read' ] = pm . io_counters ( ) . read_bytes
        except psutil . _exceptions . AccessDenied : # occassionally pid temp files that hold this information are unvailable to be read so set to zero
            d [ 'psutil_process_disk_write' ] = 0
            d [ 'psutil_process_disk_read' ] = 0
        for child in children :
            for k , v in child . as_dict ( attrs = summable_values ) . items ( ) :
                d [ 'psutil_process_' + str ( k ) ] += v
            d [ 'psutil_process_time_user' ] += child . cpu_times ( ) . user
            d [ 'psutil_process_time_system' ] += child . cpu_times ( ) . system
            d [ 'psutil_process_memory_virtual' ] += child . memory_info ( ) . vms
            d [ 'psutil_process_memory_resident' ] += child . memory_info ( ) . rss
            try :
                d [ 'psutil_process_disk_write' ] += child . io_counters ( ) . write_bytes
                d [ 'psutil_process_disk_read' ] += child . io_counters ( ) . read_bytes
            except psutil . _exceptions . AccessDenied : # occassionally pid temp files that hold this information are unvailable to be read so add zero
                d [ 'psutil_process_disk_write' ] += 0
                d [ 'psutil_process_disk_read' ] += 0
    finally :
        radio . send ( MessageType . TASK_INFO , task_id , d )
        time . sleep ( sleep_dur )
        first_msg = False


signature:
def send ( self , message_type , task_id , message ) :
    """Sends a message to the UDP receiver
 Parameter
 message _ type : monitoring . MessageType ( enum )
 In this case message type is RESOURCE _ INFO most often
 task _ id : int
 Task identifier of the task for which resource monitoring is being reported
 message : object
 Arbitrary pickle - able object that is to be sent
 Returns :
 # bytes sent"""

implementation:
x = 0
try :
    buffer = pickle . dumps ( ( self . source_id , # Identifier for manager
    int ( time . time ( ) ) , # epoch timestamp
    message_type , message ) )
except Exception as e :
    print ( "Exception during pickling {}" . format ( e ) )
    return
try :
    x = self . sock . sendto ( buffer , ( self . ip , self . port ) )
except socket . timeout :
    print ( "Could not send message within timeout limit" )
    return False
return x


signature:
def monitor_wrapper ( f , task_id , monitoring_hub_url , run_id , sleep_dur ) :
    """Internal
 Wrap the Parsl app with a function that will call the monitor function and point it at the correct pid when the task begins ."""

implementation:
def wrapped ( * args , ** kwargs ) :
    p = Process ( target = monitor , args = ( os . getpid ( ) , task_id , monitoring_hub_url , run_id , sleep_dur ) )
    p . start ( )
    try :
        return f ( * args , ** kwargs )
    finally : # There ' s a chance of zombification if the workers are killed by some signals
        p . terminate ( )
        p . join ( )
return wrapped


signature:
def execute_wait ( self , cmd , walltime = 2 , envs = { } ) :
    '''Synchronously execute a commandline string on the shell .
 Args :
 - cmd ( string ) : Commandline string to execute
 - walltime ( int ) : walltime in seconds
 Kwargs :
 - envs ( dict ) : Dictionary of env variables
 Returns :
 - retcode : Return code from the execution , - 1 on fail
 - stdout : stdout string
 - stderr : stderr string
 Raises :
 None .'''

implementation:
# Execute the command
stdin , stdout , stderr = self . ssh_client . exec_command ( self . prepend_envs ( cmd , envs ) , bufsize = - 1 , timeout = walltime )
# Block on exit status from the command
exit_status = stdout . channel . recv_exit_status ( )
return exit_status , stdout . read ( ) . decode ( "utf-8" ) , stderr . read ( ) . decode ( "utf-8" )


signature:
def execute_no_wait ( self , cmd , walltime = 2 , envs = { } ) :
    '''Execute asynchronousely without waiting for exitcode
 Args :
 - cmd ( string ) : Commandline string to be executed on the remote side
 - walltime ( int ) : timeout to exec _ command
 KWargs :
 - envs ( dict ) : A dictionary of env variables
 Returns :
 - None , stdout ( readable stream ) , stderr ( readable stream )
 Raises :
 - ChannelExecFailed ( reason )'''

implementation:
# Execute the command
stdin , stdout , stderr = self . ssh_client . exec_command ( self . prepend_envs ( cmd , envs ) , bufsize = - 1 , timeout = walltime )
return None , stdout , stderr


signature:
def push_file ( self , local_source , remote_dir ) :
    '''Transport a local file to a directory on a remote machine
 Args :
 - local _ source ( string ) : Path
 - remote _ dir ( string ) : Remote path
 Returns :
 - str : Path to copied file on remote machine
 Raises :
 - BadScriptPath : if script path on the remote side is bad
 - BadPermsScriptPath : You do not have perms to make the channel script dir
 - FileCopyException : FileCopy failed .'''

implementation:
remote_dest = remote_dir + '/' + os . path . basename ( local_source )
try :
    self . makedirs ( remote_dir , exist_ok = True )
except IOError as e :
    logger . exception ( "Pushing {0} to {1} failed" . format ( local_source , remote_dir ) )
    if e . errno == 2 :
        raise BadScriptPath ( e , self . hostname )
    elif e . errno == 13 :
        raise BadPermsScriptPath ( e , self . hostname )
    else :
        logger . exception ( "File push failed due to SFTP client failure" )
        raise FileCopyException ( e , self . hostname )
try :
    self . sftp_client . put ( local_source , remote_dest , confirm = True )
    # Set perm because some systems require the script to be executable
    self . sftp_client . chmod ( remote_dest , 0o777 )
except Exception as e :
    logger . exception ( "File push from local source {} to remote destination {} failed" . format ( local_source , remote_dest ) )
    raise FileCopyException ( e , self . hostname )
return remote_dest


signature:
def pull_file ( self , remote_source , local_dir ) :
    '''Transport file on the remote side to a local directory
 Args :
 - remote _ source ( string ) : remote _ source
 - local _ dir ( string ) : Local directory to copy to
 Returns :
 - str : Local path to file
 Raises :
 - FileExists : Name collision at local directory .
 - FileCopyException : FileCopy failed .'''

implementation:
local_dest = local_dir + '/' + os . path . basename ( remote_source )
try :
    os . makedirs ( local_dir )
except OSError as e :
    if e . errno != errno . EEXIST :
        logger . exception ( "Failed to create script_dir: {0}" . format ( script_dir ) )
        raise BadScriptPath ( e , self . hostname )
# Easier to check this than to waste time trying to pull file and
# realize there ' s a problem .
if os . path . exists ( local_dest ) :
    logger . exception ( "Remote file copy will overwrite a local file:{0}" . format ( local_dest ) )
    raise FileExists ( None , self . hostname , filename = local_dest )
try :
    self . sftp_client . get ( remote_source , local_dest )
except Exception as e :
    logger . exception ( "File pull failed" )
    raise FileCopyException ( e , self . hostname )
return local_dest


signature:
def isdir ( self , path ) :
    """Return true if the path refers to an existing directory .
 Parameters
 path : str
 Path of directory on the remote side to check ."""

implementation:
result = True
try :
    self . sftp_client . lstat ( path )
except FileNotFoundError :
    result = False
return result


signature:
def makedirs ( self , path , mode = 511 , exist_ok = False ) :
    """Create a directory on the remote side .
 If intermediate directories do not exist , they will be created .
 Parameters
 path : str
 Path of directory on the remote side to create .
 mode : int
 Permissions ( posix - style ) for the newly - created directory .
 exist _ ok : bool
 If False , raise an OSError if the target directory already exists ."""

implementation:
if exist_ok is False and self . isdir ( path ) :
    raise OSError ( 'Target directory {} already exists' . format ( path ) )
self . execute_wait ( 'mkdir -p {}' . format ( path ) )
self . sftp_client . chmod ( path , mode )


signature:
def run ( self , message ) :
    """This function needs to be fast at the same time aware of the possibility of
 ZMQ pipes overflowing .
 The timeout increases slowly if contention is detected on ZMQ pipes .
 We could set copy = False and get slightly better latency but this results
 in ZMQ sockets reaching a broken state once there are ~ 10k tasks in flight .
 This issue can be magnified if each the serialized buffer itself is larger ."""

implementation:
self . zmq_socket . send_pyobj ( message , copy = True )
reply = self . zmq_socket . recv_pyobj ( )
return reply


signature:
def put ( self , message ) :
    """This function needs to be fast at the same time aware of the possibility of
 ZMQ pipes overflowing .
 The timeout increases slowly if contention is detected on ZMQ pipes .
 We could set copy = False and get slightly better latency but this results
 in ZMQ sockets reaching a broken state once there are ~ 10k tasks in flight .
 This issue can be magnified if each the serialized buffer itself is larger ."""

implementation:
timeout_ms = 0
while True :
    socks = dict ( self . poller . poll ( timeout = timeout_ms ) )
    if self . zmq_socket in socks and socks [ self . zmq_socket ] == zmq . POLLOUT : # The copy option adds latency but reduces the risk of ZMQ overflow
        self . zmq_socket . send_pyobj ( message , copy = True )
        return
    else :
        timeout_ms += 1
        logger . debug ( "Not sending due to full zmq pipe, timeout: {} ms" . format ( timeout_ms ) )


signature:
def _wake_up_timer ( self , kill_event ) :
    """Internal . This is the function that the thread will execute .
 waits on an event so that the thread can make a quick exit when close ( ) is called
 Args :
 - kill _ event ( threading . Event ) : Event to wait on"""

implementation:
while True :
    prev = self . _wake_up_time
    # Waiting for the event returns True only when the event
    # is set , usually by the parent thread
    time_to_die = kill_event . wait ( float ( max ( prev - time . time ( ) , 0 ) ) )
    if time_to_die :
        return
    if prev == self . _wake_up_time :
        self . make_callback ( kind = 'timer' )
    else :
        print ( "Sleeping a bit more" )


signature:
def notify ( self , event_id ) :
    """Let the FlowControl system know that there is an event ."""

implementation:
self . _event_buffer . extend ( [ event_id ] )
self . _event_count += 1
if self . _event_count >= self . threshold :
    logger . debug ( "Eventcount >= threshold" )
    self . make_callback ( kind = "event" )


signature:
def make_callback ( self , kind = None ) :
    """Makes the callback and resets the timer .
 KWargs :
 - kind ( str ) : Default = None , used to pass information on what
 triggered the callback"""

implementation:
self . _wake_up_time = time . time ( ) + self . interval
self . callback ( tasks = self . _event_buffer , kind = kind )
self . _event_buffer = [ ]


signature:
def make_callback ( self , kind = None ) :
    """Makes the callback and resets the timer ."""

implementation:
self . _wake_up_time = time . time ( ) + self . interval
self . callback ( * self . cb_args )


signature:
def address_by_interface ( ifname ) :
    """Returns the IP address of the given interface name , e . g . ' eth0'
 Parameters
 ifname : str
 Name of the interface whose address is to be returned . Required .
 Taken from this Stack Overflow answer : https : / / stackoverflow . com / questions / 24196932 / how - can - i - get - the - ip - address - of - eth0 - in - python # 24196955"""

implementation:
s = socket . socket ( socket . AF_INET , socket . SOCK_DGRAM )
return socket . inet_ntoa ( fcntl . ioctl ( s . fileno ( ) , 0x8915 , # SIOCGIFADDR
struct . pack ( '256s' , bytes ( ifname [ : 15 ] , 'utf-8' ) ) ) [ 20 : 24 ] )


signature:
def submit ( self , cmd_string , blocksize , tasks_per_node , job_name = "parsl" ) :
    """Submit a job
 Args :
 - cmd _ string : ( String ) - Name of the container to initiate
 - blocksize : ( float ) - Number of replicas
 - tasks _ per _ node ( int ) : command invocations to be launched per node
 Kwargs :
 - job _ name ( String ) : Name for job , must be unique
 Returns :
 - None : At capacity , cannot provision more
 - job _ id : ( string ) Identifier for the job"""

implementation:
if not self . resources :
    cur_timestamp = str ( time . time ( ) * 1000 ) . split ( "." ) [ 0 ]
    job_name = "{0}-{1}" . format ( job_name , cur_timestamp )
    if not self . deployment_name :
        deployment_name = '{}-deployment' . format ( job_name )
    else :
        deployment_name = '{}-{}-deployment' . format ( self . deployment_name , cur_timestamp )
    formatted_cmd = template_string . format ( command = cmd_string , worker_init = self . worker_init )
    self . deployment_obj = self . _create_deployment_object ( job_name , self . image , deployment_name , cmd_string = formatted_cmd , replicas = self . init_blocks , volumes = self . persistent_volumes )
    logger . debug ( "Deployment name :{}" . format ( deployment_name ) )
    self . _create_deployment ( self . deployment_obj )
    self . resources [ deployment_name ] = { 'status' : 'RUNNING' , 'pods' : self . init_blocks }
return deployment_name


signature:
def _create_deployment_object ( self , job_name , job_image , deployment_name , port = 80 , replicas = 1 , cmd_string = None , engine_json_file = '~/.ipython/profile_default/security/ipcontroller-engine.json' , engine_dir = '.' , volumes = [ ] ) :
    """Create a kubernetes deployment for the job .
 Args :
 - job _ name ( string ) : Name of the job and deployment
 - job _ image ( string ) : Docker image to launch
 KWargs :
 - port ( integer ) : Container port
 - replicas : Number of replica containers to maintain
 Returns :
 - True : The deployment object to launch"""

implementation:
# sorry , quick hack that doesn ' t pass this stuff through to test it works .
# TODO it also doesn ' t only add what is set : (
security_context = None
if self . user_id and self . group_id :
    security_context = client . V1SecurityContext ( run_as_group = self . group_id , run_as_user = self . user_id , run_as_non_root = self . run_as_non_root )
# Create the enviornment variables and command to initiate IPP
environment_vars = client . V1EnvVar ( name = "TEST" , value = "SOME DATA" )
launch_args = [ "-c" , "{0}; /app/deploy.sh;" . format ( cmd_string ) ]
volume_mounts = [ ]
# Create mount paths for the volumes
for volume in volumes :
    volume_mounts . append ( client . V1VolumeMount ( mount_path = volume [ 1 ] , name = volume [ 0 ] ) )
# Configureate Pod template container
container = None
if security_context :
    container = client . V1Container ( name = job_name , image = job_image , ports = [ client . V1ContainerPort ( container_port = port ) ] , volume_mounts = volume_mounts , command = [ '/bin/bash' ] , args = launch_args , env = [ environment_vars ] , security_context = security_context )
else :
    container = client . V1Container ( name = job_name , image = job_image , ports = [ client . V1ContainerPort ( container_port = port ) ] , volume_mounts = volume_mounts , command = [ '/bin/bash' ] , args = launch_args , env = [ environment_vars ] )
# Create a secret to enable pulling images from secure repositories
secret = None
if self . secret :
    secret = client . V1LocalObjectReference ( name = self . secret )
# Create list of volumes from ( pvc , mount ) tuples
volume_defs = [ ]
for volume in volumes :
    volume_defs . append ( client . V1Volume ( name = volume [ 0 ] , persistent_volume_claim = client . V1PersistentVolumeClaimVolumeSource ( claim_name = volume [ 0 ] ) ) )
# Create and configurate a spec section
template = client . V1PodTemplateSpec ( metadata = client . V1ObjectMeta ( labels = { "app" : job_name } ) , spec = client . V1PodSpec ( containers = [ container ] , image_pull_secrets = [ secret ] , volumes = volume_defs ) )
# Create the specification of deployment
spec = client . ExtensionsV1beta1DeploymentSpec ( replicas = replicas , template = template )
# Instantiate the deployment object
deployment = client . ExtensionsV1beta1Deployment ( api_version = "extensions/v1beta1" , kind = "Deployment" , metadata = client . V1ObjectMeta ( name = deployment_name ) , spec = spec )
return deployment


signature:
def _create_deployment ( self , deployment ) :
    """Create the kubernetes deployment"""

implementation:
api_response = self . kube_client . create_namespaced_deployment ( body = deployment , namespace = self . namespace )
logger . debug ( "Deployment created. status='{0}'" . format ( str ( api_response . status ) ) )


signature:
def _delete_deployment ( self , deployment_name ) :
    """Delete deployment"""

implementation:
api_response = self . kube_client . delete_namespaced_deployment ( name = deployment_name , namespace = self . namespace , body = client . V1DeleteOptions ( propagation_policy = 'Foreground' , grace_period_seconds = 5 ) )
logger . debug ( "Deployment deleted. status='{0}'" . format ( str ( api_response . status ) ) )


signature:
def initialize_scaling ( self ) :
    """Compose the launch command and call the scale _ out
 This should be implemented in the child classes to take care of
 executor specific oddities ."""

implementation:
debug_opts = "--debug" if self . worker_debug else ""
max_workers = "" if self . max_workers == float ( 'inf' ) else "--max_workers={}" . format ( self . max_workers )
worker_logdir = "{}/{}" . format ( self . run_dir , self . label )
if self . worker_logdir_root is not None :
    worker_logdir = "{}/{}" . format ( self . worker_logdir_root , self . label )
l_cmd = self . launch_cmd . format ( debug = debug_opts , prefetch_capacity = self . prefetch_capacity , task_url = self . worker_task_url , result_url = self . worker_result_url , cores_per_worker = self . cores_per_worker , max_workers = max_workers , nodes_per_block = self . provider . nodes_per_block , heartbeat_period = self . heartbeat_period , heartbeat_threshold = self . heartbeat_threshold , poll_period = self . poll_period , logdir = worker_logdir )
self . launch_cmd = l_cmd
logger . debug ( "Launch command: {}" . format ( self . launch_cmd ) )
self . _scaling_enabled = self . provider . scaling_enabled
logger . debug ( "Starting HighThroughputExecutor with provider:\n%s" , self . provider )
if hasattr ( self . provider , 'init_blocks' ) :
    try :
        self . scale_out ( blocks = self . provider . init_blocks )
    except Exception as e :
        logger . error ( "Scaling out failed: {}" . format ( e ) )
        raise e


signature:
def start ( self ) :
    """Create the Interchange process and connect to it ."""

implementation:
self . outgoing_q = zmq_pipes . TasksOutgoing ( "127.0.0.1" , self . interchange_port_range )
self . incoming_q = zmq_pipes . ResultsIncoming ( "127.0.0.1" , self . interchange_port_range )
self . command_client = zmq_pipes . CommandClient ( "127.0.0.1" , self . interchange_port_range )
self . is_alive = True
self . _executor_bad_state = threading . Event ( )
self . _executor_exception = None
self . _queue_management_thread = None
self . _start_queue_management_thread ( )
self . _start_local_queue_process ( )
logger . debug ( "Created management thread: {}" . format ( self . _queue_management_thread ) )
if self . provider :
    self . initialize_scaling ( )
else :
    self . _scaling_enabled = False
    logger . debug ( "Starting HighThroughputExecutor with no provider" )


signature:
def _queue_management_worker ( self ) :
    """Listen to the queue for task status messages and handle them .
 Depending on the message , tasks will be updated with results , exceptions ,
 or updates . It expects the following messages :
 . . code : : python
 " task _ id " : < task _ id >
 " result " : serialized result object , if task succeeded
 . . . more tags could be added later
 " task _ id " : < task _ id >
 " exception " : serialized exception object , on failure
 We do not support these yet , but they could be added easily .
 . . code : : python
 " task _ id " : < task _ id >
 " cpu _ stat " : < >
 " mem _ stat " : < >
 " io _ stat " : < >
 " started " : tstamp
 The ` None ` message is a die request ."""

implementation:
logger . debug ( "[MTHREAD] queue management worker starting" )
while not self . _executor_bad_state . is_set ( ) :
    try :
        msgs = self . incoming_q . get ( timeout = 1 )
        # logger . debug ( " [ MTHREAD ] get has returned { } " . format ( len ( msgs ) ) )
    except queue . Empty :
        logger . debug ( "[MTHREAD] queue empty" )
        # Timed out .
        pass
    except IOError as e :
        logger . exception ( "[MTHREAD] Caught broken queue with exception code {}: {}" . format ( e . errno , e ) )
        return
    except Exception as e :
        logger . exception ( "[MTHREAD] Caught unknown exception: {}" . format ( e ) )
        return
    else :
        if msgs is None :
            logger . debug ( "[MTHREAD] Got None, exiting" )
            return
        else :
            for serialized_msg in msgs :
                try :
                    msg = pickle . loads ( serialized_msg )
                    tid = msg [ 'task_id' ]
                except pickle . UnpicklingError :
                    raise BadMessage ( "Message received could not be unpickled" )
                except Exception :
                    raise BadMessage ( "Message received does not contain 'task_id' field" )
                if tid == - 1 and 'exception' in msg :
                    logger . warning ( "Executor shutting down due to exception from interchange" )
                    self . _executor_exception , _ = deserialize_object ( msg [ 'exception' ] )
                    logger . exception ( "Exception: {}" . format ( self . _executor_exception ) )
                    # Set bad state to prevent new tasks from being submitted
                    self . _executor_bad_state . set ( )
                    # We set all current tasks to this exception to make sure that
                    # this is raised in the main context .
                    for task in self . tasks :
                        self . tasks [ task ] . set_exception ( self . _executor_exception )
                    break
                task_fut = self . tasks [ tid ]
                if 'result' in msg :
                    result , _ = deserialize_object ( msg [ 'result' ] )
                    task_fut . set_result ( result )
                elif 'exception' in msg :
                    try :
                        s , _ = deserialize_object ( msg [ 'exception' ] )
                        # s should be a RemoteExceptionWrapper . . . so we can reraise it
                        try :
                            s . reraise ( )
                        except Exception as e :
                            task_fut . set_exception ( e )
                    except Exception as e : # TODO could be a proper wrapped exception ?
                        task_fut . set_exception ( DeserializationError ( "Received exception, but handling also threw an exception: {}" . format ( e ) ) )
                else :
                    raise BadMessage ( "Message received is neither result or exception" )
    if not self . is_alive :
        break
logger . info ( "[MTHREAD] queue management worker finished" )


signature:
def _start_local_queue_process ( self ) :
    """Starts the interchange process locally
 Starts the interchange process locally and uses an internal command queue to
 get the worker task and result ports that the interchange has bound to ."""

implementation:
comm_q = Queue ( maxsize = 10 )
self . queue_proc = Process ( target = interchange . starter , args = ( comm_q , ) , kwargs = { "client_ports" : ( self . outgoing_q . port , self . incoming_q . port , self . command_client . port ) , "worker_ports" : self . worker_ports , "worker_port_range" : self . worker_port_range , "logdir" : "{}/{}" . format ( self . run_dir , self . label ) , "suppress_failure" : self . suppress_failure , "heartbeat_threshold" : self . heartbeat_threshold , "poll_period" : self . poll_period , "logging_level" : logging . DEBUG if self . worker_debug else logging . INFO } , )
self . queue_proc . start ( )
try :
    ( worker_task_port , worker_result_port ) = comm_q . get ( block = True , timeout = 120 )
except queue . Empty :
    logger . error ( "Interchange has not completed initialization in 120s. Aborting" )
    raise Exception ( "Interchange failed to start" )
self . worker_task_url = "tcp://{}:{}" . format ( self . address , worker_task_port )
self . worker_result_url = "tcp://{}:{}" . format ( self . address , worker_result_port )


signature:
def hold_worker ( self , worker_id ) :
    """Puts a worker on hold , preventing scheduling of additional tasks to it .
 This is called " hold " mostly because this only stops scheduling of tasks ,
 and does not actually kill the worker .
 Parameters
 worker _ id : str
 Worker id to be put on hold"""

implementation:
c = self . command_client . run ( "HOLD_WORKER;{}" . format ( worker_id ) )
logger . debug ( "Sent hold request to worker: {}" . format ( worker_id ) )
return c


signature:
def _hold_block ( self , block_id ) :
    """Sends hold command to all managers which are in a specific block
 Parameters
 block _ id : str
 Block identifier of the block to be put on hold"""

implementation:
managers = self . connected_managers
for manager in managers :
    if manager [ 'block_id' ] == block_id :
        logger . debug ( "[HOLD_BLOCK]: Sending hold to manager:{}" . format ( manager [ 'manager' ] ) )
        self . hold_worker ( manager [ 'manager' ] )


signature:
def scale_out ( self , blocks = 1 ) :
    """Scales out the number of blocks by " blocks "
 Raises :
 NotImplementedError"""

implementation:
r = [ ]
for i in range ( blocks ) :
    if self . provider :
        external_block_id = str ( len ( self . blocks ) )
        launch_cmd = self . launch_cmd . format ( block_id = external_block_id )
        internal_block = self . provider . submit ( launch_cmd , 1 , 1 )
        logger . debug ( "Launched block {}->{}" . format ( external_block_id , internal_block ) )
        if not internal_block :
            raise ( ScalingFailed ( self . provider . label , "Attempts to provision nodes via provider has failed" ) )
        r . extend ( [ external_block_id ] )
        self . blocks [ external_block_id ] = internal_block
    else :
        logger . error ( "No execution provider available" )
        r = None
return r


signature:
def scale_in ( self , blocks = None , block_ids = [ ] ) :
    """Scale in the number of active blocks by specified amount .
 The scale in method here is very rude . It doesn ' t give the workers
 the opportunity to finish current tasks or cleanup . This is tracked
 in issue # 530
 Parameters
 blocks : int
 Number of blocks to terminate and scale _ in by
 block _ ids : list
 List of specific block ids to terminate . Optional
 Raises :
 NotImplementedError"""

implementation:
if block_ids :
    block_ids_to_kill = block_ids
else :
    block_ids_to_kill = list ( self . blocks . keys ( ) ) [ : blocks ]
# Hold the block
for block_id in block_ids_to_kill :
    self . _hold_block ( block_id )
# Now kill via provider
to_kill = [ self . blocks . pop ( bid ) for bid in block_ids_to_kill ]
if self . provider :
    r = self . provider . cancel ( to_kill )
return r


signature:
def status ( self ) :
    """Return status of all blocks ."""

implementation:
status = [ ]
if self . provider :
    status = self . provider . status ( self . blocks . values ( ) )
return status


signature:
def shutdown ( self , hub = True , targets = 'all' , block = False ) :
    """Shutdown the executor , including all workers and controllers .
 This is not implemented .
 Kwargs :
 - hub ( Bool ) : Whether the hub should be shutdown , Default : True ,
 - targets ( list of ints | ' all ' ) : List of block id ' s to kill , Default : ' all '
 - block ( Bool ) : To block for confirmations or not
 Raises :
 NotImplementedError"""

implementation:
logger . info ( "Attempting HighThroughputExecutor shutdown" )
# self . outgoing _ q . close ( )
# self . incoming _ q . close ( )
self . queue_proc . terminate ( )
logger . info ( "Finished HighThroughputExecutor shutdown attempt" )
return True


signature:
def readinto ( self , buf , ** kwargs ) :
    """Read into ` ` buf ` ` from the device . The number of bytes read will be the
 length of ` ` buf ` ` .
 If ` ` start ` ` or ` ` end ` ` is provided , then the buffer will be sliced
 as if ` ` buf [ start : end ] ` ` . This will not cause an allocation like
 ` ` buf [ start : end ] ` ` will so it saves memory .
 : param bytearray buffer : buffer to write into
 : param int start : Index to start writing at
 : param int end : Index to write up to but not include"""

implementation:
self . i2c . readfrom_into ( self . device_address , buf , ** kwargs )
if self . _debug :
    print ( "i2c_device.readinto:" , [ hex ( i ) for i in buf ] )


signature:
def write ( self , buf , ** kwargs ) :
    """Write the bytes from ` ` buffer ` ` to the device . Transmits a stop bit if
 ` ` stop ` ` is set .
 If ` ` start ` ` or ` ` end ` ` is provided , then the buffer will be sliced
 as if ` ` buffer [ start : end ] ` ` . This will not cause an allocation like
 ` ` buffer [ start : end ] ` ` will so it saves memory .
 : param bytearray buffer : buffer containing the bytes to write
 : param int start : Index to start writing from
 : param int end : Index to read up to but not include
 : param bool stop : If true , output an I2C stop condition after the buffer is written"""

implementation:
self . i2c . writeto ( self . device_address , buf , ** kwargs )
if self . _debug :
    print ( "i2c_device.write:" , [ hex ( i ) for i in buf ] )


signature:
def write_then_readinto ( self , out_buffer , in_buffer , * , out_start = 0 , out_end = None , in_start = 0 , in_end = None , stop = True ) :
    """Write the bytes from ` ` out _ buffer ` ` to the device , then immediately
 reads into ` ` in _ buffer ` ` from the device . The number of bytes read
 will be the length of ` ` in _ buffer ` ` .
 Transmits a stop bit after the write , if ` ` stop ` ` is set .
 If ` ` out _ start ` ` or ` ` out _ end ` ` is provided , then the output buffer
 will be sliced as if ` ` out _ buffer [ out _ start : out _ end ] ` ` . This will
 not cause an allocation like ` ` buffer [ out _ start : out _ end ] ` ` will so
 it saves memory .
 If ` ` in _ start ` ` or ` ` in _ end ` ` is provided , then the input buffer
 will be sliced as if ` ` in _ buffer [ in _ start : in _ end ] ` ` . This will not
 cause an allocation like ` ` in _ buffer [ in _ start : in _ end ] ` ` will so
 it saves memory .
 : param bytearray out _ buffer : buffer containing the bytes to write
 : param bytearray in _ buffer : buffer containing the bytes to read into
 : param int out _ start : Index to start writing from
 : param int out _ end : Index to read up to but not include
 : param int in _ start : Index to start writing at
 : param int in _ end : Index to write up to but not include
 : param bool stop : If true , output an I2C stop condition after the buffer is written"""

implementation:
if out_end is None :
    out_end = len ( out_buffer )
if in_end is None :
    in_end = len ( in_buffer )
if hasattr ( self . i2c , 'writeto_then_readfrom' ) :
    if self . _debug :
        print ( "i2c_device.writeto_then_readfrom.out_buffer:" , [ hex ( i ) for i in out_buffer [ out_start : out_end ] ] )
    # In linux , at least , this is a special kernel function call
    self . i2c . writeto_then_readfrom ( self . device_address , out_buffer , in_buffer , out_start = out_start , out_end = out_end , in_start = in_start , in_end = in_end , stop = stop )
    if self . _debug :
        print ( "i2c_device.writeto_then_readfrom.in_buffer:" , [ hex ( i ) for i in in_buffer [ in_start : in_end ] ] )
else : # If we don ' t have a special implementation , we can fake it with two calls
    self . write ( out_buffer , start = out_start , end = out_end , stop = stop )
    if self . _debug :
        print ( "i2c_device.write_then_readinto.write.out_buffer:" , [ hex ( i ) for i in out_buffer [ out_start : out_end ] ] )
    self . readinto ( in_buffer , start = in_start , end = in_end )
    if self . _debug :
        print ( "i2c_device.write_then_readinto.readinto.in_buffer:" , [ hex ( i ) for i in in_buffer [ in_start : in_end ] ] )


signature:
def read_sample ( filename ) :
    """@ brief Returns data sample from simple text file .
 @ details This function should be used for text file with following format :
 @ code
 point _ 1 _ coord _ 1 point _ 1 _ coord _ 2 . . . point _ 1 _ coord _ n
 point _ 2 _ coord _ 1 point _ 2 _ coord _ 2 . . . point _ 2 _ coord _ n
 @ endcode
 @ param [ in ] filename ( string ) : Path to file with data .
 @ return ( list ) Points where each point represented by list of coordinates ."""

implementation:
file = open ( filename , 'r' )
sample = [ [ float ( val ) for val in line . split ( ) ] for line in file if len ( line . strip ( ) ) > 0 ]
file . close ( )
return sample


signature:
def calculate_distance_matrix ( sample ) :
    """@ brief Calculates distance matrix for data sample ( sequence of points ) using Euclidean distance as a metric .
 @ param [ in ] sample ( array _ like ) : Data points that are used for distance calculation .
 @ return ( list ) Matrix distance between data points ."""

implementation:
amount_rows = len ( sample ) ;
return [ [ euclidean_distance ( sample [ i ] , sample [ j ] ) for j in range ( amount_rows ) ] for i in range ( amount_rows ) ] ;


signature:
def read_image ( filename ) :
    """@ brief Returns image as N - dimension ( depends on the input image ) matrix , where one element of list describes pixel .
 @ param [ in ] filename ( string ) : Path to image .
 @ return ( list ) Pixels where each pixel described by list of RGB - values ."""

implementation:
with Image . open ( filename ) as image_source :
    data = [ list ( pixel ) for pixel in image_source . getdata ( ) ]
    return data


signature:
def rgb2gray ( image_rgb_array ) :
    """@ brief Returns image as 1 - dimension ( gray colored ) matrix , where one element of list describes pixel .
 @ details Luma coding is used for transformation and that is calculated directly from gamma - compressed primary intensities as a weighted sum :
 \f [ Y = 0.2989R + 0.587G + 0.114B \f ]
 @ param [ in ] image _ rgb _ array ( list ) : Image represented by RGB list .
 @ return ( list ) Image as gray colored matrix , where one element of list describes pixel .
 @ code
 colored _ image = read _ image ( file _ name ) ;
 gray _ image = rgb2gray ( colored _ image ) ;
 @ endcode
 @ see read _ image ( )"""

implementation:
image_gray_array = [ 0.0 ] * len ( image_rgb_array ) ;
for index in range ( 0 , len ( image_rgb_array ) , 1 ) :
    image_gray_array [ index ] = float ( image_rgb_array [ index ] [ 0 ] ) * 0.2989 + float ( image_rgb_array [ index ] [ 1 ] ) * 0.5870 + float ( image_rgb_array [ index ] [ 2 ] ) * 0.1140 ;
return image_gray_array ;


signature:
def stretch_pattern ( image_source ) :
    """@ brief Returns stretched content as 1 - dimension ( gray colored ) matrix with size of input image .
 @ param [ in ] image _ source ( Image ) : PIL Image instance .
 @ return ( list , Image ) Stretched image as gray colored matrix and source image ."""

implementation:
wsize , hsize = image_source . size ;
# Crop digit exactly
( ws , hs , we , he ) = gray_pattern_borders ( image_source ) ;
image_source = image_source . crop ( ( ws , hs , we , he ) ) ;
# Stretch it to initial sizes
image_source = image_source . resize ( ( wsize , hsize ) , Image . ANTIALIAS ) ;
# Transform image to simple array
data = [ pixel for pixel in image_source . getdata ( ) ] ;
image_pattern = rgb2gray ( data ) ;
return ( image_pattern , image_source ) ;


signature:
def gray_pattern_borders ( image ) :
    """@ brief Returns coordinates of gray image content on the input image .
 @ param [ in ] image ( Image ) : PIL Image instance that is processed .
 @ return ( tuple ) Returns coordinates of gray image content as ( width _ start , height _ start , width _ end , height _ end ) ."""

implementation:
width , height = image . size ;
width_start = width ;
width_end = 0 ;
height_start = height ;
height_end = 0 ;
row , col = 0 , 0 ;
for pixel in image . getdata ( ) :
    value = float ( pixel [ 0 ] ) * 0.2989 + float ( pixel [ 1 ] ) * 0.5870 + float ( pixel [ 2 ] ) * 0.1140 ;
    if ( value < 128 ) :
        if ( width_end < col ) :
            width_end = col ;
        if ( height_end < row ) :
            height_end = row ;
        if ( width_start > col ) :
            width_start = col ;
        if ( height_start > row ) :
            height_start = row ;
    col += 1 ;
    if ( col >= width ) :
        col = 0 ;
        row += 1 ;
return ( width_start , height_start , width_end + 1 , height_end + 1 ) ;


signature:
def average_neighbor_distance ( points , num_neigh ) :
    """@ brief Returns average distance for establish links between specified number of nearest neighbors .
 @ param [ in ] points ( list ) : Input data , list of points where each point represented by list .
 @ param [ in ] num _ neigh ( uint ) : Number of neighbors that should be used for distance calculation .
 @ return ( double ) Average distance for establish links between ' num _ neigh ' in data set ' points ' ."""

implementation:
if num_neigh > len ( points ) - 1 :
    raise NameError ( 'Impossible to calculate average distance to neighbors when number of object is less than number of neighbors.' ) ;
dist_matrix = [ [ 0.0 for i in range ( len ( points ) ) ] for j in range ( len ( points ) ) ] ;
for i in range ( 0 , len ( points ) , 1 ) :
    for j in range ( i + 1 , len ( points ) , 1 ) :
        distance = euclidean_distance ( points [ i ] , points [ j ] ) ;
        dist_matrix [ i ] [ j ] = distance ;
        dist_matrix [ j ] [ i ] = distance ;
    dist_matrix [ i ] = sorted ( dist_matrix [ i ] ) ;
total_distance = 0 ;
for i in range ( 0 , len ( points ) , 1 ) : # start from 0 - first element is distance to itself .
    for j in range ( 0 , num_neigh , 1 ) :
        total_distance += dist_matrix [ i ] [ j + 1 ] ;
return ( total_distance / ( num_neigh * len ( points ) ) ) ;


signature:
def medoid ( data , indexes = None , ** kwargs ) :
    """@ brief Calculate medoid for input points using Euclidean distance .
 @ param [ in ] data ( list ) : Set of points for that median should be calculated .
 @ param [ in ] indexes ( list ) : Indexes of input set of points that will be taken into account during median calculation .
 @ param [ in ] * * kwargs : Arbitrary keyword arguments ( available arguments : ' metric ' , ' data _ type ' ) .
 < b > Keyword Args : < / b > < br >
 - metric ( distance _ metric ) : Metric that is used for distance calculation between two points .
 - data _ type ( string ) : Data type of input sample ' data ' ( available values : ' points ' , ' distance _ matrix ' ) .
 @ return ( uint ) index of point in input set that corresponds to median ."""

implementation:
index_median = None
distance = float ( 'Inf' )
metric = kwargs . get ( 'metric' , type_metric . EUCLIDEAN_SQUARE )
data_type = kwargs . get ( 'data_type' , 'points' )
if data_type == 'points' :
    calculator = lambda index1 , index2 : metric ( data [ index1 ] , data [ index2 ] )
elif data_type == 'distance_matrix' :
    if isinstance ( data , numpy . matrix ) :
        calculator = lambda index1 , index2 : data . item ( index1 , index2 )
    else :
        calculator = lambda index1 , index2 : data [ index1 ] [ index2 ]
else :
    raise TypeError ( "Unknown type of data is specified '%s'." % data_type )
if indexes is None :
    range_points = range ( len ( data ) )
else :
    range_points = indexes
for index_candidate in range_points :
    distance_candidate = 0.0
    for index in range_points :
        distance_candidate += calculator ( index_candidate , index )
    if distance_candidate < distance :
        distance = distance_candidate
        index_median = index_candidate
return index_median


signature:
def euclidean_distance_square ( a , b ) :
    """@ brief Calculate square Euclidian distance between vector a and b .
 @ param [ in ] a ( list ) : The first vector .
 @ param [ in ] b ( list ) : The second vector .
 @ return ( double ) Square Euclidian distance between two vectors ."""

implementation:
if ( ( ( type ( a ) == float ) and ( type ( b ) == float ) ) or ( ( type ( a ) == int ) and ( type ( b ) == int ) ) ) :
    return ( a - b ) ** 2.0 ;
distance = 0.0 ;
for i in range ( 0 , len ( a ) ) :
    distance += ( a [ i ] - b [ i ] ) ** 2.0 ;
return distance ;


signature:
def manhattan_distance ( a , b ) :
    """@ brief Calculate Manhattan distance between vector a and b .
 @ param [ in ] a ( list ) : The first cluster .
 @ param [ in ] b ( list ) : The second cluster .
 @ return ( double ) Manhattan distance between two vectors ."""

implementation:
if ( ( ( type ( a ) == float ) and ( type ( b ) == float ) ) or ( ( type ( a ) == int ) and ( type ( b ) == int ) ) ) :
    return abs ( a - b ) ;
distance = 0.0 ;
dimension = len ( a ) ;
for i in range ( 0 , dimension ) :
    distance += abs ( a [ i ] - b [ i ] ) ;
return distance ;


signature:
def average_inter_cluster_distance ( cluster1 , cluster2 , data = None ) :
    """@ brief Calculates average inter - cluster distance between two clusters .
 @ details Clusters can be represented by list of coordinates ( in this case data shouldn ' t be specified ) ,
 or by list of indexes of points from the data ( represented by list of points ) , in this case
 data should be specified .
 @ param [ in ] cluster1 ( list ) : The first cluster where each element can represent index from the data or object itself .
 @ param [ in ] cluster2 ( list ) : The second cluster where each element can represent index from the data or object itself .
 @ param [ in ] data ( list ) : If specified than elements of clusters will be used as indexes ,
 otherwise elements of cluster will be considered as points .
 @ return ( double ) Average inter - cluster distance between two clusters ."""

implementation:
distance = 0.0 ;
if ( data is None ) :
    for i in range ( len ( cluster1 ) ) :
        for j in range ( len ( cluster2 ) ) :
            distance += euclidean_distance_square ( cluster1 [ i ] , cluster2 [ j ] ) ;
else :
    for i in range ( len ( cluster1 ) ) :
        for j in range ( len ( cluster2 ) ) :
            distance += euclidean_distance_square ( data [ cluster1 [ i ] ] , data [ cluster2 [ j ] ] ) ;
distance /= float ( len ( cluster1 ) * len ( cluster2 ) ) ;
return distance ** 0.5 ;


signature:
def average_intra_cluster_distance ( cluster1 , cluster2 , data = None ) :
    """@ brief Calculates average intra - cluster distance between two clusters .
 @ details Clusters can be represented by list of coordinates ( in this case data shouldn ' t be specified ) ,
 or by list of indexes of points from the data ( represented by list of points ) , in this case
 data should be specified .
 @ param [ in ] cluster1 ( list ) : The first cluster .
 @ param [ in ] cluster2 ( list ) : The second cluster .
 @ param [ in ] data ( list ) : If specified than elements of clusters will be used as indexes ,
 otherwise elements of cluster will be considered as points .
 @ return ( double ) Average intra - cluster distance between two clusters ."""

implementation:
distance = 0.0
for i in range ( len ( cluster1 ) + len ( cluster2 ) ) :
    for j in range ( len ( cluster1 ) + len ( cluster2 ) ) :
        if data is None : # the first point
            if i < len ( cluster1 ) :
                first_point = cluster1 [ i ]
            else :
                first_point = cluster2 [ i - len ( cluster1 ) ]
            # the second point
            if j < len ( cluster1 ) :
                second_point = cluster1 [ j ]
            else :
                second_point = cluster2 [ j - len ( cluster1 ) ]
        else : # the first point
            if i < len ( cluster1 ) :
                first_point = data [ cluster1 [ i ] ]
            else :
                first_point = data [ cluster2 [ i - len ( cluster1 ) ] ]
            if j < len ( cluster1 ) :
                second_point = data [ cluster1 [ j ] ]
            else :
                second_point = data [ cluster2 [ j - len ( cluster1 ) ] ]
        distance += euclidean_distance_square ( first_point , second_point )
distance /= float ( ( len ( cluster1 ) + len ( cluster2 ) ) * ( len ( cluster1 ) + len ( cluster2 ) - 1.0 ) )
return distance ** 0.5


signature:
def variance_increase_distance ( cluster1 , cluster2 , data = None ) :
    """@ brief Calculates variance increase distance between two clusters .
 @ details Clusters can be represented by list of coordinates ( in this case data shouldn ' t be specified ) ,
 or by list of indexes of points from the data ( represented by list of points ) , in this case
 data should be specified .
 @ param [ in ] cluster1 ( list ) : The first cluster .
 @ param [ in ] cluster2 ( list ) : The second cluster .
 @ param [ in ] data ( list ) : If specified than elements of clusters will be used as indexes ,
 otherwise elements of cluster will be considered as points .
 @ return ( double ) Average variance increase distance between two clusters ."""

implementation:
# calculate local sum
if data is None :
    member_cluster1 = [ 0.0 ] * len ( cluster1 [ 0 ] )
    member_cluster2 = [ 0.0 ] * len ( cluster2 [ 0 ] )
else :
    member_cluster1 = [ 0.0 ] * len ( data [ 0 ] )
    member_cluster2 = [ 0.0 ] * len ( data [ 0 ] )
for i in range ( len ( cluster1 ) ) :
    if data is None :
        member_cluster1 = list_math_addition ( member_cluster1 , cluster1 [ i ] )
    else :
        member_cluster1 = list_math_addition ( member_cluster1 , data [ cluster1 [ i ] ] )
for j in range ( len ( cluster2 ) ) :
    if data is None :
        member_cluster2 = list_math_addition ( member_cluster2 , cluster2 [ j ] )
    else :
        member_cluster2 = list_math_addition ( member_cluster2 , data [ cluster2 [ j ] ] )
member_cluster_general = list_math_addition ( member_cluster1 , member_cluster2 )
member_cluster_general = list_math_division_number ( member_cluster_general , len ( cluster1 ) + len ( cluster2 ) )
member_cluster1 = list_math_division_number ( member_cluster1 , len ( cluster1 ) )
member_cluster2 = list_math_division_number ( member_cluster2 , len ( cluster2 ) )
# calculate global sum
distance_general = 0.0
distance_cluster1 = 0.0
distance_cluster2 = 0.0
for i in range ( len ( cluster1 ) ) :
    if data is None :
        distance_cluster1 += euclidean_distance_square ( cluster1 [ i ] , member_cluster1 )
        distance_general += euclidean_distance_square ( cluster1 [ i ] , member_cluster_general )
    else :
        distance_cluster1 += euclidean_distance_square ( data [ cluster1 [ i ] ] , member_cluster1 )
        distance_general += euclidean_distance_square ( data [ cluster1 [ i ] ] , member_cluster_general )
for j in range ( len ( cluster2 ) ) :
    if data is None :
        distance_cluster2 += euclidean_distance_square ( cluster2 [ j ] , member_cluster2 )
        distance_general += euclidean_distance_square ( cluster2 [ j ] , member_cluster_general )
    else :
        distance_cluster2 += euclidean_distance_square ( data [ cluster2 [ j ] ] , member_cluster2 )
        distance_general += euclidean_distance_square ( data [ cluster2 [ j ] ] , member_cluster_general )
return distance_general - distance_cluster1 - distance_cluster2


signature:
def calculate_ellipse_description ( covariance , scale = 2.0 ) :
    """@ brief Calculates description of ellipse using covariance matrix .
 @ param [ in ] covariance ( numpy . array ) : Covariance matrix for which ellipse area should be calculated .
 @ param [ in ] scale ( float ) : Scale of the ellipse .
 @ return ( float , float , float ) Return ellipse description : angle , width , height ."""

implementation:
eigh_values , eigh_vectors = numpy . linalg . eigh ( covariance )
order = eigh_values . argsort ( ) [ : : - 1 ]
values , vectors = eigh_values [ order ] , eigh_vectors [ order ]
angle = numpy . degrees ( numpy . arctan2 ( * vectors [ : , 0 ] [ : : - 1 ] ) )
if 0.0 in values :
    return 0 , 0 , 0
width , height = 2.0 * scale * numpy . sqrt ( values )
return angle , width , height


signature:
def data_corners ( data , data_filter = None ) :
    """@ brief Finds maximum and minimum corner in each dimension of the specified data .
 @ param [ in ] data ( list ) : List of points that should be analysed .
 @ param [ in ] data _ filter ( list ) : List of indexes of the data that should be analysed ,
 if it is ' None ' then whole ' data ' is analysed to obtain corners .
 @ return ( list ) Tuple of two points that corresponds to minimum and maximum corner ( min _ corner , max _ corner ) ."""

implementation:
dimensions = len ( data [ 0 ] )
bypass = data_filter
if bypass is None :
    bypass = range ( len ( data ) )
maximum_corner = list ( data [ bypass [ 0 ] ] [ : ] )
minimum_corner = list ( data [ bypass [ 0 ] ] [ : ] )
for index_point in bypass :
    for index_dimension in range ( dimensions ) :
        if data [ index_point ] [ index_dimension ] > maximum_corner [ index_dimension ] :
            maximum_corner [ index_dimension ] = data [ index_point ] [ index_dimension ]
        if data [ index_point ] [ index_dimension ] < minimum_corner [ index_dimension ] :
            minimum_corner [ index_dimension ] = data [ index_point ] [ index_dimension ]
return minimum_corner , maximum_corner


signature:
def norm_vector ( vector ) :
    """@ brief Calculates norm of an input vector that is known as a vector length .
 @ param [ in ] vector ( list ) : The input vector whose length is calculated .
 @ return ( double ) vector norm known as vector length ."""

implementation:
length = 0.0
for component in vector :
    length += component * component
length = length ** 0.5
return length


signature:
def timedcall ( executable_function , * args ) :
    """@ brief Executes specified method or function with measuring of execution time .
 @ param [ in ] executable _ function ( pointer ) : Pointer to function or method .
 @ param [ in ] args ( * ) : Arguments of called function or method .
 @ return ( tuple ) Execution time and result of execution of function or method ( execution _ time , result _ execution ) ."""

implementation:
time_start = time . clock ( ) ;
result = executable_function ( * args ) ;
time_end = time . clock ( ) ;
return ( time_end - time_start , result ) ;


signature:
def extract_number_oscillations ( osc_dyn , index = 0 , amplitude_threshold = 1.0 ) :
    """@ brief Extracts number of oscillations of specified oscillator .
 @ param [ in ] osc _ dyn ( list ) : Dynamic of oscillators .
 @ param [ in ] index ( uint ) : Index of oscillator in dynamic .
 @ param [ in ] amplitude _ threshold ( double ) : Amplitude threshold when oscillation is taken into account , for example ,
 when oscillator amplitude is greater than threshold then oscillation is incremented .
 @ return ( uint ) Number of oscillations of specified oscillator ."""

implementation:
number_oscillations = 0 ;
waiting_differential = False ;
threshold_passed = False ;
high_level_trigger = True if ( osc_dyn [ 0 ] [ index ] > amplitude_threshold ) else False ;
for values in osc_dyn :
    if ( ( values [ index ] >= amplitude_threshold ) and ( high_level_trigger is False ) ) :
        high_level_trigger = True ;
        threshold_passed = True ;
    elif ( ( values [ index ] < amplitude_threshold ) and ( high_level_trigger is True ) ) :
        high_level_trigger = False ;
        threshold_passed = True ;
    if ( threshold_passed is True ) :
        threshold_passed = False ;
        if ( waiting_differential is True and high_level_trigger is False ) :
            number_oscillations += 1 ;
            waiting_differential = False ;
        else :
            waiting_differential = True ;
return number_oscillations ;


signature:
def allocate_sync_ensembles ( dynamic , tolerance = 0.1 , threshold = 1.0 , ignore = None ) :
    """@ brief Allocate clusters in line with ensembles of synchronous oscillators where each
 synchronous ensemble corresponds to only one cluster .
 @ param [ in ] dynamic ( dynamic ) : Dynamic of each oscillator .
 @ param [ in ] tolerance ( double ) : Maximum error for allocation of synchronous ensemble oscillators .
 @ param [ in ] threshold ( double ) : Amlitude trigger when spike is taken into account .
 @ param [ in ] ignore ( bool ) : Set of indexes that shouldn ' t be taken into account .
 @ return ( list ) Grours ( lists ) of indexes of synchronous oscillators , for example ,
 [ [ index _ osc1 , index _ osc3 ] , [ index _ osc2 ] , [ index _ osc4 , index _ osc5 ] ] ."""

implementation:
descriptors = [ [ ] for _ in range ( len ( dynamic [ 0 ] ) ) ] ;
# Check from the end for obtaining result
for index_dyn in range ( 0 , len ( dynamic [ 0 ] ) , 1 ) :
    if ( ( ignore is not None ) and ( index_dyn in ignore ) ) :
        continue ;
    time_stop_simulation = len ( dynamic ) - 1 ;
    active_state = False ;
    if ( dynamic [ time_stop_simulation ] [ index_dyn ] > threshold ) :
        active_state = True ;
    # if active state is detected , it means we don ' t have whole oscillatory period for the considered oscillator , should be skipped .
    if ( active_state is True ) :
        while ( ( dynamic [ time_stop_simulation ] [ index_dyn ] > threshold ) and ( time_stop_simulation > 0 ) ) :
            time_stop_simulation -= 1 ;
        # if there are no any oscillation than let ' s consider it like noise
        if ( time_stop_simulation == 0 ) :
            continue ;
        # reset
        active_state = False ;
    desc = [ 0 , 0 , 0 ] ;
    # end , start , average time of oscillation
    for t in range ( time_stop_simulation , 0 , - 1 ) :
        if ( ( dynamic [ t ] [ index_dyn ] > threshold ) and ( active_state is False ) ) :
            desc [ 0 ] = t ;
            active_state = True ;
        elif ( ( dynamic [ t ] [ index_dyn ] < threshold ) and ( active_state is True ) ) :
            desc [ 1 ] = t ;
            active_state = False ;
            break ;
    if ( desc == [ 0 , 0 , 0 ] ) :
        continue ;
    desc [ 2 ] = desc [ 1 ] + ( desc [ 0 ] - desc [ 1 ] ) / 2.0 ;
    descriptors [ index_dyn ] = desc ;
# Cluster allocation
sync_ensembles = [ ] ;
desc_sync_ensembles = [ ] ;
for index_desc in range ( 0 , len ( descriptors ) , 1 ) :
    if ( descriptors [ index_desc ] == [ ] ) :
        continue ;
    if ( len ( sync_ensembles ) == 0 ) :
        desc_ensemble = descriptors [ index_desc ] ;
        reducer = ( desc_ensemble [ 0 ] - desc_ensemble [ 1 ] ) * tolerance ;
        desc_ensemble [ 0 ] = desc_ensemble [ 2 ] + reducer ;
        desc_ensemble [ 1 ] = desc_ensemble [ 2 ] - reducer ;
        desc_sync_ensembles . append ( desc_ensemble ) ;
        sync_ensembles . append ( [ index_desc ] ) ;
    else :
        oscillator_captured = False ;
        for index_ensemble in range ( 0 , len ( sync_ensembles ) , 1 ) :
            if ( ( desc_sync_ensembles [ index_ensemble ] [ 0 ] > descriptors [ index_desc ] [ 2 ] ) and ( desc_sync_ensembles [ index_ensemble ] [ 1 ] < descriptors [ index_desc ] [ 2 ] ) ) :
                sync_ensembles [ index_ensemble ] . append ( index_desc ) ;
                oscillator_captured = True ;
                break ;
        if ( oscillator_captured is False ) :
            desc_ensemble = descriptors [ index_desc ] ;
            reducer = ( desc_ensemble [ 0 ] - desc_ensemble [ 1 ] ) * tolerance ;
            desc_ensemble [ 0 ] = desc_ensemble [ 2 ] + reducer ;
            desc_ensemble [ 1 ] = desc_ensemble [ 2 ] - reducer ;
            desc_sync_ensembles . append ( desc_ensemble ) ;
            sync_ensembles . append ( [ index_desc ] ) ;
return sync_ensembles ;


signature:
def draw_dynamics ( t , dyn , x_title = None , y_title = None , x_lim = None , y_lim = None , x_labels = True , y_labels = True , separate = False , axes = None ) :
    """@ brief Draw dynamics of neurons ( oscillators ) in the network .
 @ details It draws if matplotlib is not specified ( None ) , othewise it should be performed manually .
 @ param [ in ] t ( list ) : Values of time ( used by x axis ) .
 @ param [ in ] dyn ( list ) : Values of output of oscillators ( used by y axis ) .
 @ param [ in ] x _ title ( string ) : Title for Y .
 @ param [ in ] y _ title ( string ) : Title for X .
 @ param [ in ] x _ lim ( double ) : X limit .
 @ param [ in ] y _ lim ( double ) : Y limit .
 @ param [ in ] x _ labels ( bool ) : If True - shows X labels .
 @ param [ in ] y _ labels ( bool ) : If True - shows Y labels .
 @ param [ in ] separate ( list ) : Consists of lists of oscillators where each such list consists of oscillator indexes that will be shown on separated stage .
 @ param [ in ] axes ( ax ) : If specified then matplotlib axes will be used for drawing and plot will not be shown .
 @ return ( ax ) Axes of matplotlib ."""

implementation:
number_lines = 0 ;
stage_xlim = None ;
if ( x_lim is not None ) :
    stage_xlim = x_lim ;
elif ( len ( t ) > 0 ) :
    stage_xlim = [ 0 , t [ len ( t ) - 1 ] ] ;
if ( ( isinstance ( separate , bool ) is True ) and ( separate is True ) ) :
    if ( isinstance ( dyn [ 0 ] , list ) is True ) :
        number_lines = len ( dyn [ 0 ] ) ;
    else :
        number_lines = 1 ;
elif ( isinstance ( separate , list ) is True ) :
    number_lines = len ( separate ) ;
else :
    number_lines = 1 ;
dysplay_result = False ;
if ( axes is None ) :
    dysplay_result = True ;
    ( fig , axes ) = plt . subplots ( number_lines , 1 ) ;
# Check if we have more than one dynamic
if ( isinstance ( dyn [ 0 ] , list ) is True ) :
    num_items = len ( dyn [ 0 ] ) ;
    for index in range ( 0 , num_items , 1 ) :
        y = [ item [ index ] for item in dyn ] ;
        if ( number_lines > 1 ) :
            index_stage = - 1 ;
            # Find required axes for the y
            if ( isinstance ( separate , bool ) is True ) :
                index_stage = index ;
            elif ( isinstance ( separate , list ) is True ) :
                for index_group in range ( 0 , len ( separate ) , 1 ) :
                    if ( index in separate [ index_group ] ) :
                        index_stage = index_group ;
                        break ;
            if ( index_stage != - 1 ) :
                if ( index_stage != number_lines - 1 ) :
                    axes [ index_stage ] . get_xaxis ( ) . set_visible ( False ) ;
                axes [ index_stage ] . plot ( t , y , 'b-' , linewidth = 0.5 ) ;
                set_ax_param ( axes [ index_stage ] , x_title , y_title , stage_xlim , y_lim , x_labels , y_labels , True ) ;
        else :
            axes . plot ( t , y , 'b-' , linewidth = 0.5 ) ;
            set_ax_param ( axes , x_title , y_title , stage_xlim , y_lim , x_labels , y_labels , True ) ;
else :
    axes . plot ( t , dyn , 'b-' , linewidth = 0.5 ) ;
    set_ax_param ( axes , x_title , y_title , stage_xlim , y_lim , x_labels , y_labels , True ) ;
if ( dysplay_result is True ) :
    plt . show ( ) ;
return axes ;


signature:
def set_ax_param ( ax , x_title = None , y_title = None , x_lim = None , y_lim = None , x_labels = True , y_labels = True , grid = True ) :
    """@ brief Sets parameters for matplotlib ax .
 @ param [ in ] ax ( Axes ) : Axes for which parameters should applied .
 @ param [ in ] x _ title ( string ) : Title for Y .
 @ param [ in ] y _ title ( string ) : Title for X .
 @ param [ in ] x _ lim ( double ) : X limit .
 @ param [ in ] y _ lim ( double ) : Y limit .
 @ param [ in ] x _ labels ( bool ) : If True - shows X labels .
 @ param [ in ] y _ labels ( bool ) : If True - shows Y labels .
 @ param [ in ] grid ( bool ) : If True - shows grid ."""

implementation:
from matplotlib . font_manager import FontProperties ;
from matplotlib import rcParams ;
if ( _platform == "linux" ) or ( _platform == "linux2" ) :
    rcParams [ 'font.sans-serif' ] = [ 'Liberation Serif' ] ;
else :
    rcParams [ 'font.sans-serif' ] = [ 'Arial' ] ;
rcParams [ 'font.size' ] = 12 ;
surface_font = FontProperties ( ) ;
if ( _platform == "linux" ) or ( _platform == "linux2" ) :
    surface_font . set_name ( 'Liberation Serif' ) ;
else :
    surface_font . set_name ( 'Arial' ) ;
surface_font . set_size ( '12' ) ;
if ( y_title is not None ) :
    ax . set_ylabel ( y_title , fontproperties = surface_font ) ;
if ( x_title is not None ) :
    ax . set_xlabel ( x_title , fontproperties = surface_font ) ;
if ( x_lim is not None ) :
    ax . set_xlim ( x_lim [ 0 ] , x_lim [ 1 ] ) ;
if ( y_lim is not None ) :
    ax . set_ylim ( y_lim [ 0 ] , y_lim [ 1 ] ) ;
if ( x_labels is False ) :
    ax . xaxis . set_ticklabels ( [ ] ) ;
if ( y_labels is False ) :
    ax . yaxis . set_ticklabels ( [ ] ) ;
ax . grid ( grid ) ;


signature:
def draw_dynamics_set ( dynamics , xtitle = None , ytitle = None , xlim = None , ylim = None , xlabels = False , ylabels = False ) :
    """@ brief Draw lists of dynamics of neurons ( oscillators ) in the network .
 @ param [ in ] dynamics ( list ) : List of network outputs that are represented by values of output of oscillators ( used by y axis ) .
 @ param [ in ] xtitle ( string ) : Title for Y .
 @ param [ in ] ytitle ( string ) : Title for X .
 @ param [ in ] xlim ( double ) : X limit .
 @ param [ in ] ylim ( double ) : Y limit .
 @ param [ in ] xlabels ( bool ) : If True - shows X labels .
 @ param [ in ] ylabels ( bool ) : If True - shows Y labels ."""

implementation:
# Calculate edge for confortable representation .
number_dynamics = len ( dynamics ) ;
if ( number_dynamics == 1 ) :
    draw_dynamics ( dynamics [ 0 ] [ 0 ] , dynamics [ 0 ] [ 1 ] , xtitle , ytitle , xlim , ylim , xlabels , ylabels ) ;
    return ;
number_cols = int ( numpy . ceil ( number_dynamics ** 0.5 ) ) ;
number_rows = int ( numpy . ceil ( number_dynamics / number_cols ) ) ;
real_index = 0 , 0 ;
double_indexer = True ;
if ( ( number_cols == 1 ) or ( number_rows == 1 ) ) :
    real_index = 0 ;
    double_indexer = False ;
( _ , axarr ) = plt . subplots ( number_rows , number_cols ) ;
# plt . setp ( [ ax for ax in axarr ] , visible = False ) ;
for dynamic in dynamics :
    axarr [ real_index ] = draw_dynamics ( dynamic [ 0 ] , dynamic [ 1 ] , xtitle , ytitle , xlim , ylim , xlabels , ylabels , axes = axarr [ real_index ] ) ;
    # plt . setp ( axarr [ real _ index ] , visible = True ) ;
    if ( double_indexer is True ) :
        real_index = real_index [ 0 ] , real_index [ 1 ] + 1 ;
        if ( real_index [ 1 ] >= number_cols ) :
            real_index = real_index [ 0 ] + 1 , 0 ;
    else :
        real_index += 1 ;
plt . show ( ) ;


signature:
def draw_image_color_segments ( source , clusters , hide_axes = True ) :
    """@ brief Shows image segments using colored image .
 @ details Each color on result image represents allocated segment . The first image is initial and other is result of segmentation .
 @ param [ in ] source ( string ) : Path to image .
 @ param [ in ] clusters ( list ) : List of clusters ( allocated segments of image ) where each cluster
 consists of indexes of pixel from source image .
 @ param [ in ] hide _ axes ( bool ) : If True then axes will not be displayed ."""

implementation:
image_source = Image . open ( source ) ;
image_size = image_source . size ;
( fig , axarr ) = plt . subplots ( 1 , 2 ) ;
plt . setp ( [ ax for ax in axarr ] , visible = False ) ;
available_colors = [ ( 0 , 162 , 232 ) , ( 34 , 177 , 76 ) , ( 237 , 28 , 36 ) , ( 255 , 242 , 0 ) , ( 0 , 0 , 0 ) , ( 237 , 28 , 36 ) , ( 255 , 174 , 201 ) , ( 127 , 127 , 127 ) , ( 185 , 122 , 87 ) , ( 200 , 191 , 231 ) , ( 136 , 0 , 21 ) , ( 255 , 127 , 39 ) , ( 63 , 72 , 204 ) , ( 195 , 195 , 195 ) , ( 255 , 201 , 14 ) , ( 239 , 228 , 176 ) , ( 181 , 230 , 29 ) , ( 153 , 217 , 234 ) , ( 112 , 146 , 180 ) ] ;
image_color_segments = [ ( 255 , 255 , 255 ) ] * ( image_size [ 0 ] * image_size [ 1 ] ) ;
for index_segment in range ( len ( clusters ) ) :
    for index_pixel in clusters [ index_segment ] :
        image_color_segments [ index_pixel ] = available_colors [ index_segment ] ;
stage = array ( image_color_segments , numpy . uint8 ) ;
stage = numpy . reshape ( stage , ( image_size [ 1 ] , image_size [ 0 ] ) + ( ( 3 ) , ) ) ;
# ( ( 3 ) , ) it ' s size of RGB - third dimension .
image_cluster = Image . fromarray ( stage , 'RGB' ) ;
axarr [ 0 ] . imshow ( image_source , interpolation = 'none' ) ;
axarr [ 1 ] . imshow ( image_cluster , interpolation = 'none' ) ;
for i in range ( 2 ) :
    plt . setp ( axarr [ i ] , visible = True ) ;
    if ( hide_axes is True ) :
        axarr [ i ] . xaxis . set_ticklabels ( [ ] ) ;
        axarr [ i ] . yaxis . set_ticklabels ( [ ] ) ;
        axarr [ i ] . xaxis . set_ticks_position ( 'none' ) ;
        axarr [ i ] . yaxis . set_ticks_position ( 'none' ) ;
plt . show ( ) ;


signature:
def draw_image_mask_segments ( source , clusters , hide_axes = True ) :
    """@ brief Shows image segments using black masks .
 @ details Each black mask of allocated segment is presented on separate plot .
 The first image is initial and others are black masks of segments .
 @ param [ in ] source ( string ) : Path to image .
 @ param [ in ] clusters ( list ) : List of clusters ( allocated segments of image ) where each cluster
 consists of indexes of pixel from source image .
 @ param [ in ] hide _ axes ( bool ) : If True then axes will not be displayed ."""

implementation:
if ( len ( clusters ) == 0 ) :
    print ( "Warning: Nothing to draw - list of clusters is empty." )
    return ;
image_source = Image . open ( source ) ;
image_size = image_source . size ;
# Calculate edge for confortable representation .
number_clusters = len ( clusters ) + 1 ;
# show with the source image
number_cols = int ( numpy . ceil ( number_clusters ** 0.5 ) ) ;
number_rows = int ( numpy . ceil ( number_clusters / number_cols ) ) ;
real_index = 0 , 0 ;
double_indexer = True ;
if ( ( number_cols == 1 ) or ( number_rows == 1 ) ) :
    real_index = 0 ;
    double_indexer = False ;
( fig , axarr ) = plt . subplots ( number_rows , number_cols ) ;
plt . setp ( [ ax for ax in axarr ] , visible = False ) ;
axarr [ real_index ] . imshow ( image_source , interpolation = 'none' ) ;
plt . setp ( axarr [ real_index ] , visible = True ) ;
if ( hide_axes is True ) :
    axarr [ real_index ] . xaxis . set_ticklabels ( [ ] ) ;
    axarr [ real_index ] . yaxis . set_ticklabels ( [ ] ) ;
    axarr [ real_index ] . xaxis . set_ticks_position ( 'none' ) ;
    axarr [ real_index ] . yaxis . set_ticks_position ( 'none' ) ;
if ( double_indexer is True ) :
    real_index = 0 , 1 ;
else :
    real_index += 1 ;
for cluster in clusters :
    stage_cluster = [ ( 255 , 255 , 255 ) ] * ( image_size [ 0 ] * image_size [ 1 ] ) ;
    for index in cluster :
        stage_cluster [ index ] = ( 0 , 0 , 0 ) ;
    stage = array ( stage_cluster , numpy . uint8 ) ;
    stage = numpy . reshape ( stage , ( image_size [ 1 ] , image_size [ 0 ] ) + ( ( 3 ) , ) ) ;
    # ( ( 3 ) , ) it ' s size of RGB - third dimension .
    image_cluster = Image . fromarray ( stage , 'RGB' ) ;
    axarr [ real_index ] . imshow ( image_cluster , interpolation = 'none' ) ;
    plt . setp ( axarr [ real_index ] , visible = True ) ;
    if ( hide_axes is True ) :
        axarr [ real_index ] . xaxis . set_ticklabels ( [ ] ) ;
        axarr [ real_index ] . yaxis . set_ticklabels ( [ ] ) ;
        axarr [ real_index ] . xaxis . set_ticks_position ( 'none' ) ;
        axarr [ real_index ] . yaxis . set_ticks_position ( 'none' ) ;
    if ( double_indexer is True ) :
        real_index = real_index [ 0 ] , real_index [ 1 ] + 1 ;
        if ( real_index [ 1 ] >= number_cols ) :
            real_index = real_index [ 0 ] + 1 , 0 ;
    else :
        real_index += 1 ;
plt . show ( ) ;