Merge pull request #62 from LightForm-group/feat/iterations

Feat/iterations

CHANGELOG.md

@@ -1,16 +1,32 @@
 # Change Log
 
-## [0.x.xx] - xxxx.xx.xx
+## [0.2.12] - 2020.12.16
 
 ### Added
 
-- Figures definitions
-- Workflow metadata attribute for storing arbitrary metadata
+- Add `Workflow.figures` attribute for storing associated figure definitions.
+- Add `Workflow.metadata` attribute for storing arbitrary metadata (will later be used for Zenodo archiving).
+- Add various `Workflow` static methods to help with retrieving information in the viewer without loading the whole workflow via `hickle`.
+- Add `get_task_schemas` to API to load the available task schemas without generating a workflow.
+- Add `refresh` bool parameter to `Config.set_config`, to force a reload of the configuration.
+- Support inputs as dependencies as well as outputs.
+- Support "parameter modifying" tasks (a task which outputs a parameter that is also an input to that task).
+- Add `iterate_run_options` to Workflow.
+- Add new methods for finding dependent and dependency tasks/parameters, upstream/downstream parameter values associated with a given element.
+- Add input option: `include_all_iterations`. If True, inputs from all iterations are passed to input map functions.
 
 ### Fixed
 
 - Only save input/output map files if they exist!
 - Fix bug in propagating groups correctly
+- Various code formatting issues
+- Fix failure to raise on invalid schemas.
+- Fix bug when the same file is to be saved from multiple output maps.
+
+### Changed
+- Redo task sorting algorithm such that minimal ordering changes are made.
+- Set `stats` bool to False by default.
+- Bump hpcflow version to v0.1.12.
 
 ## [0.2.11] - 2020.09.29
 

matflow/__init__.py

@@ -8,4 +8,5 @@
     append_schema_source,
     prepend_schema_source,
     validate,
+    get_task_schemas,
 )

matflow/_version.py

@@ -1 +1 @@
-__version__ = '0.2.11'
+__version__ = '0.2.12'

matflow/api.py

@@ -45,6 +45,13 @@ def make_workflow(profile_path, directory=None, write_dirs=True):
         profile_str = handle.read()
 
     profile = {'file': profile_str, 'parsed': copy.deepcopy(workflow_dict)}
+
+    iterate_run_opts = {
+        **Config.get('default_sticky_iterate_run_options'),
+        **Config.get('default_iterate_run_options'),
+    }
+    workflow_dict.update({'iterate_run_options': iterate_run_opts})
+
     workflow = Workflow(**workflow_dict, stage_directory=directory, profile=profile)
     workflow.set_ids()
 
@@ -101,11 +108,13 @@ def load_workflow(directory, full_path=False):
     return workflow
 
 
-def prepare_task(task_idx, directory, is_array=False):
-    """Prepare a task for execution by setting inputs and running input maps."""
+def prepare_task(task_idx, iteration_idx, directory, is_array=False):
+    """Prepare a task (iteration) for execution by setting inputs and running input
+    maps."""
+
     load_extensions()
     workflow = load_workflow(directory)
-    workflow.prepare_task(task_idx, is_array=is_array)
+    workflow.prepare_task(task_idx, iteration_idx, is_array=is_array)
 
 
 def prepare_task_element(task_idx, element_idx, directory, is_array=False):
@@ -115,11 +124,11 @@ def prepare_task_element(task_idx, element_idx, directory, is_array=False):
     workflow.prepare_task_element(task_idx, element_idx, is_array=is_array)
 
 
-def process_task(task_idx, directory, is_array=False):
-    """Process a completed task by running the output map."""
+def process_task(task_idx, iteration_idx, directory, is_array=False):
+    """Process a completed task (iteration) by running the output map."""
     load_extensions()
     workflow = load_workflow(directory)
-    workflow.process_task(task_idx, is_array=is_array)
+    workflow.process_task(task_idx, iteration_idx, is_array=is_array)
 
 
 def process_task_element(task_idx, element_idx, directory, is_array=False):
@@ -136,11 +145,11 @@ def run_python_task(task_idx, element_idx, directory):
     workflow.run_python_task(task_idx, element_idx)
 
 
-def prepare_sources(task_idx, directory):
+def prepare_sources(task_idx, iteration_idx, directory):
     """Prepare source files."""
     load_extensions()
     workflow = load_workflow(directory)
-    workflow.prepare_sources(task_idx)
+    workflow.prepare_sources(task_idx, iteration_idx)
 
 
 def append_schema_source(schema_source_path):
@@ -165,3 +174,21 @@ def kill(directory):
 def cloud_connect(provider):
     Config.set_config()
     hpcflow_cloud_connect(provider, config_dir=Config.get('hpcflow_config_dir'))
+
+
+def write_element_directories(iteration_idx, directory):
+    'Generate element directories for a given iteration.'
+    load_extensions()
+    workflow = load_workflow(directory)
+    if workflow.iterate:
+        num_iters = workflow.iterate['num_iterations']
+    else:
+        num_iters = workflow.num_iterations
+    if iteration_idx < num_iters:
+        workflow.write_element_directories(iteration_idx)
+        workflow.prepare_iteration(iteration_idx)
+
+
+def get_task_schemas():
+    Config.set_config()
+    return Config.get('task_schemas')

matflow/cli.py

@@ -35,11 +35,12 @@ def go(workflow_path, directory=None):
 
 @cli.command()
 @click.option('--task-idx', '-t', type=click.INT, required=True)
+@click.option('--iteration-idx', '-i', type=click.INT, required=True)
 @click.option('--directory', '-d', type=click.Path(exists=True))
 @click.option('--array', is_flag=True)
-def prepare_task(task_idx, directory=None, array=False):
+def prepare_task(task_idx, iteration_idx, directory=None, array=False):
     print('matflow.cli.prepare_task', flush=True)
-    api.prepare_task(task_idx, directory, is_array=array)
+    api.prepare_task(task_idx, iteration_idx, directory, is_array=array)
 
 
 @cli.command()
@@ -54,11 +55,12 @@ def prepare_task_element(task_idx, element_idx, directory=None, array=False):
 
 @cli.command()
 @click.option('--task-idx', '-t', type=click.INT, required=True)
+@click.option('--iteration-idx', '-i', type=click.INT, required=True)
 @click.option('--directory', '-d', type=click.Path(exists=True))
 @click.option('--array', is_flag=True)
-def process_task(task_idx, directory=None, array=False):
+def process_task(task_idx, iteration_idx, directory=None, array=False):
     print('matflow.cli.process_task', flush=True)
-    api.process_task(task_idx, directory, is_array=array)
+    api.process_task(task_idx, iteration_idx, directory, is_array=array)
 
 
 @cli.command()
@@ -82,10 +84,11 @@ def run_python_task(task_idx, element_idx, directory=None):
 
 @cli.command()
 @click.option('--task-idx', '-t', type=click.INT, required=True)
+@click.option('--iteration-idx', '-i', type=click.INT, required=True)
 @click.option('--directory', '-d', type=click.Path(exists=True))
-def prepare_sources(task_idx, directory=None):
+def prepare_sources(task_idx, iteration_idx, directory=None):
     print('matflow.cli.prepare_sources', flush=True)
-    api.prepare_sources(task_idx, directory)
+    api.prepare_sources(task_idx, iteration_idx, directory)
 
 
 @cli.command()
@@ -119,5 +122,12 @@ def kill(directory):
     api.kill(directory)
 
 
+@cli.command()
+@click.option('--iteration-idx', '-i', type=click.INT, required=True)
+@click.option('--directory', '-d', type=click.Path(exists=True))
+def write_element_directories(iteration_idx, directory=None):
+    api.write_element_directories(iteration_idx, directory)
+
+
 if __name__ == '__main__':
     cli()

matflow/config.py

@@ -18,9 +18,11 @@ class Config(object):
         'default_run_options',
         'default_preparation_run_options',
         'default_processing_run_options',
+        'default_iterate_run_options',
         'default_sticky_run_options',
         'default_sticky_preparation_run_options',
         'default_sticky_processing_run_options',
+        'default_sticky_iterate_run_options',
         'parallel_modes',
         'archive_locations',
     ]
@@ -112,15 +114,16 @@ def get_config_file(config_dir):
         return config_dat, config_file
 
     @staticmethod
-    def set_config(config_dir=None, raise_on_set=False):
+    def set_config(config_dir=None, raise_on_set=False, refresh=False):
         """Load configuration from a YAML file."""
 
         config_dir = Config.resolve_config_dir(config_dir)
 
         if Config._is_set:
             if raise_on_set:
                 raise ConfigurationError('Configuration is already set.')
-            return
+            elif not refresh:
+                return
 
         config_dat, _ = Config.get_config_file(config_dir)
         schema_sources = [Path(i).expanduser() for i in config_dat['task_schema_sources']]
@@ -229,9 +232,11 @@ def set_config(config_dir=None, raise_on_set=False):
             'default_run_options',
             'default_preparation_run_options',
             'default_processing_run_options',
+            'default_iterate_run_options',
             'default_sticky_run_options',
             'default_sticky_preparation_run_options',
             'default_sticky_processing_run_options',
+            'default_sticky_iterate_run_options',
         ]:
             Config.__conf[i] = config_dat.get(i, {})
 

matflow/errors.py

@@ -81,3 +81,7 @@ class UnexpectedSourceMapReturnError(Exception):
 
 class CommandError(Exception):
     """For problems with command groups and commands."""
+
+
+class WorkflowIterationError(Exception):
+    """For issues with resolving requested iterations."""