<commit_before>#!/usr/bin/env python
import argparse
import csv
import glob
import os
import shutil
import subprocess

import requests

from datetime import datetime

from pm.log import loggers
from pm.utils.filesystem import chdir
from pm.utils import config as cf
from pm.utils import parsers

DESCRIPTION =(" Script to keep track and pre-process Illumina X Ten runs. "

"The script will work only with X Ten runs. X Ten runs generate different file "
"structure and naming than HiSeq 2000/2500. To run this script you will also need "
"bcl2fastq V >= 2.1."

"Once a run is completed and it has been pre-processed, demultiplexed data will be "
"sent to the processing server/HPC indicated in the configuration file.")

LOG = loggers.minimal_logger('Run tracker')

def check_config_options(config):
    """ Check that all needed configuration sections/config are present

    :param dict config: Parsed configuration file
    """
    try:
        config['preprocessing']
        config['preprocessing']['hiseq_data']
        config['preprocessing']['miseq_data']
        config['preprocessing']['mfs']
        config['preprocessing']['bcl2fastq']['path']
        config['preprocessing']['status_dir']
        config['preprocessing']['samplesheets_dir']
        config['preprocessing']['sync']
        config['preprocessing']['sync']['user']
        config['preprocessing']['sync']['host']
        config['preprocessing']['sync']['data_archive']
    except KeyError:
        raise RuntimeError(("Required configuration config not found, please "
            "refer to the README file."))


def is_finished(run):
    """ Checks if a run is finished or not. Check corresponding status file

    :param str run: Run directory
    """
    return os.path.exists(os.path.join(run, 'RTAComplete.txt'))


def processing_status(run):
    """ Returns the processing status of a sequencing run. Status are:

        TO_START - The BCL conversion and demultiplexing process has not yet started 
        IN_PROGRESS - The BCL conversion and demultiplexing process is started but not completed
        COMPLETED - The BCL conversion and demultiplexing process is completed

    :param str run: Run directory
    """
    demux_dir = os.path.join(run, 'Demultiplexing')
    if not os.path.exists(demux_dir):
        return 'TO_START'
    elif os.path.exists(os.path.join(demux_dir, 'Stats', 'DemultiplexingStats.xml')):
        return 'COMPLETED'
    else:
        return 'IN_PROGRESS'


def is_transferred(run, transfer_file):
    """ Checks wether a run has been transferred to the analysis server or not

    :param str run: Run directory
    :param str transfer_file: Path to file with information about transferred runs
    """
    try:
        with open(transfer_file, 'r') as f:
            t_f = csv.reader(f, delimiter='\t')
            for row in t_f:
                #Rows have two columns: run and transfer date
                if row[0] == run:
                    return True
        if os.path.exists(os.path.join(run, '.transferring')):
            return True
        return False
    except IOError:
        return False


def transfer_run(run, config, analysis=True):
    """ Transfer a run to the analysis server. Will add group R/W permissions to
    the run directory in the destination server so that the run can be processed
    by any user/account in that group (i.e a functional account...)

    :param str run: Run directory
    :param dict config: Parsed configuration
    :param bool analysis: Trigger analysis on remote server
    """

    with chdir(run):
        cl = ['rsync', '-av']
        # Add R/W permissions to the group
        cl.append('--chmod=g+rw')
        # rsync works in a really funny way, if you don't understand this, refer to 
        # this note: http://silentorbit.com/notes/2013/08/rsync-by-extension/
        cl.append("--include=*/")
        for to_include in config['sync']['include']:
            cl.append("--include={}".format(to_include))
        cl.extend(["--exclude=*", "--prune-empty-dirs"])
        r_user = config['sync']['user']
        r_host = config['sync']['host']
        r_dir = config['sync']['data_archive']
        remote = "{}@{}:{}".format(r_user, r_host, r_dir)
        cl.extend([run, remote])

        with open('rsync.out', 'w') as rsync_out, open('rsync.err', 'w') as rsync_err:
            try:
                # Create temp file indicating that the run is being transferred
                open('.transferring', 'w').close()
                started = ("Started transfer of run {} on {}".format(os.path.basename(run), datetime.now()))
                LOG.info(started)
                rsync_out.write(started + '\n')
                rsync_out.write('Command: {}\n'.format(' '.join(cl)))
                rsync_out.write(''.join(['=']*len(cl)) + '\n')
                subprocess.check_call(cl, stdout=rsync_out, stderr=rsync_err)
            except subprocess.CalledProcessError, e:
                error_msg = ("Transfer for run {} FAILED (exit code {}), "
                             "please check log files rsync.out and rsync.err".format(
                                                        os.path.basename(run), str(e.returncode)))
                shutil.rmtree('.transferring')
                raise e

        t_file = os.path.join(config['status_dir'], 'transfer.tsv')
        LOG.info('Adding run {} to {}'.format(os.path.basename(run), t_file))
        with open(t_file, 'a') as tf:
            tsv_writer = csv.writer(tf, delimiter='\t')
            tsv_writer.writerow([os.path.basename(run), str(datetime.now())])
        shutil.rmtree('.transferring')

        if analysis:
            trigger_analysis(run, config)


def trigger_analysis(run, config):
    """ Trigger the analysis of the flowcell in the analysis sever.

    :param str run: Run directory
    :param dict config: Parsed configuration
    """
    if not config.get('analysis'):
        LOG.warn(("No configuration found for remote analysis server. Not triggering"
                  "analysis of {}".format(os.path.basename(run))))
    else:
        url = "http://{}:{}/flowcell_analysis/{}".format(config['analysis']['host'],
                                                         config['analysis']['port'],
                                                         os.path.basename(run))
        r = requests.get(url)
        if r.status_code != requests.status_codes.codes.OK:
            LOG.warn(("Something went wrong when triggering the analysis of {}. Please "
                      "check the logfile and make sure to start the analysis!".format(os.path.basename(run))))
        else:
            LOG.info('Analysis of flowcell {} triggered in {}'.format(os.path.basename(run),<commit_msg>Remove with os<commit_after>#!/usr/bin/env python
import argparse
import csv
import glob
import os
import shutil
import subprocess

import requests

from datetime import datetime

from pm.log import loggers
from pm.utils.filesystem import chdir
from pm.utils import config as cf
from pm.utils import parsers

DESCRIPTION =(" Script to keep track and pre-process Illumina X Ten runs. "

"The script will work only with X Ten runs. X Ten runs generate different file "
"structure and naming than HiSeq 2000/2500. To run this script you will also need "
"bcl2fastq V >= 2.1."

"Once a run is completed and it has been pre-processed, demultiplexed data will be "
"sent to the processing server/HPC indicated in the configuration file.")

LOG = loggers.minimal_logger('Run tracker')

def check_config_options(config):
    """ Check that all needed configuration sections/config are present

    :param dict config: Parsed configuration file
    """
    try:
        config['preprocessing']
        config['preprocessing']['hiseq_data']
        config['preprocessing']['miseq_data']
        config['preprocessing']['mfs']
        config['preprocessing']['bcl2fastq']['path']
        config['preprocessing']['status_dir']
        config['preprocessing']['samplesheets_dir']
        config['preprocessing']['sync']
        config['preprocessing']['sync']['user']
        config['preprocessing']['sync']['host']
        config['preprocessing']['sync']['data_archive']
    except KeyError:
        raise RuntimeError(("Required configuration config not found, please "
            "refer to the README file."))


def is_finished(run):
    """ Checks if a run is finished or not. Check corresponding status file

    :param str run: Run directory
    """
    return os.path.exists(os.path.join(run, 'RTAComplete.txt'))


def processing_status(run):
    """ Returns the processing status of a sequencing run. Status are:

        TO_START - The BCL conversion and demultiplexing process has not yet started 
        IN_PROGRESS - The BCL conversion and demultiplexing process is started but not completed
        COMPLETED - The BCL conversion and demultiplexing process is completed

    :param str run: Run directory
    """
    demux_dir = os.path.join(run, 'Demultiplexing')
    if not os.path.exists(demux_dir):
        return 'TO_START'
    elif os.path.exists(os.path.join(demux_dir, 'Stats', 'DemultiplexingStats.xml')):
        return 'COMPLETED'
    else:
        return 'IN_PROGRESS'


def is_transferred(run, transfer_file):
    """ Checks wether a run has been transferred to the analysis server or not

    :param str run: Run directory
    :param str transfer_file: Path to file with information about transferred runs
    """
    try:
        with open(transfer_file, 'r') as f:
            t_f = csv.reader(f, delimiter='\t')
            for row in t_f:
                #Rows have two columns: run and transfer date
                if row[0] == run:
                    return True
        if os.path.exists(os.path.join(run, '.transferring')):
            return True
        return False
    except IOError:
        return False


def transfer_run(run, config, analysis=True):
    """ Transfer a run to the analysis server. Will add group R/W permissions to
    the run directory in the destination server so that the run can be processed
    by any user/account in that group (i.e a functional account...)

    :param str run: Run directory
    :param dict config: Parsed configuration
    :param bool analysis: Trigger analysis on remote server
    """

    with chdir(run):
        cl = ['rsync', '-av']
        # Add R/W permissions to the group
        cl.append('--chmod=g+rw')
        # rsync works in a really funny way, if you don't understand this, refer to 
        # this note: http://silentorbit.com/notes/2013/08/rsync-by-extension/
        cl.append("--include=*/")
        for to_include in config['sync']['include']:
            cl.append("--include={}".format(to_include))
        cl.extend(["--exclude=*", "--prune-empty-dirs"])
        r_user = config['sync']['user']
        r_host = config['sync']['host']
        r_dir = config['sync']['data_archive']
        remote = "{}@{}:{}".format(r_user, r_host, r_dir)
        cl.extend([run, remote])

        with open('rsync.out', 'w') as rsync_out, open('rsync.err', 'w') as rsync_err:
            try:
                # Create temp file indicating that the run is being transferred
                open('.transferring', 'w').close()
                started = ("Started transfer of run {} on {}".format(os.path.basename(run), datetime.now()))
                LOG.info(started)
                rsync_out.write(started + '\n')
                rsync_out.write('Command: {}\n'.format(' '.join(cl)))
                rsync_out.write(''.join(['=']*len(cl)) + '\n')
                subprocess.check_call(cl, stdout=rsync_out, stderr=rsync_err)
            except subprocess.CalledProcessError, e:
                error_msg = ("Transfer for run {} FAILED (exit code {}), "
                             "please check log files rsync.out and rsync.err".format(
                                                        os.path.basename(run), str(e.returncode)))
                os.remove('.transferring')
                raise e

        t_file = os.path.join(config['status_dir'], 'transfer.tsv')
        LOG.info('Adding run {} to {}'.format(os.path.basename(run), t_file))
        with open(t_file, 'a') as tf:
            tsv_writer = csv.writer(tf, delimiter='\t')
            tsv_writer.writerow([os.path.basename(run), str(datetime.now())])
        os.remove('.transferring')

        if analysis:
            trigger_analysis(run, config)


def trigger_analysis(run, config):
    """ Trigger the analysis of the flowcell in the analysis sever.

    :param str run: Run directory
    :param dict config: Parsed configuration
    """
    if not config.get('analysis'):
        LOG.warn(("No configuration found for remote analysis server. Not triggering"
                  "analysis of {}".format(os.path.basename(run))))
    else:
        url = "http://{}:{}/flowcell_analysis/{}".format(config['analysis']['host'],
                                                         config['analysis']['port'],
                                                         os.path.basename(run))
        r = requests.get(url)
        if r.status_code != requests.status_codes.codes.OK:
            LOG.warn(("Something went wrong when triggering the analysis of {}. Please "
                      "check the logfile and make sure to start the analysis!".format(os.path.basename(run))))
        else:
            LOG.info('Analysis of flowcell {} triggered in {}'.format(os.path.basename(run),
{'fraction_non_alphanumeric': 0.11911993480998592, 'fraction_numerical': 0.0028150233350618566, 'mean_word_length': 2.4820737683776115, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '623844', 'n_tokens_mistral': 3797, 'n_tokens_neox': 3656, 'n_words': 1151}