damian.py

#!/usr/bin/env python3

import os
import sys
import logging
import tempfile
import shutil
import datetime
from pathlib import Path
import psycopg2

from config import *

from lib.options import Options
from lib.wrap.trimmomatic import Trimmomatic
from lib.wrap.bowtie2 import Bowtie2
from lib.wrap.hmmsearch import HMMsearch
from lib.wrap.assembler import Assembler
from lib.wrap.trinity import Trinity
from lib.wrap.spades import SPAdes
from lib.wrap.idba import IDBA
from lib.wrap.no_assembler import No_Assembler
from lib.wrap.blast import Blast
from lib.wrap.blastn import Blastn
from lib.wrap.blastp import Blastp
from lib.usage import Usage
from lib.prepare_run import PrepareRun
from lib.contig_ops import ContigOps
from lib.ranking_ops import RankingOps
from lib.task_manager import TaskManager
from lib.taxonomy_ops import TaxonomyOps
from lib.check_dependencies import CheckDependencies

START_TIME = datetime.datetime.now()

options = Options.parse(sys.argv[1:])
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(sys.stderr)
logger.addHandler(handler)

ContigOps = ContigOps()
RankingOps = RankingOps()
TaskManager = TaskManager()

checkpoint = 0
trimmomatic = Trimmomatic(TRIMMOMATIC_PATH, TRIMMOMATIC_PARAMS, TRIMMOMATIC_PE_PARAMS,
                          TRIMMOMATIC_SE_PARAMS, options.threads, logger)
bowtie2 = Bowtie2(BOWTIE2_PATH, BOWTIE2_PARAMS, options.threads, logger)

assembler = None
if options.no_assembly:
    assembler = No_Assembler(None, None, options.threads, logger)
else:
    if ASSEMBLER == "trinity":
        assembler = Trinity(TRINITY_PATH, TRINITY_PARAMS, options.threads, logger)
    elif ASSEMBLER == "spades":
        assembler = SPAdes(SPADES_PATH, SPADES_PARAMS, options.threads, logger)
    elif ASSEMBLER == "idba":
        assembler = IDBA_ud(IDBA_PATH, IDBA_PARAMS, options.threads, logger)

hmmsearch = HMMsearch(HMMSEARCH_PATH, f"{HMMSEARCH_PARAMS} --domE {HMMSEARCH_EVALUE}",
                      PFAM_PATH, options.threads, logger)
blastn = Blastn(BLASTN_PATH, BLASTN_PARAMS, BLASTN_NT_PATH, options.threads, logger)
blastp = Blastp(BLASTP_PATH, BLASTP_PARAMS, BLASTP_NR_PATH, options.threads, logger)

def finish(failure_msg = nil)
  if @options.clean
    begin
      FileUtils.rm_rf(@options.derived[:sample_folder])
    rescue => e
      @logger.error "#{e}\nFailed to remove folder '#{@options.derived[:sample_folder]}'. Please remove it manually."
    end
  end

  pp @options.derived if @options.debug

  if failure_msg
    checkpoint(failure_msg, 20)
    exit(false)
  else
    checkpoint('Analysis Complete')
    cmd = "INSERT INTO analysis (finished) VALUES ('now')"
    @logger.debug(cmd)
    @conn.exec(cmd)
    exit(true)
  end
end

# ... (other parts of the script) ...

checkpoint('Generating summary report')
require_relative File.join('lib', 'wrap', 'report')
report = Report.new(@options, @logger)
report.create

# Upload results to the user-specified storage (e.g., AWS S3, Google Cloud Storage)
checkpoint('Uploading results')
if @options.upload_results
  require_relative File.join('lib', 'wrap', 'upload_results')
  uploader = UploadResults.new(@options, @logger)
  uploader.upload
end

finish()

like unlike

点赞本文的读者

还没有人对此文章表态

本文有评论

没有评论

看文章，发评论，不要沉默