<commit_before>"""Server which listens for finished NGS runs, processing for upload to galaxy.

Usage:
    analyze_finished_sqn.py <Galaxy config file> <Post-processing config file>

Need to configure the RabbitMQ server with:

    rabbitmqctl add_user galaxy password
    rabbitmqctl add_vhost galaxy_messaging_engine
    rabbitmqctl set_permissions -p galaxy_messaging_engine galaxy '.*' '.*' '.*'
"""
import os
import sys
import ConfigParser
import json
import subprocess
import contextlib

import yaml
from amqplib import client_0_8 as amqp

def main(galaxy_config, processing_config):
    amqp_config = _read_amqp_config(galaxy_config)
    with open(processing_config) as in_handle:
        config = yaml.load(in_handle)
    process_tag = config["msg_process_tag"]
    handlers = [(process_tag,
        analysis_handler(config, process_tag, processing_config))]
    message_reader(handlers, amqp_config)

def copy_and_analyze(remote_info, config, config_file):
    """Remote copy an output directory, process it, and upload to Galaxy.
    """
    print remote_info
    fc_dir = _remote_copy(remote_info, config["local_sqn_dir"])
    print fc_dir
    analysis_dir = os.path.join(config["analysis"]["base_dir"],
                                os.path.basename(remote_info["directory"]))
    if not config_file.startswith("/"):
        config_file = os.path.join(os.getcwd(), config_file)
    with _make_and_chdir(analysis_dir):
        cl = [config["analysis"]["process_program"], config_file, fc_dir]
        subprocess.check_call(cl)
    cl = [config["analysis"]["upload_program"], config_file, fc_dir, analysis_dir]
    subprocess.check_call(cl)

def _remote_copy(remote_info, local_sqn_dir):
    """Securely copy files from remote directory to the processing server.

    This requires ssh public keys to be setup so that no password entry
    is necessary.
    """
    fc_dir = os.path.join(local_sqn_dir,
            os.path.basename(remote_info['directory']))
    if not os.path.exists(fc_dir):
        os.makedirs(fc_dir)
    for fcopy in remote_info['to_copy']:
        target_loc = os.path.join(fc_dir, fcopy)
        if not os.path.exists(target_loc):
            target_dir = os.path.dirname(target_loc)
            if not os.path.exists(target_dir):
                os.makedirs(target_dir)
            cl = ["rsync", "-craz", "%s@%s:%s/%s" % (remote_info["user"],
                      remote_info["hostname"], remote_info["directory"], fcopy),
                  target_loc]
            subprocess.check_call(cl)
    return fc_dir

def analysis_handler(processing_config, tag_name, config_file):
    def receive_msg(msg):
        if msg.properties['application_headers'].get('msg_type') == tag_name:
            copy_and_analyze(json.loads(msg.body), processing_config,
                config_file)
    return receive_msg

def message_reader(handlers, config):
    """Wait for messages with the give tag, passing on to the supplied handler.
    """
    conn = amqp.Connection(host=config['host'] + ":" + config['port'],
                           userid=config['userid'], password=config['password'],
                           virtual_host=config['virtual_host'], insist=False)
    chan = conn.channel()
    for tag_name, handler in handlers:
        chan.queue_declare(queue=tag_name, exclusive=False, auto_delete=False,
                durable=True)
        chan.exchange_declare(exchange=config['exchange'], type="fanout", durable=True,
                auto_delete=False)
        chan.queue_bind(queue=tag_name, exchange=config['exchange'],
                        routing_key=config['routing_key'])
        chan.basic_consume(queue=tag_name, no_ack=True,
                           callback=handler, consumer_tag=tag_name)
    while True:<commit_msg>Switch back to scp; rsync is creating additional directories<commit_after>"""Server which listens for finished NGS runs, processing for upload to galaxy.

Usage:
    analyze_finished_sqn.py <Galaxy config file> <Post-processing config file>

Need to configure the RabbitMQ server with:

    rabbitmqctl add_user galaxy password
    rabbitmqctl add_vhost galaxy_messaging_engine
    rabbitmqctl set_permissions -p galaxy_messaging_engine galaxy '.*' '.*' '.*'
"""
import os
import sys
import ConfigParser
import json
import subprocess
import contextlib

import yaml
from amqplib import client_0_8 as amqp

def main(galaxy_config, processing_config):
    amqp_config = _read_amqp_config(galaxy_config)
    with open(processing_config) as in_handle:
        config = yaml.load(in_handle)
    process_tag = config["msg_process_tag"]
    handlers = [(process_tag,
        analysis_handler(config, process_tag, processing_config))]
    message_reader(handlers, amqp_config)

def copy_and_analyze(remote_info, config, config_file):
    """Remote copy an output directory, process it, and upload to Galaxy.
    """
    print remote_info
    fc_dir = _remote_copy(remote_info, config["local_sqn_dir"])
    print fc_dir
    analysis_dir = os.path.join(config["analysis"]["base_dir"],
                                os.path.basename(remote_info["directory"]))
    if not config_file.startswith("/"):
        config_file = os.path.join(os.getcwd(), config_file)
    with _make_and_chdir(analysis_dir):
        cl = [config["analysis"]["process_program"], config_file, fc_dir]
        subprocess.check_call(cl)
    cl = [config["analysis"]["upload_program"], config_file, fc_dir, analysis_dir]
    subprocess.check_call(cl)

def _remote_copy(remote_info, local_sqn_dir):
    """Securely copy files from remote directory to the processing server.

    This requires ssh public keys to be setup so that no password entry
    is necessary.
    """
    fc_dir = os.path.join(local_sqn_dir,
            os.path.basename(remote_info['directory']))
    if not os.path.exists(fc_dir):
        os.makedirs(fc_dir)
    for fcopy in remote_info['to_copy']:
        target_loc = os.path.join(fc_dir, fcopy)
        if not os.path.exists(target_loc):
            target_dir = os.path.dirname(target_loc)
            if not os.path.exists(target_dir):
                os.makedirs(target_dir)
            cl = ["scp", "-r", "%s@%s:%s/%s" % (remote_info["user"],
                      remote_info["hostname"], remote_info["directory"], fcopy),
                  target_loc]
            subprocess.check_call(cl)
    return fc_dir

def analysis_handler(processing_config, tag_name, config_file):
    def receive_msg(msg):
        if msg.properties['application_headers'].get('msg_type') == tag_name:
            copy_and_analyze(json.loads(msg.body), processing_config,
                config_file)
    return receive_msg

def message_reader(handlers, config):
    """Wait for messages with the give tag, passing on to the supplied handler.
    """
    conn = amqp.Connection(host=config['host'] + ":" + config['port'],
                           userid=config['userid'], password=config['password'],
                           virtual_host=config['virtual_host'], insist=False)
    chan = conn.channel()
    for tag_name, handler in handlers:
        chan.queue_declare(queue=tag_name, exclusive=False, auto_delete=False,
                durable=True)
        chan.exchange_declare(exchange=config['exchange'], type="fanout", durable=True,
                auto_delete=False)
        chan.queue_bind(queue=tag_name, exchange=config['exchange'],
                        routing_key=config['routing_key'])
        chan.basic_consume(queue=tag_name, no_ack=True,
                           callback=handler, consumer_tag=tag_name)
    while True:
{'fraction_non_alphanumeric': 0.10611653045083047, 'fraction_numerical': 0.0005272871078302136, 'mean_word_length': 3.2314556609035137, 'pattern_counts': {'":': 2, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3199193', 'n_tokens_mistral': 2308, 'n_tokens_neox': 2258, 'n_words': 501}