Sbachmei/mic 5786/embarrassingly parallel sections (#189)

* rename user_configurable attr to directly_implemented

* break reliance on splitting and aggregating node being the same

* Add a new custom-schema integration test (#173)

* set e.p.'s step parent; remove i/o slots during e.p. step construction

* simplify EmbarrassinglyParallelStep construction

* Sbachmei/mic 5987/implement splitter steps and aggregator steps (#183)

* save here: first working cut

* remove unused splitter and aggregator attrs from I/OSlots

* remove old commented out code

* bugfix InputSlot.__eq__

* fix broken tests; add docstrings

* add new StandaloneStep class for IOSteps, SplitterSteps, and AggregatorSteps

* small tweaks

* fix doc build failures

* rename splitter_step_name to splitter_node_name

* remove unused arg to _check_embarrassingly_parallel_details

* assert we only have one output slot when aggregating and stop looping

* modify docstring

* special-case e.p. steps to prevent duplicative naming conventions

* test against embarrassingly parallel parallel steps (#185)

* test against embarrassingly parallel parallel steps

* use string representation of input_files list

* add test against embarrassingly parallel loop steps (#186)

* add test against embarrassingly parallel HierarchicalSteps (#187)

* add test against embarrassingly parallel HierarchicalSteps

* Update tests/integration/test_compositions.py

Co-authored-by: Zeb Burke-Conte <zmbc@users.noreply.github.com>

---------

Co-authored-by: Zeb Burke-Conte <zmbc@users.noreply.github.com>

* changelog

---------

Co-authored-by: Zeb Burke-Conte <zmbc@users.noreply.github.com>

Author: stevebachmeier

CHANGELOG.rst

@@ -1,3 +1,7 @@
+**0.1.14 - 5/1/25**
+
+ - Add support for EmbarrassinglyParallelSteps to accept sections (i.e. non-leaf steps)
+
 **0.1.13 - 4/21/25**
 
  - remove graphviz from doc build install

setup.py

@@ -56,6 +56,9 @@
         # TODO MIC-4963: Resolve quoting issue and remove pin
         "snakemake-interface-executor-plugins<9.0.0",
         "snakemake-executor-plugin-slurm",
+        # Type stubs
+        "pandas-stubs",
+        "pyarrow-stubs",
     ]
 
     setup_requires = ["setuptools_scm"]

src/easylink/configuration.py

@@ -274,10 +274,10 @@ def _validate_environment(self) -> dict[Any, Any]:
 
 
 def load_params_from_specification(
-    pipeline_specification: str,
-    input_data: str,
-    computing_environment: str | None,
-    results_dir: str,
+    pipeline_specification: str | Path,
+    input_data: str | Path,
+    computing_environment: str | Path | None,
+    results_dir: str | Path,
 ) -> dict[str, Any]:
     """Gathers together all specification data.
 
@@ -325,7 +325,7 @@ def _load_input_data_paths(
 
 
 def _load_computing_environment(
-    computing_environment_specification_path: str | None,
+    computing_environment_specification_path: str | Path | None,
 ) -> dict[Any, Any]:
     """Loads the computing environment specification file and returns the contents as a dict."""
     if not computing_environment_specification_path:

src/easylink/graph_components.py

@@ -18,6 +18,12 @@
 
 import networkx as nx
 
+from easylink.implementation import (
+    NullAggregatorImplementation,
+    NullImplementation,
+    NullSplitterImplementation,
+)
+
 if TYPE_CHECKING:
     from easylink.implementation import Implementation
     from easylink.step import Step
@@ -42,48 +48,38 @@ class InputSlot:
     env_var: str | None
     """The environment variable that is used to pass a list  of data filepaths to 
     an ``Implementation``."""
-    validator: Callable[[str], None] = field(compare=False)
+    validator: Callable[[str], None] | None = field(compare=False)
     """A function that validates the input data being passed into the pipeline via 
     this ``InputSlot``. If the data is invalid, the function should raise an exception 
     with a descriptive error message which will then be reported to the user.
     **Note that the function *must* be defined in the** :mod:`easylink.utilities.validation_utils` 
     **module!**"""
-    splitter: Callable[[list[str], str, Any], None] | None = field(
-        default=None, compare=False
-    )
-    """A function that splits the incoming data to this ``InputSlot`` into smaller
-    pieces. The primary purpose of this functionality is to run sections of the 
-    pipeline in an embarrassingly parallel manner. **Note that the function *must* 
-    be defined in the **:mod:`easylink.utilities.splitter_utils`** module!**"""
 
     def __eq__(self, other: Any) -> bool | NotImplementedType:
-        """Checks if two ``InputSlots`` are equal.
-
-        Two ``InputSlots`` are considered equal if their names, ``env_vars``, and
-        names of their ``validators`` and ``splitters`` are all the same.
-        """
+        """Checks if two ``InputSlots`` are equal."""
         if not isinstance(other, InputSlot):
             return NotImplemented
-        splitter_name = self.splitter.__name__ if self.splitter else None
-        other_splitter_name = other.splitter.__name__ if other.splitter else None
+        validator_name = self.validator.__name__ if self.validator else None
+        other_validator_name = other.validator.__name__ if other.validator else None
         return (
             self.name == other.name
             and self.env_var == other.env_var
-            and self.validator.__name__ == other.validator.__name__
-            and splitter_name == other_splitter_name
+            and validator_name == other_validator_name
         )
 
     def __hash__(self) -> int:
-        """Hashes an ``InputSlot``.
-
-        The hash is based on the name of the ``InputSlot``, its ``env_var``, and
-        the names of its ``validator`` and ``splitter``.
-        """
-        splitter_name = self.splitter.__name__ if self.splitter else None
-        return hash((self.name, self.env_var, self.validator.__name__, splitter_name))
+        """Hashes an ``InputSlot``."""
+        validator_name = self.validator.__name__ if self.validator else None
+        return hash(
+            (
+                self.name,
+                self.env_var,
+                validator_name,
+            )
+        )
 
 
-@dataclass()
+@dataclass(frozen=True)
 class OutputSlot:
     """A single output slot from a specific node.
 
@@ -104,31 +100,6 @@ class OutputSlot:
 
     name: str
     """The name of the ``OutputSlot``."""
-    aggregator: Callable[[list[str], str], None] = field(default=None, compare=False)
-    """A function that aggregates all of the generated data to be passed out via this
-    ``OutputSlot``. The primary purpose of this functionality is to run sections
-    of the pipeline in an embarrassingly parallel manner. **Note that the function 
-    *must* be defined in the **:py:mod:`easylink.utilities.aggregator_utils`** module!**"""
-
-    def __eq__(self, other: Any) -> bool | NotImplementedType:
-        """Checks if two ``OutputSlots`` are equal.
-
-        Two ``OutputSlots`` are considered equal if their names and the names of their
-        ``aggregators`` are the same.
-        """
-        if not isinstance(other, OutputSlot):
-            return NotImplemented
-        aggregator_name = self.aggregator.__name__ if self.aggregator else None
-        other_aggregator_name = other.aggregator.__name__ if other.aggregator else None
-        return self.name == other.name and aggregator_name == other_aggregator_name
-
-    def __hash__(self) -> int:
-        """Hashes an ``OutputSlot``.
-
-        The hash is based on the name of the ``OutputSlot`` and the name of its ``aggregator``.
-        """
-        aggregator_name = self.aggregator.__name__ if self.aggregator else None
-        return hash((self.name, aggregator_name))
 
 
 @dataclass(frozen=True)
@@ -263,7 +234,33 @@ class ImplementationGraph(nx.MultiDiGraph):
     def implementation_nodes(self) -> list[str]:
         """The topologically sorted list of ``Implementation`` names."""
         ordered_nodes = list(nx.topological_sort(self))
-        return [node for node in ordered_nodes if node != "input_data" and node != "results"]
+        # Remove nodes that do not actually have implementations
+        null_implementations = [
+            node
+            for node in ordered_nodes
+            if isinstance(self.nodes[node]["implementation"], NullImplementation)
+        ]
+        return [node for node in ordered_nodes if node not in null_implementations]
+
+    @property
+    def splitter_nodes(self) -> list[str]:
+        """The topologically sorted list of splitter nodes (which have no implementations)."""
+        ordered_nodes = list(nx.topological_sort(self))
+        return [
+            node
+            for node in ordered_nodes
+            if isinstance(self.nodes[node]["implementation"], NullSplitterImplementation)
+        ]
+
+    @property
+    def aggregator_nodes(self) -> list[str]:
+        """The topologically sorted list of aggregator nodes (which have no implementations)."""
+        ordered_nodes = list(nx.topological_sort(self))
+        return [
+            node
+            for node in ordered_nodes
+            if isinstance(self.nodes[node]["implementation"], NullAggregatorImplementation)
+        ]
 
     @property
     def implementations(self) -> list[Implementation]:

src/easylink/implementation.py

@@ -9,15 +9,20 @@
 
 """
 
+from __future__ import annotations
+
 from collections.abc import Iterable
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 from layered_config_tree import LayeredConfigTree
 
-from easylink.graph_components import InputSlot, OutputSlot
 from easylink.utilities import paths
 from easylink.utilities.data_utils import load_yaml
 
+if TYPE_CHECKING:
+    from easylink.graph_components import InputSlot, OutputSlot
+
 
 class Implementation:
     """A representation of an actual container that will be executed for a :class:`~easylink.step.Step`.
@@ -43,8 +48,8 @@ def __init__(
         self,
         schema_steps: list[str],
         implementation_config: LayeredConfigTree,
-        input_slots: Iterable["InputSlot"] = (),
-        output_slots: Iterable["OutputSlot"] = (),
+        input_slots: Iterable[InputSlot] = (),
+        output_slots: Iterable[OutputSlot] = (),
         is_embarrassingly_parallel: bool = False,
     ):
         self.name = implementation_config.name
@@ -137,9 +142,8 @@ def outputs(self) -> dict[str, list[str]]:
 class NullImplementation:
     """A partial :class:`Implementation` interface when no container is needed to run.
 
-    The primary use case for this class is when adding an
-    :class:`~easylink.step.IOStep` - which does not have a corresponding
-    ``Implementation`` - to an :class:`~easylink.graph_components.ImplementationGraph`
+    The primary use case for this class is to be able to add a :class:`~easylink.step.Step`
+    that does *not* have a corresponding ``Implementation`` to an :class:`~easylink.graph_components.ImplementationGraph`
     since adding any new node requires an object with :class:`~easylink.graph_components.InputSlot`
     and :class:`~easylink.graph_components.OutputSlot` names.
 
@@ -151,13 +155,14 @@ class NullImplementation:
         All required ``InputSlots``.
     output_slots
         All required ``OutputSlots``.
+
     """
 
     def __init__(
         self,
         name: str,
-        input_slots: Iterable["InputSlot"] = (),
-        output_slots: Iterable["OutputSlot"] = (),
+        input_slots: Iterable[InputSlot] = (),
+        output_slots: Iterable[OutputSlot] = (),
     ):
         self.name = name
         """The name of this ``NullImplementation``."""
@@ -172,6 +177,61 @@ def __init__(
         is a constituent. This is definitionally None."""
 
 
+class NullSplitterImplementation(NullImplementation):
+    """A type of :class:`NullImplementation` specifically for :class:`SplitterSteps<easylink.step.SplitterStep>`.
+
+    See ``NullImplementation`` for inherited attributes.
+
+    Parameters
+    ----------
+    splitter_func_name
+        The name of the splitter function to use.
+
+    """
+
+    def __init__(
+        self,
+        name: str,
+        input_slots: Iterable[InputSlot],
+        output_slots: Iterable[OutputSlot],
+        splitter_func_name: str,
+    ):
+        super().__init__(name, input_slots, output_slots)
+        self.splitter_func_name = splitter_func_name
+        """The name of the splitter function to use."""
+
+
+class NullAggregatorImplementation(NullImplementation):
+    """A type of :class:`NullImplementation` specifically for :class:`AggregatorSteps<easylink.step.AggregatorStep>`.
+
+    See ``NullImplementation`` for inherited attributes.
+
+    Parameters
+    ----------
+    aggregator_func_name
+        The name of the aggregation function to use.
+    splitter_node_name
+        The name of the :class:`~easylink.step.SplitterStep` and its corresponding
+        :class:`NullSplitterImplementation` that did the splitting.
+
+    """
+
+    def __init__(
+        self,
+        name: str,
+        input_slots: Iterable[InputSlot],
+        output_slots: Iterable[OutputSlot],
+        aggregator_func_name: str,
+        splitter_node_name: str,
+    ):
+        super().__init__(name, input_slots, output_slots)
+        self.aggregator_func_name = aggregator_func_name
+        """The name of the aggregation function to use."""
+        self.splitter_node_name = splitter_node_name
+        """The name of the :class:`~easylink.step.SplitterStep` and its corresponding
+        :class:`NullSplitterImplementation` that did the splitting."""
+
+
 class PartialImplementation:
     """One part of a combined implementation that spans multiple :class:`Steps<easylink.step.Step>`.
 
@@ -205,8 +265,8 @@ def __init__(
         self,
         combined_name: str,
         schema_step: str,
-        input_slots: Iterable["InputSlot"] = (),
-        output_slots: Iterable["OutputSlot"] = (),
+        input_slots: Iterable[InputSlot] = (),
+        output_slots: Iterable[OutputSlot] = (),
     ):
         self.combined_name = combined_name
         """The name of the combined implementation of which this ``PartialImplementation``