diff --git a/src/pyrecest/experimental/dvs/normal_flow.py b/src/pyrecest/experimental/dvs/normal_flow.py index 0441bf0a5..ad0822cbd 100644 --- a/src/pyrecest/experimental/dvs/normal_flow.py +++ b/src/pyrecest/experimental/dvs/normal_flow.py @@ -2,6 +2,8 @@ from __future__ import annotations +import math + import numpy as np from .active_contour import ( @@ -185,14 +187,23 @@ def infer_polarity_contrast_sign( if flows.shape != polarities.shape: raise ValueError("event_polarities must have one value per signed normal flow") - score = 0.0 + contributions: list[float] = [] for signed_flow, event_polarity in zip(flows, polarities, strict=True): if signed_scalar_sign(signed_flow, zero_tolerance=tolerance) == 0.0: continue - score += event_polarity_sign(event_polarity) * float(signed_flow) - if abs(score) <= tolerance: + contributions.append( + event_polarity_sign(event_polarity) * float(signed_flow) + ) + if not contributions: + return 1.0 + + scale = max(abs(contribution) for contribution in contributions) + scaled_score = math.fsum( + contribution / scale for contribution in contributions + ) + if abs(scaled_score) <= tolerance / scale: return 1.0 - return 1.0 if score > 0.0 else -1.0 + return 1.0 if scaled_score > 0.0 else -1.0 def polarity_consistency_for_signed_flow( diff --git a/tests/experimental/test_dvs_normal_flow.py b/tests/experimental/test_dvs_normal_flow.py index 613445b06..37b074d2b 100644 --- a/tests/experimental/test_dvs_normal_flow.py +++ b/tests/experimental/test_dvs_normal_flow.py @@ -73,6 +73,17 @@ def test_infer_polarity_contrast_sign_from_batch(): assert infer_polarity_contrast_sign(signed_flows, polarities, -1.0) == -1.0 +def test_infer_polarity_contrast_sign_avoids_overflow(): + largest = np.finfo(float).max + signed_flows = largest * np.array([0.75, 0.75, -0.75, -0.9]) + polarities = np.ones(signed_flows.shape) + + with np.errstate(over="raise", invalid="raise"): + inferred = infer_polarity_contrast_sign(signed_flows, polarities, "infer") + + assert inferred == -1.0 + + def test_infer_polarity_contrast_sign_rejects_invalid_batches(): with pytest.raises(ValueError, match="signed_normal_flows"): infer_polarity_contrast_sign([1.0, np.nan], [1.0, 0.0], "infer")