Feat/new snaps producer

depthai_nodes/message/__init__.py

@@ -10,6 +10,7 @@
 from .map import Map2D
 from .prediction import Prediction, Predictions
 from .segmentation import SegmentationMask
+from .snap_data import SnapData
 
 __all__ = [
     "ImgDetectionExtended",
@@ -26,4 +27,5 @@
     "Prediction",
     "Predictions",
     "GatheredData",
+    "SnapData",
 ]

depthai_nodes/message/snap_data.py

@@ -0,0 +1,40 @@
+from typing import Dict, List, Optional
+
+import depthai as dai
+
+
+class SnapData(dai.Buffer):
+    """DepthAI-compatible message for representing a single snap event.
+
+    Attributes
+    ----------
+    snap_name : str
+        Logical name of the snap.
+    file_name : str
+        File name for the snap image.
+    frame : dai.ImgFrame
+        Captured image frame associated with the snap.
+    detections : Optional[dai.ImgDetections]
+        Optional detection data.
+    tags : List[str]
+        Optional list of tags to include.
+    extras : Dict[str, str]
+        Additional metadata.
+    """
+
+    def __init__(
+        self,
+        snap_name: str,
+        frame: dai.ImgFrame,
+        file_name: str = "",
+        detections: Optional[dai.ImgDetections] = None,
+        tags: Optional[List[str]] = None,
+        extras: Optional[Dict[str, str]] = None,
+    ):
+        super().__init__()
+        self.snap_name = snap_name
+        self.file_name = file_name
+        self.frame = frame
+        self.detections = detections
+        self.tags = tags or []
+        self.extras = extras or {}

depthai_nodes/node/__init__.py

@@ -30,7 +30,9 @@
 from .parsers.yolo import YOLOExtendedParser
 from .parsers.yunet import YuNetParser
 from .parsing_neural_network import ParsingNeuralNetwork
-from .snaps_producer import SnapsProducer, SnapsProducerFrameOnly
+from .snaps_uploader import (
+    SnapsUploader,
+)
 from .tiles_patcher import TilesPatcher
 from .tiling import Tiling
 
@@ -69,7 +71,6 @@
     "ImgFrameOverlay",
     "ImgDetectionsBridge",
     "ImgDetectionsFilter",
-    "SnapsProducerFrameOnly",
-    "SnapsProducer",
+    "SnapsUploader",
     "BaseHostNode",
 ]

depthai_nodes/node/snaps_uploader.py

@@ -0,0 +1,44 @@
+import os
+
+import depthai as dai
+
+from depthai_nodes.message import SnapData
+from depthai_nodes.node.base_host_node import BaseHostNode
+
+
+class SnapsUploader(BaseHostNode):
+    """Host node responsible for receiving SnapData messages and sending snaps to
+    DepthAI Hub Events API."""
+
+    def __init__(self):
+        super().__init__()
+        self._em = dai.EventsManager()
+
+    def set_token(self, token: str):
+        os.environ.setdefault("DEPTHAI_HUB_API_KEY", token)
+
+    def set_url(self, url: str):
+        os.environ.setdefault("DEPTHAI_HUB_EVENTS_BASE_URL", url)
+
+    def build(self, snaps: dai.Node.Output):
+        self.link_args(snaps)
+        return self
+
+    def process(self, snap: dai.Buffer):
+        assert isinstance(snap, SnapData), f"Expected SnapData, got {type(snap)}"
+
+        self._logger.debug(f"Sending snap: {snap.snap_name} -> {snap.file_name}")
+
+        success = self._em.sendSnap(
+            name=snap.snap_name,
+            fileName=snap.file_name,
+            imgFrame=snap.frame,
+            imgDetections=snap.detections,
+            tags=snap.tags,
+            extras=snap.extras,
+        )
+
+        if success:
+            self._logger.info(f"Snap '{snap.snap_name}' sent successfully.")
+        else:
+            self._logger.error(f"Failed to send snap '{snap.snap_name}'.")

requirements.txt

@@ -1,3 +1,3 @@
-depthai>=3.0.0rc2,<=3.0.0
+depthai>=3.0.0.dev0,<3.1
 opencv-python-headless~=4.10.0
 numpy>=1.22