<commit_before>#!/usr/bin/python
# encoding: utf-8
"""
SGE.py
Fragment generation submission module for the Sun Grid Engine.

Created by <NAME> 2013

Use this as a basis for your cluster system module. The following functionality should be provided:
    - A SingleTask class for single job runs;
    - A MultipleTask class for parallel/multiple/array job runs;
    - a submit function which takes a command_filename e.g. cluster script and a working directory in which to run the script;
    - a query function which takes a logfile path and an optional job ID and returns information, via the cluster scheduler, about all jobs (if job ID is not specified) or else the specified job;
    - a check function which takes a logfile path and an optional job ID. This is similar to the query function but describes whether the job has finished or not.

"""

import sys
import os
import re
import subprocess

from ClusterTask import ClusterTask

sys.path.insert(0, "..")
from utils import colorprinter, JobInitializationException

# todo: move these into a common file e.g. utils.py
ERRCODE_ARGUMENTS = 1
ERRCODE_CLUSTER = 2
ERRCODE_OLDRESULTS = 3
ERRCODE_CONFIG = 4
ERRCODE_NOOUTPUT = 5
ERRCODE_JOBFAILED = 6
ERRCODE_MISSING_FILES = 1

ClusterType = "SGE"

class SingleTask(ClusterTask):

    submission_script ='''
#!/usr/bin/bash
#$ -N %(jobname)s
#$ -o %(outpath)s
#$ -e %(outpath)s
#$ -cwd
#$ -r y
#$ -l arch=linux-x64
'''

    python_script_preamble = '''
import sys
from time import strftime
import socket
import os
import platform
import subprocess
import tempfile
import shutil
import glob

class ProcessOutput(object):

    def __init__(self, stdout, stderr, errorcode):
        self.stdout = stdout
        self.stderr = stderr
        self.errorcode = errorcode
    
    def getError(self):
        if self.errorcode != 0:
            return("Errorcode: %%d\\n%%s" %% (self.errorcode, self.stderr))
        return None

def Popen(outdir, args):
    subp = subprocess.Popen([str(arg) for arg in args], stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=outdir, env={'SPARKSXDIR' : '/netapp/home/klabqb3backrub/tools/sparks-x'})
    output = subp.communicate()
    return ProcessOutput(output[0], output[1], subp.returncode) # 0 is stdout, 1 is stderr

def create_scratch_path():
    path = tempfile.mkdtemp(dir = '/scratch')
    if not os.path.isdir(path):
        raise os.error
    return path

print("<make_fragments>")

print("<start_time>")
print(strftime("%%Y-%%m-%%d %%H:%%M:%%S"))
print("</start_time>")

print("<host>")
print(socket.gethostname())
print("</host>")

print("<arch>")
print(platform.machine() + ', ' + platform.processor() + ', ' + platform.platform())
print("</arch>")

task_id = os.environ.get('SGE_TASK_ID')
'''

    python_script_postamble = '''
print("<end_time>")
print(strftime("%%Y-%%m-%%d %%H:%%M:%%S"))
print("</end_time>")

echo "<maxvmem>"
if [ -n "${SGE_TASK_ID+x}" ]
  then
    qstat -j $JOB_ID | grep -E "usage +$SGE_TASK_ID" | sed "s/.*maxvmem=//"
  else
    echo "Need to write the case for when there are no tasks"
fi
echo "</maxvmem>"

print("</make_fragments>")

if subp.errorcode != 0:
    sys.stderr.write(subp.stderr)
    sys.exit(subp.errorcode)
'''

    def __init__(self, make_fragments_perl_script, options, test_mode = False):
        '''options must contain jobname, outpath, no_homologs, NumTasks, [pdb_ids], [chains], and [fasta_files].'''
        super(SingleTask, self).__init__(make_fragments_perl_script, options)

        # NOTE: This if/elif block is specific to the QB3 cluster system and is only included for the benefit of QB3 users. Delete this line or alter it to fit your SGE cluster queue names.
        if options['queue'] not in ['long.q', 'lab.q', 'short.q']: 
            raise JobInitializationException("The queue you specified (%s) is invalid. Please use either long.q, lab.q, or short.q." % options['queue'])
        elif options['queue'] == 'short.q':
            test_mode = True

        self.test_mode = test_mode
        self.submission_script = self.__class__.submission_script
        self.set_run_length_and_queue()
        self.set_memory_and_scratch()
        self.submission_script += '\npython python_script.py\n\n'

    def set_run_length_and_queue(self):
        if self.test_mode:
            self.submission_script += '#$ -l h_rt=0:29:00\n'
        else:<commit_msg>Untested changes to SGE code - we now print out the memory and CPU usages per task.<commit_after>#!/usr/bin/python
# encoding: utf-8
"""
SGE.py
Fragment generation submission module for the Sun Grid Engine.

Created by <NAME> 2013

Use this as a basis for your cluster system module. The following functionality should be provided:
    - A SingleTask class for single job runs;
    - A MultipleTask class for parallel/multiple/array job runs;
    - a submit function which takes a command_filename e.g. cluster script and a working directory in which to run the script;
    - a query function which takes a logfile path and an optional job ID and returns information, via the cluster scheduler, about all jobs (if job ID is not specified) or else the specified job;
    - a check function which takes a logfile path and an optional job ID. This is similar to the query function but describes whether the job has finished or not.

"""

import sys
import os
import re
import subprocess

from ClusterTask import ClusterTask

sys.path.insert(0, "..")
from utils import colorprinter, JobInitializationException

# todo: move these into a common file e.g. utils.py
ERRCODE_ARGUMENTS = 1
ERRCODE_CLUSTER = 2
ERRCODE_OLDRESULTS = 3
ERRCODE_CONFIG = 4
ERRCODE_NOOUTPUT = 5
ERRCODE_JOBFAILED = 6
ERRCODE_MISSING_FILES = 1

ClusterType = "SGE"

class SingleTask(ClusterTask):

    submission_script ='''
#!/usr/bin/bash
#$ -N %(jobname)s
#$ -o %(outpath)s
#$ -e %(outpath)s
#$ -cwd
#$ -r y
#$ -l arch=linux-x64
'''

    python_script_preamble = '''
import sys
from time import strftime
import socket
import os
import platform
import subprocess
import tempfile
import shutil
import glob
import re
import shlex
import traceback

class ProcessOutput(object):

    def __init__(self, stdout, stderr, errorcode):
        self.stdout = stdout
        self.stderr = stderr
        self.errorcode = errorcode
    
    def getError(self):
        if self.errorcode != 0:
            return("Errorcode: %%d\\n%%s" %% (self.errorcode, self.stderr))
        return None

def Popen(outdir, args):
    subp = subprocess.Popen([str(arg) for arg in args], stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=outdir, env={'SPARKSXDIR' : '/netapp/home/klabqb3backrub/tools/sparks-x'})
    output = subp.communicate()
    return ProcessOutput(output[0], output[1], subp.returncode) # 0 is stdout, 1 is stderr

def shell_execute(command_line):
    subp = subprocess.Popen(shlex.split(command_line), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    output = subp.communicate()
    return ProcessOutput(output[0], output[1], subp.returncode) # 0 is stdout, 1 is stderr

def create_scratch_path():
    path = tempfile.mkdtemp(dir = '/scratch')
    if not os.path.isdir(path):
        raise os.error
    return path

print("<make_fragments>")

print("<start_time>")
print(strftime("%%Y-%%m-%%d %%H:%%M:%%S"))
print("</start_time>")

print("<host>")
print(socket.gethostname())
print("</host>")

print("<arch>")
print(platform.machine() + ', ' + platform.processor() + ', ' + platform.platform())
print("</arch>")

task_id = os.environ.get('SGE_TASK_ID')
job_id = os.environ.get('JOB_ID')
'''

    python_script_postamble = '''
print("<end_time>")
print(strftime("%%Y-%%m-%%d %%H:%%M:%%S"))
print("</end_time>")

task_usages = shell_execute('qstat -j %%d' %% job_id)
if task_usages.errorcode == 0:
  try:
    mtchs = re.match('.*?usage\s*(\d+):(.*?)\\n.*', task_usages.stdout, re.DOTALL)
    if mtchs and str(mtchs.group(1)) == str(task_id):
      task_properties = [s.strip() for s in mtchs.group(2).strip().split(",")]
      for tp in task_properties:
         if tp:
           prp=tp.split('=')[0]
           v=tp.split('=')[1]
           print('<task_%%s>%%s</task_%%s>' %% (prp, v, prp))
  except Exception, e:
    print('<qstat_parse_error>')
    print(str(e))
    print(traceback.format_exc())
    print('</qstat_parse_error>')
else:
  print("<qstat_error>")
  print(task_usages.stderr)
  print("</qstat_error>")

print("</make_fragments>")

if subp.errorcode != 0:
    sys.stderr.write(subp.stderr)
    sys.exit(subp.errorcode)
'''

    def __init__(self, make_fragments_perl_script, options, test_mode = False):
        '''options must contain jobname, outpath, no_homologs, NumTasks, [pdb_ids], [chains], and [fasta_files].'''
        super(SingleTask, self).__init__(make_fragments_perl_script, options)

        # NOTE: This if/elif block is specific to the QB3 cluster system and is only included for the benefit of QB3 users. Delete this line or alter it to fit your SGE cluster queue names.
        if options['queue'] not in ['long.q', 'lab.q', 'short.q']: 
            raise JobInitializationException("The queue you specified (%s) is invalid. Please use either long.q, lab.q, or short.q." % options['queue'])
        elif options['queue'] == 'short.q':
            test_mode = True

        self.test_mode = test_mode
        self.submission_script = self.__class__.submission_script
        self.set_run_length_and_queue()
        self.set_memory_and_scratch()
        self.submission_script += '\npython python_script.py\n\n'

    def set_run_length_and_queue(self):
        if self.test_mode:
            self.submission_script += '#$ -l h_rt=0:29:00\n'
        else:
{'fraction_non_alphanumeric': 0.13047094292546002, 'fraction_numerical': 0.006965381016737706, 'mean_word_length': 3.722631320569465, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1426205', 'n_tokens_mistral': 3230, 'n_tokens_neox': 3168, 'n_words': 910}