<commit_before># radical.pilot script to run MDAnalysis Path Similarity Analysis
#
# Provide a JSON file with two lists of files:
#
# [
#  [topol1, topol1, topol1, ..., topol2, ...],
#  [trj1, trj2, trj3, ...]
# ]
#
# where each trj file name has a corresponding topology filename;
# the topology filenames can (and should) be repeated as often as
# necessary.
#
# The entries must be absolute paths.
#
# For this example, no additional superposition is performed and
# the trajectories are used as is. CA atoms are selected.
#
# The Frechet distance matrix between ALL trajectories (all-vs-all)
# is calculated.

import os
from os.path import basename
import time

os.environ['RADICAL_PILOT_VERBOSE']='DEBUG'

import sys
import radical.pilot as rp

import json

#----------------------------------------------------------------------------
if __name__ == "__main__":

    MY_STAGING_AREA = 'staging:///'
    SHARED_MDA_SCRIPT = 'mdanalysis_psa_partial.py'
    FILELIST = sys.argv([1])
    WINDOW_SIZE = int(sys.argv[2])
    cores = int(sys.argv[3])
    session_name = sys.argv[4]

    try:
        session   = rp.Session (name=session_name)
        c         = rp.Context ('ssh')
        session.add_context (c)

        print "initialize pilot manager ..."
        pmgr = rp.PilotManager (session=session)
        #pmgr.register_callback (pilot_state_cb)

        pdesc = rp.ComputePilotDescription ()
        pdesc.resource = ""
        pdesc.runtime  = 20 # minutes
        pdesc.cores    = cores
        pdesc.project  = "" #Project allocation
        pdesc.cleanup  = False

        # submit the pilot.
        pilot = pmgr.submit_pilots (pdesc)

        #initialize unit manager
        umgr = rp.UnitManager  (session=session, scheduler=rp.SCHED_DIRECT_SUBMISSION)

        #add pilot to unit manager
        umgr.add_pilots(pilot)

        # get ALL topology and trajectory files
        with open(FILELIST) as inp:
            topologies, trajectories = json.load(inp)

        fshared_list   = []
        fname_stage = []
        # stage all files to the staging area
        src_url = 'file://%s/%s' % (os.getcwd(), SHARED_MDA_SCRIPT)

        #print src_url

        sd_pilot = {'source': src_url,
                    'target': os.path.join(MY_STAGING_AREA, SHARED_MDA_SCRIPT),
                    'action': rp.TRANSFER,
        }
        fname_stage.append(sd_pilot)

        # Synchronously stage the data to the pilot
        pilot.stage_in(fname_stage)

        # we create one CU for a block of the distance matrix
        cudesc_list = []

        for i in range(len(trajectories), WINDOW_SIZE):
            for j in range(i+1, len(trajectories), WINDOW_SIZE):
                fshared = list()
                shared = {'source': os.path.join(MY_STAGING_AREA, SHARED_MDA_SCRIPT),
                         'target': SHARED_MDA_SCRIPT,
                         'action': rp.LINK}
                fshared.append(shared)

                if i == j: # never happesn since j >= i+1
                    shared = [{'source': 'file://{0}'.format(trajectory),
                              'target' : basename(trajectory),
                               'action' : rp.LINK}
                              for trajectory in trajectories[i:i+WINDOW_SIZE]]
                    fshared.extend(shared)
                else:
                    shared = [{'source': 'file://{0}'.format(trajectory),
                              'target' : basename(trajectory),
                              'action' : rp.LINK}
                              for trajectory in trajectories[i:i+WINDOW_SIZE]]
                    fshared.extend(shared)
                    shared = [{'source': 'file://{0}'.format(trajectory),
                              'target' : basename(trajectory),
                              'action' : rp.LINK}
                              for trajectory in trajectories[j:j+WINDOW_SIZE]]
                    fshared.extend(shared)
                # always copy all unique topology files
                shared = [{'source': 'file://{0}'.format(topology),
                           'target' : basename(topology),
                           'action' : rp.LINK}
                          for topology in set(topologies)]
                fshared.extend(shared)

                # block of topology / trajectory pairs
                #   block[:nsplit] + block[nsplit:]
                # The MDA script wants one long list of trajectories and the index nsplit
                # that indicates where to split the list to create the two groups of 
                # trajectories that are compared against each other.
                block_top = topologies[i:i+WINDOW_SIZE] + topologies[j:j+WINDOW_SIZE]
                block_trj = trajectories[i:i+WINDOW_SIZE] + trajectories[j:j+WINDOW_SIZE]
                block = [block_top, block_trj]
                nsplit = len(trajectories[i:i+WINDOW_SIZE])
                delta_i = len(trajectories[i:i+WINDOW_SIZE]) - i + 1
                delta_j = len(trajectories[j:j+WINDOW_SIZE]) - j + 1
                # should remember i, delta_i and j_delta_j because we calculate the
                # submatrix D[i:i+di, j:j+dj] in this CU.
                block_json = "block-{0}-{1}__{2}-{3}.json".format(
                    i, delta_i, j, delta_j)
                block_matrixfile = 'subdistances_{0}-{1}__{2}-{3}.npy'.format(
                    i, delta_i, j, delta_j)

                # create input file for the cu and add share it
                with open(block_json, "w") as out:
                    json.dump(block, out)

                # share input json file
                shared = [{'source': 'file://{0}'.format(os.path.realpath(block_json)),
                           'target' : basename(block_json),
                           'action' : rp.LINK}
                ]
                fshared.extend(shared)

            # define the compute unit, to compute over the trajectory pair
                cudesc = rp.ComputeUnitDescription()
                cudesc.executable    = "python"
                cudesc.pre_exec      = ["module load python"] #Only for Stampede
                cudesc.input_staging = fshared
                cudesc.arguments     = [SHARED_MDA_SCRIPT, '--nsplit', nsplit,
                                        '--inputfile', block_json,
                                        '--outfile', block_matrixfile,
                ]
                cudesc.cores         = 1

                cudesc_list.append (cudesc)

        # submit, run and wait and...
        #print "submit units to unit manager ..."
        units = umgr.submit_units (cudesc_list)

        #print "wait for units ..."
        umgr.wait_units()

        print "Creating Profile"
        with open('{0}.csv'.format(session.name),'w') as ProfFile:
            ProfFile.write('CU,New,StageIn,Allocate,Exec,StageOut,Done\n')
            for cu in units:
                timing_str=[cu.uid,'N/A','N/A','N/A','N/A','N/A','N/A']
                for states in cu.state_history:
                    if states.as_dict()['state']=='Scheduling':
                        timing_str[1]= (states.as_dict()['timestamp']-pilot.start_time).__str__()
                    elif states.as_dict()['state']=='AgentStagingInput':
                        timing_str[2]= (states.as_dict()['timestamp']-pilot.start_time).__str__()
                    elif states.as_dict()['state']=='Allocating':
                        timing_str[3]= (states.as_dict()['timestamp']-pilot.start_time).__str__()
                    elif states.as_dict()['state']=='Executing':
                        timing_str[4]= (states.as_dict()['timestamp']-pilot.start_time).__str__()
                    elif states.as_dict()['state']=='AgentStagingOutput':
                        timing_str[5]= (states.as_dict()['timestamp']-pilot.start_time).__str__()
                    elif states.as_dict()['state']=='Done':
                        timing_str[6]= (states.as_dict()['timestamp']-pilot.start_time).__str__()

                ProfFile.write(timing_str[0]+','+timing_str[1]+','+
                               timing_str[2]+','+timing_str[3]+','+
                               timing_str[4]+','+timing_str[5]+','+
                               timing_str[6]+'\n')


    except (KeyboardInterrupt, SystemExit) as e:
        # the callback called sys.exit(), and we can here catch the
        # corresponding KeyboardInterrupt exception for shutdown.  We also catch
        # SystemExit (which gets raised if the main threads exits for some other
        # reason).
        print "need to exit now: %s" % e

    except Exception as e:
        # Something unexpected happened in the pilot code above
        print "caught Exception: %s" % e
        raise

    finally :
        print "Closing session, exiting now ..."
        session.close(cleanup=False)

<commit_msg>rp_psa.py configured for TACC stampede<commit_after># radical.pilot script to run MDAnalysis Path Similarity Analysis
#
# =======================================
# configured for XSEDE TACC stampede
# =======================================
#
# USAGE: python rp_psa.py trajectories.json <blocksize> <cores> <session-name>
#
# ENVIRONMENT VARIABLES must be set:
#
#   export RADICAL_PILOT_PROJECT=TG-xxxxxx
#   export RADICAL_PILOT_DBURL="mongodb://user:pass@host:port/dbname"
#
#
# Provide a JSON file with two lists of files:
#
# [
#  [topol1, topol1, topol1, ..., topol2, ...],
#  [trj1, trj2, trj3, ...]
# ]
#
# where each trj file name has a corresponding topology filename;
# the topology filenames can (and should) be repeated as often as
# necessary.
#
# The entries must be absolute paths.
#
# For this example, no additional superposition is performed and
# the trajectories are used as is. CA atoms are selected.
#
# The Frechet distance matrix between ALL trajectories (all-vs-all)
# is calculated.

import os
from os.path import basename
import time

os.environ['RADICAL_PILOT_VERBOSE']='DEBUG'

import sys
import radical.pilot as rp

import json

#----------------------------------------------------------------------------
if __name__ == "__main__":

    MY_STAGING_AREA = 'staging:///'
    SHARED_MDA_SCRIPT = 'mdanalysis_psa_partial.py'
    FILELIST = sys.argv[1]
    WINDOW_SIZE = int(sys.argv[2])
    cores = int(sys.argv[3])
    session_name = sys.argv[4]

    try:
    	PROJECT = os.environ['RADICAL_PILOT_PROJECT']
        if not PROJECT:
	    raise ValueError
    except (KeyError, ValueError):
	raise RuntimeError("Set RADICAL_PILOT_PROJECT to your XSEDE allocation")        

    try:
        session   = rp.Session (name=session_name)
        c         = rp.Context ('ssh')
        session.add_context (c)

        print "initialize pilot manager ..."
        pmgr = rp.PilotManager (session=session)
        #pmgr.register_callback (pilot_state_cb)

        pdesc = rp.ComputePilotDescription ()
        pdesc.resource = "xsede.stampede"   # xsede.stampede or xsede.comet or ... check docs!
        pdesc.runtime  = 20 # minutes
        pdesc.cores    = cores
        pdesc.project  = PROJECT #Project allocation, pass through env var PROJECT
        pdesc.cleanup  = False

        # submit the pilot.
        pilot = pmgr.submit_pilots (pdesc)

        #initialize unit manager
        umgr = rp.UnitManager  (session=session, scheduler=rp.SCHED_DIRECT_SUBMISSION)

        #add pilot to unit manager
        umgr.add_pilots(pilot)

        # get ALL topology and trajectory files
        with open(FILELIST) as inp:
            topologies, trajectories = json.load(inp)

        fshared_list   = []
        fname_stage = []
        # stage all files to the staging area
        src_url = 'file://%s/%s' % (os.getcwd(), SHARED_MDA_SCRIPT)

        #print src_url

        sd_pilot = {'source': src_url,
                    'target': os.path.join(MY_STAGING_AREA, SHARED_MDA_SCRIPT),
                    'action': rp.TRANSFER,
        }
        fname_stage.append(sd_pilot)

        # Synchronously stage the data to the pilot
        pilot.stage_in(fname_stage)

        # we create one CU for a block of the distance matrix
        cudesc_list = []

        for i in range(len(trajectories), WINDOW_SIZE):
            for j in range(i+1, len(trajectories), WINDOW_SIZE):
                fshared = list()
                shared = {'source': os.path.join(MY_STAGING_AREA, SHARED_MDA_SCRIPT),
                         'target': SHARED_MDA_SCRIPT,
                         'action': rp.LINK}
                fshared.append(shared)

                if i == j: # never happesn since j >= i+1
                    shared = [{'source': 'file://{0}'.format(trajectory),
                              'target' : basename(trajectory),
                               'action' : rp.LINK}
                              for trajectory in trajectories[i:i+WINDOW_SIZE]]
                    fshared.extend(shared)
                else:
                    shared = [{'source': 'file://{0}'.format(trajectory),
                              'target' : basename(trajectory),
                              'action' : rp.LINK}
                              for trajectory in trajectories[i:i+WINDOW_SIZE]]
                    fshared.extend(shared)
                    shared = [{'source': 'file://{0}'.format(trajectory),
                              'target' : basename(trajectory),
                              'action' : rp.LINK}
                              for trajectory in trajectories[j:j+WINDOW_SIZE]]
                    fshared.extend(shared)
                # always copy all unique topology files
                shared = [{'source': 'file://{0}'.format(topology),
                           'target' : basename(topology),
                           'action' : rp.LINK}
                          for topology in set(topologies)]
                fshared.extend(shared)

                # block of topology / trajectory pairs
                #   block[:nsplit] + block[nsplit:]
                # The MDA script wants one long list of trajectories and the index nsplit
                # that indicates where to split the list to create the two groups of 
                # trajectories that are compared against each other.
                block_top = topologies[i:i+WINDOW_SIZE] + topologies[j:j+WINDOW_SIZE]
                block_trj = trajectories[i:i+WINDOW_SIZE] + trajectories[j:j+WINDOW_SIZE]
                block = [block_top, block_trj]
                nsplit = len(trajectories[i:i+WINDOW_SIZE])
                delta_i = len(trajectories[i:i+WINDOW_SIZE]) - i + 1
                delta_j = len(trajectories[j:j+WINDOW_SIZE]) - j + 1
                # should remember i, delta_i and j_delta_j because we calculate the
                # submatrix D[i:i+di, j:j+dj] in this CU.
                block_json = "block-{0}-{1}__{2}-{3}.json".format(
                    i, delta_i, j, delta_j)
                block_matrixfile = 'subdistances_{0}-{1}__{2}-{3}.npy'.format(
                    i, delta_i, j, delta_j)

                # create input file for the cu and add share it
                with open(block_json, "w") as out:
                    json.dump(block, out)

                # share input json file
                shared = [{'source': 'file://{0}'.format(os.path.realpath(block_json)),
                           'target' : basename(block_json),
                           'action' : rp.LINK}
                ]
                fshared.extend(shared)

            # define the compute unit, to compute over the trajectory pair
                cudesc = rp.ComputeUnitDescription()
                cudesc.executable    = "python"
                cudesc.pre_exec      = ["module load python"] #Only for Stampede
                cudesc.input_staging = fshared
                cudesc.arguments     = [SHARED_MDA_SCRIPT, '--nsplit', nsplit,
                                        '--inputfile', block_json,
                                        '--outfile', block_matrixfile,
                ]
                cudesc.cores         = 1

                cudesc_list.append (cudesc)

        # submit, run and wait and...
        #print "submit units to unit manager ..."
        units = umgr.submit_units (cudesc_list)

        #print "wait for units ..."
        umgr.wait_units()

        print "Creating Profile"
        with open('{0}.csv'.format(session.name),'w') as ProfFile:
            ProfFile.write('CU,New,StageIn,Allocate,Exec,StageOut,Done\n')
            for cu in units:
                timing_str=[cu.uid,'N/A','N/A','N/A','N/A','N/A','N/A']
                for states in cu.state_history:
                    if states.as_dict()['state']=='Scheduling':
                        timing_str[1]= (states.as_dict()['timestamp']-pilot.start_time).__str__()
                    elif states.as_dict()['state']=='AgentStagingInput':
                        timing_str[2]= (states.as_dict()['timestamp']-pilot.start_time).__str__()
                    elif states.as_dict()['state']=='Allocating':
                        timing_str[3]= (states.as_dict()['timestamp']-pilot.start_time).__str__()
                    elif states.as_dict()['state']=='Executing':
                        timing_str[4]= (states.as_dict()['timestamp']-pilot.start_time).__str__()
                    elif states.as_dict()['state']=='AgentStagingOutput':
                        timing_str[5]= (states.as_dict()['timestamp']-pilot.start_time).__str__()
                    elif states.as_dict()['state']=='Done':
                        timing_str[6]= (states.as_dict()['timestamp']-pilot.start_time).__str__()

                ProfFile.write(timing_str[0]+','+timing_str[1]+','+
                               timing_str[2]+','+timing_str[3]+','+
                               timing_str[4]+','+timing_str[5]+','+
                               timing_str[6]+'\n')


    except (KeyboardInterrupt, SystemExit) as e:
        # the callback called sys.exit(), and we can here catch the
        # corresponding KeyboardInterrupt exception for shutdown.  We also catch
        # SystemExit (which gets raised if the main threads exits for some other
        # reason).
        print "need to exit now: %s" % e

    except Exception as e:
        # Something unexpected happened in the pilot code above
        print "caught Exception: %s" % e
        raise

    finally :
        print "Closing session, exiting now ..."
        session.close(cleanup=False)


{'fraction_non_alphanumeric': 0.12317780306725194, 'fraction_numerical': 0.004877255730775484, 'mean_word_length': 1.736764051609076, 'pattern_counts': {'":': 2, '<': 6, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4172813', 'n_tokens_mistral': 5552, 'n_tokens_neox': 5235, 'n_words': 1234}