<commit_before>#!/usr/bin/env python
import os, subprocess
from shlex import quote
from shutil import copytree
from fnmatch import fnmatch
from igf_data.utils.fileutils import get_temp_dir,remove_dir
from igf_data.utils.fileutils import create_file_manifest_for_dir
from igf_data.utils.fileutils import prepare_file_archive
from ehive.runnable.IGFBaseProcess import IGFBaseProcess
from igf_data.utils.tools.cellranger.cellranger_count_utils import get_cellranger_count_input_list
from igf_data.utils.tools.cellranger.cellranger_count_utils import check_cellranger_count_output
from igf_data.utils.tools.reference_genome_utils import Reference_genome_utils
from igf_data.utils.analysis_collection_utils import Analysis_collection_utils

class RunCellrangerCount(IGFBaseProcess):
  '''
  A ehive process class for running cellranger count pipeline
  '''
  def param_defaults(self):
    params_dict=super(RunCellrangerCount,self).param_defaults()
    params_dict.update({
        'force_overwrite':True,
        'cellranger_exe':None,
        'fastq_collection_type':'demultiplexed_fastq',
        'reference_type':'TRANSCRIPTOME_TENX',
        'cellranger_options':'{"--nopreflight":"","--disable-ui":"", "--jobmode":"pbspro","--localcores":"1","--localmem":"4","--mempercore":"4","--maxjobs":"20"}',
        'job_timeout':43200,
        'manifest_filename':'file_manifest.csv',
        'analysis_name':'cellranger_count',
        'collection_type':'CELLRANGER_RESULTS',
        'collection_table':'experiment',
      })
    return params_dict

  def run(self):
    '''
    A method for running the cellranger count for a given sample using ehive pipeline
    
    :param project_igf_id: A project igf id
    :param experiment_igf_id: An experiment igf id
    :param sample_igf_id: A sample igf id
    :param igf_session_class: A database session class
    :param cellranger_exe: Cellranger executable path
    :param cellranger_options: Cellranger parameters
                               
                               List of default parameters
                                 --jobmode=pbspro
                                 --localcores=1
                                 --localmem=4
                                 --mempercore=4
                                 --maxjobs=20
    
    :param base_work_dir: Base work directory path
    :param fastq_collection_type: Collection type name for input fastq files, default demultiplexed_fastq
    :param species_name: Reference genome collection name
    :param reference_type: Reference genome collection type, default TRANSCRIPTOME_TENX
    :param job_timeout: Timeout for cellranger job, default 24hrs
    :returns: Adding cellranger_output to the dataflow_params
    '''
    try:
      project_igf_id=self.param_required('project_igf_id')
      experiment_igf_id=self.param_required('experiment_igf_id')
      sample_igf_id=self.param_required('sample_igf_id')
      sample_submitter_id=self.param_required('sample_submitter_id')
      igf_session_class=self.param_required('igf_session_class')
      cellranger_exe=self.param_required('cellranger_exe')
      cellranger_options=self.param_required('cellranger_options')
      base_work_dir=self.param_required('base_work_dir')
      base_result_dir=self.param_required('base_results_dir')
      fastq_collection_type=self.param_required('fastq_collection_type')
      job_timeout=self.param_required('job_timeout')
      species_name=self.param('species_name')
      reference_type=self.param('reference_type')
      manifest_filename=self.param('manifest_filename')
      analysis_name=self.param('analysis_name')
      collection_type=self.param('collection_type')
      collection_table=self.param('collection_table')
      # setup work dir for run
      work_dir=False
      work_dir_prefix=os.path.join(base_work_dir,
                                   project_igf_id,
                                   sample_igf_id,
                                   experiment_igf_id)
      work_dir=self.get_job_work_dir(work_dir=work_dir_prefix)                  # replace this with temp dir while running in queue
      # setup env for run
      os.chdir(work_dir)                                                        # move to work dir
      os.environ['PATH'] += '{0}{1}'.format(os.pathsep,
                                            os.path.dirname(cellranger_exe))    # add cellranger location to env PATH
      # collect reference genome for run
      ref_genome=Reference_genome_utils(genome_tag=species_name,
                                        dbsession_class=igf_session_class,
                                        tenx_ref_type=reference_type)
      # collect fastq input for run
      cellranger_ref_transcriptome=ref_genome.get_transcriptome_tenx()          # fetch tenx ref transcriptome from db
      input_fastq_dirs=get_cellranger_count_input_list(\
                         db_session_class=igf_session_class,
                         experiment_igf_id=experiment_igf_id,
                         fastq_collection_type=fastq_collection_type)           # fetch fastq dir paths as list for run
      # configure cellranger count command for run
      cellranger_options=self.format_tool_options(cellranger_options,
                                                  separator='=')
      cellranger_cmd=[cellranger_exe,
                      'count',
                      '{0}={1}'.format('--fastqs',
                                       quote(','.join(input_fastq_dirs))),
                      '{0}={1}'.format('--id',
                                       quote(experiment_igf_id)),
                      '{0}={1}'.format('--transcriptome',
                                       quote(cellranger_ref_transcriptome)),
                     ]                                                          # set initial parameters
      cellranger_cmd.extend(cellranger_options)                                 # add optional parameters
      # log before job submission
      message='started cellranger count for {0}, {1} {2}'.\
              format(project_igf_id,
                     sample_igf_id,
                     experiment_igf_id)
      self.post_message_to_slack(message,reaction='pass')                       # send log to slack
      self.comment_asana_task(task_name=project_igf_id, comment=message)        # send comment to Asana
      message=' '.join(cellranger_cmd)
      self.comment_asana_task(task_name=project_igf_id, comment=message)        # send cellranger command to Asana
      # start job execution
      subprocess.check_call(cellranger_cmd,
                            timeout=job_timeout)                                # run cellranger count
      # prepare output after cellranger run
      cellranger_output=os.path.join(work_dir,<commit_msg>running cellranger command using shell<commit_after>#!/usr/bin/env python
import os, subprocess
from shlex import quote
from shutil import copytree
from fnmatch import fnmatch
from igf_data.utils.fileutils import get_temp_dir,remove_dir
from igf_data.utils.fileutils import create_file_manifest_for_dir
from igf_data.utils.fileutils import prepare_file_archive
from ehive.runnable.IGFBaseProcess import IGFBaseProcess
from igf_data.utils.tools.cellranger.cellranger_count_utils import get_cellranger_count_input_list
from igf_data.utils.tools.cellranger.cellranger_count_utils import check_cellranger_count_output
from igf_data.utils.tools.reference_genome_utils import Reference_genome_utils
from igf_data.utils.analysis_collection_utils import Analysis_collection_utils

class RunCellrangerCount(IGFBaseProcess):
  '''
  A ehive process class for running cellranger count pipeline
  '''
  def param_defaults(self):
    params_dict=super(RunCellrangerCount,self).param_defaults()
    params_dict.update({
        'force_overwrite':True,
        'cellranger_exe':None,
        'fastq_collection_type':'demultiplexed_fastq',
        'reference_type':'TRANSCRIPTOME_TENX',
        'cellranger_options':'{"--nopreflight":"","--disable-ui":"", "--jobmode":"pbspro","--localcores":"1","--localmem":"4","--mempercore":"4","--maxjobs":"20"}',
        'job_timeout':43200,
        'manifest_filename':'file_manifest.csv',
        'analysis_name':'cellranger_count',
        'collection_type':'CELLRANGER_RESULTS',
        'collection_table':'experiment',
      })
    return params_dict

  def run(self):
    '''
    A method for running the cellranger count for a given sample using ehive pipeline
    
    :param project_igf_id: A project igf id
    :param experiment_igf_id: An experiment igf id
    :param sample_igf_id: A sample igf id
    :param igf_session_class: A database session class
    :param cellranger_exe: Cellranger executable path
    :param cellranger_options: Cellranger parameters
                               
                               List of default parameters
                                 --jobmode=pbspro
                                 --localcores=1
                                 --localmem=4
                                 --mempercore=4
                                 --maxjobs=20
    
    :param base_work_dir: Base work directory path
    :param fastq_collection_type: Collection type name for input fastq files, default demultiplexed_fastq
    :param species_name: Reference genome collection name
    :param reference_type: Reference genome collection type, default TRANSCRIPTOME_TENX
    :param job_timeout: Timeout for cellranger job, default 24hrs
    :returns: Adding cellranger_output to the dataflow_params
    '''
    try:
      project_igf_id=self.param_required('project_igf_id')
      experiment_igf_id=self.param_required('experiment_igf_id')
      sample_igf_id=self.param_required('sample_igf_id')
      sample_submitter_id=self.param_required('sample_submitter_id')
      igf_session_class=self.param_required('igf_session_class')
      cellranger_exe=self.param_required('cellranger_exe')
      cellranger_options=self.param_required('cellranger_options')
      base_work_dir=self.param_required('base_work_dir')
      base_result_dir=self.param_required('base_results_dir')
      fastq_collection_type=self.param_required('fastq_collection_type')
      job_timeout=self.param_required('job_timeout')
      species_name=self.param('species_name')
      reference_type=self.param('reference_type')
      manifest_filename=self.param('manifest_filename')
      analysis_name=self.param('analysis_name')
      collection_type=self.param('collection_type')
      collection_table=self.param('collection_table')
      # setup work dir for run
      work_dir=False
      work_dir_prefix=os.path.join(base_work_dir,
                                   project_igf_id,
                                   sample_igf_id,
                                   experiment_igf_id)
      work_dir=self.get_job_work_dir(work_dir=work_dir_prefix)                  # replace this with temp dir while running in queue
      # setup env for run
      os.chdir(work_dir)                                                        # move to work dir
      os.environ['PATH'] += '{0}{1}'.format(os.pathsep,
                                            os.path.dirname(cellranger_exe))    # add cellranger location to env PATH
      # collect reference genome for run
      ref_genome=Reference_genome_utils(genome_tag=species_name,
                                        dbsession_class=igf_session_class,
                                        tenx_ref_type=reference_type)
      # collect fastq input for run
      cellranger_ref_transcriptome=ref_genome.get_transcriptome_tenx()          # fetch tenx ref transcriptome from db
      input_fastq_dirs=get_cellranger_count_input_list(\
                         db_session_class=igf_session_class,
                         experiment_igf_id=experiment_igf_id,
                         fastq_collection_type=fastq_collection_type)           # fetch fastq dir paths as list for run
      # configure cellranger count command for run
      cellranger_options=self.format_tool_options(cellranger_options,
                                                  separator='=')
      cellranger_cmd=[cellranger_exe,
                      'count',
                      '{0}={1}'.format('--fastqs',
                                       quote(','.join(input_fastq_dirs))),
                      '{0}={1}'.format('--id',
                                       quote(experiment_igf_id)),
                      '{0}={1}'.format('--transcriptome',
                                       quote(cellranger_ref_transcriptome)),
                     ]                                                          # set initial parameters
      cellranger_cmd.extend(cellranger_options)                                 # add optional parameters
      # log before job submission
      message='started cellranger count for {0}, {1} {2}'.\
              format(project_igf_id,
                     sample_igf_id,
                     experiment_igf_id)
      self.post_message_to_slack(message,reaction='pass')                       # send log to slack
      self.comment_asana_task(task_name=project_igf_id, comment=message)        # send comment to Asana
      message=' '.join(cellranger_cmd)
      self.comment_asana_task(task_name=project_igf_id, comment=message)        # send cellranger command to Asana
      # start job execution
      cellranger_cmd=' '.join(cellranger_cmd)                                   # create shell command string
      subprocess.check_call(cellranger_cmd,
                            shell=True,
                            timeout=job_timeout)                                # run cellranger count using shell
      # prepare output after cellranger run
      cellranger_output=os.path.join(work_dir,
{'fraction_non_alphanumeric': 0.07760307224114195, 'fraction_numerical': 0.004057676979928991, 'mean_word_length': 2.300334768053563, 'pattern_counts': {'":': 14, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 49, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '9648', 'n_tokens_mistral': 3889, 'n_tokens_neox': 3727, 'n_words': 690}