From f8ccd3b8137c7184ede04df848f23a45ced29570 Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Thu, 3 Aug 2023 17:37:19 +0200
Subject: [PATCH] Add dataflow example

---
 examples/python-dataflow/control_node.py      | 32 +++++++++++++++++++
 examples/python-dataflow/dataflow.yaml        | 19 +++++++++++
 examples/python-dataflow/random_turtle.py     | 15 ++++++++-
 .../python-standalone-bridge/random_turtle.py | 30 +++++++++++++++++
 4 files changed, 95 insertions(+), 1 deletion(-)
 create mode 100644 examples/python-dataflow/control_node.py
 create mode 100644 examples/python-dataflow/dataflow.yaml
 create mode 100644 examples/python-standalone-bridge/random_turtle.py

diff --git a/examples/python-dataflow/control_node.py b/examples/python-dataflow/control_node.py
new file mode 100644
index 00000000..ab55c251
--- /dev/null
+++ b/examples/python-dataflow/control_node.py
@@ -0,0 +1,32 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import random
+from dora import Node
+import pyarrow as pa
+
+node = Node()
+
+event = node.next()
+direction = {
+    "linear": {
+        "x": random.random() + 1,
+    },
+    "angular": {
+        "z": (random.random() - 0.5) * 5,
+    },
+}
+if event["type"] == "INPUT":
+    print(
+        f"""Node received:
+    id: {event["id"]},
+    value: {event["data"]},
+    metadata: {event["metadata"]}"""
+    )
+    node.send_output(
+        "direction",
+        pa.array(
+            [random.random() + 1, 0, 0, 0, 0, random.random() - 0.5],
+            type=pa.uint8(),
+        ),
+    )
diff --git a/examples/python-dataflow/dataflow.yaml b/examples/python-dataflow/dataflow.yaml
new file mode 100644
index 00000000..3aa7d007
--- /dev/null
+++ b/examples/python-dataflow/dataflow.yaml
@@ -0,0 +1,19 @@
+nodes:
+  - id: dora_ros2_bridge
+    custom:
+      source: python
+      python: random_turtle.py
+      inputs:
+        direction: control/direction
+        timer: dora/timer/millis/50
+      outputs:
+        - turtle_pose
+
+  - id: control
+    custom:
+      source: python
+      args: ./control_node.py
+      inputs:
+        turtle_pose: dora_ros2_bridge/turtle_pose
+      outputs:
+        - direction
\ No newline at end of file
diff --git a/examples/python-dataflow/random_turtle.py b/examples/python-dataflow/random_turtle.py
index 75d909e8..2f000561 100644
--- a/examples/python-dataflow/random_turtle.py
+++ b/examples/python-dataflow/random_turtle.py
@@ -1,6 +1,7 @@
 import time, random
 from dora_ros2_bridge import *
 from dora import Node
+import pyarrow as pa
 
 node = Node()
 context = Ros2Context()
@@ -34,4 +35,16 @@ for event in node:
                     pose = pose_reader.next()
                     if pose == None:
                         break
-                    node.send_output(pose)
+                    node.send_output(
+                        pa.array(
+                            [
+                                pose["x"],
+                                pose["y"],
+                                pose["z"],
+                                pose["theta"],
+                                pose["linar_velocity"],
+                                pose["angular_velocity"],
+                            ],
+                            type=pa.uint8(),
+                        )
+                    )
diff --git a/examples/python-standalone-bridge/random_turtle.py b/examples/python-standalone-bridge/random_turtle.py
new file mode 100644
index 00000000..2b68dd2d
--- /dev/null
+++ b/examples/python-standalone-bridge/random_turtle.py
@@ -0,0 +1,30 @@
+from dora_ros2_bridge import *
+import time, random
+
+context = Ros2Context()
+node = context.new_node("turtle_teleop", "/ros2_demo", Ros2NodeOptions(rosout = True))
+
+topic_qos = Ros2QosPolicies(reliable = True, max_blocking_time = 0.1)
+
+turtle_twist_topic = node.create_topic("/turtle1/cmd_vel", "geometry_msgs::Twist", topic_qos)
+twist_writer = node.create_publisher(turtle_twist_topic)
+
+turtle_pose_topic = node.create_topic("/turtle1/pose", "turtlesim::Pose", topic_qos)
+pose_reader = node.create_subscription(turtle_pose_topic)
+
+for i in range(500):
+    direction = {
+        'linear': {
+            'x': random.random() + 1,
+        },
+        'angular': {
+            'z': (random.random() - 0.5) * 5,
+        }
+    }
+    twist_writer.publish(direction);
+    while True:
+        pose = pose_reader.next()
+        if pose == None:
+            break
+        print(pose)
+    time.sleep(0.5)