The previous behaviour with regex() where mismatches fail even if no substitution is made is retained by the use of re.subn(). This is a corner case but I didn’t want user code to break

# filter on ".txt"
input_filenames = ["a.wrong", "b.txt"]
regex("(.txt)$")

# fails, no substitution possible
r"\1"

# fails anyway even through regular expression matches not referenced...
r"output.filename"

• The database is Jacob Sondergaard’s dbdict which is a nosql / key-value store wrapper around sqlite

  job_history = dbdict.open(RUFFUS_HISTORY_FILE, picklevalues=True)
  

• The key is the output file name, so it is important not to confuse Ruffus by having different tasks generate the same output file!

• Is it possible to abstract this so that jobs get timestamped as well?

• If we should ever want to abstract out dbdict, we need to have a similar key-value store class, and make sure that a single instance of dbdict is used through pipeline_run which is passed up and down the function call chain. dbdict would then be drop-in replaceable by our custom (e.g. flat-file-based) dbdict alternative.

To peek into the database:

$ sqlite3 .ruffus_history.sqlite
sqlite> .tables
data
sqlite> .schema data
CREATE TABLE data (key PRIMARY KEY,value);
sqlite> select key from data order by key;

Allows Ruffus to avoid adding an extra 1 second pause between tasks to guard against file systems with low timestamp granularity.

• If the local file time looks to be in sync with the underlying file system, saved system time is used instead of file timestamps

Parameters in Ruffus v 2.x are obtained using a “pull” model.

Each task has its self.param_generator_func() This is an iterator function which yields param and descriptive_param per iteration:

   for param, descriptive_param in self.param_generator_func(runtime_data):
       pass


``param`` and ``descriptive_param`` are basically the same except that globs are not expanded in ``descriptive_param`` because
they are used for display.

The iterator functions have all the state they need to generate their input, output and extra parameters (only runtime_data) is added at run time. These closures are generated as nested functions inside “factory” functions defined in file_name_parameters.py

Each task type has its own factory function. For example:

args_param_factory (orig_args)
files_param_factory (input_files_task_globs, flatten_input, do_not_expand_single_job_tasks, output_extras)
split_param_factory (input_files_task_globs, output_files_task_globs, *extra_params)
merge_param_factory (input_files_task_globs, output_param, *extra_params)
originate_param_factory (list_output_files_task_globs, extras)

The following factory files delegate most of their work to yield_io_params_per_job:

to support:

• inputs(), add_inputs() input parameter supplementing
• extra inputs, outputs, extra parameter replacement with suffix(), regex() and formatter

collate_param_factory       (input_files_task_globs,      flatten_input,                              file_names_transform, extra_input_files_task_globs, replace_inputs, output_pattern,          *extra_specs)
transform_param_factory     (input_files_task_globs,      flatten_input,                              file_names_transform, extra_input_files_task_globs, replace_inputs, output_pattern,          *extra_specs)
combinatorics_param_factory (input_files_task_globs,      flatten_input, combinatorics_type, k_tuple, file_names_transform, extra_input_files_task_globs, replace_inputs, output_pattern,          *extra_specs)
subdivide_param_factory     (input_files_task_globs,      flatten_input,                              file_names_transform, extra_input_files_task_globs, replace_inputs, output_files_task_globs, *extra_specs)
product_param_factory       (list_input_files_task_globs, flatten_input,                              file_names_transform, extra_input_files_task_globs, replace_inputs, output_pattern,          *extra_specs)


yield_io_params_per_job (input_params, file_names_transform, extra_input_files_task_globs, replace_inputs, output_pattern, extra_specs, runtime_data, iterator, expand_globs_in_output = False):

1. The first thing they do is to get a list of input parameters, either directly, or by expanding globs or by query upstream tasks:

   file_names_from_tasks_globs(files_task_globs, runtime_data, do_not_expand_single_job_tasks = True_if_split_or_merge)
   

   Note

   True_if_split_or_merge is a wierd parameter which directly queries the upstream dependency for its output files if it is a single task...

   This is legacy code. Probably should be refactored out of existence...

2. They then convert the input parameters to a flattened list of file names (passing through unchanged the original input parameters structure)

   input_param_to_file_name_list()
   # combinatorics and product call:
   list_input_param_to_file_name_list()
   

   This is done at the iterator level because the combinatorics decorators do not have just a list of input parameters (They have combinations, permutations, products of input parameters etc) but a list of lists of input parameters.

   transform, collate, subdivide => list of strings. combinatorics / product => list of lists of strings

3. yield_io_params_per_job yields pairs of param sets by

   • Replacing or supplementing input parameters for the indicator objects inputs() and add_inputs()
   • Expanding extra parameters
   • Expanding output parameters (with or without expanding globs)

   In each case:

   • If these contains objects which look like strings, we do regular expression / file component substitution
   • If they contain tasks, these are queries for output files

   Note

   This should be changed:

   If the flattened list of input file names is empty, ie. if the input parameters contain just other stuff, then the entire parameter is ignored.

All strings in input (or output parameters) are treated as file names unless they are wrapped with output_from in which case they are Task, Pipeline or function names.

A list of strings for ready for substitution to output parameters is obtained from the ruffus_utility.get_strings_in_flattened_sequence()

This is called from:

file_name_parameters

1. Either to check that input files exist: check_input_files_exist() needs_update_check_directory_missing() needs_update_check_exist() needs_update_check_modify_time()

2. Or to generate parameters from the various param factories

   product_param_factory() transform_param_factory() collate_param_factory() combinatorics_param_factory() subdivide_param_factory()

These first call file_names_from_tasks_globs() to get the input parameters, then pass a flattened list of strings to yield_io_params_per_job()

-> file_names_from_tasks_globs() -> yield_io_params_per_job( input_param_to_file_name_list() / list_input_param_to_file_name_list() )

task

3. to obtain a list of file names to touch

   job_wrapper_io_files

4. to make directories

   job_wrapper_mkdir

5. update / remove files in job_history if job succeeded or failed

   pipeline_run

We want to expand objects with ruffus_params only when doing output parameter substitution, i.e. Case (2) above. They are not file names: cases (1), (3), (4), (5).

Therefore: Expand in file_names_from_tasks_globs() which also handles inputs() and add_inputs and @split outputs.

Non-recursive Substitution in all:

construct new list where each item is replaced referring to the original and then assign

extra_inputs() “[INPUTS[1][2]]” refers to the original input output / extras “[INPUTS[1][2]]” refers to substituted input

In addition to the flattened input paramters, we need to pass in the unflattened input and extra parameters

In file_name_parameters.py.: yield_io_params_per_job

From: .. code-block:: python

extra_inputs = extra_input_files_task_globs.file_names_transformed (filenames, file_names_transform) extra_params = tuple( file_names_transform.substitute(filenames, p) for p in extra_specs) output_pattern_transformed = output_pattern.file_names_transformed (filenames, file_names_transform) output_param = file_names_transform.substitute_output_files(filenames, output_pattern)

To: .. code-block:: python

extra_inputs = extra_input_files_task_globs.file_names_transformed (orig_input_param, extra_specs, filenames, file_names_transform) extra_params = tuple( file_names_transform.substitute(input_param, extra_specs, filenames, p) for p in extra_specs) output_pattern_transformed = output_pattern.file_names_transformed (input_param, extra_specs, filenames, file_names_transform) output_param = file_names_transform.substitute_output_files(input_param, extra_specs, filenames, output_pattern)

In other words, we need two extra parameters for inputs and extras

class t_file_names_transform(object):
    def substitute (self, input_param, extra_param, starting_file_names, pattern):
        pass
    def substitute_output_files (self, input_param, extra_param, starting_file_names, pattern):
        pass


class t_params_tasks_globs_run_time_data(object):
    def file_names_transformed (self, input_param, extra_param, filenames, file_names_transform):
        pass

• what happens to get_outputs or checkpointing when the job completes but the output files are not made?

• either_or matches

  • the only alternative to have all files existing
  • the alternative with the most recent file

• either_or behaves as list() in file_name_parameters.py. : file_names_from_tasks_globs

• Handled to check that input files exist:

  check_input_files_exist() needs_update_check_directory_missing() needs_update_check_exist() needs_update_check_modify_time()

• Handled to update / remove files in job_history if job succeeded or failed

• Only first either_or is used to obtain list of file names to touch

  task.job_wrapper_io_files

• Only first either_or is used to obtain list of file names to make directories

  job_wrapper_mkdir

• What happens in task.get_output_files()?

1. Refactored to remove unused “flattened” code paths / parameters

2. Prefix all attributes for Task into underscore so that help(Task) is not overloaded with details

3. Named parameters

   • parse named parameters in order filling in from unnamed

   • save parameters in dict Task.parsed_args

   • call setup_task_func() afterwards which knows how to setup:

     • poor man’s OOP but
     • allows type to be changed after constructor: Because can’t guarantee that @transform @merge is the first Ruffus decorator to be encountered.

   • setup_task_func() is called for every task before pipeline_xxx()

4. Much more informative messages for errors when parsing decorator arguments

5. Pipeline decorator methods renamed to decorator_xxx as in decorator_follows

6. Task.get_task_name() * rename to Task.get_display_name() * distinguish between decorator and OO interface

7. Rename _task to Task

8. Identifying tasks from t_job_result:

   • job results do not contain references to Task so that it can be marshalled more easily
   • we need to look up task at job completion
   • use _node_index from graph.py so we have always a unique identifier for each Task

9. Parse arguments using ruffus_utility.parse_task_arguments

   • Reveals full hackiness and inconsistency between add_inputs and inputs. The latter only takes a single argument. Each of the elements of the former gets added along side the existing inputs.

10. Add Pipeline class * Create global called "main" (accessed by Pipeline.pipelines[“main”])

11. Task name lookup

    • Task names are unique (Otherwise Ruffus will complain at Task creation)
    • Can also lookup by fully qualified or unqualified function name but these can be ambiguous
    • Ambiguous lookups give a list of tasks only so we can have nice diagnostic messages ... UI trumps clean design

12. Look up strings across pipelines #. Is pipeline name qualified? Check that #. Check default (current) pipeline #. Check if pipeline name. In which case returns all tail functions #. Check all pipelines

    • Will blow up at any instance of ambiguity in any particular pipeline
    • Will blow up at any instance of ambiguity across pipelines
    • Note that mis-spellings will cause problems but if this were c++, I would enforce stricter checking

13. Look up functions across pipelines * Try current pipeline first, then all pipelines * Will blow up at any instance of ambiguity in any particular pipeline * Will blow up at any instance of ambiguity across pipelines (if not in current pipeline)

14. @mkdir, @follows(mkdir)

15. Pipeline.get_head_tasks(self) (including tasks with mkdir())

16. Pipeline.get_tail_tasks(self)

17. Pipeline._complete_task_setup() which follows chain of dependencies for each task in a pipeline

• Share code as far as possible between decorator and OOP syntax

• Cannot use textbook OOP inheritance hierarchy easily because @decorators are not necessarily given in order.

  Pipeline.transform
      _do_create_task_by_OOP()
  
  @transform
      Pipeline._create_task()
      task._decorator_transform
  
          task._prepare_transform()
              self.setup_task_func = self._transform_setup
              parse_task_arguments
  
  
  Pipeline.run
      pipeline._complete_task_setup()
          # walk up ancestors of all task and call setup_task_func
          unprocessed_tasks = Pipeline.tasks
          while len(unprocessed_tasks):
              ancestral_tasks = setup_task_func()
              if not already processed:
                  unprocessed_tasks.append(ancestral_tasks)
  
          Call _complete_task_setup() for all the pipelines of each task
  

task._complete_setup()
    task._remove_all_parents()
    task._deferred_connect_parents()
    task._setup_task_func()
        task._handle_tasks_globs_in_inputs()
            task._connect_parents()
                # re-lookup task from names in current pipeline so that pipeline.clone() works

• Task dependencies are normally deferred and saved to Task.deferred_follow_params
• If Task dependencies call for a new Task (follows/follows(mkdir)), this takes place immediately
• The parameters in Task.deferred_follow_params are updated with the created Task when this happens
• Task._prepare_preceding_mkdir() has a defer flag to prevent it from updating Task.deferred_follow_params when it is called to resolve deferred dependencies from Task._connect_parents(). Otherwise we will have two copies of each deferred dependency...
• Task.deferred_follow_params must be deep-copied otherwise cloned pipelines will interfere with each other when dependencies are resolved...