diff --git a/openpilot/starpilot/speed_limits/vision/model.py b/openpilot/starpilot/speed_limits/vision/model.py index 43955ee3a6..7e60fba347 100644 --- a/openpilot/starpilot/speed_limits/vision/model.py +++ b/openpilot/starpilot/speed_limits/vision/model.py @@ -458,6 +458,21 @@ class VisionModelCore: return detection + @staticmethod + def _is_yellow_advisory(sign_crop): + if sign_crop.size == 0: + return False + sign_crop = _primary_regulatory_panel(sign_crop) + hsv = cv2.cvtColor(sign_crop, cv2.COLOR_BGR2HSV) + hue, saturation, value = hsv[:, :, 0], hsv[:, :, 1], hsv[:, :, 2] + white_mask = ((value >= REGULATORY_WHITE_VALUE_MIN) & (saturation <= REGULATORY_WHITE_SAT_MAX)) + yellow_mask = ((hue >= REGULATORY_YELLOW_HUE_MIN) & + (hue <= REGULATORY_YELLOW_HUE_MAX) & + (saturation >= REGULATORY_YELLOW_SAT_MIN) & + (value >= REGULATORY_YELLOW_VALUE_MIN)) + return float((yellow_mask & ~white_mask).astype(np.uint8).mean()) > .5 + + def _is_regulatory_speed_sign(self, sign_crop): if sign_crop.size == 0: return False @@ -849,6 +864,10 @@ class VisionModelCore: proposal_area_ratio < DETECTOR_CLASSIFIER_TINY_LOW_CONF_AREA_RATIO and proposal_confidence < DETECTOR_CLASSIFIER_TINY_LOW_CONF_MIN_CONFIDENCE ) + if class_id == 0 and self._is_yellow_advisory( + frame_bgr[max(int(y1), 0):min(int(y2), frame_height), max(int(x1), 0):min(int(x2), frame_width)]): + rejected_advisory_boxes.append((x1, y1, x2, y2)) + continue self._remember_detector_proposal(proposal_confidence, class_id, (x1, y1, x2, y2)) if class_id == 2: diff --git a/openpilot/starpilot/speed_limits/vision/tests/test_explicit_yellow_advisory.py b/openpilot/starpilot/speed_limits/vision/tests/test_explicit_yellow_advisory.py new file mode 100644 index 0000000000..b6fc8fff29 --- /dev/null +++ b/openpilot/starpilot/speed_limits/vision/tests/test_explicit_yellow_advisory.py @@ -0,0 +1,84 @@ +"""Pure pixel/fixture outputs; no model loading or DNN inference.""" +import unittest +from unittest.mock import patch + +import numpy as np + +from openpilot.starpilot.speed_limits.vision import model +from openpilot.starpilot.speed_limits.vision.tests import test_terminal_material_false as fixture + + +@unittest.skipIf(model.cv2 is None, 'OpenCV is optional') +class TestExplicitYellowAdvisory(unittest.TestCase): + def case(self, color, *, class_id=0, proposals=None): + if proposals is None: + proposals = [(.7, class_id, (1400, 100, 1490, 237))] + core, face, proposals, frame = fixture.TestTerminalMaterialFalse.setup_case(self, proposals=proposals) + frame[100:237, 1400:1490] = color + return core, face, proposals, frame + + def run_case(self, case): + return fixture.TestTerminalMaterialFalse.run_case(self, case) + + def test_clear_yellow_is_rejected_before_remember_or_reader(self): + case = self.case((0, 255, 255)) + with patch.object(case[0], '_remember_detector_proposal', side_effect=AssertionError('yellow must not be remembered')): + self.assertIsNone(self.run_case(case)[0]) + self.assertEqual(case[0].classifier_net.calls, 0) + self.assertFalse(case[0].last_classifier_advisory_rejected) + self.assertFalse(case[0].last_classifier_reject_dominant) + + def test_no_yellow_white_gray_and_yellow_tint_keep_rescue(self): + for color in ((255, 255, 255), (90, 90, 90), (170, 210, 210)): + with self.subTest(color=color): + case = self.case(color) + self.assertFalse(case[0]._is_yellow_advisory(case[3][100:237, 1400:1490])) + result, _ = self.run_case(case) + self.assertIsNotNone(result) + self.assertEqual(result.speed_limit_mph, 40) + self.assertGreater(case[0].classifier_net.calls, 0) + + def test_existing_primary_panel_keeps_separate_white_maximum(self): + compound = np.zeros((220, 100, 3), np.uint8) + compound[:120] = 255 + compound[128:] = (0, 255, 255) + panel = model._primary_regulatory_panel(compound) + self.assertLess(panel.shape[0], compound.shape[0]) + self.assertFalse(model.VisionModelCore._is_yellow_advisory(compound)) + self.assertTrue(model.VisionModelCore._is_yellow_advisory(compound[128:])) + + def test_class2_school_path_does_not_use_class0_veto(self): + case = self.case((0, 255, 255), class_id=2) + with patch.object(case[0], '_is_yellow_advisory', side_effect=AssertionError('school behavior unchanged')): + self.run_case(case) + self.assertGreater(case[0].classifier_net.calls, 0) + + def test_near_duplicate_yellow_proposal_uses_existing_overlap_suppression(self): + proposals = [(.7, 0, (1400, 100, 1490, 237)), (.69, 0, (1401, 101, 1491, 238))] + case = self.case((0, 255, 255), proposals=proposals) + original = case[0]._is_yellow_advisory + with patch.object(case[0], '_is_yellow_advisory', wraps=original) as predicate: + self.assertIsNone(self.run_case(case)[0]) + self.assertEqual(predicate.call_count, 1) + self.assertEqual(case[0].classifier_net.calls, 0) + + def test_strict_majority_disjoint_yellow_boundary(self): + self.assertFalse(model.VisionModelCore._is_yellow_advisory(np.empty((0, 0, 3), np.uint8))) + for yellow_rows, expected in ((12, False), (30, False), (50, False), (51, True)): + with self.subTest(yellow_rows=yellow_rows): + crop = np.full((100, 100, 3), 255, np.uint8) + crop[:yellow_rows] = (0, 255, 255) + self.assertEqual(model.VisionModelCore._is_yellow_advisory(crop), expected) + + def test_minority_yellow_keeps_reader_rescue(self): + case = self.case((255, 255, 255)) + case[3][100:140, 1400:1490] = (0, 255, 255) + self.assertFalse(case[0]._is_yellow_advisory(case[3][100:237, 1400:1490])) + result, _ = self.run_case(case) + self.assertIsNotNone(result) + self.assertEqual(result.speed_limit_mph, 40) + self.assertGreater(case[0].classifier_net.calls, 0) + + +if __name__ == '__main__': + unittest.main() diff --git a/prebuilt.json b/prebuilt.json index bd88846fbf..12fcebd952 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "fb09ca16dad2d54909ef424bfeff46750e7879cc1d8d1b01ddcbe36b5760f601", "schema_version": 1} +{"protected_inventory_sha256": "3edbe6b51ec2bf04a8a9ab30ea433bde6b4ff7cb12ca5462ecef1074e8b9e91f", "schema_version": 1}