<commit_before>require 'open3'
require 'shellwords'

module PipelineRunsHelper
  STEP_DESCRIPTIONS = {
    PipelineRunStage::HOST_FILTERING_STAGE_NAME => {
      "stage" => "Filter out host reads and conduct quality control.",
      "steps" => {
        "validate_input_out" => "Validates the input files and truncates to max fragments.",
        "star_out" => "Remove host reads using STAR.",
        "trimmomatic_out" => "Trim Illumina adapters using trimmomatic.",
        "priceseq_out" => "Remove low-quality reads using PriceSeqFilter.",
        "cdhitdup_out" => "Remove duplicate reads using CD-HIT-DUP.",
        "lzw_out" => "Remove low-complexity reads using LZW compression filter.",
        "bowtie2_out" => "Remove remaining host reads using Bowtie2.",
        "star_human_out" => "Remove human reads using STAR.",
        "bowtie2_human_out" => "Remove residual human reads using Bowtie2.",
        "subsampled_out" => "Subsample if there are too many remaining reads.",
        "gsnap_filter_out" => "Remove residual human reads using GSNAP.",
      },
    },
    PipelineRunStage::ALIGNMENT_STAGE_NAME => {
      "stage" => "Align reads to NCBI nucleotide and protein databases.",
      "steps" => {
        "gsnap_out" => "Align remaining reads to the NCBI NT database using GSNAP.",
        "rapsearch2_out" => "Align remaining reads to the NCBI NR database using RAPSearch2.",
        "taxon_count_out" => "Count preliminary taxon hits.",
        "annotated_out" => "Annotate non-host FASTA with preliminary NCBI accession IDs.",
      },
    },
    PipelineRunStage::POSTPROCESS_STAGE_NAME => {
      "stage" => "Assemble reads and refine read assignments.",
      "steps" => {
        "taxid_fasta_out" => "Annotate non-host FASTA with preliminary Taxonomy IDs.",
        "taxid_locator_out" => "Sort preliminary annotated FASTA by Taxonomy IDs and store the byte range of each Taxonomy ID in a JSON.",
        "alignment_viz_out" => "Record number of unique accessions matched.",
        "assembly_out" => "Assemble non-host reads using SPAdes.",
        "coverage_out" => "Calculate coverage statistics for assembled contigs",
        "gsnap_accessions_out" => "Generate FASTA of candidate references matched during GSNAP / NT alignment.",
        "rapsearch2_accessions_out" => "Generate FASTA of candidate references matched during RAPSearch2 / NR alignment.",
        "refined_gsnap_out" => "BLAST assembled contigs against candidate references from NT; reassign corresponding reads to the matched taxon.",
        "refined_rapsearch2_out" => "BLAST assembled contigs against candidate references from NR; reassign corresponding reads to the matched taxon.",
        "refined_taxon_count_out" => "Count taxon hits after the reassignment based on the BLAST results.",
        "contig_summary_out" => "Record statistics on the assembled contigs.",
        "refined_annotated_out" => "Annotate non-host FASTA with NCBI accession IDs after the BLAST-based match refinement.",
        "refined_taxid_locator_out" => "Annotate non-host FASTA with revised Taxonomy IDs after the BLAST-based match refinement.",
      },
    },
    PipelineRunStage::EXPT_STAGE_NAME => {
      "stage" => "Generate additional experimental output.",
      "steps" => {
        "nonhost_fastq_out" => "Filter original fastq/fasta input files to only contain non-host reads processed by IdSeq.",
      },
    },
  }.freeze

  PIPELINE_RUN_STILL_RUNNING_ERROR = "PIPELINE_RUN_STILL_RUNNING_ERROR".freeze
  PIPELINE_RUN_FAILED_ERROR = "PIPELINE_RUN_FAILED_ERROR".freeze

  def aegea_batch_submit_command(base_command,
                                 memory: Sample::DEFAULT_MEMORY_IN_MB,
                                 vcpus: Sample::DEFAULT_VCPUS,
                                 job_queue: Sample::DEFAULT_QUEUE,
                                 docker_image: "idseq_dag",
                                 sample_id: 0,
                                 stage_name: "misc")
    command = "aegea batch submit --command=#{Shellwords.escape(base_command)} "
    command += " --name=idseq-#{Rails.env}-#{sample_id}-#{stage_name} "
    command += " --ecr-image #{Shellwords.escape(docker_image)} --memory #{memory} --queue #{Shellwords.escape(job_queue)} --vcpus #{vcpus} --job-role idseq-pipeline "
    command += if job_queue == Sample::DEFAULT_QUEUE_HIMEM
                 " --mount-instance-storage "
               else
                 " --storage /mnt=#{Sample::DEFAULT_STORAGE_IN_GB} --volume-type gp2 "
               end
    command
  end

  def job_info(job_id, run_id)
    job_status = nil
    job_log_id = nil
    job_hash = nil
    stdout, stderr, status = Open3.capture3("aegea", "batch", "describe", job_id.to_s)
    if status.exitstatus.zero?
      job_description = stdout
      job_hash = JSON.parse(job_description)
      job_status = job_hash['status']
      if job_hash['container'] && job_hash['container']['logStreamName']
        job_log_id = job_hash['container']['logStreamName']
      end
    else
      LogUtil.log_err_and_airbrake("Error for update job status for record #{run_id} with error #{stderr}")
      job_status = PipelineRunStage::STATUS_ERROR # transient error, job is still "in progress"
      job_status = PipelineRunStage::STATUS_FAILED if stderr =~ /IndexError/ # job no longer exists
    end<commit_msg>Use instance storage on lomem CE as well as himem (#2705)<commit_after>require 'open3'
require 'shellwords'

module PipelineRunsHelper
  STEP_DESCRIPTIONS = {
    PipelineRunStage::HOST_FILTERING_STAGE_NAME => {
      "stage" => "Filter out host reads and conduct quality control.",
      "steps" => {
        "validate_input_out" => "Validates the input files and truncates to max fragments.",
        "star_out" => "Remove host reads using STAR.",
        "trimmomatic_out" => "Trim Illumina adapters using trimmomatic.",
        "priceseq_out" => "Remove low-quality reads using PriceSeqFilter.",
        "cdhitdup_out" => "Remove duplicate reads using CD-HIT-DUP.",
        "lzw_out" => "Remove low-complexity reads using LZW compression filter.",
        "bowtie2_out" => "Remove remaining host reads using Bowtie2.",
        "star_human_out" => "Remove human reads using STAR.",
        "bowtie2_human_out" => "Remove residual human reads using Bowtie2.",
        "subsampled_out" => "Subsample if there are too many remaining reads.",
        "gsnap_filter_out" => "Remove residual human reads using GSNAP.",
      },
    },
    PipelineRunStage::ALIGNMENT_STAGE_NAME => {
      "stage" => "Align reads to NCBI nucleotide and protein databases.",
      "steps" => {
        "gsnap_out" => "Align remaining reads to the NCBI NT database using GSNAP.",
        "rapsearch2_out" => "Align remaining reads to the NCBI NR database using RAPSearch2.",
        "taxon_count_out" => "Count preliminary taxon hits.",
        "annotated_out" => "Annotate non-host FASTA with preliminary NCBI accession IDs.",
      },
    },
    PipelineRunStage::POSTPROCESS_STAGE_NAME => {
      "stage" => "Assemble reads and refine read assignments.",
      "steps" => {
        "taxid_fasta_out" => "Annotate non-host FASTA with preliminary Taxonomy IDs.",
        "taxid_locator_out" => "Sort preliminary annotated FASTA by Taxonomy IDs and store the byte range of each Taxonomy ID in a JSON.",
        "alignment_viz_out" => "Record number of unique accessions matched.",
        "assembly_out" => "Assemble non-host reads using SPAdes.",
        "coverage_out" => "Calculate coverage statistics for assembled contigs",
        "gsnap_accessions_out" => "Generate FASTA of candidate references matched during GSNAP / NT alignment.",
        "rapsearch2_accessions_out" => "Generate FASTA of candidate references matched during RAPSearch2 / NR alignment.",
        "refined_gsnap_out" => "BLAST assembled contigs against candidate references from NT; reassign corresponding reads to the matched taxon.",
        "refined_rapsearch2_out" => "BLAST assembled contigs against candidate references from NR; reassign corresponding reads to the matched taxon.",
        "refined_taxon_count_out" => "Count taxon hits after the reassignment based on the BLAST results.",
        "contig_summary_out" => "Record statistics on the assembled contigs.",
        "refined_annotated_out" => "Annotate non-host FASTA with NCBI accession IDs after the BLAST-based match refinement.",
        "refined_taxid_locator_out" => "Annotate non-host FASTA with revised Taxonomy IDs after the BLAST-based match refinement.",
      },
    },
    PipelineRunStage::EXPT_STAGE_NAME => {
      "stage" => "Generate additional experimental output.",
      "steps" => {
        "nonhost_fastq_out" => "Filter original fastq/fasta input files to only contain non-host reads processed by IdSeq.",
      },
    },
  }.freeze

  PIPELINE_RUN_STILL_RUNNING_ERROR = "PIPELINE_RUN_STILL_RUNNING_ERROR".freeze
  PIPELINE_RUN_FAILED_ERROR = "PIPELINE_RUN_FAILED_ERROR".freeze

  def aegea_batch_submit_command(base_command,
                                 memory: Sample::DEFAULT_MEMORY_IN_MB,
                                 vcpus: Sample::DEFAULT_VCPUS,
                                 job_queue: Sample::DEFAULT_QUEUE,
                                 docker_image: "idseq_dag",
                                 sample_id: 0,
                                 stage_name: "misc")
    command = "aegea batch submit --command=#{Shellwords.escape(base_command)} "
    command += " --name=idseq-#{Rails.env}-#{sample_id}-#{stage_name} "
    command += " --ecr-image #{Shellwords.escape(docker_image)} --memory #{memory} --queue #{Shellwords.escape(job_queue)} --vcpus #{vcpus} --job-role idseq-pipeline "
    command += " --mount-instance-storage "
    command
  end

  def job_info(job_id, run_id)
    job_status = nil
    job_log_id = nil
    job_hash = nil
    stdout, stderr, status = Open3.capture3("aegea", "batch", "describe", job_id.to_s)
    if status.exitstatus.zero?
      job_description = stdout
      job_hash = JSON.parse(job_description)
      job_status = job_hash['status']
      if job_hash['container'] && job_hash['container']['logStreamName']
        job_log_id = job_hash['container']['logStreamName']
      end
    else
      LogUtil.log_err_and_airbrake("Error for update job status for record #{run_id} with error #{stderr}")
      job_status = PipelineRunStage::STATUS_ERROR # transient error, job is still "in progress"
      job_status = PipelineRunStage::STATUS_FAILED if stderr =~ /IndexError/ # job no longer exists
    end
{'fraction_non_alphanumeric': 0.10399089961133756, 'fraction_numerical': 0.0029386671722438145, 'mean_word_length': 3.065510597302505, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 85, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5377083', 'n_tokens_mistral': 3134, 'n_tokens_neox': 3011, 'n_words': 862}