Add TwoStageSync node #194

HonzaCuhel · 2025-03-24T14:49:20Z

Purpose

Adding TwoStageSync node

Specification

Adding TwoStageSync and unittests

Dependencies & Potential Impact

None / not applicable

Deployment Plan

None / not applicable

Testing & Validation

None / not applicable

depthai_nodes/message/detected_recognitions.py

depthai_nodes/node/detections_recognitions_sync.py

codecov-commenter · 2025-03-27T12:38:45Z

Codecov Report

Attention: Patch coverage is 85.16129% with 23 lines in your changes missing coverage. Please review.

Project coverage is 51.44%. Comparing base (e8aeb49) to head (9e7a4e5).
Report is 1 commits behind head on main.

✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
depthai_nodes/node/two_stage_sync.py	85.12%	18 Missing ⚠️
depthai_nodes/message/detected_recognitions.py	84.37%	5 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #194      +/-   ##
==========================================
+ Coverage   50.29%   51.44%   +1.15%     
==========================================
  Files          79       81       +2     
  Lines        4575     4731     +156     
==========================================
+ Hits         2301     2434     +133     
- Misses       2274     2297      +23

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

kkeroo · 2025-03-27T15:02:45Z

Can we come up with a better name? Because the node can sync any other message not only recognitions?

dominik737 · 2025-03-27T15:27:42Z

This node actually can sync only ImgDetections and ImgDetectionsExtended. It relies on the len(ImgDetections.detections) to determine for how many (recognition - NNData) messages to wait.

There is another GatheredData node that can sync any messages, the drawback is that it's not as plug&play and you have to have some type of a converter to get the number of messages you need to wait for.

We've agreed that this node will in the future inherit from the GatheredData since the logic is the same. But for now it is not necessary, because the API will stay the same when we decide for the refactor.

kkeroo · 2025-03-27T15:32:32Z

I meant that this node can also sync detections with Keypoints msg. for example, right?

dominik737 · 2025-03-27T15:47:08Z

Actually, you are right. You can supply other objects beside the NNData. We just assume and typehint for that.

If we want to make it more generic we should change the typehints and then the name should follow.

jkbmrz · 2025-04-07T07:16:49Z

depthai_nodes/message/detected_recognitions.py

+from .img_detections import ImgDetectionsExtended
+
+
+class DetectedRecognitions(dai.Buffer):


Should we make this message more general, i.e. to store arbitrary message types?

We could yea. Preferably using generics.

jkbmrz · 2025-04-07T07:18:53Z

depthai_nodes/node/two_stage_sync.py

+class TwoStageSync(dai.node.ThreadedHostNode):
+    FPS_TOLERANCE_DIVISOR = 2.0
+    INPUT_CHECKS_PER_FPS = 100
+    """A class for synchronizing detections and recognitions.


"synchronizing detections and recognitions" - I thought we want to make this node more general. If that's the case, the docstring should adapt to the change in the node name.

And same for parameter names below - they should adapt to the changes that are not aiming to only sync recognitions and detections.

jkbmrz · 2025-04-07T07:22:03Z

depthai_nodes/node/two_stage_sync.py

+        self.out = self.createOutput()
+
+    def build(self, camera_fps: int) -> "TwoStageSync":
+        if camera_fps <= 0:


Maybe rather move this check to the self.set_camera_fps() setter.

jkbmrz · 2025-04-07T07:23:39Z

depthai_nodes/node/two_stage_sync.py

+        self._camera_fps = camera_fps
+        return self
+
+    def set_camera_fps(self, fps: int) -> None:


Rename to setCameraFPS() to better match the naming in other nodes.

jkbmrz · 2025-04-07T07:36:43Z

depthai_nodes/node/two_stage_sync.py

+        Output for detected recognitions.
+    """
+
+    def __init__(self, *args, **kwargs) -> None:


Should we rather define the expected params? Might be more convenient for the user to know exactly what is expected.

There are no params the node itself needs. You can delete the (k)args. I think I just passed it down to dai.ThreadedHostNode in case in future there is passed something important, but doesn't seem like it.

jkbmrz · 2025-04-07T07:38:05Z

depthai_nodes/node/two_stage_sync.py

+        kwargs: Keyword arguments to be passed to the ThreadedHostNode class.
+        """
+        super().__init__(*args, **kwargs)
+        self._camera_fps = 30


Why do we set a default _camera_fps value if we enforce setting it inside the build() method?

jkbmrz · 2025-04-07T07:44:06Z

depthai_nodes/node/two_stage_sync.py

+        for unmatched_recognition in unmatched_recognitions_to_remove:
+            self._unmatched_recognitions.remove(unmatched_recognition)
+
+    def _get_total_seconds_ts(self, buffer_like: dai.Buffer) -> float:


Do we really need this method? It's only abstracts the .getTimestamp().total_seconds() call, which is quite straight-forward to me.

Also, why the .total_seconds() is needed. Cannot we simply compare the timestamps?

The getTimestamp gives you datetime.timedelta object. With the object you cannot really easily do arithmetics like substracting plain number. For that reason the .total_seconds() is used instead.

When you reference something that is repeating (e.g. multiple method calls) you should create a method, otherwise you violate DRY principle. The _get_total_seconds_ts is referenced several times in the code. If the way to obtain total seconds would change somehow the replacement would have to be made in all cases. Also you would have to write these two methods every time you want to get the timestamp total seconds.

jkbmrz · 2025-04-07T07:53:56Z

depthai_nodes/node/two_stage_sync.py

+        The camera FPS.
+    _unmatched_recognitions: List[dai.Buffer]
+        List of unmatched recognitions.
+    _recognitions_by_detection_ts: Dict[float, List[dai.Buffer]]


Just a suggestion, might it be more clear to have:

_unmatched_detections ([float, List[dai.Buffer]])

_unmatched_recognitions ([float, List[dai.Buffer]])

_matched_messages ([float, Tuple[dai.Buffer, dai.Buffer]]

This way, the parameter names are more straight-forward, and the _update_ready_timestamps parameter is not needed anymore.

jkbmrz · 2025-04-07T08:03:07Z

depthai_nodes/node/two_stage_sync.py

+
+    def _timestamps_in_tolerance(self, timestamp1: float, timestamp2: float) -> bool:
+        difference = abs(timestamp1 - timestamp2)
+        return difference < (1 / self._camera_fps / self.FPS_TOLERANCE_DIVISOR)


Should we allow the option to modify the self.FPS_TOLERANCE_DIVISOR param? Because, if understanding correctly, the delay depends on the "recognition" model which might be lower/higher depending on the specific architecture used.

The FPS_TOLERANCE_DIVISOR is basically saying how much deviation there can be in the timestamps to consider the detection/recognition belong together.

I've set it such because that is exactly splitting the interval into 2, meaning that consecutive detections do not have any overlapping timestamps. If you would set it to 3 you would have some timestamp space where the recognitions would not be matched to any detection. If set to 1.5 there would be overlap between two consecutive detections.

Hence, I view this as a constant. I doubt there would be a good reason to change it.

jkbmrz · 2025-04-07T08:05:46Z

depthai_nodes/node/two_stage_sync.py

+
+    def _update_ready_timestamps(self, timestamp: float) -> None:
+        if not self._timestamp_ready(timestamp):
+            return


Why is this return needed?

It's a guard clause. It can be used to increase readability, although works well in my experience only if the function is short.

In this case if there is not ready timestamp then there is nothing to update so the return makes the executor jump out of the function.

Wouldn't it be much simpler (and also had same efficiency) if written as:

def _update_ready_timestamps(self, timestamp: float) -> None: if self._timestamp_ready(timestamp): self._ready_timestamps.put(timestamp)

Efficiency is the same and negligible in this case, let's put that aside.

Simplicity is quite subjective. The one who is used to guard clauses, looks at the method and sees there are clauses that prevent further execution of the method. The method logic is less horizontally nested with the clause, which increases the readability.

If in the future there are more conditions why the ready timestamps should not be updated, it can be solved by simply adding another guard clause. Another guard clause will not make the condition more complicated and won't claim more horizontal space, this makes it quite extensible.

aljazkonec1

Added some small comments about the structure that dont affect the logic.

The logic is sound, so LGTM

aljazkonec1 · 2025-04-07T08:32:33Z

depthai_nodes/message/detected_recognitions.py

+        """Initializes the DetectedRecognitions object."""
+        super().__init__()
+        self._img_detections = None
+        self._recognitions_data = []


Add type hints

aljazkonec1 · 2025-04-07T09:00:02Z

depthai_nodes/node/two_stage_sync.py

+
+        self._ready_timestamps.put(timestamp)
+
+    def _timestamp_ready(self, timestamp: float) -> bool:


Function is called once in line 143. The code can just be put in that function instead. Makes the code more readable

The use-case for functions is not to only make parts of code reusable. It's also great tool to create hierarchy and abstractions, as demonstrated here.

When you read the _update_ready_timestamps you can quickly see that the function is checking whether there are any timestamps ready and then it updates ready timestamps. The implementation logic of retrieving ready timestamps is just a detail of lower hierarchy - it has nothing to do with updating ready timestamps.

aljazkonec1 · 2025-04-07T09:41:27Z

depthai_nodes/node/two_stage_sync.py

+        self._clear_unmatched_recognitions(current_timestamp)
+        self._clear_old_detections(current_timestamp)
+
+    def _clear_unmatched_recognitions(self, current_timestamp) -> None:


Called once 3 lines above. I would move it there to make the function _clear_old_data better readable

See the comment above.

In the abstraction hierarchy level of clearing old data there are two tasks:

clearing unmatched recognitions

clearing old detections

The implementation details of either are lower abstraction hierarchy level.

dominik737 · 2025-04-07T15:54:43Z

depthai_nodes/message/detected_recognitions.py

+from .img_detections import ImgDetectionsExtended
+
+
+class DetectedRecognitions(dai.Buffer):


We could yea. Preferably using generics.

dominik737 · 2025-04-07T16:03:47Z

depthai_nodes/message/detected_recognitions.py

+        self._recognitions_data = []
+
+    @property
+    def img_detections(self) -> Union[dai.ImgDetections, ImgDetectionsExtended]:


We might consider using typing.Protocol rather than Union. We basically only need the detections: list property. This way we won't have to add any other types to the Union in future. Also if we need to change the property name we depend upon it will be in once place (in the Protocol).

dominik737 · 2025-04-07T16:05:53Z

depthai_nodes/node/two_stage_sync.py

+    _camera_fps: int
+        The camera FPS.
+    _unmatched_recognitions: List[dai.Buffer]
+        List of unmatched recognitions.
+    _recognitions_by_detection_ts: Dict[float, List[dai.Buffer]]
+        Dictionary of recognitions by detection timestamp.
+    _detections: Dict[float, Union[dai.ImgDetections, dai.SpatialImgDetections, ImgDetectionsExtended]]
+        Dictionary of detections.
+    _ready_timestamps: PriorityQueue
+        Priority queue of ready timestamps.


I think private attributes should not be part of the docstrings.

dominik737 · 2025-04-07T16:07:34Z

depthai_nodes/node/two_stage_sync.py

+        Output for detected recognitions.
+    """
+
+    def __init__(self, *args, **kwargs) -> None:


There are no params the node itself needs. You can delete the (k)args. I think I just passed it down to dai.ThreadedHostNode in case in future there is passed something important, but doesn't seem like it.

dominik737 · 2025-04-07T16:22:27Z

depthai_nodes/node/two_stage_sync.py

+
+    def _timestamps_in_tolerance(self, timestamp1: float, timestamp2: float) -> bool:
+        difference = abs(timestamp1 - timestamp2)
+        return difference < (1 / self._camera_fps / self.FPS_TOLERANCE_DIVISOR)


The FPS_TOLERANCE_DIVISOR is basically saying how much deviation there can be in the timestamps to consider the detection/recognition belong together.

I've set it such because that is exactly splitting the interval into 2, meaning that consecutive detections do not have any overlapping timestamps. If you would set it to 3 you would have some timestamp space where the recognitions would not be matched to any detection. If set to 1.5 there would be overlap between two consecutive detections.

Hence, I view this as a constant. I doubt there would be a good reason to change it.

dominik737 · 2025-04-07T16:26:31Z

depthai_nodes/node/two_stage_sync.py

+
+    def _update_ready_timestamps(self, timestamp: float) -> None:
+        if not self._timestamp_ready(timestamp):
+            return


It's a guard clause. It can be used to increase readability, although works well in my experience only if the function is short.

In this case if there is not ready timestamp then there is nothing to update so the return makes the executor jump out of the function.

dominik737 · 2025-04-07T16:35:26Z

depthai_nodes/node/two_stage_sync.py

+
+        self._ready_timestamps.put(timestamp)
+
+    def _timestamp_ready(self, timestamp: float) -> bool:


The use-case for functions is not to only make parts of code reusable. It's also great tool to create hierarchy and abstractions, as demonstrated here.

When you read the _update_ready_timestamps you can quickly see that the function is checking whether there are any timestamps ready and then it updates ready timestamps. The implementation logic of retrieving ready timestamps is just a detail of lower hierarchy - it has nothing to do with updating ready timestamps.

dominik737 · 2025-04-07T16:40:00Z

depthai_nodes/node/two_stage_sync.py

+        self._clear_unmatched_recognitions(current_timestamp)
+        self._clear_old_detections(current_timestamp)
+
+    def _clear_unmatched_recognitions(self, current_timestamp) -> None:


See the comment above.

In the abstraction hierarchy level of clearing old data there are two tasks:

clearing unmatched recognitions

clearing old detections

The implementation details of either are lower abstraction hierarchy level.

dominik737 · 2025-04-07T17:13:31Z

depthai_nodes/node/two_stage_sync.py

+        for unmatched_recognition in unmatched_recognitions_to_remove:
+            self._unmatched_recognitions.remove(unmatched_recognition)
+
+    def _get_total_seconds_ts(self, buffer_like: dai.Buffer) -> float:


The getTimestamp gives you datetime.timedelta object. With the object you cannot really easily do arithmetics like substracting plain number. For that reason the .total_seconds() is used instead.

When you reference something that is repeating (e.g. multiple method calls) you should create a method, otherwise you violate DRY principle. The _get_total_seconds_ts is referenced several times in the code. If the way to obtain total seconds would change somehow the replacement would have to be made in all cases. Also you would have to write these two methods every time you want to get the timestamp total seconds.

dominik737 · 2025-04-11T18:55:03Z

Closing the PR with successor #203.

Add DetectionsRecognitionsSync

d1720a2

HonzaCuhel self-assigned this Mar 24, 2025

HonzaCuhel requested review from jkbmrz, kkeroo, klemen1999 and tersekmatija as code owners March 24, 2025 14:49

HonzaCuhel marked this pull request as draft March 24, 2025 14:49

github-actions bot added the enhancement New feature or request label Mar 24, 2025

kkeroo requested changes Mar 24, 2025

View reviewed changes

depthai_nodes/message/detected_recognitions.py Outdated Show resolved Hide resolved

Update typings to support py3.8

e015fba

jkbmrz reviewed Mar 25, 2025

View reviewed changes

HonzaCuhel and others added 4 commits March 26, 2025 10:37

Update DetectionsRecognitionsSync & tests

fce76e0

Fix input, output

5ad4b2e

Fix tests.

e9c06d4

Pre commit fix.

608c402

klemen1999 requested a review from dominik737 March 27, 2025 13:46

HonzaCuhel and others added 7 commits April 3, 2025 20:55

Update DetectionsRecognitions

e8ba945

Merge branch 'main' into feat/add-detections-recognitions-sync

ca9012b

Rename property nn_data -> recognitions_data

5fe4342

Format code

65a7a41

Import update.

2de2923

_node appendix.

d4db905

Rename DetectionsRecognitionsSync -> TwoStageSync & refactor

9e7a4e5

HonzaCuhel marked this pull request as ready for review April 5, 2025 10:55

HonzaCuhel changed the title ~~[Draft] Add DetectionsRecognitionsSync~~ Add TwoStageSync node Apr 5, 2025

HonzaCuhel requested review from jkbmrz and kkeroo April 5, 2025 10:59

jkbmrz suggested changes Apr 7, 2025

View reviewed changes

HonzaCuhel requested a review from aljazkonec1 April 7, 2025 09:56

aljazkonec1 reviewed Apr 7, 2025

View reviewed changes

dominik737 mentioned this pull request Apr 7, 2025

Gen3 - AgeGender & HumanPose: Adding TwoStageSync Node luxonis/oak-examples#641

Closed

dominik737 reviewed Apr 7, 2025

View reviewed changes

dominik737 closed this Apr 11, 2025

klemen1999 deleted the feat/add-detections-recognitions-sync branch June 5, 2025 10:10

		from .img_detections import ImgDetectionsExtended


		class DetectedRecognitions(dai.Buffer):


		self._ready_timestamps.put(timestamp)

		def _timestamp_ready(self, timestamp: float) -> bool:

Add TwoStageSync node #194

Add TwoStageSync node #194

Uh oh!

Conversation

HonzaCuhel commented Mar 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Specification

Dependencies & Potential Impact

Deployment Plan

Testing & Validation

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

codecov-commenter commented Mar 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

kkeroo commented Mar 27, 2025

Uh oh!

dominik737 commented Mar 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

kkeroo commented Mar 27, 2025

Uh oh!

dominik737 commented Mar 27, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aljazkonec1 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HonzaCuhel commented Mar 24, 2025 •

edited

Loading

codecov-commenter commented Mar 27, 2025 •

edited

Loading

dominik737 commented Mar 27, 2025 •

edited

Loading