<commit_before>import luigi
import time
import random
import string
from collections import namedtuple

# ==============================================================================
# Methods for simplifying creation of (output) targets

def create_file_targets(target_spec=None, **kwargs):
    if len(kwargs) > 0:
        return {name : luigi.LocalTarget(path) for name, path in kwargs.iteritems()}
    else:
        return {name : luigi.LocalTarget(path) for name, path in target_spec.iteritems()}

# ==============================================================================

# Named tuple, used for sending specification of which target, from which
# task, to use, when stitching workflow tasks' outputs and inputs together.
TargetSpec = namedtuple('TargetSpec', ['task', 'output'], rename=True)

# ==============================================================================

class TargetSpecParameter(luigi.Parameter):
    '''
    Parameter whose value is a Target, or actually a TargetSpec
    '''

    def parse(self, s):
        # One could maybe do something more fancy here?
        return s

# ==============================================================================

class DependencyHelpers():
    '''
    Mixin implementing methods for supporting dynamic, and target-based
    workflow definition, as opposed to the task-based one in vanilla luigi.
    '''

    def requires(self):
        return self._upstream_tasks()

    def _upstream_tasks(self):
        upstream_tasks = []
        for param_val in self.param_args:
            if type(param_val) is TargetSpec:
                upstream_tasks.append(param_val[0])
        return upstream_tasks

    # Methods for dynamic wiring of workflow

    def output_spec(self, output_name):
        '''
        Return a specification for an output of a task, to be injected
        into the target-parameters of downstream tasks, whereafter
        the specified task can be obtained by the get_input() method
        of that (downstream task.
        '''
        #return { 'upstream' : { 'task' : self, 'port' : portname } }
        return TargetSpec(task=self, output=output_name)

    def outspec(self, output_name):
        '''
        Short version of output_spec()
        '''
        return self.output_spec(output_name)

    def input(self, input_name):
        '''
        Retrieve the task
        '''
        param = self.param_kwargs[input_name]
        if type(param) is TargetSpec:
            return param[0].output()[param[1]]
        else:
            return param

    def get_path(self, input_name):
        return self.input(input_name).path


# ==============================================================================

class SciLuigiTask(DependencyHelpers, luigi.Task):
    pass

# ==============================================================================

class SciLuigiExternalTask(DependencyHelpers, luigi.ExternalTask):
    pass

# ==============================================================================

class WorkflowTask(luigi.Task):

    def output(self):
        timestamp = time.strftime('%Y%m%d_%H%M%S', time.localtime())
        clsname = self.__class__.__name__
        return luigi.LocalTarget('workflow_' + clsname.lower() + '_completed_at_{t}'.format(t=timestamp))

    def run(self):
        timestamp = time.strftime('%Y%m%d.%H%M%S', time.localtime())
        with self.output().open('w') as outfile:
            outfile.write('workflow finished at {t}'.format(t=timestamp))

# ==============================================================================

class HPCHelpers():
    '''
    Mixin with various convenience methods that most tasks need, such as for executing SLURM
    commands
    '''
    accounted_project = luigi.Parameter()

    # Main Execution methods
    def execute_in_configured_mode(self, command):
        '''Execute either locally or via SLURM, depending on config'''

        if self.get_task_config("runmode") == "local":
            self.execute_command(command)

        elif self.get_task_config("runmode") == "nodejob":
            train_size = "NA"
            if hasattr(self, 'train_size'):
                train_size = self.train_size
            replicate_id = "NA"
            if hasattr(self, 'replicate_id'):
                replicate_id = self.replicate_id

            self.execute_hpcjob(command,
                    accounted_project = self.accounted_project,
                    time_limit = self.get_task_config("time_limit"),
                    partition  = self.get_task_config("partition"),<commit_msg>Time stamp formatting<commit_after>import luigi
import time
import random
import string
from collections import namedtuple

# ==============================================================================
# Methods for simplifying creation of (output) targets

def create_file_targets(target_spec=None, **kwargs):
    if len(kwargs) > 0:
        return {name : luigi.LocalTarget(path) for name, path in kwargs.iteritems()}
    else:
        return {name : luigi.LocalTarget(path) for name, path in target_spec.iteritems()}

# ==============================================================================

# Named tuple, used for sending specification of which target, from which
# task, to use, when stitching workflow tasks' outputs and inputs together.
TargetSpec = namedtuple('TargetSpec', ['task', 'output'], rename=True)

# ==============================================================================

class TargetSpecParameter(luigi.Parameter):
    '''
    Parameter whose value is a Target, or actually a TargetSpec
    '''

    def parse(self, s):
        # One could maybe do something more fancy here?
        return s

# ==============================================================================

class DependencyHelpers():
    '''
    Mixin implementing methods for supporting dynamic, and target-based
    workflow definition, as opposed to the task-based one in vanilla luigi.
    '''

    def requires(self):
        return self._upstream_tasks()

    def _upstream_tasks(self):
        upstream_tasks = []
        for param_val in self.param_args:
            if type(param_val) is TargetSpec:
                upstream_tasks.append(param_val[0])
        return upstream_tasks

    # Methods for dynamic wiring of workflow

    def output_spec(self, output_name):
        '''
        Return a specification for an output of a task, to be injected
        into the target-parameters of downstream tasks, whereafter
        the specified task can be obtained by the get_input() method
        of that (downstream task.
        '''
        #return { 'upstream' : { 'task' : self, 'port' : portname } }
        return TargetSpec(task=self, output=output_name)

    def outspec(self, output_name):
        '''
        Short version of output_spec()
        '''
        return self.output_spec(output_name)

    def input(self, input_name):
        '''
        Retrieve the task
        '''
        param = self.param_kwargs[input_name]
        if type(param) is TargetSpec:
            return param[0].output()[param[1]]
        else:
            return param

    def get_path(self, input_name):
        return self.input(input_name).path


# ==============================================================================

class SciLuigiTask(DependencyHelpers, luigi.Task):
    pass

# ==============================================================================

class SciLuigiExternalTask(DependencyHelpers, luigi.ExternalTask):
    pass

# ==============================================================================

class WorkflowTask(luigi.Task):

    def output(self):
        timestamp = time.strftime('%Y%m%d_%H%M%S', time.localtime())
        clsname = self.__class__.__name__
        return luigi.LocalTarget('workflow_' + clsname.lower() + '_completed_at_{t}'.format(t=timestamp))

    def run(self):
        timestamp = time.strftime('%Y-%m-%d, %H:%M:%S', time.localtime())
        with self.output().open('w') as outfile:
            outfile.write('workflow finished at {t}'.format(t=timestamp))

# ==============================================================================

class HPCHelpers():
    '''
    Mixin with various convenience methods that most tasks need, such as for executing SLURM
    commands
    '''
    accounted_project = luigi.Parameter()

    # Main Execution methods
    def execute_in_configured_mode(self, command):
        '''Execute either locally or via SLURM, depending on config'''

        if self.get_task_config("runmode") == "local":
            self.execute_command(command)

        elif self.get_task_config("runmode") == "nodejob":
            train_size = "NA"
            if hasattr(self, 'train_size'):
                train_size = self.train_size
            replicate_id = "NA"
            if hasattr(self, 'replicate_id'):
                replicate_id = self.replicate_id

            self.execute_hpcjob(command,
                    accounted_project = self.accounted_project,
                    time_limit = self.get_task_config("time_limit"),
                    partition  = self.get_task_config("partition"),
{'fraction_non_alphanumeric': 0.22241585883890644, 'fraction_numerical': 0.0008713647750789674, 'mean_word_length': 3.276665114112715, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 5, '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': '2444896', 'n_tokens_mistral': 2414, 'n_tokens_neox': 2263, 'n_words': 624}