<commit_before>'''
Main wrapper script for Genomel Recalibration workflow
'''
import os
import time
import argparse
import logging
import sys
import uuid
import tempfile
import utils.s3
import utils.pipeline
import datetime
import socket
import json

import postgres.status
import postgres.utils
import postgres.mixins

from sqlalchemy.exc import NoSuchTableError

def is_nat(x):
    '''
    Checks that a value is a natural number.
    '''
    if int(x) > 0:
        return int(x)
    raise argparse.ArgumentTypeError('%s must be positive, non-zero' % x)

def get_args():
    '''
    Loads the parser
    '''
    # Main parser
    parser = argparse.ArgumentParser(description="Genomel Recalibration Pipeline")
    # Args
    required = parser.add_argument_group("Required input parameters")
    
    # Metadata from input table
    required.add_argument("--input_id", default=None, help="INPUT_ID, internal production id.")
    required.add_argument("--project", default=None, help="PROJECT, project name.")
    required.add_argument("--input_table", default=None, help="Input table name")    
    required.add_argument("--md5", default=None, help="MD5, md5 of input file.")
    required.add_argument("--s3_url", default=None, help="S3_URL, s3 url of the input.")
    required.add_argument("--s3_profile", required=True, help="S3 profile name for project tenant.")
    required.add_argument("--s3_endpoint", required=True, help="S3 endpoint url for project tenant.")
    
    # Parameters for pipeline
    required.add_argument("--basedir", default="/mnt/SCRATCH/", help="Base directory for computations.")
    required.add_argument("--refdir", required=True, help="Path to reference directory.")
    required.add_argument("--cwl", required=True, help="Path to Genomel Recalibration CWL workflow yaml.")
    required.add_argument("--bam_s3dir", default="s3://", help="S3 dir for uploading bam output files.")
    required.add_argument("--fastq_s3dir", default="s3://", help="S3 dir for uploading fastq output files.")
    required.add_argument('--output_id', required=True, help='UUID for the output')
    required.add_argument('--thread_count', type=is_nat, default=8, help='Threads count.')
    required.add_argument('--java_heap', default="20g", help='Jave heap memory')

    # Tools parameters
    required.add_argument('--end_mode', default="PE", help="End mode in alignment")
    required.add_argument('--phred', default="33", help="Trimmomatic phred")
    required.add_argument('--illuminaclip', default="2:30:12", help="Trimmomatic Illumina clip lengths")
    required.add_argument('--slidingwindow', default="4:15", help="Trimmomatic sliding windows length")    
    required.add_argument('--leading', default=12, help="Trimmomatic leading")
    required.add_argument('--trailing', default=12, help="Trimmomatic trailing")
    required.add_argument('--minlen', default=36, help="Trimmomatic minimal acceptable length")
    required.add_argument('--format', default="ILM1.8", help="Alignment technology format")    
    required.add_argument('--length', default="300,125", help="Alignment read length")
    required.add_argument('--output_format', default="SAM", help="Alignment output format")

    return parser.parse_args()


def run_pipeline(args, harmo_statusclass, harmo_metricsclass):
    '''
    Executes the CWL pipeline and record status/metrics tables
    '''
    if not os.path.isdir(args.basedir):
        raise Exception("Could not find path to base directory: %s" %args.basedir)
    
    # Get output uuid
    job_id = args.output_id
    # Get hostname
    hostname = socket.gethostname()
    # Get datetime start
    datetime_start = str(datetime.datetime.now())
    
    # Create directory structure
    jobdir = tempfile.mkdtemp(prefix="harmonization_%s_" % str(job_id), dir=args.basedir)
    workdir = tempfile.mkdtemp(prefix="workdir_", dir=jobdir)
    inputdir = tempfile.mkdtemp(prefix="input_", dir=jobdir)
    resultdir = tempfile.mkdtemp(prefix="result_", dir=jobdir)
    refdir = args.refdir
    
    # Setup logger
    log_file = os.path.join(resultdir, "%s.harmonization.cwl.log" % str(job_id))
    logger = utils.pipeline.setup_logging(logging.INFO, str(job_id), log_file)
    
    # Logging inputs
    logger.info("hostname: %s" % (hostname))
    logger.info("input_id: %s" % (args.input_id))
    logger.info("project: %s" % (args.project))
    logger.info("md5: %s" % (args.md5))
    logger.info("datetime_start: %s" % (datetime_start))
    
    # Setup start point
    cwl_start = time.time()
    
    # Getting refs
    logger.info("getting resources")
    reference_data        = utils.pipeline.load_reference_json('etc/reference.json')
    reference_fasta_path  = os.path.join(refdir, reference_data["reference_fasta"])
    reference_indel_vcf   = os.path.join(refdir, reference_data["reference_indel_vcf"])
    reference_snp_vcf     = os.path.join(refdir, reference_data["reference_snp_vcf"])    
    postgres_config       = os.path.join(refdir, reference_data["pg_config"])
    
    # Logging pipeline info
    cwl_version    = reference_data["cwl_version"]
    docker_version = reference_data["docker_version"]
    logger.info("cwl_version: %s" % (cwl_version))
    logger.info("docker_version: %s" % (docker_version))
    
    # Download input bam
    input_url = args.s3_url
    input_bam = os.path.join(inputdir, os.path.basename(input_url))
    project_s3_profile = args.s3_profile
    project_s3_endpoint_url = args.s3_endpoint
    download_exit_code = utils.s3.aws_s3_get(logger, input_url, inputdir,
                                             project_s3_profile, project_s3_endpoint_url, recursive=False)<commit_msg>feat(slurm): add ref file args<commit_after>'''
Main wrapper script for Genomel Recalibration workflow
'''
import os
import time
import argparse
import logging
import sys
import uuid
import tempfile
import utils.s3
import utils.pipeline
import datetime
import socket
import json

import postgres.status
import postgres.utils
import postgres.mixins

from sqlalchemy.exc import NoSuchTableError

def is_nat(x):
    '''
    Checks that a value is a natural number.
    '''
    if int(x) > 0:
        return int(x)
    raise argparse.ArgumentTypeError('%s must be positive, non-zero' % x)

def get_args():
    '''
    Loads the parser
    '''
    # Main parser
    parser = argparse.ArgumentParser(description="Genomel Recalibration Pipeline")
    # Args
    required = parser.add_argument_group("Required input parameters")
    
    # Metadata from input table
    required.add_argument("--input_id", default=None, help="INPUT_ID, internal production id.")
    required.add_argument("--project", default=None, help="PROJECT, project name.")
    required.add_argument("--input_table", default=None, help="Input table name")    
    required.add_argument("--md5", default=None, help="MD5, md5 of input file.")
    required.add_argument("--s3_url", default=None, help="S3_URL, s3 url of the input.")
    required.add_argument("--s3_profile", required=True, help="S3 profile name for project tenant.")
    required.add_argument("--s3_endpoint", required=True, help="S3 endpoint url for project tenant.")
    
    # Parameters for pipeline
    required.add_argument("--basedir", default="/mnt/SCRATCH/", help="Base directory for computations.")
    required.add_argument("--refdir", required=True, help="Path to reference directory.")
    required.add_argument("--cwl", required=True, help="Path to Genomel Recalibration CWL workflow yaml.")
    required.add_argument("--bam_s3dir", default="s3://", help="S3 dir for uploading bam output files.")
    required.add_argument("--fastq_s3dir", default="s3://", help="S3 dir for uploading fastq output files.")
    required.add_argument('--output_id', required=True, help='UUID for the output')
    required.add_argument('--thread_count', type=is_nat, default=8, help='Threads count.')
    required.add_argument('--java_heap', default="20g", help='Jave heap memory')

    # Tools parameters
    required.add_argument('--end_mode', default="PE", help="End mode in alignment")
    required.add_argument('--phred', default="33", help="Trimmomatic phred")
    required.add_argument('--illuminaclip', default="2:30:12", help="Trimmomatic Illumina clip lengths")
    required.add_argument('--slidingwindow', default="4:15", help="Trimmomatic sliding windows length")    
    required.add_argument('--leading', default=12, help="Trimmomatic leading")
    required.add_argument('--trailing', default=12, help="Trimmomatic trailing")
    required.add_argument('--minlen', default=36, help="Trimmomatic minimal acceptable length")
    required.add_argument('--format', default="ILM1.8", help="Alignment technology format")    
    required.add_argument('--length', default="300,125", help="Alignment read length")
    required.add_argument('--output_format', default="SAM", help="Alignment output format")

    return parser.parse_args()


def run_pipeline(args, harmo_statusclass, harmo_metricsclass):
    '''
    Executes the CWL pipeline and record status/metrics tables
    '''
    if not os.path.isdir(args.basedir):
        raise Exception("Could not find path to base directory: %s" %args.basedir)
    
    # Get output uuid
    job_id = args.output_id
    # Get hostname
    hostname = socket.gethostname()
    # Get datetime start
    datetime_start = str(datetime.datetime.now())
    
    # Create directory structure
    jobdir = tempfile.mkdtemp(prefix="harmonization_%s_" % str(job_id), dir=args.basedir)
    workdir = tempfile.mkdtemp(prefix="workdir_", dir=jobdir)
    inputdir = tempfile.mkdtemp(prefix="input_", dir=jobdir)
    resultdir = tempfile.mkdtemp(prefix="result_", dir=jobdir)
    refdir = args.refdir
    
    # Setup logger
    log_file = os.path.join(resultdir, "%s.harmonization.cwl.log" % str(job_id))
    logger = utils.pipeline.setup_logging(logging.INFO, str(job_id), log_file)
    
    # Logging inputs
    logger.info("hostname: %s" % (hostname))
    logger.info("input_id: %s" % (args.input_id))
    logger.info("project: %s" % (args.project))
    logger.info("md5: %s" % (args.md5))
    logger.info("datetime_start: %s" % (datetime_start))
    
    # Setup start point
    cwl_start = time.time()
    
    # Getting refs
    logger.info("getting resources")
    reference_data        = utils.pipeline.load_reference_json('etc/reference.json')
    reference_fasta_path  = os.path.join(refdir, reference_data["reference_fasta"])
    reference_database    = os.path.join(refdir, reference_data["reference_database"]) 
    adapters_file         = os.path.join(refdir, reference_data["adapters_file"])     
    postgres_config       = os.path.join(refdir, reference_data["pg_config"])
    
    # Logging pipeline info
    cwl_version    = reference_data["cwl_version"]
    docker_version = reference_data["docker_version"]
    logger.info("cwl_version: %s" % (cwl_version))
    logger.info("docker_version: %s" % (docker_version))
    
    # Download input bam
    input_url = args.s3_url
    input_bam = os.path.join(inputdir, os.path.basename(input_url))
    project_s3_profile = args.s3_profile
    project_s3_endpoint_url = args.s3_endpoint
    download_exit_code = utils.s3.aws_s3_get(logger, input_url, inputdir,
                                             project_s3_profile, project_s3_endpoint_url, recursive=False)
{'fraction_non_alphanumeric': 0.12951223807008277, 'fraction_numerical': 0.009860891001936961, 'mean_word_length': 4.112061206120612, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '918106', 'n_tokens_mistral': 3552, 'n_tokens_neox': 3504, 'n_words': 769}