Merge branch 'develop' into test

Author: yuanchi2807

.github/pull_request_template.md

@@ -0,0 +1,25 @@
+**Related Issue**
+
+<!-- List any issues that this pr is related to, fixes or supports -->
+
+Supports #ISSUE_NUMBER
+
+**Related PRs**
+
+<!-- Does this pr depend on or replace an existing pr? Link it here -->
+
+This PR is not dependent on any other PR
+
+**What does this PR do?**
+
+<!-- The intent of this section is to help team members understand the scope of your changes and why you are making said changes. If there are any concerns or risks associated with your pull request, they should be documented here so your team members can give informative feedback in their reviews. -->
+
+**Description of Changes**
+
+<!-- The intent of this section is to describe the implementation details of the changes you made. Any architectural decisions made should be noted here as well as a short description of why you made that decision. -->
+
+**What gif most accurately describes how I feel towards this PR?**
+
+<!-- Cute animals are also acceptable. -->
+
+![Example of a gif](https://media.github.ibm.com/user/59/files/074da280-78ba-11ea-81d1-49ce0654c29d)

codeflare/pipelines/Datamodel.py

@@ -1,6 +1,12 @@
-from sklearn.base import BaseEstimator
 from abc import ABC, abstractmethod
+from enum import Enum
+
+import sklearn.base as base
+from sklearn.base import TransformerMixin
+from sklearn.base import BaseEstimator
 
+import ray
+import codeflare.pipelines.Exceptions as pe
 
 class Xy:
     """
@@ -35,9 +41,12 @@ class XYRef:
     computed), these holders are essential to the pipeline constructs.
     """
 
-    def __init__(self, Xref, yref):
+    def __init__(self, Xref, yref, prev_node_state_ref=None, curr_node_state_ref=None, prev_Xyrefs = None):
         self.__Xref__ = Xref
         self.__yref__ = yref
+        self.__prev_node_state_ref__ = prev_node_state_ref
+        self.__curr_node_state_ref__ = curr_node_state_ref
+        self.__prev_Xyrefs__ = prev_Xyrefs
 
     def get_Xref(self):
         """
@@ -51,6 +60,32 @@ def get_yref(self):
         """
         return self.__yref__
 
+    def get_prev_node_state_ref(self):
+        return self.__prev_node_state_ref__
+
+    def get_curr_node_state_ref(self):
+        return self.__curr_node_state_ref__
+
+    def get_prev_xyrefs(self):
+        return self.__prev_Xyrefs__
+
+
+class NodeInputType(Enum):
+    OR = 0,
+    AND = 1
+
+
+class NodeFiringType(Enum):
+    ANY = 0,
+    ALL = 1
+
+
+class NodeStateType(Enum):
+    STATELESS = 0,
+    IMMUTABLE = 1,
+    MUTABLE_SEQUENTIAL = 2,
+    MUTABLE_AGGREGATE = 3
+
 
 class Node(ABC):
     """
@@ -59,12 +94,27 @@ class Node(ABC):
     node name and the type of the node match.
     """
 
+    def __init__(self, node_name, node_input_type: NodeInputType, node_firing_type: NodeFiringType, node_state_type: NodeStateType):
+        self.__node_name__ = node_name
+        self.__node_input_type__ = node_input_type
+        self.__node_firing_type__ = node_firing_type
+        self.__node_state_type__ = node_state_type
+
     def __str__(self):
         return self.__node_name__
 
+    def get_node_input_type(self):
+        return self.__node_input_type__
+
+    def get_node_firing_type(self):
+        return self.__node_firing_type__
+
+    def get_node_state_type(self):
+        return self.__node_state_type__
+
     @abstractmethod
-    def get_and_flag(self):
-        raise NotImplementedError("Please implement this method")
+    def clone(self):
+        raise NotImplementedError("Please implement the clone method")
 
     def __hash__(self):
         """
@@ -88,12 +138,11 @@ def __eq__(self, other):
         )
 
 
-class OrNode(Node):
+class EstimatorNode(Node):
     """
     Or node, which is the basic node that would be the equivalent of any SKlearn pipeline
     stage. This node is initialized with an estimator that needs to extend sklearn.BaseEstimator.
     """
-    __estimator__ = None
 
     def __init__(self, node_name: str, estimator: BaseEstimator):
         """
@@ -102,7 +151,8 @@ def __init__(self, node_name: str, estimator: BaseEstimator):
         :param node_name: Name of the node
         :param estimator: The base estimator
         """
-        self.__node_name__ = node_name
+
+        super().__init__(node_name, NodeInputType.OR, NodeFiringType.ANY, NodeStateType.IMMUTABLE)
         self.__estimator__ = estimator
 
     def get_estimator(self) -> BaseEstimator:
@@ -113,37 +163,33 @@ def get_estimator(self) -> BaseEstimator:
         """
         return self.__estimator__
 
-    def get_and_flag(self):
-        """
-        A flag to check if node is AND or not. By definition, this is NOT
-        an AND node.
-        :return: False, always
-        """
-        return False
+    def clone(self):
+        cloned_estimator = base.clone(self.__estimator__)
+        return EstimatorNode(self.__node_name__, cloned_estimator)
 
 
-class AndFunc(ABC):
-    """
-    Or nodes are init-ed from the
-    """
+class AndTransform(TransformerMixin, BaseEstimator):
+    @abstractmethod
+    def transform(self, xy_list: list) -> Xy:
+        raise NotImplementedError("Please implement this method")
 
+
+class GeneralTransform(TransformerMixin, BaseEstimator):
     @abstractmethod
-    def eval(self, xy_list: list) -> Xy:
+    def transform(self, xy: Xy) -> Xy:
         raise NotImplementedError("Please implement this method")
 
 
 class AndNode(Node):
-    __andfunc__ = None
-
-    def __init__(self, node_name: str, and_func: AndFunc):
-        self.__node_name__ = node_name
+    def __init__(self, node_name: str, and_func: AndTransform):
+        super().__init__(node_name, NodeInputType.AND, NodeFiringType.ANY, NodeStateType.STATELESS)
         self.__andfunc__ = and_func
 
-    def get_and_func(self) -> AndFunc:
+    def get_and_func(self) -> AndTransform:
         return self.__andfunc__
 
-    def get_and_flag(self):
-        return True
+    def clone(self):
+        return AndNode(self.__node_name__, self.__andfunc__)
 
 
 class Edge:
@@ -322,5 +368,69 @@ def get_post_edges(self, node: Node):
         return post_edges
 
     def is_terminal(self, node: Node):
-        node_post_edges = self.get_post_edges(node)
-        return len(node_post_edges) == 0
+        post_nodes = self.__post_graph__[node]
+        return not post_nodes
+
+    def get_terminal_nodes(self):
+        # dict from level to nodes
+        terminal_nodes = []
+        for node in self.__pre_graph__.keys():
+            if self.is_terminal(node):
+                terminal_nodes.append(node)
+        return terminal_nodes
+
+
+class PipelineOutput:
+    """
+    Pipeline output to keep reference counters so that pipelines can be materialized
+    """
+    def __init__(self, out_args, edge_args):
+        self.__out_args__ = out_args
+        self.__edge_args__ = edge_args
+
+    def get_xyrefs(self, node: Node):
+        if node in self.__out_args__:
+            xyrefs_ptr = self.__out_args__[node]
+        elif node in self.__edge_args__:
+            xyrefs_ptr = self.__edge_args__[node]
+        else:
+            raise pe.PipelineNodeNotFoundException("Node " + str(node) + " not found")
+
+        xyrefs = ray.get(xyrefs_ptr)
+        return xyrefs
+
+    def get_edge_args(self):
+        return self.__edge_args__
+
+
+class PipelineInput:
+    """
+    in_args is a dict from a node -> [Xy]
+    """
+    def __init__(self):
+        self.__in_args__ = {}
+
+    def add_xyref_ptr_arg(self, node: Node, xyref_ptr):
+        if node not in self.__in_args__:
+            self.__in_args__[node] = []
+
+        self.__in_args__[node].append(xyref_ptr)
+
+    def add_xyref_arg(self, node: Node, xyref: XYRef):
+        if node not in self.__in_args__:
+            self.__in_args__[node] = []
+
+        xyref_ptr = ray.put(xyref)
+        self.__in_args__[node].append(xyref_ptr)
+
+    def add_xy_arg(self, node: Node, xy: Xy):
+        if node not in self.__in_args__:
+            self.__in_args__[node] = []
+
+        x_ref = ray.put(xy.get_x())
+        y_ref = ray.put(xy.get_y())
+        xyref = XYRef(x_ref, y_ref)
+        self.add_xyref_arg(node, xyref)
+
+    def get_in_args(self):
+        return self.__in_args__

codeflare/pipelines/Exceptions.py

@@ -0,0 +1,17 @@
+class BasePipelineException(Exception):
+    pass
+
+
+class PipelineSaveException(BasePipelineException):
+    def __init__(self, message):
+        self.message = message
+
+
+class PipelineNodeNotFoundException(BasePipelineException):
+    def __init__(self, message):
+        self.message = message
+
+
+class PipelineException(BasePipelineException):
+    def __init__(self, message):
+        self.message = message