samuell

class TaskA(luigi.Task):
  param1 = luigi.Parameter()
  pass # We leave out the details for now


class TaskB(luigi.Task):
  param1 = luigi.Parameter()
  
  # Add a "hard-wired" depenency on TaskA
  def requires(self):

samuell

# Note: SomeOtherUpstreamTask - which is a different task than TaskA
class SomeOtherUpstreamTask(luigi.Task):
  param1 = luigi.Parameter()
  pass # We leave out the details for now
 
# Here we sub-class the original TaskB and overrides its requires method:
class MyOwnTaskB(TaskB):
  param1 = luigi.Parameter()
  # Override the requires method:
  def requires(self):

samuell

class TaskA(luigi.Task):
  param1 = luigi.Parameter()
  ... 
 
class TaskB(luigi.Task):
  param1 = luigi.Parameter()
  param2 = luigi.Parameter()  
  def requires(self):
    return TaskA(param1=self.param1)
  ...

samuell

class TaskA(luigi.Task):
  def output(self):
    # Multiple, named output TARGETS
    return { 'out1' : luigi.LocalTarget('out1.txt'),
             'out2' : luigi.LocalTarget('out2.txt') }
  ...

class TaskB(luigi.Task):
  def output(self):
    # Multiple, named output TARGETS

samuell

###### Meta class ######

class DependencyMetaTask(luigi.Task):
    # METHODS FOR AUTOMATING DEPENDENCY MANAGEMENT
    def requires(self):
        upstream_tasks = []
        for param_val in self.param_args:
            if type(param_val) is dict:
                if 'upstream' in param_val:
                    upstream_tasks.append(param_val['upstream']['task'])

samuell

use std::io::BufferedReader;
use std::io::File;
use std::str::StrSlice;

fn main() {
    let path = Path::new("Homo_sapiens.GRCh37.67.dna_rm.chromosome.Y.fa");
    let mut file = BufferedReader::new(File::open(&path));
    let mut gc = 0i; 
    let mut at = 0i; 
    for line in file.lines() {

samuell

use std::io::BufferedReader;
use std::io::File;
use std::str::StrSlice;
 
fn main() {
    let path = Path::new("Homo_sapiens.GRCh37.67.dna_rm.chromosome.Y.fa");
    let mut file = BufferedReader::new(File::open(&path));
    let mut gc = 0i; 
    let mut at = 0i; 
    for line in file.lines().filter_map(|result| result.ok()) {

samuell

class TimingMixin():
    start_time = None 
    end_time = None 
    exec_time = None 

    @luigi.Task.event_handler(luigi.Event.DEPENDENCY_DISCOVERED)
    def save_start_time(self):
        self.start_time = time.time()
        self.log("Dependency present. Starting task")

samuell

import luigi
import time

class TimeTaskMixin(object):
    '''
    A mixin that when added to a luigi task, will print out
    the tasks execution time to standard out, when the task is
    finished
    '''
    @luigi.Task.event_handler(luigi.Event.PROCESSING_TIME)

samuell

ERROR:root:Command failed: salloc -A b2013262 -p devel -n 16 -t 1:00:00 -J nana60GenerateFingerprint srun -n 1 -c 16 java -jar jars/FingerprintsGenerator.jar -fp ecfbit -inputfile /pica/v6/b2013262_nobackup/workflows/workflows/data/training_60.data -parser 1 -outputfile /pica/v6/b2013262_nobackup/workflows/workflows/data/training_60.data
ERROR:root:OUTPUT OF FAILED COMMAND: salloc: Granted job allocation 2698088; 
srun: Job step created; 
Input file: /pica/v6/b2013262_nobackup/workflows/workflows/data/training_60.data; 
Output file: /pica/v6/b2013262_nobackup/workflows/workflows/data/training_60.data.ecfbit.csr; 
Key-value log file: /pica/v6/b2013262_nobackup/workflows/workflows/data/training_60.data.ecfbit.csr.log; 
Fingerprint: null; 
Working threads: 16; 
Max non hydrogen atoms: 50; 
Min non hydrogen atoms: 5;