Repository: skypyproject/skypy

Issue:
cross match
created a base class for cross matching

Discussion:

Patch:
diff --git a/skypy/pipeline/driver.py b/skypy/pipeline/driver.py
index dd8dbbc..e834c00 100644
--- a/skypy/pipeline/driver.py
+++ b/skypy/pipeline/driver.py
@@ -4,7 +4,8 @@ This module provides methods to run pipelines of functions with dependencies
 and handle their results.
 """
 
-from astropy.table import Table
+from collections.abc import Mapping
+from copy import deepcopy
 import networkx
 import re
 
@@ -18,20 +19,20 @@ class SkyPyDriver:
     r'''Class for running pipelines.
 
     This is the main class for running pipelines of functions with dependencies
-    and using their results to generate tables.
+    and using their results to generate variables and tables.
     '''
 
-    def execute(self, config, file_format=None):
+    def execute(self, configuration, file_format=None):
         r'''Run a pipeline.
 
-        This function runs a pipeline of functions to generate the cosmology
-        and the columns of a set of tables. It uses a Directed Acyclic Graph to
+        This function runs a pipeline of functions to generate variables and
+        the columns of a set of tables. It uses a Directed Acyclic Graph to
         determine a non-blocking order of execution that resolves any
         dependencies, see [1]_. Tables can optionally be written to file.
 
         Parameters
         ----------
-        config : dict-like
+        configuration : dict-like
             Configuration for the pipeline.
         file_format : str
             File format used to write tables. Files are written using the
@@ -52,14 +53,15 @@ class SkyPyDriver:
         'requires' specifices the names of previous steps in the pipeline and
         uses their return values as keyword arguments.
 
-        'config' should contain 'cosmology' and/or 'tables'. 'cosmology' should
-        return a dictionary configuring a function that returns an
-        astropy.cosmology.Cosmology object. 'tables' should contain a set of
-        nested dictionaries, first giving the names of each table, then the
-        names of each column within each table. Each column should return a
-        dictionary configuring a function that returns an array-like object.
+        'configuration' should contain the name and configuration of each
+        variable and/or an entry named 'tables'. 'tables' should contain a set
+        of nested dictionaries, first containing the name of each table, then
+        the name and configuration of each column and optionally an entry named
+        'init' with a configuration that initialises the table. If 'init' is
+        not specificed the table will be initialised as an empty astropy Table
+        by default.
 
-        See [3]_ for examples of pipeline configurations in yaml format.
+        See [3]_ for examples of pipeline configurations in YAML format.
 
         References
         ----------
@@ -68,37 +70,66 @@ class SkyPyDriver:
         .. [3] https://github.com/skypyproject/skypy/tree/master/examples
         '''
 
+        # config contains settings for all variables and table initialisation
+        # table_config contains settings for all table columns
+        config = deepcopy(configuration)
+        table_config = config.pop('tables', {})
+        default_table = {'module': 'astropy.table', 'function': 'Table'}
+        config.update({k: v.pop('init', default_table)
+                      for k, v in table_config.items()})
+
         # Create a Directed Acyclic Graph of all jobs and dependencies
         dag = networkx.DiGraph()
-        if 'cosmology' in config:
-            dag.add_node('cosmology')
-        for table, columns in config.get('tables', {}).items():
-            dag.add_node(table)
-            for column, settings in columns.items():
-                node = '.'.join((table, column))
-                dag.add_node(node)
-        for table, columns in config.get('tables', {}).items():
+
+        # Variables initialised by value don't require function evaluations
+        def isfunction(f):
+            return isinstance(f, Mapping) and 'module' in f and 'function' in f
+        variables = {k: v for k, v in config.items() if not isfunction(v)}
+        for v in variables:
+            dag.add_node(v)
+            setattr(self, v, config.pop(v))
+
+        # Add nodes for each variable, table and column
+        for job in config:
+            dag.add_node(job)
+        for table, columns in table_config.items():
+            table_complete = '.'.join((table, 'complete'))
+            dag.add_node(table_complete)
+            for column in columns.keys():
+                job = '.'.join((table, column))
+                dag.add_node(job)
+
+        # Add edges for all requirements and dependencies
+        for job, settings in config.items():
+            dependencies = settings.get('depends', [])
+            dependencies += settings.get('requires', {}).values()
+            dag.add_edges_from((d, job) for d in dependencies)
+        for table, columns in table_config.items():
+            table_complete = '.'.join((table, 'complete'))
+            dag.add_edge(table, table_complete)
             for column, settings in columns.items():
-                node = '.'.join((table, column))
-                dag.add_edge(table, node)
-                requirements = settings.get('requires', {}).values()
-                dag.add_edges_from((r, node) for r in requirements)
+                job = '.'.join((table, column))
+                dag.add_edge(table, job)
+                dag.add_edge(job, table_complete)
+                dependencies = settings.get('depends', [])
+                dependencies += settings.get('requires', {}).values()
+                dag.add_edges_from((d, job) for d in dependencies)
 
         # Execute jobs in order that resolves dependencies
         for job in networkx.topological_sort(dag):
-            if job == 'cosmology':
-                settings = config.get('cosmology')
-                self.cosmology = self._call_from_config(settings)
-            elif job in config.get('tables', {}):
-                setattr(self, job, Table())
+            if job in variables or job.endswith('.complete'):
+                continue
+            elif job in config:
+                settings = config.get(job)
+                setattr(self, job, self._call_from_config(settings))
             else:
                 table, column = job.split('.')
-                settings = config['tables'][table][column]
+                settings = table_config[table][column]
                 getattr(self, table)[column] = self._call_from_config(settings)
 
         # Write tables to file
         if file_format:
-            for table in config.get('tables', {}).keys():
+            for table in table_config.keys():
                 filename = '.'.join((table, file_format))
                 getattr(self, table).write(filename)
 
diff --git a/skypy/pipeline/scripts/skypy.py b/skypy/pipeline/scripts/skypy.py
index b2d6d59..f1a6e2b 100644
--- a/skypy/pipeline/scripts/skypy.py
+++ b/skypy/pipeline/scripts/skypy.py
@@ -55,6 +55,7 @@ def main(args=None):
 
     args = parser.parse_args(args or ['--help'])
     config = yaml.safe_load(args.config)
+    config = {} if config is None else config
     driver = SkyPyDriver()
     driver.execute(config, file_format=args.format)
     return(0)


Test patch:
diff --git a/skypy/pipeline/tests/data/cosmology_config.yaml b/skypy/pipeline/tests/data/cosmology_config.yaml
deleted file mode 100644
index 2b81fac..0000000
--- a/skypy/pipeline/tests/data/cosmology_config.yaml
+++ /dev/null
@@ -1,6 +0,0 @@
-cosmology:
-  module: astropy.cosmology
-  function: FlatLambdaCDM
-  kwargs:
-    H0: 67.74
-    Om0: 0.3075
diff --git a/skypy/pipeline/tests/data/empty_config.yaml b/skypy/pipeline/tests/data/empty_config.yml
similarity index 100%
rename from skypy/pipeline/tests/data/empty_config.yaml
rename to skypy/pipeline/tests/data/empty_config.yml
diff --git a/skypy/pipeline/tests/data/test_config.yml b/skypy/pipeline/tests/data/test_config.yml
new file mode 100644
index 0000000..1f7a612
--- /dev/null
+++ b/skypy/pipeline/tests/data/test_config.yml
@@ -0,0 +1,33 @@
+test_int: 1
+test_float: 1.0
+test_cosmology:
+  module: astropy.cosmology
+  function: FlatLambdaCDM
+  kwargs:
+    H0: 67.74
+    Om0: 0.3075
+tables:
+  test_table_1:
+    test_column_1:
+      module: numpy.random
+      function: uniform
+      kwargs:
+        size: 10
+    test_column_2:
+      module: numpy.random
+      function: uniform
+      requires:
+        low: test_table_1.test_column_1
+    test_column_3:
+      module: numpy.random
+      function: uniform
+      requires:
+        low: test_table_1.test_column_1
+        high: test_table_1.test_column_2
+  test_table_2:
+    init:
+      module: astropy.table
+      function: QTable
+      kwargs:
+        names: ['column_1', 'column_2']
+      depends: [test_table_1.complete]
diff --git a/skypy/pipeline/tests/test_driver.py b/skypy/pipeline/tests/test_driver.py
index 945cc0a..84d8a18 100644
--- a/skypy/pipeline/tests/test_driver.py
+++ b/skypy/pipeline/tests/test_driver.py
@@ -11,13 +11,13 @@ from skypy.pipeline.driver import SkyPyDriver
 def test_driver():
 
     # Evaluate and store the default astropy cosmology.
-    config = {'cosmology': {
+    config = {'test_cosmology': {
                 'module': 'astropy.cosmology',
                 'function': 'default_cosmology.get'}}
 
     driver = SkyPyDriver()
     driver.execute(config)
-    assert driver.cosmology == default_cosmology.get()
+    assert driver.test_cosmology == default_cosmology.get()
 
     # Generate a simple two column table with a dependency. Also write the
     # table to a fits file and check it's contents.
@@ -46,20 +46,30 @@ def test_driver():
     config['tables']['test_table']['column1']['requires'] = \
         {'low': 'test_table.column1'}
     with pytest.raises(networkx.NetworkXUnfeasible):
-        driver.execute(config)
+        SkyPyDriver().execute(config)
 
     # Check for failure if 'column1' and 'column2' both require each other
     # creating a cyclic dependency graph
     config['tables']['test_table']['column1']['requires'] = \
         {'low': 'test_table.column2'}
     with pytest.raises(networkx.NetworkXUnfeasible):
-        driver.execute(config)
+        SkyPyDriver().execute(config)
 
     # Check for failure if 'column1' is removed from the config so that the
     # requirements for 'column2' are not satisfied.
     del config['tables']['test_table']['column1']
     with pytest.raises(KeyError):
-        driver.execute(config)
+        SkyPyDriver().execute(config)
+
+    # Check variables intialised by value
+    config = {'test_int': 1,
+              'test_float': 1.0}
+    driver = SkyPyDriver()
+    driver.execute(config)
+    assert isinstance(driver.test_int, int)
+    assert isinstance(driver.test_float, float)
+    assert driver.test_int == 1
+    assert driver.test_float == 1.0
 
 
 def teardown_module(module):
diff --git a/skypy/pipeline/tests/test_skypy.py b/skypy/pipeline/tests/test_skypy.py
index bde0f93..b64656d 100644
--- a/skypy/pipeline/tests/test_skypy.py
+++ b/skypy/pipeline/tests/test_skypy.py
@@ -26,10 +26,9 @@ def test_skypy():
     assert e.value.code == 2
 
     # Process empty config file
-    filename = get_pkg_data_filename('data/empty_config.yaml')
-    with pytest.raises(TypeError) as e:
-        assert skypy.main(['--config', filename]) == 0
+    filename = get_pkg_data_filename('data/empty_config.yml')
+    assert skypy.main(['--config', filename]) == 0
 
-    # Process cosmology config file
-    filename = get_pkg_data_filename('data/cosmology_config.yaml')
+    # Process test config file
+    filename = get_pkg_data_filename('data/test_config.yml')
     assert skypy.main(['--config', filename]) == 0