From 65ad948db40970c6b128cbe32f21355e80eb72b7 Mon Sep 17 00:00:00 2001
From: Bertty Contreras-Rojas <bertty@databloom.ai>
Date: Wed, 6 Apr 2022 17:03:24 +0200
Subject: [PATCH] [WAYANG-#8] pywy - Graph add tuple option

Signed-off-by: bertty <bertty@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 5544b3e6b..ce9d1a9d2 100644
--- a/python/src/pywy/dataquanta.py
+++ b/python/src/pywy/dataquanta.py
@@ -55,8 +55,8 @@ def flatmap(self: "DataQuanta[I]", f: FlatmapFunction) -> "DataQuanta[IterableO]
     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 aa748e3c4..005ef5f49 100644
--- a/python/src/pywy/graph/graphtypes.py
+++ b/python/src/pywy/graph/graphtypes.py
@@ -23,4 +23,27 @@ def __init__(self, nodes: List[WyOperator]):
         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 1ba41ba64..220f44169 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 @@ def __init__(self, plugins: Set[Plugin], sinks: Iterable[SinkOperator]):
         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 @@ def print_plan(current: NodeOperator, previous: NodeOperator):
         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)