cjs3/cjs/automate.py

#!/usr/bin/env python3

import threading
import click
import subprocess
import shlex
import sys
import queue

import utils.io
from utils.io import read_json, write_csv
import utils.net
from utils.net import send_email

import signal

def signal_handler(sig, frame):
    print('Exiting automate.')
    sys.exit(0)

signal.signal(signal.SIGINT, signal_handler)

# from https://www.endpoint.com/blog/2015/01/28/getting-realtime-output-using-python
def run_command(command):
    return subprocess.call(command)

# main-entry point to the automate feature
@click.group(name='automate')
@click.option('-w', '--workers', 
              type=int, default=1,
              help='Set number of worker threads.')
@click.pass_context
def automate(ctx, workers):
    '''Bulk test systems.'''
    ctx.obj['workers'] = workers

def send_error(subject, message):
    send_email('resources/mailer.json', 
               'cucjsoftware@gmail.com', 
               subject, message)

def _efm(q):
    from core.vault import next_job, all_jobs
    # load the custom headless euroformix package
    '''
    rjson = importr('rjson')
    euroformix = importr('euroformix.headless')
    '''
    worker_id = 1

    while True:
        # get next job
        evidence_table, comparison_table = q.get()

        # open input files from job
        evidence = read_json(evidence_table.path)
        comparison = read_json(comparison_table.path)

        # convert them to euroformix format
        # evidence: reqbt -> euroformix
        # initialize csv with header
        evidence_header = [
            'Sample  Name', 'Marker', # NOTE: THIS HAS TWO SPACES - IS THAT OK?
            'Allele 1', 'Allele 2', 'Allele 3', 
            'Allele 4', 'Allele 5', 'Allele 6',
            'Height 1', 'Height 2', 'Height 3',
            'Height 4', 'Height 5', 'Height 6',
            'ADO', 'UD1'
        ]
        comparison_header = [
            'SampleName',
            'Marker',
            'Allele1',
            'Allele2'
        ]

        evidence_loci = [
            'AMEL',
            'D3S1358',
            'TH01',
            'D21S11',
            'D18S51',
            'D10S1248',
            'D1S1656',
            'D2S1338',
            'D16S539',
            'D22S1045',
            'vWA',
            'D8S1179',
            'FGA',
            'D2S441',
            'D12S391',
            'D19S433',
            'SE33'
        ]
        comparison_loci = [
            'D3S1358',
            'TH01',
            'D21S11',
            'D18S51',
            'D10S1248',
            'D1S1656',
            'D2S1338',
            'D16S539',
            'D22S1045',
            'VWA', # NOTE: caps?
            'D8S1179',
            'FGA',
            'D2S441',
            'D12S391',
            'D19S433',
            'SE33'
        ]

        from pprint import pprint

        for replicate in evidence['replicates']:
            tmp_evidence = []
            max_values = 6
            for i in range(17):
                tmp_evidence.append([evidence['name']])
            for locus, row in zip(evidence_loci, tmp_evidence):
                row.append(locus)
                try:
                    values = replicate[locus]
                    curr_values = 0
                    for v in values:
                        if curr_values > 5: # only allow 6 values
                            print('Warning: %s may have more than 6 loci values.' % evidence['name'])
                            break
                        row.append(v)
                        curr_values += 1
                except KeyError:
                    row.append('false')
                except Exception as e:
                    print(e)
                    print('Something went seriously wrong.')
                    sys.exit(1)
            break # NOTE: only get first replicate

        pprint(tmp_evidence)
        print('----------------------------------------------------------------')
        tmp_comparison = []
        for i in range(len(comparison_loci)):
            tmp_comparison.append([comparison['name']])
        for locus, row in zip(comparison_loci, tmp_comparison):
            row.append(locus)
            try:
                if locus == 'VWA':
                    locus = 'vWA'
                values = comparison[locus] # NOTE: use uppercase here
                for v in values:
                    row.append(v)
            except KeyError:
                row.append('') # append 2 blank values
                row.append('')
                print('Bad key: %s: Does this exist?' % locus)
                pass
            except Exception as e:
                print(e)
                print('Something went seriously wrong.')
                sys.exit(1)
        pprint(tmp_comparison)





        # comparison: reqbt -> euroformix
        sys.exit()

        try:
            # call the entry-point function
            # TODO: put settings.json into vault table
            #print(euroformix.headless_efm(evidence.path, comparison.path, 'resources/settings.json'))
            print('\n' + '-'*64)
            err = run_command(['euroformix.headless/R/efm_headless.R', evidence.path, comparison.path])
            if err:
                send_error('Worker %d: RRuntimeError' % worker_id, 
                           'Something went wrong!\nSorry,\nWorker %d' % (worker_id))
                break           
            print('\n' + '-'*64)
        except Exception as e:
            send_error('Worker %d: RRuntimeError' % worker_id, 
                           'An exception occurred: %s\nSorry,\nWorker %d' % (e, worker_id))

# main-entry point to the euroformix subcommand
@automate.command()
@click.argument('vault')
@click.pass_context
def efm(ctx, vault):
    from core.vault import all_jobs
    num_workers = ctx.obj['workers']

    q = queue.Queue()
    threads = []
    for i in range(0, num_workers):
        t = threading.Thread(target=_efm, args=(q,))
        t.start()
        threads.append(t)
    jobs = all_jobs(vault)
    print('''
\    / _ _  _  _ | _  _
 \/\/ | (_|| |(_||(/_| 
               _|       v0.1.0
     System: EuroForMix v2.1.0
     Total Jobs: %d
     Workers: %d
    ''' % (len(jobs), num_workers), end='')
    for job in jobs:
        q.put(job)

    q.join()
    for i in range(num_workers):
        q.put(None)
    for t in threads:
        t.join()