diff --git a/workflows/CHANGELOG.md b/workflows/CHANGELOG.md index eb82f556fc..06df200b13 100644 --- a/workflows/CHANGELOG.md +++ b/workflows/CHANGELOG.md @@ -58,7 +58,7 @@ for contributor and maintainer responsibilities. ### Fixed - `roboflow_core/asset_library_attributes@v1` normalizes metadata keys before sending them: characters outside `[A-Za-z0-9_.]` become `_`, and a leading `.` gets a `_` prefix. If two keys of one image normalize to the same key (for example `traffic light` and `traffic_light`), the block raises an error instead of letting one value overwrite the other. Before, one invalid key (for example a class name with a space) made the API reject every attribute for that image. - +- Rate Limiter (`roboflow_core/rate_limiter@v1`) falls back to wall-clock time when `video_reference_image` carries no video metadata, for example a still image sent over HTTP or the output of a step that builds a new image (`stitch_images@v1`, `grid_visualization@v1`, the Stability AI blocks). The block used to read the placeholder metadata such images report (frame 0 at 30 FPS), so its time never advanced and it suppressed `next_steps` on every run after the first. Video time (`frame_number / fps`) is now used only for frames from a video file with a known, non-zero FPS; stream frames, frames whose source is unknown and video files without FPS use the frame timestamp, as Time in Zone and Velocity do. Streams that report FPS used to be timed by `frame_number / fps`. The new `WorkflowImageData.has_video_metadata()` tells real video metadata from the placeholder. - Root-coordinate recovery (`sv_detections_to_root_coordinates`, applied to workflow outputs by default) re-encodes `rle_mask` from the re-anchored dense masks. A block that ran on a crop and returns RLE with dense masks, such as SAM 3 v3 with `output_format: rle`, used to serialize root-coordinate boxes next to an RLE still sized and placed in the crop. RLE-only predictions (`mask=None`) are unchanged, as in `scale_sv_detections`. - Video-state blocks now accept an image from a step output as well as from a Workflow input, so they keep working inside an embedded `roboflow_core/inner_workflow@v1` whose `image` is bound to a parent step (for example a preprocessing block). Inlining replaces the child's `$inputs.image` with that step selector, and the merged definition used to fail with `Could not parse workflow definition` even though each Workflow compiled on its own. Affected blocks: `time_in_zone@v2`/`@v3`, `velocity@v1`, `line_counter@v2`, `path_deviation_analytics@v2`, `detection_event_log@v1`, `byte_tracker@v2`, `cache_get@v1`, `cache_set@v1`, `delta_filter@v1` and `rate_limiter@v1` (`video_reference_image`). The blocks read `video_metadata` from the image, which step outputs built with `WorkflowImageData.copy_and_replace` keep, so per-video state is unchanged. diff --git a/workflows/roboflow_workflows/core_steps/flow_control/rate_limiter/v1.py b/workflows/roboflow_workflows/core_steps/flow_control/rate_limiter/v1.py index 8eb1f6f82d..19d8fcf56b 100644 --- a/workflows/roboflow_workflows/core_steps/flow_control/rate_limiter/v1.py +++ b/workflows/roboflow_workflows/core_steps/flow_control/rate_limiter/v1.py @@ -39,8 +39,9 @@ 1. Takes a cooldown period (in seconds), a `depends_on` reference, and `next_steps` as input 2. Tracks the timestamp of the last execution using an internal state variable 3. Calculates the current time: - - For video processing: Uses video metadata (frame number and FPS) to compute a video-time-based timestamp when `video_reference_image` is provided - - For other contexts: Uses system clock time (datetime.now()) + - For video files: Uses video metadata (frame number and FPS) to compute a video-time-based timestamp when `video_reference_image` is provided + - For streams, or video without a known FPS: Uses the frame timestamp from the video metadata of `video_reference_image` + - For other contexts (no `video_reference_image`, or an image without video metadata, such as a still image): Uses system clock time (datetime.now()) 4. Compares the time elapsed since the last execution against the `cooldown_seconds` threshold 5. If sufficient time has passed (elapsed time >= cooldown_seconds): - Updates the last execution timestamp @@ -96,7 +97,7 @@ class RateLimiterManifest(WorkflowBlockManifest): } ) cooldown_seconds: float = Field( - description="Minimum number of seconds that must elapse between consecutive executions of the next_steps blocks. The rate limiter tracks the last execution timestamp and only allows execution to continue if at least this many seconds have passed since the previous execution. Must be greater than or equal to 0.0. For video workflows, this cooldown period is enforced based on video time (calculated from frame number and FPS) when video_reference_image is provided, rather than wall-clock time.", + description="Minimum number of seconds that must elapse between consecutive executions of the next_steps blocks. The rate limiter tracks the last execution timestamp and only allows execution to continue if at least this many seconds have passed since the previous execution. Must be greater than or equal to 0.0. For video files, this cooldown period is enforced based on video time (calculated from frame number and FPS) when video_reference_image is provided, rather than wall-clock time; for streams it is enforced based on frame timestamps.", examples=[1.0], default=1.0, ge=0.0, @@ -113,7 +114,7 @@ class RateLimiterManifest(WorkflowBlockManifest): examples=[["$steps.upload"]], ) video_reference_image: Optional[Selector(kind=[IMAGE_KIND])] = Field( - description="Optional reference to a video frame image to use for video-time-based timestamp generation. When provided, the rate limiter calculates timestamps based on video metadata (frame number and FPS) rather than system clock time. This is useful when processing video faster than real-time, ensuring rate limiting works correctly relative to video playback time rather than processing speed. If not provided (None), the block uses system clock time (datetime.now()). Only applicable for video processing workflows.", + description="Optional reference to a video frame image to use for video-time-based timestamp generation. When provided, the rate limiter calculates timestamps based on video metadata rather than system clock time: frame number and FPS for video files, and the frame timestamp for streams or when FPS is unknown. This is useful when processing video faster than real-time, ensuring rate limiting works correctly relative to video playback time rather than processing speed. If not provided (None), or if the image carries no video metadata (for example a still image), the block uses system clock time (datetime.now()). Only applicable for video processing workflows.", examples=["$inputs.image"], default=None, ) @@ -161,7 +162,7 @@ def discover_dependent_resources(self) -> List[DependentResource]: class RateLimiterBlockV1(WorkflowBlock): def __init__(self): super().__init__() - self._last_executed_at: Optional[datetime] = None + self._last_executed_at: Optional[float] = None @classmethod def get_manifest(cls) -> Type[WorkflowBlockManifest]: @@ -174,22 +175,28 @@ def run( next_steps: List[StepSelector], video_reference_image: Optional[WorkflowImageData] = None, ) -> BlockResult: - current_time = datetime.now() - try: - metadata = video_reference_image.video_metadata - current_time = datetime.fromtimestamp( - 1 / metadata.fps * metadata.frame_number - ) - except Exception: - # reference not passed, metadata not set, or not a video frame - pass - + current_time = _get_current_time_seconds(video_reference_image) should_throttle = False if self._last_executed_at is not None: - should_throttle = ( - current_time - self._last_executed_at - ).total_seconds() < cooldown_seconds + should_throttle = (current_time - self._last_executed_at) < cooldown_seconds if should_throttle: return FlowControl(mode="terminate_branch") self._last_executed_at = current_time return FlowControl(mode="select_step", context=next_steps) + + +def _get_current_time_seconds( + video_reference_image: Optional[WorkflowImageData], +) -> float: + # Wall-clock time unless the image carries real video metadata: the + # placeholder `video_metadata` of a still image is always frame 0, which would + # pin the time and throttle every frame after the first. + if video_reference_image is None or not video_reference_image.has_video_metadata(): + return datetime.now().timestamp() + + metadata = video_reference_image.video_metadata + if metadata.comes_from_video_file and metadata.fps: + video_time = metadata.frame_number / metadata.fps + return video_time + frame_time = metadata.frame_timestamp.timestamp() + return frame_time diff --git a/workflows/roboflow_workflows/execution_engine/entities/base.py b/workflows/roboflow_workflows/execution_engine/entities/base.py index 48be8f1ad8..e68058d4cf 100644 --- a/workflows/roboflow_workflows/execution_engine/entities/base.py +++ b/workflows/roboflow_workflows/execution_engine/entities/base.py @@ -742,6 +742,16 @@ def base64_image(self) -> str: ).decode("ascii") return self._base64_image + def has_video_metadata(self) -> bool: + """Whether the image carries video metadata supplied by its producer. + + ``video_metadata`` never fails: for an image without metadata (a still image, + or a step output built without copying the origin's metadata) it synthesises + a placeholder with ``frame_number=0`` and ``fps=30``. Blocks that derive time + from video metadata use this check to tell the placeholder from a real frame. + """ + return self._video_metadata is not None + @property def video_metadata(self) -> VideoMetadata: if self._video_metadata is not None: diff --git a/workflows/tests/unit_tests/core_steps/control_flow/test_rate_limiter.py b/workflows/tests/unit_tests/core_steps/control_flow/test_rate_limiter.py index 4258077c84..a74dee5f5d 100644 --- a/workflows/tests/unit_tests/core_steps/control_flow/test_rate_limiter.py +++ b/workflows/tests/unit_tests/core_steps/control_flow/test_rate_limiter.py @@ -1,8 +1,23 @@ +from datetime import datetime, timezone +from typing import List, Optional +from unittest import mock +from unittest.mock import MagicMock + +import numpy as np import pytest from pydantic import ValidationError +from roboflow_workflows.core_steps.flow_control.rate_limiter import ( + v1 as rate_limiter_v1, +) from roboflow_workflows.core_steps.flow_control.rate_limiter.v1 import ( + RateLimiterBlockV1, RateLimiterManifest, ) +from roboflow_workflows.execution_engine.entities.base import ( + ImageParentMetadata, + VideoMetadata, + WorkflowImageData, +) @pytest.mark.parametrize( @@ -46,3 +61,157 @@ def test_rate_limiter_manifest_parsing_when_cooldown_is() -> None: # when with pytest.raises(ValidationError): _ = RateLimiterManifest.model_validate(raw_manifest) + + +def _image(video_metadata: Optional[VideoMetadata] = None) -> WorkflowImageData: + return WorkflowImageData( + parent_metadata=ImageParentMetadata(parent_id="some"), + numpy_image=np.zeros((32, 32, 3), dtype=np.uint8), + video_metadata=video_metadata, + ) + + +def _video_frame( + frame_number: int, + *, + fps: Optional[float] = 1, + comes_from_video_file: Optional[bool] = True, + frame_timestamp: Optional[datetime] = None, +) -> WorkflowImageData: + if frame_timestamp is None: + frame_timestamp = datetime.fromtimestamp(1726570875, tz=timezone.utc) + video_metadata = VideoMetadata( + video_identifier="video", + frame_number=frame_number, + frame_timestamp=frame_timestamp, + fps=fps, + comes_from_video_file=comes_from_video_file, + ) + image = _image(video_metadata=video_metadata) + + return image + + +def _run_frames( + images: List[Optional[WorkflowImageData]], + *, + cooldown_seconds: float, +) -> List[bool]: + block = RateLimiterBlockV1() + passed = [] + for image in images: + result = block.run( + cooldown_seconds=cooldown_seconds, + depends_on="$steps.model", + next_steps=["$steps.upload"], + video_reference_image=image, + ) + passed.append(result.context == ["$steps.upload"]) + + return passed + + +@mock.patch.object(rate_limiter_v1, "datetime") +def test_rate_limiter_uses_wall_clock_for_image_without_video_metadata( + datetime_mock: MagicMock, +) -> None: + # given + wall_clock = [ + datetime.fromtimestamp(1726570875 + offset) for offset in (0, 0.5, 1.0, 2.5) + ] + datetime_mock.now.side_effect = wall_clock + images = [_image() for _ in wall_clock] + + # when + result = _run_frames(images, cooldown_seconds=1.0) + + # then + assert result == [ + True, + False, + True, + True, + ], "Expected still images to be throttled by wall-clock time, not by frame 0 of placeholder metadata" + + +@mock.patch.object(rate_limiter_v1, "datetime") +def test_rate_limiter_uses_wall_clock_when_reference_image_is_not_provided( + datetime_mock: MagicMock, +) -> None: + # given + wall_clock = [ + datetime.fromtimestamp(1726570875 + offset) for offset in (0, 0.5, 1.0, 2.5) + ] + datetime_mock.now.side_effect = wall_clock + + # when + result = _run_frames([None] * len(wall_clock), cooldown_seconds=1.0) + + # then + assert result == [True, False, True, True] + + +@mock.patch.object(rate_limiter_v1, "datetime") +def test_rate_limiter_uses_video_time_for_video_file_frames( + datetime_mock: MagicMock, +) -> None: + # given + datetime_mock.now.return_value = datetime.fromtimestamp(1726570875) + images = [_video_frame(frame_number) for frame_number in (1, 3, 5)] + + # when + result = _run_frames(images, cooldown_seconds=3.0) + + # then + assert result == [ + True, + False, + True, + ], "Expected frames 1 and 5 at 1 FPS to be 4 seconds of video time apart" + + +@mock.patch.object(rate_limiter_v1, "datetime") +def test_rate_limiter_lets_video_file_frames_through_once_cooldown_elapses( + datetime_mock: MagicMock, +) -> None: + # given + datetime_mock.now.return_value = datetime.fromtimestamp(1726570875) + images = [_video_frame(frame_number) for frame_number in (1, 3, 5)] + + # when + result = _run_frames(images, cooldown_seconds=2.0) + + # then + assert result == [True, True, True] + + +@pytest.mark.parametrize( + "comes_from_video_file, fps", + [(False, 30), (None, 30), (True, 0), (True, None)], + ids=["stream", "unknown_source", "video_file_with_zero_fps", "video_file_no_fps"], +) +@mock.patch.object(rate_limiter_v1, "datetime") +def test_rate_limiter_uses_frame_timestamp_when_video_time_is_not_available( + datetime_mock: MagicMock, + comes_from_video_file: Optional[bool], + fps: Optional[float], +) -> None: + # given + datetime_mock.now.return_value = datetime.fromtimestamp(1726570875) + images = [ + _video_frame( + frame_number, + fps=fps, + comes_from_video_file=comes_from_video_file, + frame_timestamp=datetime.fromtimestamp( + 1726570875 + offset, tz=timezone.utc + ), + ) + for frame_number, offset in ((1, 0), (2, 0.5), (3, 1.0), (4, 2.5)) + ] + + # when + result = _run_frames(images, cooldown_seconds=1.0) + + # then + assert result == [True, False, True, True] diff --git a/workflows/tests/unit_tests/execution_engine/entities/test_base.py b/workflows/tests/unit_tests/execution_engine/entities/test_base.py index 2c00f8e966..f4a5f82b1d 100644 --- a/workflows/tests/unit_tests/execution_engine/entities/test_base.py +++ b/workflows/tests/unit_tests/execution_engine/entities/test_base.py @@ -717,6 +717,34 @@ def test_workflow_image_video_metadata_preservation() -> None: assert result is metadata +def test_workflow_image_has_video_metadata_only_when_metadata_is_provided() -> None: + # given + metadata = VideoMetadata( + video_identifier="parent", + frame_number=7, + frame_timestamp=datetime.now(), + fps=40, + ) + image_with_metadata = WorkflowImageData( + parent_metadata=ImageParentMetadata(parent_id="parent"), + numpy_image=np.zeros((192, 168, 3), dtype=np.uint8), + video_metadata=metadata, + ) + image_without_metadata = WorkflowImageData( + parent_metadata=ImageParentMetadata(parent_id="parent"), + numpy_image=np.zeros((192, 168, 3), dtype=np.uint8), + ) + + # when + _ = image_without_metadata.video_metadata + + # then + assert image_with_metadata.has_video_metadata() is True + assert ( + image_without_metadata.has_video_metadata() is False + ), "Expected placeholder video metadata not to count as real metadata" + + def test_workflow_image_replace_image_operation() -> None: # given metadata = VideoMetadata(