[incubator-wayang] 11/32: [WAYANG-#8] pywy - Graph add tuple option

[be...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

bertty pushed a commit to branch python-platform
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git

commit 65ad948db40970c6b128cbe32f21355e80eb72b7
Author: Bertty Contreras-Rojas <be...@databloom.ai>
AuthorDate: Wed Apr 6 17:03:24 2022 +0200

    [WAYANG-#8] pywy - Graph add tuple option
    
    Signed-off-by: bertty <be...@apache.org>
---
 python/src/pywy/dataquanta.py        |  4 ++--
 python/src/pywy/graph/graphtypes.py  | 25 ++++++++++++++++++++++++-
 python/src/pywy/wayangplan/wayang.py | 23 +++++++++++++++++++++--
 3 files changed, 47 insertions(+), 5 deletions(-)

diff --git a/python/src/pywy/dataquanta.py b/python/src/pywy/dataquanta.py
index 5544b3e6..ce9d1a9d 100644
--- a/python/src/pywy/dataquanta.py
+++ b/python/src/pywy/dataquanta.py
@@ -55,8 +55,8 @@ class DataQuanta(GenericTco):
     def storeTextFile(self: "DataQuanta[I]", path: str) :
         last = self.__connect(TextFileSink(path))
         plan = PywyPlan(self.context.plugins, [last])
-        plan.print()
-
+        #plan.print()
+        plan.printTuple()
         # TODO add the logic to execute the plan
 
     def __connect(self, op:WyOperator, port_op: int = 0) -> WyOperator:
diff --git a/python/src/pywy/graph/graphtypes.py b/python/src/pywy/graph/graphtypes.py
index aa748e3c..005ef5f4 100644
--- a/python/src/pywy/graph/graphtypes.py
+++ b/python/src/pywy/graph/graphtypes.py
@@ -23,4 +23,27 @@ class WGraphOfOperator(WayangGraph[NodeOperator]):
         super(WGraphOfOperator, self).__init__(nodes)
 
     def build_node(self, t:WyOperator) -> NodeOperator:
-        return NodeOperator(t)
\ No newline at end of file
+        return NodeOperator(t)
+
+
+class NodeTuple(GraphNode[Tuple[WyOperator, WyOperator]]):
+
+    def __init__(self, op: WyOperator):
+        super(NodeTuple, self).__init__((op, None))
+
+    def getadjacents(self) -> Iterable[Tuple[WyOperator, WyOperator]]:
+        operator: WyOperator = self.current[0]
+        if operator is None or operator.inputs == 0:
+            return []
+        return operator.inputOperator
+
+    def build_node(self, t:WyOperator) -> 'NodeTuple':
+        return NodeTuple(t)
+
+class WGraphOfTuple(WayangGraph[NodeTuple]):
+
+    def __init__(self, nodes: List[WyOperator]):
+        super(WGraphOfTuple, self).__init__(nodes)
+
+    def build_node(self, t:WyOperator) -> NodeTuple:
+        return NodeTuple(t)
\ No newline at end of file
diff --git a/python/src/pywy/wayangplan/wayang.py b/python/src/pywy/wayangplan/wayang.py
index 1ba41ba6..220f4416 100644
--- a/python/src/pywy/wayangplan/wayang.py
+++ b/python/src/pywy/wayangplan/wayang.py
@@ -1,7 +1,7 @@
 from typing import Iterable, Set
 
 from pywy.graph.graph import WayangGraph
-from pywy.graph.graphtypes import WGraphOfOperator, NodeOperator
+from pywy.graph.graphtypes import WGraphOfOperator, NodeOperator, WGraphOfTuple, NodeTuple
 from pywy.wayangplan.sink import SinkOperator
 from pywy.platforms.basic.plugin import Plugin
 
@@ -16,7 +16,7 @@ class PywyPlan:
         self.set_graph()
 
     def set_graph(self):
-        self.graph = WGraphOfOperator(self.sinks)
+        self.graph = WGraphOfTuple(self.sinks)
 
     def print(self):
         def print_plan(current: NodeOperator, previous: NodeOperator):
@@ -39,4 +39,23 @@ class PywyPlan:
         self.graph.traversal(None, self.graph.starting_nodes, print_plan)
 
 
+    def printTuple(self):
+        def print_plan(current: NodeTuple, previous: NodeTuple):
+            if current is None:
+                print("this is source")
+                print(previous.current)
+                return
+            if previous is None:
+                print("this is sink")
+                print(current.current)
+                return
+
+            print(
+                "############\n{}\n@@@@@ => previous is\n{}\n############\n"
+                    .format(
+                        current.current,
+                        previous.current
+                     )
+            )
+        self.graph.traversal(None, self.graph.starting_nodes, print_plan)