Upate webcam operator example to make use of opentelemetry jaeger

.github/workflows/ci-python.yml

@@ -6,6 +6,8 @@ on:
     paths:
       - apis/python/**
       - binaries/runtime/**
+      - examples/python-dataflow/**
+      - examples/python-operator-dataflow/**
   pull_request:
   workflow_dispatch:
 

examples/python-dataflow/webcam.py

@@ -4,6 +4,7 @@
 import time
 
 import cv2
+
 from dora import Node
 
 node = Node()
@@ -20,7 +21,11 @@
         case "INPUT":
             ret, frame = video_capture.read()
             if ret:
-                node.send_output("image", cv2.imencode(".jpg", frame)[1].tobytes())
+                node.send_output(
+                    "image",
+                    cv2.imencode(".jpg", frame)[1].tobytes(),
+                    event["metadata"],
+                )
         case "STOP":
             print("received stop")
             break

examples/python-operator-dataflow/dataflow.yml

@@ -4,8 +4,8 @@ communication:
 
 nodes:
   - id: webcam
-    custom:
-      source: webcam.py
+    operator:
+      python: webcam.py
       inputs:
         tick: dora/timer/millis/100
       outputs:

examples/python-operator-dataflow/object_detection.py

@@ -8,7 +8,7 @@
 import torch
 
 from dora import DoraStatus
-    
+
 
 class Operator:
     """

examples/python-operator-dataflow/webcam.py

@@ -1,31 +1,46 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
 import time
+from typing import Callable
 
 import cv2
-from dora import Node
-
-node = Node()
-
-video_capture = cv2.VideoCapture(0)
-
-start = time.time()
-
-# Run for 20 seconds
-while time.time() - start < 10:
-    # Wait next dora_input
-    event = node.next()
-    match event["type"]:
-        case "INPUT":
-            ret, frame = video_capture.read()
-            if ret:
-                node.send_output("image", cv2.imencode(".jpg", frame)[1].tobytes())
-        case "STOP":
-            print("received stop")
-            break
-        case other:
-            print("received unexpected event:", other)
-            break
-
-video_capture.release()
+
+from dora import DoraStatus
+
+
+class Operator:
+    """
+    Sending image from webcam to the dataflow
+    """
+
+    def __init__(self):
+        self.video_capture = cv2.VideoCapture(0)
+        self.start_time = time.time()
+
+    def on_event(
+        self,
+        dora_event: dict,
+        send_output: Callable[[str, bytes], None],
+    ) -> DoraStatus:
+        match dora_event["type"]:
+            case "INPUT":
+                ret, frame = self.video_capture.read()
+                if ret:
+                    send_output(
+                        "image",
+                        cv2.imencode(".jpg", frame)[1].tobytes(),
+                        dora_event["metadata"],
+                    )
+            case "STOP":
+                print("received stop")
+            case other:
+                print("received unexpected event:", other)
+
+        if time.time() - self.start_time < 20:
+            return DoraStatus.CONTINUE
+        else:
+            return DoraStatus.STOP
+
+    def __del__(self):
+        self.video_capture.release()