diff --git a/bridge/test_vision_service.py b/bridge/test_vision_service.py index 1ce7a87..36e34bc 100644 --- a/bridge/test_vision_service.py +++ b/bridge/test_vision_service.py @@ -83,7 +83,7 @@ def test_hash_pinned_yunet_model_loads_and_rejects_blank_frame(self) -> None: with self.assertRaises(RuntimeError): verify_yunet_model(bad_model) - def test_preflight_validates_local_runtime_without_fetching_or_leaking_pairing_code(self) -> None: + def _run_preflight(self, debug_payload): with tempfile.TemporaryDirectory() as directory: pairing = Path(directory) / "pairing.txt" pairing.write_text("123456\n", encoding="ascii") @@ -95,15 +95,52 @@ def test_preflight_validates_local_runtime_without_fetching_or_leaking_pairing_c str(pairing), "--preflight", ] - with patch.object(sys, "argv", argv), patch("builtins.print") as emit: + with patch.object(sys, "argv", argv), patch("builtins.print") as emit, patch.object( + CameraVisionService, "_get", **debug_payload + ): result = main() + return result, json.loads(emit.call_args.args[0]) + + def test_preflight_reports_ready_without_leaking_pairing_code(self) -> None: + result, payload = self._run_preflight( + {"return_value": b'{"compiled_enable_camera":1,"compiled_enable_camera_host_vision":1}'} + ) - payload = json.loads(emit.call_args.args[0]) self.assertEqual(0, result) self.assertTrue(payload["ready"]) self.assertFalse(payload["raw_frame_persistence"]) self.assertNotIn("123456", json.dumps(payload)) + def test_preflight_refuses_a_camera_less_image(self) -> None: + # F3's signature failure: the worker runs happily against firmware with the + # camera compiled out and returns zero detections forever, which is + # indistinguishable from a detector that simply sees nobody. + result, payload = self._run_preflight( + {"return_value": b'{"compiled_enable_camera":0,"compiled_enable_camera_host_vision":0}'} + ) + + self.assertEqual(2, result) + self.assertFalse(payload["ready"]) + self.assertIn("firmware-camera-disabled", payload["reason"]) + self.assertIn("compiled_enable_camera", payload["reason"]) + + def test_preflight_refuses_an_unreachable_robot(self) -> None: + result, payload = self._run_preflight( + {"side_effect": urllib.error.URLError("offline")} + ) + + self.assertEqual(2, result) + self.assertFalse(payload["ready"]) + self.assertIn("robot-unreachable", payload["reason"]) + + def test_preflight_treats_a_truncated_debug_response_as_unknown_not_disabled(self) -> None: + # /debug truncates by omitting fields rather than zeroing them, so an + # absent flag must not be read as "camera disabled". + result, payload = self._run_preflight({"return_value": b'{"bridge_state":"ready"}'}) + + self.assertEqual(0, result) + self.assertTrue(payload["ready"]) + def test_camera_service_retries_one_transport_miss_and_records_recovery(self) -> None: class Detector: @staticmethod diff --git a/bridge/vision_service.py b/bridge/vision_service.py index 6b98242..1613064 100644 --- a/bridge/vision_service.py +++ b/bridge/vision_service.py @@ -132,6 +132,16 @@ def encode_face_targets(pairing_code: str, faces: list[FaceTarget]) -> str: return f"/vision-target?p={pairing_code}&f={';'.join(encoded)}" +def _truthy_flag(value: object) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return value != 0 + if isinstance(value, str): + return value.strip().lower() in {"1", "true", "yes"} + return False + + def verify_yunet_model(path: str | Path) -> Path: model_path = Path(path).resolve() try: @@ -215,6 +225,36 @@ def _get_with_retry(self, path: str, timeout: float) -> bytes: time.sleep(0.05) raise RuntimeError("vision transport retry exhausted") + def camera_capability(self, timeout: float = 4.0) -> tuple[bool, str]: + """Report whether the installed firmware can serve host vision at all. + + F3 is that vision delivers nothing, and the failure is silent: a worker + started against a camera-less image polls forever and returns zero + detections, which looks identical to a detector that cannot see anyone. + The compiled flags distinguish those two cases immediately, and /debug is + the unauthenticated surface, so this needs no pairing code. + """ + try: + payload = self._get("/debug", timeout) + except (OSError, RuntimeError, urllib.error.URLError) as exc: + return False, f"robot-unreachable: {type(exc).__name__}" + try: + debug = json.loads(payload.decode("utf-8", "replace")) + except (UnicodeDecodeError, ValueError): + return False, "robot-debug-unparsable" + if not isinstance(debug, dict): + return False, "robot-debug-not-an-object" + missing = [ + flag + for flag in ("compiled_enable_camera", "compiled_enable_camera_host_vision") + # A truncated /debug response omits fields rather than zeroing them, so + # an absent flag is unknown, not disabled. Only an explicit 0 disables. + if flag in debug and not _truthy_flag(debug[flag]) + ] + if missing: + return False, "firmware-camera-disabled: " + ",".join(missing) + return True, "ok" + def step(self, timeout: float = 4.0) -> list[FaceTarget]: try: started = time.perf_counter() @@ -262,11 +302,13 @@ def main() -> int: ) service = CameraVisionService(args.robot_url, pairing_code, OpenCvYuNetDetector(args.model_path)) if args.preflight: + capable, reason = service.camera_capability() print( json.dumps( { "schema": "stackchan.local-vision-preflight.v1", - "ready": True, + "ready": capable, + "reason": reason, "robot_url": service.robot_url, "model_sha256": YUNET_MODEL_SHA256, "raw_frame_persistence": False, @@ -274,7 +316,7 @@ def main() -> int: separators=(",", ":"), ) ) - return 0 + return 0 if capable else 2 started = time.monotonic() exit_code = 0 try: