<commit_before>import logging
import luigi
import sciluigi as sl
import math
from subprocess import call
import subprocess as sub
import sys
import requests
import time

# ------------------------------------------------------------------------
# Init logging
# ------------------------------------------------------------------------

log = logging.getLogger('sciluigi-interface')

# ------------------------------------------------------------------------
# Workflow class
# ------------------------------------------------------------------------

class NGITestWF(sl.WorkflowTask):
    task = luigi.Parameter() # Task to return, chosable on commandline

    def workflow(self):
        # Rsync a folder
        rsync = sl.new_task('rsync', RSyncAFolder, self,
                src_dir_path = 'data/afolder',
                dest_dir_path = 'data/afolder_rsynced')

        # Run a program that takes 10 minutes (seconds)
        run10min = sl.new_task('run10min', Run10MinuteSleep, self)
        run10min.in_upstream = rsync.out_destdir

        # Do a web request
        webreq = sl.new_task('run10min', DoWebRequest, self)
        webreq.in_upstream = run10min.out_doneflag

        # Split a file
        rawdata = sl.new_task('rawdata', ExistingData, self,
                file_name='acgt.txt')

        split = sl.new_task('run10min', SplitAFile, self)
        split.in_data = rawdata.out_acgt

        # Run the same task on the two splits
        dosth1 = sl.new_task('dosth1', DoSomething, self)
        dosth1.in_data = split.out_part1

        dosth2 = sl.new_task('dosth2', DoSomething, self)
        dosth2.in_data = split.out_part2

        # Merge the results
        merge = sl.new_task('merge', MergeFiles, self)
        merge.in_part1 = dosth1.out_data
        merge.in_part2 = dosth2.out_data

        return locals()[self.task]

# ------------------------------------------------------------------------
# Task classes
# ------------------------------------------------------------------------

# Rsync a folder
class RSyncAFolder(sl.Task):

    # Params
    src_dir_path = luigi.Parameter()
    dest_dir_path = luigi.Parameter()

    # I/O
    def out_destdir(self):
        return sl.TargetInfo(self, self.dest_dir_path)

    # Impl
    def run(self):
        call('rsync -a {src}/ {dest}/'.format(
            src = self.src_dir_path,
            dest = self.dest_dir_path),
        shell=True)

# Run a program that takes 10 minutes (seconds now, for a try) to run
class Run10MinuteSleep(sl.Task):

    # I/O
    in_upstream = None

    def out_doneflag(self):
        return sl.TargetInfo(self, self.in_upstream().path + '.10mintask_done')

    # Impl
    def run(self):
        time.sleep(10)
        with self.out_doneflag().open('w') as flagfile:
            flagfile.write('Done!')


# Perform a web request
class DoWebRequest(sl.Task):

    # I/O
    in_upstream = None

    def out_doneflag(self):
        return sl.TargetInfo(self, self.in_upstream().path + '.webrequest_done')

    # Impl
    def run(self):
        resp = requests.get('http://bils.se')
        if resp.status_code != 200:
            raise Exception('Web request failed!')
            sys.exit()
        else:
            with self.out_doneflag().open('w') as flagfile:
                flagfile.write('Web Request Task Done!')


class ExistingData(sl.ExternalTask):

    # Params
    file_name = luigi.Parameter(default='acgt.txt')

    # I/O
    def out_acgt(self):
        return sl.TargetInfo(self, 'data/' + self.file_name)


class SplitAFile(sl.Task):

    # I/O
    in_data = None<commit_msg>Log, don't raise Exception<commit_after>import logging
import luigi
import sciluigi as sl
import math
from subprocess import call
import subprocess as sub
import sys
import requests
import time

# ------------------------------------------------------------------------
# Init logging
# ------------------------------------------------------------------------

log = logging.getLogger('sciluigi-interface')

# ------------------------------------------------------------------------
# Workflow class
# ------------------------------------------------------------------------

class NGITestWF(sl.WorkflowTask):
    task = luigi.Parameter() # Task to return, chosable on commandline

    def workflow(self):
        # Rsync a folder
        rsync = sl.new_task('rsync', RSyncAFolder, self,
                src_dir_path = 'data/afolder',
                dest_dir_path = 'data/afolder_rsynced')

        # Run a program that takes 10 minutes (seconds)
        run10min = sl.new_task('run10min', Run10MinuteSleep, self)
        run10min.in_upstream = rsync.out_destdir

        # Do a web request
        webreq = sl.new_task('run10min', DoWebRequest, self)
        webreq.in_upstream = run10min.out_doneflag

        # Split a file
        rawdata = sl.new_task('rawdata', ExistingData, self,
                file_name='acgt.txt')

        split = sl.new_task('run10min', SplitAFile, self)
        split.in_data = rawdata.out_acgt

        # Run the same task on the two splits
        dosth1 = sl.new_task('dosth1', DoSomething, self)
        dosth1.in_data = split.out_part1

        dosth2 = sl.new_task('dosth2', DoSomething, self)
        dosth2.in_data = split.out_part2

        # Merge the results
        merge = sl.new_task('merge', MergeFiles, self)
        merge.in_part1 = dosth1.out_data
        merge.in_part2 = dosth2.out_data

        return locals()[self.task]

# ------------------------------------------------------------------------
# Task classes
# ------------------------------------------------------------------------

# Rsync a folder
class RSyncAFolder(sl.Task):

    # Params
    src_dir_path = luigi.Parameter()
    dest_dir_path = luigi.Parameter()

    # I/O
    def out_destdir(self):
        return sl.TargetInfo(self, self.dest_dir_path)

    # Impl
    def run(self):
        call('rsync -a {src}/ {dest}/'.format(
            src = self.src_dir_path,
            dest = self.dest_dir_path),
        shell=True)

# Run a program that takes 10 minutes (seconds now, for a try) to run
class Run10MinuteSleep(sl.Task):

    # I/O
    in_upstream = None

    def out_doneflag(self):
        return sl.TargetInfo(self, self.in_upstream().path + '.10mintask_done')

    # Impl
    def run(self):
        time.sleep(10)
        with self.out_doneflag().open('w') as flagfile:
            flagfile.write('Done!')


# Perform a web request
class DoWebRequest(sl.Task):

    # I/O
    in_upstream = None

    def out_doneflag(self):
        return sl.TargetInfo(self, self.in_upstream().path + '.webrequest_done')

    # Impl
    def run(self):
        resp = requests.get('http://bils.se')
        if resp.status_code != 200:
            log.error('Web request failed!')
        else:
            with self.out_doneflag().open('w') as flagfile:
                flagfile.write('Web Request Task Done!')


class ExistingData(sl.ExternalTask):

    # Params
    file_name = luigi.Parameter(default='acgt.txt')

    # I/O
    def out_acgt(self):
        return sl.TargetInfo(self, 'data/' + self.file_name)


class SplitAFile(sl.Task):

    # I/O
    in_data = None
{'fraction_non_alphanumeric': 0.21514055107152796, 'fraction_numerical': 0.010854439187308655, 'mean_word_length': 1.6717472118959107, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '6477391', 'n_tokens_mistral': 2220, 'n_tokens_neox': 2110, 'n_words': 484}