gwf_decorators_with_example.py

def prepend_line(line, rest):
    return line + '\n' + rest


class Transformer:
    def __call__(self, f):
        def wrapper(*args, **kwargs):
            options, spec = f(*args, **kwargs)
            return self.transform(options, spec)
        return wrapper

    
class with_conda(Transformer):
    def __init__(self, env_name):
        self.env_name = env_name
    
    def transform(self, options, spec):
        return options, prepend_line('source activate {}'.format(self.env_name), spec)


class with_envvar(Transformer):
    def __init__(self, envvar_name, envvar_value):
        self.envvar_name = envvar_name
        self.envvar_value = envvar_value
    
    def transform(self, options, spec):
        return options, prepend_line('export {}="{}"'.format(self.envvar_name, self.envvar_value), spec)


@with_conda('rubisco')
@with_envvar('BLASTDB', '/home/das/faststorage/dbs/')
def blast(query, out, max_target_seqs=10000, evalue=10e-6, db='nr', num_threads=4):
    options = dict(inputs=[query], outputs=[out], cores=num_threads, memory='2g', walltime='04:00:00')

    spec = '''
    blastp \
        -db {db} \
        -evalue {evalue} \
        -outfmt "10 qseqid sgi pident evalue bitscore qcovs qcovhsp qcovus" \
        -max_target_seqs {max_target_seqs} \
        -num_threads {num_threads} \
        -query {query} \
        -out {out}
    '''.format(**locals())

    return options, spec