Video processing in inference server

development/stream_interface/workflows_demo.py

@@ -1,4 +1,3 @@
-import os
 from threading import Thread
 from typing import List, Optional
 
@@ -7,12 +6,13 @@
 
 from inference import InferencePipeline
 from inference.core.interfaces.camera.entities import VideoFrame
+from inference.core.interfaces.camera.video_source import BufferFillingStrategy, BufferConsumptionStrategy
 from inference.core.interfaces.stream.watchdog import PipelineWatchDog, BasePipelineWatchDog
 from inference.core.utils.drawing import create_tiles
 
 STOP = False
 ANNOTATOR = sv.BoundingBoxAnnotator()
-TARGET_PROJECT = os.environ["TARGET_PROJECT"]
+fps_monitor = sv.FPSMonitor()
 
 
 def main() -> None:
@@ -32,36 +32,25 @@ def main() -> None:
                 "confidence": 0.5,
             },
             {
-                "type": "RoboflowDatasetUpload",
-                "name": "roboflow_dataset_upload",
-                "images": "$inputs.image",
+                "type": "roboflow_core/bounding_box_visualization@v1",
+                "name": "bbox_visualiser",
                 "predictions": "$steps.step_1.predictions",
-                "target_project": TARGET_PROJECT,
-                "usage_quota_name": "upload_quota_XXX",
-                "fire_and_forget": True,
-            },
-            {
-                "type": "RoboflowCustomMetadata",
-                "name": "metadata_upload",
-                "predictions": "$steps.step_1.predictions",
-                "field_name": "dummy",
-                "field_value": "dummy",
-                "fire_and_forget": True,
-            },
+                "image": "$inputs.image"
+            }
         ],
         "outputs": [
             {"type": "JsonField", "name": "predictions", "selector": "$steps.step_1.predictions"},
-            {"type": "JsonField", "name": "upload_error", "selector": "$steps.roboflow_dataset_upload.error_status"},
-            {"type": "JsonField", "name": "upload_message", "selector": "$steps.roboflow_dataset_upload.message"},
-            {"type": "JsonField", "name": "metadata_error", "selector": "$steps.metadata_upload.error_status"},
-            {"type": "JsonField", "name": "metadata_message", "selector": "$steps.metadata_upload.message"},
+            {"type": "JsonField", "name": "preview", "selector": "$steps.bbox_visualiser.image"},
+
         ],
     }
     pipeline = InferencePipeline.init_with_workflow(
-        video_reference=[os.environ["VIDEO_REFERENCE"]] * 2,
+        video_reference=["rtsp://localhost:8554/live.stream"],
         workflow_specification=workflow_specification,
         watchdog=watchdog,
         on_prediction=workflows_sink,
+        source_buffer_filling_strategy=BufferFillingStrategy.DROP_OLDEST,
+        source_buffer_consumption_strategy=BufferConsumptionStrategy.EAGER,
     )
     control_thread = Thread(target=command_thread, args=(pipeline, watchdog))
     control_thread.start()
@@ -91,17 +80,25 @@ def workflows_sink(
     predictions: List[Optional[dict]],
     video_frames: List[Optional[VideoFrame]],
 ) -> None:
+    fps_monitor.tick()
+    if not isinstance(predictions, list):
+        predictions = [predictions]
+        video_frames = [video_frames]
     images_to_show = []
     for prediction, frame in zip(predictions, video_frames):
         if prediction is None or frame is None:
             continue
         detections: sv.Detections = prediction["predictions"]
         visualised = ANNOTATOR.annotate(frame.image.copy(), detections)
         images_to_show.append(visualised)
-        print(prediction["upload_message"], prediction["metadata_message"])
     tiles = create_tiles(images=images_to_show)
     cv2.imshow(f"Predictions", tiles)
     cv2.waitKey(1)
+    if hasattr(fps_monitor, "fps"):
+        fps_value = fps_monitor.fps
+    else:
+        fps_value = fps_monitor()
+    print(f"FPS: {fps_value}")
 
 
 if __name__ == '__main__':

docker/config/cpu_http.py

@@ -1,5 +1,8 @@
+from multiprocessing import Process
+
 from inference.core.cache import cache
 from inference.core.interfaces.http.http_api import HttpInterface
+from inference.core.interfaces.stream_manager.manager_app.app import start
 from inference.core.managers.active_learning import ActiveLearningManager, BackgroundTaskActiveLearningManager
 from inference.core.managers.base import ModelManager
 from inference.core.managers.decorators.fixed_size_cache import WithFixedSizeCache
@@ -9,7 +12,7 @@
 import os
 from prometheus_fastapi_instrumentator import Instrumentator
 
-from inference.core.env import MAX_ACTIVE_MODELS, ACTIVE_LEARNING_ENABLED, LAMBDA
+from inference.core.env import MAX_ACTIVE_MODELS, ACTIVE_LEARNING_ENABLED, LAMBDA, ENABLE_STREAM_API
 from inference.models.utils import ROBOFLOW_MODEL_TYPES
 
 model_registry = RoboflowModelRegistry(ROBOFLOW_MODEL_TYPES)
@@ -38,3 +41,9 @@
     @app.on_event("startup")
     async def _startup():
         instrumentor.expose(app)
+
+if ENABLE_STREAM_API:
+    stream_manager_process = Process(
+        target=start,
+    )
+    stream_manager_process.start()

docker/config/gpu_http.py

@@ -1,9 +1,12 @@
 import os
+from multiprocessing import Process
+
 from prometheus_fastapi_instrumentator import Instrumentator
 
 from inference.core.cache import cache
-from inference.core.env import MAX_ACTIVE_MODELS, ACTIVE_LEARNING_ENABLED, LAMBDA
+from inference.core.env import MAX_ACTIVE_MODELS, ACTIVE_LEARNING_ENABLED, LAMBDA, ENABLE_STREAM_API
 from inference.core.interfaces.http.http_api import HttpInterface
+from inference.core.interfaces.stream_manager.manager_app.app import start
 from inference.core.managers.active_learning import ActiveLearningManager, BackgroundTaskActiveLearningManager
 from inference.core.managers.base import ModelManager
 from inference.core.managers.decorators.fixed_size_cache import WithFixedSizeCache
@@ -41,3 +44,9 @@
     @app.on_event("startup")
     async def _startup():
         instrumentor.expose(app)
+
+if ENABLE_STREAM_API:
+    stream_manager_process = Process(
+        target=start,
+    )
+    stream_manager_process.start()

docker/dockerfiles/Dockerfile.onnx.jetson.4.5.0

@@ -70,5 +70,6 @@ ENV WORKFLOWS_STEP_EXECUTION_MODE=local
 ENV WORKFLOWS_MAX_CONCURRENT_STEPS=1
 ENV API_LOGGING_ENABLED=True
 ENV CORE_MODEL_TROCR_ENABLED=false
+ENV RUNS_ON_JETSON=True
 
 ENTRYPOINT uvicorn gpu_http:app --workers $NUM_WORKERS --host $HOST --port $PORT

docker/dockerfiles/Dockerfile.onnx.jetson.4.6.1

@@ -85,5 +85,6 @@ ENV WORKFLOWS_STEP_EXECUTION_MODE=local
 ENV WORKFLOWS_MAX_CONCURRENT_STEPS=1
 ENV API_LOGGING_ENABLED=True
 ENV CORE_MODEL_TROCR_ENABLED=false
+ENV RUNS_ON_JETSON=True
 
 ENTRYPOINT uvicorn gpu_http:app --workers $NUM_WORKERS --host $HOST --port $PORT

docker/dockerfiles/Dockerfile.onnx.jetson.5.1.1

@@ -81,5 +81,6 @@ ENV WORKFLOWS_STEP_EXECUTION_MODE=local
 ENV WORKFLOWS_MAX_CONCURRENT_STEPS=1
 ENV API_LOGGING_ENABLED=True
 ENV CORE_MODEL_TROCR_ENABLED=false
+ENV RUNS_ON_JETSON=True
 
 ENTRYPOINT uvicorn gpu_http:app --workers $NUM_WORKERS --host $HOST --port $PORT

docs/workflows/video_processing/overview.md

@@ -0,0 +1,7 @@
+# Video Processing with Workflows
+
+We've begun our journey into video processing using Workflows. Over time, we've expanded the number of 
+video-specific blocks (e.g., the ByteTracker block) and continue to dedicate efforts toward improving 
+their performance and robustness. The current state of this work is as follows:
+
+* 

inference/core/env.py

@@ -424,3 +424,7 @@
 DEDICATED_DEPLOYMENT_WORKSPACE_URL = os.environ.get(
     "DEDICATED_DEPLOYMENT_WORKSPACE_URL", None
 )
+
+ENABLE_STREAM_API = str2bool(os.getenv("ENABLE_STREAM_API", "True"))
+
+RUNS_ON_JETSON = str2bool(os.getenv("RUNS_ON_JETSON", "False"))

inference/core/interfaces/camera/video_source.py

@@ -17,6 +17,7 @@
     DEFAULT_BUFFER_SIZE,
     DEFAULT_MAXIMUM_ADAPTIVE_FRAMES_DROPPED_IN_ROW,
     DEFAULT_MINIMUM_ADAPTIVE_MODE_SAMPLES,
+    RUNS_ON_JETSON,
 )
 from inference.core.interfaces.camera.entities import (
     SourceProperties,
@@ -132,7 +133,10 @@ def locked_executor(video_source: "VideoSource", *args, **kwargs) -> None:
 
 class CV2VideoFrameProducer(VideoFrameProducer):
     def __init__(self, video: Union[str, int]):
-        self.stream = cv2.VideoCapture(video)
+        if _consumes_camera_on_jetson(video=video):
+            self.stream = cv2.VideoCapture(video, cv2.CAP_V4L2)
+        else:
+            self.stream = cv2.VideoCapture(video)
 
     def isOpened(self) -> bool:
         return self.stream.isOpened()
@@ -165,6 +169,14 @@ def release(self):
         self.stream.release()
 
 
+def _consumes_camera_on_jetson(video: Union[str, int]) -> bool:
+    if not RUNS_ON_JETSON:
+        return False
+    if isinstance(video, int):
+        return True
+    return video.startswith("/dev/video")
+
+
 class VideoSource:
     @classmethod
     def init(