From 6e034126bd5877346156d98b1ca68e7c9135491f Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Thu, 11 Aug 2022 18:05:33 +0200
Subject: [PATCH 01/13] Add webcam and plot python example

---
 examples/python-dataflow/dataflow.yml       | 20 +++++++++
 examples/python-dataflow/plot_cv2.py        | 49 +++++++++++++++++++++
 examples/python-dataflow/requirements.txt   |  1 +
 examples/python-dataflow/webcam_operator.py | 45 +++++++++++++++++++
 4 files changed, 115 insertions(+)
 create mode 100644 examples/python-dataflow/dataflow.yml
 create mode 100644 examples/python-dataflow/plot_cv2.py
 create mode 100644 examples/python-dataflow/requirements.txt
 create mode 100644 examples/python-dataflow/webcam_operator.py

diff --git a/examples/python-dataflow/dataflow.yml b/examples/python-dataflow/dataflow.yml
new file mode 100644
index 00000000..8561be2c
--- /dev/null
+++ b/examples/python-dataflow/dataflow.yml
@@ -0,0 +1,20 @@
+communication:
+  zenoh:
+    prefix: /example-python-dataflow
+
+nodes:
+  - id: node-1
+    operators:
+    - id: python_webcam
+      python: webcam_operator.py
+      inputs:
+        timer: dora/timer/millis/100
+      outputs:
+        - image
+
+  - id: node-2
+    operators:
+    - id: python_plot
+      python: plot_cv2.py
+      inputs:
+        image: node-1/python_webcam/image
diff --git a/examples/python-dataflow/plot_cv2.py b/examples/python-dataflow/plot_cv2.py
new file mode 100644
index 00000000..50bc9e2b
--- /dev/null
+++ b/examples/python-dataflow/plot_cv2.py
@@ -0,0 +1,49 @@
+from enum import Enum
+from typing import Callable
+
+import cv2
+import numpy as np
+
+
+class DoraStatus(Enum):
+    CONTINUE = 0
+    STOP = 1
+
+
+class Operator:
+    """
+    Example operator incrementing a counter every times its been called.
+
+    The current value of the counter is sent back to dora on `counter`.
+    """
+
+    def __init__(self):
+        self.counter = 0
+
+    def on_input(
+        self,
+        input_id: str,
+        value: bytes,
+        send_output: Callable[[str, bytes], None],
+    ) -> DoraStatus:
+        """Handle input by incrementing count by one.
+
+        Args:
+            input_id (str): Id of the input declared in the yaml configuration
+            value (bytes): Bytes message of the input
+            send_output (Callable[[str, bytes]]): Function enabling sending output back to dora.
+        """
+        self.counter += 1
+        if input_id == "image":
+            frame = np.frombuffer(value, dtype="uint8")
+            frame = np.reshape(frame, (480, 640, 3))
+            cv2.imshow("frame", frame)
+            if cv2.waitKey(1) & 0xFF == ord("q"):
+                return DoraStatus.STOP
+        if self.counter > 20:
+            return DoraStatus.STOP
+        else:
+            return DoraStatus.CONTINUE
+
+    def drop_operator(self):
+        cv2.destroyAllWindows()
diff --git a/examples/python-dataflow/requirements.txt b/examples/python-dataflow/requirements.txt
new file mode 100644
index 00000000..fb10efbf
--- /dev/null
+++ b/examples/python-dataflow/requirements.txt
@@ -0,0 +1 @@
+pip install opencv-python
\ No newline at end of file
diff --git a/examples/python-dataflow/webcam_operator.py b/examples/python-dataflow/webcam_operator.py
new file mode 100644
index 00000000..8eedc550
--- /dev/null
+++ b/examples/python-dataflow/webcam_operator.py
@@ -0,0 +1,45 @@
+from enum import Enum
+from typing import Callable
+
+import cv2
+
+
+class DoraStatus(Enum):
+    CONTINUE = 0
+    STOP = 1
+
+
+class Operator:
+    """
+    Example operator incrementing a counter every times its been called.
+
+    The current value of the counter is sent back to dora on `counter`.
+    """
+
+    def __init__(self):
+        self.video_capture = cv2.VideoCapture(0)
+
+    def on_input(
+        self,
+        input_id: str,
+        value: bytes,
+        send_output: Callable[[str, bytes], None],
+    ) -> DoraStatus:
+        """Handle input by incrementing count by one.
+
+        Args:
+            input_id (str): Id of the input declared in the yaml configuration
+            value (bytes): Bytes message of the input
+            send_output (Callable[[str, bytes]]): Function enabling sending output back to dora.
+        """
+
+        ret, frame = self.video_capture.read()
+        if ret:
+            send_output("image", frame.tobytes())
+        else:
+            print("did not sent video")
+
+        return DoraStatus.CONTINUE
+
+    def drop_operator(self):
+        self.video_capture.release()

From fa5af35a20338265f382d8ac9e039dd7b41a4eb4 Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Fri, 12 Aug 2022 11:08:14 +0200
Subject: [PATCH 02/13] adding object detection to python example

---
 examples/python-dataflow/.gitignore           |  1 +
 examples/python-dataflow/dataflow.yml         | 27 +++++---
 .../{plot_cv2.py => object_detection.py}      | 28 ++++----
 examples/python-dataflow/plot.py              | 66 +++++++++++++++++++
 examples/python-dataflow/requirements.txt     | 16 ++++-
 .../{webcam_operator.py => webcam.py}         |  5 ++
 6 files changed, 116 insertions(+), 27 deletions(-)
 create mode 100644 examples/python-dataflow/.gitignore
 rename examples/python-dataflow/{plot_cv2.py => object_detection.py} (60%)
 create mode 100644 examples/python-dataflow/plot.py
 rename examples/python-dataflow/{webcam_operator.py => webcam.py} (90%)

diff --git a/examples/python-dataflow/.gitignore b/examples/python-dataflow/.gitignore
new file mode 100644
index 00000000..eede66d8
--- /dev/null
+++ b/examples/python-dataflow/.gitignore
@@ -0,0 +1 @@
+*.pt
\ No newline at end of file
diff --git a/examples/python-dataflow/dataflow.yml b/examples/python-dataflow/dataflow.yml
index 8561be2c..26beefc0 100644
--- a/examples/python-dataflow/dataflow.yml
+++ b/examples/python-dataflow/dataflow.yml
@@ -3,18 +3,25 @@ communication:
     prefix: /example-python-dataflow
 
 nodes:
-  - id: node-1
-    operators:
-    - id: python_webcam
-      python: webcam_operator.py
+  - id: webcam
+    operator:
+      python: webcam.py
       inputs:
-        timer: dora/timer/millis/100
+        timer: dora/timer/millis/500
       outputs:
         - image
+  
+  - id: object_detection
+    operator:
+      python: object_detection.py
+      inputs:
+        image: webcam/image
+      outputs:
+        - bbox
 
-  - id: node-2
-    operators:
-    - id: python_plot
-      python: plot_cv2.py
+  - id: plot
+    operator:
+      python: plot.py
       inputs:
-        image: node-1/python_webcam/image
+        image: webcam/image
+        bbox: object_detection/bbox
diff --git a/examples/python-dataflow/plot_cv2.py b/examples/python-dataflow/object_detection.py
similarity index 60%
rename from examples/python-dataflow/plot_cv2.py
rename to examples/python-dataflow/object_detection.py
index 50bc9e2b..5737fcfc 100644
--- a/examples/python-dataflow/plot_cv2.py
+++ b/examples/python-dataflow/object_detection.py
@@ -1,8 +1,8 @@
 from enum import Enum
 from typing import Callable
 
-import cv2
 import numpy as np
+import torch
 
 
 class DoraStatus(Enum):
@@ -18,7 +18,7 @@ class Operator:
     """
 
     def __init__(self):
-        self.counter = 0
+        self.model = torch.hub.load("ultralytics/yolov5", "yolov5n")
 
     def on_input(
         self,
@@ -33,17 +33,13 @@ class Operator:
             value (bytes): Bytes message of the input
             send_output (Callable[[str, bytes]]): Function enabling sending output back to dora.
         """
-        self.counter += 1
-        if input_id == "image":
-            frame = np.frombuffer(value, dtype="uint8")
-            frame = np.reshape(frame, (480, 640, 3))
-            cv2.imshow("frame", frame)
-            if cv2.waitKey(1) & 0xFF == ord("q"):
-                return DoraStatus.STOP
-        if self.counter > 20:
-            return DoraStatus.STOP
-        else:
-            return DoraStatus.CONTINUE
-
-    def drop_operator(self):
-        cv2.destroyAllWindows()
+
+        frame = np.frombuffer(value, dtype="uint8")
+        frame = np.reshape(frame, (480, 640, 3))[
+            :, :, ::-1
+        ]  # OpenCV image (BGR to RGB)
+
+        results = self.model(frame)  # includes NMS
+        arrays = np.array(results.xyxy[0].cpu()).tobytes()
+        send_output("bbox", arrays)
+        return DoraStatus.CONTINUE
diff --git a/examples/python-dataflow/plot.py b/examples/python-dataflow/plot.py
new file mode 100644
index 00000000..2ce8be30
--- /dev/null
+++ b/examples/python-dataflow/plot.py
@@ -0,0 +1,66 @@
+from enum import Enum
+from typing import Callable
+
+import cv2
+import numpy as np
+
+
+class DoraStatus(Enum):
+    CONTINUE = 0
+    STOP = 1
+
+
+class Operator:
+    """
+    Example operator incrementing a counter every times its been called.
+
+    The current value of the counter is sent back to dora on `counter`.
+    """
+
+    def __init__(self):
+        self.image = []
+
+    def on_input(
+        self,
+        input_id: str,
+        value: bytes,
+        send_output: Callable[[str, bytes], None],
+    ) -> DoraStatus:
+        """Handle input by incrementing count by one.
+
+        Args:
+            input_id (str): Id of the input declared in the yaml configuration
+            value (bytes): Bytes message of the input
+            send_output (Callable[[str, bytes]]): Function enabling sending output back to dora.
+        """
+        if input_id == "image":
+            frame = np.frombuffer(value, dtype="uint8")
+            frame = np.reshape(frame, (480, 640, 3))
+            self.image = frame
+        elif input_id == "bbox" and len(self.image) != 0:
+            bboxs = np.frombuffer(value, dtype="float32")
+            bboxs = np.reshape(bboxs, (-1, 6))
+            for bbox in bboxs:
+                [
+                    min_x,
+                    min_y,
+                    max_x,
+                    max_y,
+                    _confidence,
+                    _class_label,
+                ] = bbox
+                cv2.rectangle(
+                    self.image,
+                    (int(min_x), int(min_y)),
+                    (int(max_x), int(max_y)),
+                    (0, 255, 0),
+                    2,
+                )
+            cv2.imshow("frame", self.image)
+            if cv2.waitKey(1) & 0xFF == ord("q"):
+                return DoraStatus.STOP
+
+        return DoraStatus.CONTINUE
+
+    def drop_operator(self):
+        cv2.destroyAllWindows()
diff --git a/examples/python-dataflow/requirements.txt b/examples/python-dataflow/requirements.txt
index fb10efbf..f3a6b71a 100644
--- a/examples/python-dataflow/requirements.txt
+++ b/examples/python-dataflow/requirements.txt
@@ -1 +1,15 @@
-pip install opencv-python
\ No newline at end of file
+# YOLOv5 requirements
+# Usage: pip install -r requirements.txt
+
+# Base ----------------------------------------
+matplotlib>=3.2.2
+numpy>=1.18.5
+opencv-python>=4.1.1
+Pillow>=7.1.2
+PyYAML>=5.3.1
+requests>=2.23.0
+scipy>=1.4.1
+torch>=1.7.0
+torchvision>=0.8.1
+tqdm>=4.64.0
+protobuf<=3.20.1  # https://github.com/ultralytics/yolov5/issues/8012
diff --git a/examples/python-dataflow/webcam_operator.py b/examples/python-dataflow/webcam.py
similarity index 90%
rename from examples/python-dataflow/webcam_operator.py
rename to examples/python-dataflow/webcam.py
index 8eedc550..ab704c13 100644
--- a/examples/python-dataflow/webcam_operator.py
+++ b/examples/python-dataflow/webcam.py
@@ -18,6 +18,7 @@ class Operator:
 
     def __init__(self):
         self.video_capture = cv2.VideoCapture(0)
+        self.counter = 0
 
     def on_input(
         self,
@@ -39,6 +40,10 @@ class Operator:
         else:
             print("did not sent video")
 
+        self.counter += 1
+        if self.counter > 100:
+            return DoraStatus.STOP
+
         return DoraStatus.CONTINUE
 
     def drop_operator(self):

From b1c25e19e7542cb5093836afc8e4b1d611572c99 Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Fri, 12 Aug 2022 11:16:01 +0200
Subject: [PATCH 03/13] Make op queue full error a warning

---
 binaries/runtime/src/operator/mod.rs | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/binaries/runtime/src/operator/mod.rs b/binaries/runtime/src/operator/mod.rs
index 852334da..a1623fa8 100644
--- a/binaries/runtime/src/operator/mod.rs
+++ b/binaries/runtime/src/operator/mod.rs
@@ -1,6 +1,7 @@
 use dora_core::descriptor::{OperatorDefinition, OperatorSource};
 use dora_node_api::config::DataId;
 use eyre::{eyre, Context};
+use log::warn;
 use std::any::Any;
 use tokio::sync::mpsc::{self, Sender};
 
@@ -56,9 +57,12 @@ impl Operator {
                 )
             })?
             .try_send(OperatorInput { id, value })
-            .map_err(|err| match err {
-                tokio::sync::mpsc::error::TrySendError::Closed(_) => eyre!("operator crashed"),
-                tokio::sync::mpsc::error::TrySendError::Full(_) => eyre!("operator queue full"),
+            .or_else(|err| match err {
+                tokio::sync::mpsc::error::TrySendError::Closed(_) => Err(eyre!("operator crashed")),
+                tokio::sync::mpsc::error::TrySendError::Full(_) => {
+                    warn!("operator queue full");
+                    Ok(())
+                }
             })
     }
 

From 7822e87b2dd11db20bd403fba9b61c064425f0db Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Fri, 12 Aug 2022 16:56:06 +0200
Subject: [PATCH 04/13] Add python installer and run script

---
 examples/python-dataflow/install.sh | 10 +++++++
 examples/python-dataflow/run.rs     | 41 +++++++++++++++++++++++++++++
 2 files changed, 51 insertions(+)
 create mode 100644 examples/python-dataflow/install.sh
 create mode 100644 examples/python-dataflow/run.rs

diff --git a/examples/python-dataflow/install.sh b/examples/python-dataflow/install.sh
new file mode 100644
index 00000000..a95bc8a4
--- /dev/null
+++ b/examples/python-dataflow/install.sh
@@ -0,0 +1,10 @@
+python3 -m venv .env
+. $(pwd)/.env/bin/activate
+# Dev dependencies
+pip install maturin
+cd ../../apis/python/node
+maturin develop
+cd ../../../examples/python-dataflow
+
+# Dependencies
+pip install -r requirements.txt
\ No newline at end of file
diff --git a/examples/python-dataflow/run.rs b/examples/python-dataflow/run.rs
new file mode 100644
index 00000000..7ce73049
--- /dev/null
+++ b/examples/python-dataflow/run.rs
@@ -0,0 +1,41 @@
+use eyre::{bail, Context};
+use std::path::Path;
+
+#[tokio::main]
+async fn main() -> eyre::Result<()> {
+    let root = Path::new(env!("CARGO_MANIFEST_DIR"));
+    std::env::set_current_dir(root.join(file!()).parent().unwrap())
+        .wrap_err("failed to set working dir")?;
+
+    build_package("dora-runtime").await?;
+
+    install_python_dependencies(root).await?;
+
+    dora_coordinator::run(dora_coordinator::Command::Run {
+        dataflow: Path::new("dataflow.yml").to_owned(),
+        runtime: Some(root.join("target").join("release").join("dora-runtime")),
+    })
+    .await?;
+
+    Ok(())
+}
+
+async fn build_package(package: &str) -> eyre::Result<()> {
+    let cargo = std::env::var("CARGO").unwrap();
+    let mut cmd = tokio::process::Command::new(&cargo);
+    cmd.arg("build").arg("--release");
+    cmd.arg("--package").arg(package);
+    if !cmd.status().await?.success() {
+        bail!("failed to build {package}");
+    };
+    Ok(())
+}
+
+async fn install_python_dependencies(root: &Path) -> eyre::Result<()> {
+    let mut install = tokio::process::Command::new("sh");
+    install.arg("./install.sh");
+    if !install.status().await?.success() {
+        bail!("failed to create venv");
+    };
+    Ok(())
+}

From 89aa70c9a6e333e4b2a115929fda4098bbc002c8 Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Fri, 12 Aug 2022 16:56:26 +0200
Subject: [PATCH 05/13] make python example usable without webcam

---
 examples/python-dataflow/no_webcam.py        | 23 ++++++++
 examples/python-dataflow/object_detection.py |  6 +-
 examples/python-dataflow/plot.py             |  3 +-
 examples/python-dataflow/webcam.py           | 59 ++++++--------------
 4 files changed, 44 insertions(+), 47 deletions(-)
 create mode 100755 examples/python-dataflow/no_webcam.py
 mode change 100644 => 100755 examples/python-dataflow/webcam.py

diff --git a/examples/python-dataflow/no_webcam.py b/examples/python-dataflow/no_webcam.py
new file mode 100755
index 00000000..d4b47c74
--- /dev/null
+++ b/examples/python-dataflow/no_webcam.py
@@ -0,0 +1,23 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import time
+import urllib.request
+
+import cv2
+import numpy as np
+from dora import Node
+
+req = urllib.request.urlopen(
+    "https://pyimagesearch.com/wp-content/uploads/2015/01/opencv_logo.png"
+)
+
+arr = np.asarray(bytearray(req.read()), dtype=np.uint8)
+node = Node()
+
+start = time.time()
+
+while time.time() - start < 20:
+    # Wait next input
+    node.next()
+    node.send_output("image", arr.tobytes())
diff --git a/examples/python-dataflow/object_detection.py b/examples/python-dataflow/object_detection.py
index 5737fcfc..1e7ea431 100644
--- a/examples/python-dataflow/object_detection.py
+++ b/examples/python-dataflow/object_detection.py
@@ -1,6 +1,7 @@
 from enum import Enum
 from typing import Callable
 
+import cv2
 import numpy as np
 import torch
 
@@ -35,9 +36,8 @@ class Operator:
         """
 
         frame = np.frombuffer(value, dtype="uint8")
-        frame = np.reshape(frame, (480, 640, 3))[
-            :, :, ::-1
-        ]  # OpenCV image (BGR to RGB)
+        frame = cv2.imdecode(frame, -1)
+        frame = frame[:, :, ::-1]  # OpenCV image (BGR to RGB)
 
         results = self.model(frame)  # includes NMS
         arrays = np.array(results.xyxy[0].cpu()).tobytes()
diff --git a/examples/python-dataflow/plot.py b/examples/python-dataflow/plot.py
index 2ce8be30..1f64c0ef 100644
--- a/examples/python-dataflow/plot.py
+++ b/examples/python-dataflow/plot.py
@@ -35,8 +35,9 @@ class Operator:
         """
         if input_id == "image":
             frame = np.frombuffer(value, dtype="uint8")
-            frame = np.reshape(frame, (480, 640, 3))
+            frame = cv2.imdecode(frame, -1)
             self.image = frame
+
         elif input_id == "bbox" and len(self.image) != 0:
             bboxs = np.frombuffer(value, dtype="float32")
             bboxs = np.reshape(bboxs, (-1, 6))
diff --git a/examples/python-dataflow/webcam.py b/examples/python-dataflow/webcam.py
old mode 100644
new mode 100755
index ab704c13..fc913236
--- a/examples/python-dataflow/webcam.py
+++ b/examples/python-dataflow/webcam.py
@@ -1,50 +1,23 @@
-from enum import Enum
-from typing import Callable
-
-import cv2
-
-
-class DoraStatus(Enum):
-    CONTINUE = 0
-    STOP = 1
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
 
+import time
 
-class Operator:
-    """
-    Example operator incrementing a counter every times its been called.
-
-    The current value of the counter is sent back to dora on `counter`.
-    """
-
-    def __init__(self):
-        self.video_capture = cv2.VideoCapture(0)
-        self.counter = 0
-
-    def on_input(
-        self,
-        input_id: str,
-        value: bytes,
-        send_output: Callable[[str, bytes], None],
-    ) -> DoraStatus:
-        """Handle input by incrementing count by one.
+import cv2
+from dora import Node
 
-        Args:
-            input_id (str): Id of the input declared in the yaml configuration
-            value (bytes): Bytes message of the input
-            send_output (Callable[[str, bytes]]): Function enabling sending output back to dora.
-        """
+node = Node()
 
-        ret, frame = self.video_capture.read()
-        if ret:
-            send_output("image", frame.tobytes())
-        else:
-            print("did not sent video")
+video_capture = cv2.VideoCapture(0)
 
-        self.counter += 1
-        if self.counter > 100:
-            return DoraStatus.STOP
+start = time.time()
 
-        return DoraStatus.CONTINUE
+# Run for 20 seconds
+while time.time() - start < 20:
+    # Wait next input
+    node.next()
+    ret, frame = video_capture.read()
+    if ret:
+        node.send_output("image", cv2.imencode(".jpg", frame)[1].tobytes())
 
-    def drop_operator(self):
-        self.video_capture.release()
+video_capture.release()

From eb3ee9d8b72013f9d8fcc9a922c7bd2439bcdd25 Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Fri, 12 Aug 2022 16:57:00 +0200
Subject: [PATCH 06/13] Add python CI without webcam

---
 .github/workflows/ci-python.yml               | 27 +++++++++++++++++++
 Cargo.toml                                    |  4 +++
 examples/python-dataflow/dataflow.yml         |  6 ++---
 .../dataflow_without_webcam.yml               | 27 +++++++++++++++++++
 4 files changed, 61 insertions(+), 3 deletions(-)
 create mode 100644 .github/workflows/ci-python.yml
 create mode 100644 examples/python-dataflow/dataflow_without_webcam.yml

diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
new file mode 100644
index 00000000..38f9e903
--- /dev/null
+++ b/.github/workflows/ci-python.yml
@@ -0,0 +1,27 @@
+
+name: CI-python
+
+on:
+  push:
+    branches:
+    - main
+    paths:
+    - apis/python
+    - binaries/runtime
+    
+jobs:
+
+  examples:
+    name: "Python Examples"
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Install Cap'n Proto
+        run: |
+          export DEBIAN_FRONTEND=noninteractive
+          sudo apt-get install -y capnproto libcapnp-dev
+      - uses: actions/setup-python@v2
+        with:
+          python-version: 3.8.10
+      - name: "Install python dependencies"
+        run: examples/python-dataflow/install.sh
\ No newline at end of file
diff --git a/Cargo.toml b/Cargo.toml
index bd234270..75912b0b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -36,3 +36,7 @@ path = "examples/rust-dataflow/run.rs"
 [[example]]
 name = "cxx-dataflow"
 path = "examples/c++-dataflow/run.rs"
+
+[[example]]
+name = "python-dataflow"
+path = "examples/python-dataflow/run.rs"
diff --git a/examples/python-dataflow/dataflow.yml b/examples/python-dataflow/dataflow.yml
index 26beefc0..0b17bb0f 100644
--- a/examples/python-dataflow/dataflow.yml
+++ b/examples/python-dataflow/dataflow.yml
@@ -4,10 +4,10 @@ communication:
 
 nodes:
   - id: webcam
-    operator:
-      python: webcam.py
+    custom:
+      run: ./webcam.py
       inputs:
-        timer: dora/timer/millis/500
+        timer: dora/timer/millis/100
       outputs:
         - image
   
diff --git a/examples/python-dataflow/dataflow_without_webcam.yml b/examples/python-dataflow/dataflow_without_webcam.yml
new file mode 100644
index 00000000..9f88a6b4
--- /dev/null
+++ b/examples/python-dataflow/dataflow_without_webcam.yml
@@ -0,0 +1,27 @@
+communication:
+  zenoh:
+    prefix: /example-python-dataflow
+
+nodes:
+  - id: no_webcam
+    custom:
+      run: ./no_webcam.py
+      inputs:
+        timer: dora/timer/millis/100
+      outputs:
+        - image
+  
+  - id: object_detection
+    operator:
+      python: object_detection.py
+      inputs:
+        image: no_webcam/image
+      outputs:
+        - bbox
+
+  - id: plot
+    operator:
+      python: plot.py
+      inputs:
+        image: no_webcam/image
+        bbox: object_detection/bbox

From b91b6db3f4db4cb9d52737b8eff7947206f7e4e4 Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Fri, 12 Aug 2022 17:07:28 +0200
Subject: [PATCH 07/13] Remove legacy python example

---
 .../examples/graphs/python_test.yml           | 29 --------------
 .../examples/nodes/python/printer.py          |  8 ----
 .../examples/nodes/python/static_string.py    | 11 ------
 examples/python-operator/op.py                | 36 -----------------
 examples/python-operator/op2.py               | 39 -------------------
 5 files changed, 123 deletions(-)
 delete mode 100644 binaries/coordinator/examples/graphs/python_test.yml
 delete mode 100644 binaries/coordinator/examples/nodes/python/printer.py
 delete mode 100644 binaries/coordinator/examples/nodes/python/static_string.py
 delete mode 100644 examples/python-operator/op.py
 delete mode 100644 examples/python-operator/op2.py

diff --git a/binaries/coordinator/examples/graphs/python_test.yml b/binaries/coordinator/examples/graphs/python_test.yml
deleted file mode 100644
index 66044ba7..00000000
--- a/binaries/coordinator/examples/graphs/python_test.yml
+++ /dev/null
@@ -1,29 +0,0 @@
-communication:
-  zenoh:
-    prefix: /foo
-
-nodes:
-  - id: static-string
-    custom:
-      run: python examples/nodes/python/static_string.py
-      outputs:
-        - string
-
-  - id: python-printer
-    custom:
-      run: python examples/nodes/python/printer.py
-      inputs:
-          string: static-string/string
-          time2: rust-timer/time
-  
-  - id: rust-timer
-    custom:
-      run: cargo run --example source_timer 
-      outputs:
-          - time
-
-  - id: rust-logger
-    custom:
-      run: cargo run --example sink_logger 
-      inputs:
-          time: static-string/string
\ No newline at end of file
diff --git a/binaries/coordinator/examples/nodes/python/printer.py b/binaries/coordinator/examples/nodes/python/printer.py
deleted file mode 100644
index 45943ed0..00000000
--- a/binaries/coordinator/examples/nodes/python/printer.py
+++ /dev/null
@@ -1,8 +0,0 @@
-from dora import Node
-
-node = Node()
-
-for id, value in node:
-    print(f"From Python, id: {id}, value: {value}") if value is not [] else None
-
-print("printer finished")
diff --git a/binaries/coordinator/examples/nodes/python/static_string.py b/binaries/coordinator/examples/nodes/python/static_string.py
deleted file mode 100644
index 8b158fd0..00000000
--- a/binaries/coordinator/examples/nodes/python/static_string.py
+++ /dev/null
@@ -1,11 +0,0 @@
-import time
-
-from dora import Node
-
-node = Node()
-
-for i in range(100):
-    node.send_output("string", b"Hello World")
-    time.sleep(0.1)
-
-print("static string finished")
diff --git a/examples/python-operator/op.py b/examples/python-operator/op.py
deleted file mode 100644
index bffe8508..00000000
--- a/examples/python-operator/op.py
+++ /dev/null
@@ -1,36 +0,0 @@
-from typing import Callable
-from enum import Enum
-
-class DoraStatus(Enum):
-    CONTINUE = 0
-    STOP = 1
-
-class Operator:
-    """
-    Example operator incrementing a counter every times its been called.
-
-    The current value of the counter is sent back to dora on `counter`.
-    """
-
-    def __init__(self, counter=0):
-        self.counter = counter
-
-    def on_input(
-        self,
-        input_id: str,
-        value: bytes,
-        send_output: Callable[[str, bytes], None],
-    ):
-        """Handle input by incrementing count by one.
-
-        Args:
-            input_id (str): Id of the input declared in the yaml configuration
-            value (bytes): Bytes message of the input
-            send_output (Callable[[str, bytes]]): Function enabling sending output back to dora.
-        """
-        val_len = len(value)
-        print(f"PYTHON received input {input_id}; value length: {val_len}")
-        send_output("counter", (self.counter % 256).to_bytes(1, "little"))
-        self.counter = self.counter + 1
-
-        return DoraStatus.OK
diff --git a/examples/python-operator/op2.py b/examples/python-operator/op2.py
deleted file mode 100644
index 0053b1a4..00000000
--- a/examples/python-operator/op2.py
+++ /dev/null
@@ -1,39 +0,0 @@
-from typing import Callable
-from enum import Enum
-
-class DoraStatus(Enum):
-    OK = 0
-    STOP = 1
-
-class Operator:
-    """
-    Example operator incrementing a counter every times its been called.
-
-    The current value of the counter is sent back to dora on `counter`.
-    """
-
-    def __init__(self, counter=0):
-        self.counter = counter
-
-    def on_input(
-        self,
-        input_id: str,
-        value: bytes,
-        send_output: Callable[[str, bytes], None],
-    ):
-        """Handle input by incrementing count by one.
-
-        Args:
-            input_id (str): Id of the input declared in the yaml configuration
-            value (bytes): Bytes message of the input
-            send_output (Callable[[str, bytes]]): Function enabling sending output back to dora.
-        """
-        val_len = len(value)
-        print(f"PYTHON received input {input_id}; value length: {val_len}")
-        send_output("counter", (self.counter % 256).to_bytes(1, "little"))
-        self.counter = self.counter + 1
-
-        if self.counter > 500:
-            return DoraStatus.STOP
-        else:
-            return DoraStatus.OK

From fe25716fbc587a8218ee35191ee256fb66e9ab50 Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Fri, 12 Aug 2022 17:52:00 +0200
Subject: [PATCH 08/13] adding `README` to `python-dataflow`

---
 examples/python-dataflow/REAMDE.md | 33 ++++++++++++++++++++++++++++++
 1 file changed, 33 insertions(+)
 create mode 100644 examples/python-dataflow/REAMDE.md

diff --git a/examples/python-dataflow/REAMDE.md b/examples/python-dataflow/REAMDE.md
new file mode 100644
index 00000000..b9ece590
--- /dev/null
+++ b/examples/python-dataflow/REAMDE.md
@@ -0,0 +1,33 @@
+# Python Dataflow Example
+
+This examples shows how to create and connect dora operators and custom nodes in Python.
+
+## Overview
+
+The [`dataflow.yml`](./dataflow.yml) defines a simple dataflow graph with the following three nodes:
+
+- a webcam node, that connects to your webcam and feed the dataflow with webcam frame as jpeg compressed bytearray.
+- an object detection node, that apply Yolo v5 on the webcam image. The model is imported from Pytorch Hub. The output is the bouding box of each object detected, the confidence and the class. You can have more info here: https://pytorch.org/hub/ultralytics_yolov5/
+- a window plotting node, that will retrieve the webcam image and the Yolov5 bounding box and join the two together.
+
+## Getting started
+
+```bash
+cargo run --example python-dataflow 
+```
+
+## Installation
+
+To install, you should run the `install.sh` script.
+
+```bash
+install.sh
+```
+
+## Run the dataflow as a standalone
+
+- Start the `dora-coordinator`, passing the paths to the dataflow file and the `dora-runtime` as arguments:
+
+```
+../../target/release/dora-coordinator run dataflow.yml ../../target/release/dora-runtime
+```

From a52bbec30b20a70d02973dae9e6f58a6d23c140c Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Fri, 12 Aug 2022 19:10:53 +0200
Subject: [PATCH 09/13] fix CI and make CI job non-interactive

---
 .github/workflows/ci-python.yml                 |  7 +++++--
 .../python-dataflow/dataflow_without_webcam.yml |  2 +-
 examples/python-dataflow/no_webcam.py           |  6 +++---
 examples/python-dataflow/object_detection.py    |  6 ++----
 examples/python-dataflow/plot.py                | 17 ++++++++++-------
 examples/python-dataflow/run.rs                 | 10 ++++++++--
 6 files changed, 29 insertions(+), 19 deletions(-)

diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index 38f9e903..9389fd60 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -23,5 +23,8 @@ jobs:
       - uses: actions/setup-python@v2
         with:
           python-version: 3.8.10
-      - name: "Install python dependencies"
-        run: examples/python-dataflow/install.sh
\ No newline at end of file
+      - name: "Python Dataflow example"
+        uses: actions-rs/cargo@v1
+        with:
+          command: run
+          args: --example python-dataflow
\ No newline at end of file
diff --git a/examples/python-dataflow/dataflow_without_webcam.yml b/examples/python-dataflow/dataflow_without_webcam.yml
index 9f88a6b4..944847f0 100644
--- a/examples/python-dataflow/dataflow_without_webcam.yml
+++ b/examples/python-dataflow/dataflow_without_webcam.yml
@@ -1,6 +1,6 @@
 communication:
   zenoh:
-    prefix: /example-python-dataflow
+    prefix: /example-python-no-webcam-dataflow
 
 nodes:
   - id: no_webcam
diff --git a/examples/python-dataflow/no_webcam.py b/examples/python-dataflow/no_webcam.py
index d4b47c74..464cd77c 100755
--- a/examples/python-dataflow/no_webcam.py
+++ b/examples/python-dataflow/no_webcam.py
@@ -8,9 +8,7 @@ import cv2
 import numpy as np
 from dora import Node
 
-req = urllib.request.urlopen(
-    "https://pyimagesearch.com/wp-content/uploads/2015/01/opencv_logo.png"
-)
+req = urllib.request.urlopen("https://ultralytics.com/images/zidane.jpg")
 
 arr = np.asarray(bytearray(req.read()), dtype=np.uint8)
 node = Node()
@@ -21,3 +19,5 @@ while time.time() - start < 20:
     # Wait next input
     node.next()
     node.send_output("image", arr.tobytes())
+
+time.sleep(1)
diff --git a/examples/python-dataflow/object_detection.py b/examples/python-dataflow/object_detection.py
index 1e7ea431..784bc70e 100644
--- a/examples/python-dataflow/object_detection.py
+++ b/examples/python-dataflow/object_detection.py
@@ -13,9 +13,7 @@ class DoraStatus(Enum):
 
 class Operator:
     """
-    Example operator incrementing a counter every times its been called.
-
-    The current value of the counter is sent back to dora on `counter`.
+    Infering object from images
     """
 
     def __init__(self):
@@ -27,7 +25,7 @@ class Operator:
         value: bytes,
         send_output: Callable[[str, bytes], None],
     ) -> DoraStatus:
-        """Handle input by incrementing count by one.
+        """Handle image
 
         Args:
             input_id (str): Id of the input declared in the yaml configuration
diff --git a/examples/python-dataflow/plot.py b/examples/python-dataflow/plot.py
index 1f64c0ef..fe9d2a8a 100644
--- a/examples/python-dataflow/plot.py
+++ b/examples/python-dataflow/plot.py
@@ -1,9 +1,12 @@
+import os
 from enum import Enum
 from typing import Callable
 
 import cv2
 import numpy as np
 
+CI = os.environ.get("CI")
+
 
 class DoraStatus(Enum):
     CONTINUE = 0
@@ -12,9 +15,7 @@ class DoraStatus(Enum):
 
 class Operator:
     """
-    Example operator incrementing a counter every times its been called.
-
-    The current value of the counter is sent back to dora on `counter`.
+    Plot image and bounding box
     """
 
     def __init__(self):
@@ -26,7 +27,8 @@ class Operator:
         value: bytes,
         send_output: Callable[[str, bytes], None],
     ) -> DoraStatus:
-        """Handle input by incrementing count by one.
+        """
+        Put image and bounding box on cv2 window.
 
         Args:
             input_id (str): Id of the input declared in the yaml configuration
@@ -57,9 +59,10 @@ class Operator:
                     (0, 255, 0),
                     2,
                 )
-            cv2.imshow("frame", self.image)
-            if cv2.waitKey(1) & 0xFF == ord("q"):
-                return DoraStatus.STOP
+            if CI != "true":
+                cv2.imshow("frame", self.image)
+                if cv2.waitKey(1) & 0xFF == ord("q"):
+                    return DoraStatus.STOP
 
         return DoraStatus.CONTINUE
 
diff --git a/examples/python-dataflow/run.rs b/examples/python-dataflow/run.rs
index 7ce73049..2be87ea0 100644
--- a/examples/python-dataflow/run.rs
+++ b/examples/python-dataflow/run.rs
@@ -1,5 +1,5 @@
 use eyre::{bail, Context};
-use std::path::Path;
+use std::{env, path::Path};
 
 #[tokio::main]
 async fn main() -> eyre::Result<()> {
@@ -11,8 +11,14 @@ async fn main() -> eyre::Result<()> {
 
     install_python_dependencies(root).await?;
 
+    let dataflow = if env::var("CI").is_ok() {
+        Path::new("dataflow_without_webcam.yml").to_owned()
+    } else {
+        Path::new("dataflow.yml").to_owned()
+    };
+
     dora_coordinator::run(dora_coordinator::Command::Run {
-        dataflow: Path::new("dataflow.yml").to_owned(),
+        dataflow,
         runtime: Some(root.join("target").join("release").join("dora-runtime")),
     })
     .await?;

From 8d66e35add61e9316e121705ab0b8292a1931cc7 Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Fri, 12 Aug 2022 19:29:35 +0200
Subject: [PATCH 10/13] Fix CI filter pattern and clippy warning

---
 .github/workflows/ci-python.yml | 9 +++++++--
 examples/python-dataflow/run.rs | 2 +-
 2 files changed, 8 insertions(+), 3 deletions(-)

diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index 9389fd60..b673478a 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -1,13 +1,18 @@
 
 name: CI-python
 
+# Filter CI as this job will take time.
 on:
   push:
+    paths:
+    - apis/python/**
+    - binaries/runtime/**
+  pull_request:
     branches:
     - main
     paths:
-    - apis/python
-    - binaries/runtime
+    - apis/python/**
+    - binaries/runtime/**
     
 jobs:
 
diff --git a/examples/python-dataflow/run.rs b/examples/python-dataflow/run.rs
index 2be87ea0..9fec64ff 100644
--- a/examples/python-dataflow/run.rs
+++ b/examples/python-dataflow/run.rs
@@ -37,7 +37,7 @@ async fn build_package(package: &str) -> eyre::Result<()> {
     Ok(())
 }
 
-async fn install_python_dependencies(root: &Path) -> eyre::Result<()> {
+async fn install_python_dependencies(_root: &Path) -> eyre::Result<()> {
     let mut install = tokio::process::Command::new("sh");
     install.arg("./install.sh");
     if !install.status().await?.success() {

From ff814e0a2a0f1776c1a85e7489d4760aa9198b01 Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Sat, 13 Aug 2022 10:53:09 +0200
Subject: [PATCH 11/13] Run python exemple inside of python env

---
 examples/python-dataflow/run.rs               | 24 +++++--------------
 .../python-dataflow/{install.sh => run.sh}    |  4 +++-
 2 files changed, 9 insertions(+), 19 deletions(-)
 rename examples/python-dataflow/{install.sh => run.sh} (62%)

diff --git a/examples/python-dataflow/run.rs b/examples/python-dataflow/run.rs
index 9fec64ff..8f9874a3 100644
--- a/examples/python-dataflow/run.rs
+++ b/examples/python-dataflow/run.rs
@@ -9,19 +9,7 @@ async fn main() -> eyre::Result<()> {
 
     build_package("dora-runtime").await?;
 
-    install_python_dependencies(root).await?;
-
-    let dataflow = if env::var("CI").is_ok() {
-        Path::new("dataflow_without_webcam.yml").to_owned()
-    } else {
-        Path::new("dataflow.yml").to_owned()
-    };
-
-    dora_coordinator::run(dora_coordinator::Command::Run {
-        dataflow,
-        runtime: Some(root.join("target").join("release").join("dora-runtime")),
-    })
-    .await?;
+    run(root).await?;
 
     Ok(())
 }
@@ -37,11 +25,11 @@ async fn build_package(package: &str) -> eyre::Result<()> {
     Ok(())
 }
 
-async fn install_python_dependencies(_root: &Path) -> eyre::Result<()> {
-    let mut install = tokio::process::Command::new("sh");
-    install.arg("./install.sh");
-    if !install.status().await?.success() {
-        bail!("failed to create venv");
+async fn run(_root: &Path) -> eyre::Result<()> {
+    let mut run = tokio::process::Command::new("sh");
+    run.arg("./run.sh");
+    if !run.status().await?.success() {
+        bail!("failed to run python example.");
     };
     Ok(())
 }
diff --git a/examples/python-dataflow/install.sh b/examples/python-dataflow/run.sh
similarity index 62%
rename from examples/python-dataflow/install.sh
rename to examples/python-dataflow/run.sh
index a95bc8a4..7f11bda9 100644
--- a/examples/python-dataflow/install.sh
+++ b/examples/python-dataflow/run.sh
@@ -7,4 +7,6 @@ maturin develop
 cd ../../../examples/python-dataflow
 
 # Dependencies
-pip install -r requirements.txt
\ No newline at end of file
+pip install -r requirements.txt
+
+cargo run -p dora-coordinator --release -- run dataflow_without_webcam.yml
\ No newline at end of file

From e22fecc35d98dc0da79ecb0ddb95d468480bc648 Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Sat, 13 Aug 2022 12:38:36 +0200
Subject: [PATCH 12/13] fix requirements

---
 examples/python-dataflow/requirements.txt | 28 +++++++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/examples/python-dataflow/requirements.txt b/examples/python-dataflow/requirements.txt
index f3a6b71a..8a934ca1 100644
--- a/examples/python-dataflow/requirements.txt
+++ b/examples/python-dataflow/requirements.txt
@@ -13,3 +13,31 @@ torch>=1.7.0
 torchvision>=0.8.1
 tqdm>=4.64.0
 protobuf<=3.20.1  # https://github.com/ultralytics/yolov5/issues/8012
+
+# Logging -------------------------------------
+tensorboard>=2.4.1
+# wandb
+# clearml
+
+# Plotting ------------------------------------
+pandas>=1.1.4
+seaborn>=0.11.0
+
+# Export --------------------------------------
+# coremltools>=5.2  # CoreML export
+# onnx>=1.9.0  # ONNX export
+# onnx-simplifier>=0.4.1  # ONNX simplifier
+# nvidia-pyindex  # TensorRT export
+# nvidia-tensorrt  # TensorRT export
+# scikit-learn==0.19.2  # CoreML quantization
+# tensorflow>=2.4.1  # TFLite export (or tensorflow-cpu, tensorflow-aarch64)
+# tensorflowjs>=3.9.0  # TF.js export
+# openvino-dev  # OpenVINO export
+
+# Extras --------------------------------------
+ipython  # interactive notebook
+psutil  # system utilization
+thop>=0.1.1  # FLOPs computation
+# albumentations>=1.0.3
+# pycocotools>=2.0  # COCO mAP
+# roboflow
\ No newline at end of file

From 79d07a81a105aa6b0c5eeae1f92986db80ee753f Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Tue, 16 Aug 2022 18:39:38 +0200
Subject: [PATCH 13/13] add text label to plot

---
 examples/python-dataflow/plot.py  | 20 +++++++-
 examples/python-dataflow/utils.py | 82 +++++++++++++++++++++++++++++++
 2 files changed, 100 insertions(+), 2 deletions(-)
 create mode 100644 examples/python-dataflow/utils.py

diff --git a/examples/python-dataflow/plot.py b/examples/python-dataflow/plot.py
index fe9d2a8a..82b4c895 100644
--- a/examples/python-dataflow/plot.py
+++ b/examples/python-dataflow/plot.py
@@ -5,8 +5,12 @@ from typing import Callable
 import cv2
 import numpy as np
 
+from utils import LABELS
+
 CI = os.environ.get("CI")
 
+font = cv2.FONT_HERSHEY_SIMPLEX
+
 
 class DoraStatus(Enum):
     CONTINUE = 0
@@ -49,8 +53,8 @@ class Operator:
                     min_y,
                     max_x,
                     max_y,
-                    _confidence,
-                    _class_label,
+                    confidence,
+                    label,
                 ] = bbox
                 cv2.rectangle(
                     self.image,
@@ -59,6 +63,18 @@ class Operator:
                     (0, 255, 0),
                     2,
                 )
+
+                cv2.putText(
+                    self.image,
+                    LABELS[int(label)] + f", {confidence:0.2f}",
+                    (int(max_x), int(max_y)),
+                    font,
+                    0.75,
+                    (0, 255, 0),
+                    2,
+                    1,
+                )
+
             if CI != "true":
                 cv2.imshow("frame", self.image)
                 if cv2.waitKey(1) & 0xFF == ord("q"):
diff --git a/examples/python-dataflow/utils.py b/examples/python-dataflow/utils.py
new file mode 100644
index 00000000..dabc915e
--- /dev/null
+++ b/examples/python-dataflow/utils.py
@@ -0,0 +1,82 @@
+LABELS = [
+    "ABC",
+    "bicycle",
+    "car",
+    "motorcycle",
+    "airplane",
+    "bus",
+    "train",
+    "truck",
+    "boat",
+    "traffic light",
+    "fire hydrant",
+    "stop sign",
+    "parking meter",
+    "bench",
+    "bird",
+    "cat",
+    "dog",
+    "horse",
+    "sheep",
+    "cow",
+    "elephant",
+    "bear",
+    "zebra",
+    "giraffe",
+    "backpack",
+    "umbrella",
+    "handbag",
+    "tie",
+    "suitcase",
+    "frisbee",
+    "skis",
+    "snowboard",
+    "sports ball",
+    "kite",
+    "baseball bat",
+    "baseball glove",
+    "skateboard",
+    "surfboard",
+    "tennis racket",
+    "bottle",
+    "wine glass",
+    "cup",
+    "fork",
+    "knife",
+    "spoon",
+    "bowl",
+    "banana",
+    "apple",
+    "sandwich",
+    "orange",
+    "broccoli",
+    "carrot",
+    "hot dog",
+    "pizza",
+    "donut",
+    "cake",
+    "chair",
+    "couch",
+    "potted plant",
+    "bed",
+    "dining table",
+    "toilet",
+    "tv",
+    "laptop",
+    "mouse",
+    "remote",
+    "keyboard",
+    "cell phone",
+    "microwave",
+    "oven",
+    "toaster",
+    "sink",
+    "refrigerator",
+    "book",
+    "clock",
+    "vase",
+    "scissors",
+    "teddy bear",
+    "hair drier",
+    "toothbrush",
+]