feat(console): auto-select cuda for detection workflows

shuishen

6 days ago deac984180e54dcb904f415c8f2e095b8b1661a7

 tests/test_serve_workbench_console.py

@@ -239,6 +239,80 @@
                with self.assertRaisesRegex(MODULE.ApiError, "CUDA was requested"):
                    MODULE.pointcloud_execution_environment(root, "cuda")

    def test_object_detection_execution_uses_fixed_gpu_environment_when_probe_succeeds(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            gpu_python = root / ".venvs" / MODULE.OBJECT_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
            gpu_python.parent.mkdir(parents=True)
            gpu_python.write_bytes(b"fixed-interpreter")
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": true, "torch": "2.11.0+cu128"}\n', "")):
                execution = MODULE.object_detection_execution_environment(root)
            self.assertEqual(execution["device"], "cuda")
            self.assertEqual(execution["environment"], MODULE.OBJECT_DETECTION_GPU_ENVIRONMENT)

    def test_run_discovery_uses_actual_device_in_case_note(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            detection = root / "shared" / "outputs" / "01-object-detection" / "gpu-case"
            raw = root / "shared" / "data" / "raw" / "01-object-detection" / "gpu-case"
            detection.mkdir(parents=True)
            raw.mkdir(parents=True)
            (detection / "detections.json").write_text("{}", encoding="utf-8")
            (detection / "run_metadata.json").write_text(json.dumps({"input_dir": str(raw), "created_at": "2026-08-25", "device": "cuda:0"}), encoding="utf-8")
            self.assertIn("GPU", MODULE.detection_runs(root)[0]["note"])

            change = root / "shared" / "outputs" / "00-change-detection" / "gpu-case"
            change_raw = root / "shared" / "data" / "raw" / "00-change-detection" / "gpu-case"
            before = change_raw / "before.jpg"
            after = change_raw / "after.jpg"
            change.mkdir(parents=True)
            change_raw.mkdir(parents=True)
            before.write_bytes(b"before")
            after.write_bytes(b"after")
            (change / "overlay.jpg").write_bytes(b"overlay")
            (change / "changes.geojson").write_text("{}", encoding="utf-8")
            (change / "run_metadata.json").write_text(json.dumps({"capability": "00-change-detection", "schema_version": 1, "created_at": "2026-08-25", "device": "cuda", "input_files": ["before.jpg", "after.jpg"], "raw_input_dir": change_raw.relative_to(root).as_posix(), "raw_before": before.relative_to(root).as_posix(), "raw_after": after.relative_to(root).as_posix(), "artifacts": {"overlay": "overlay.jpg", "vector": "changes.geojson"}}), encoding="utf-8")
            self.assertIn("GPU", MODULE.change_runs(root)[0]["note"])

    def test_object_detection_execution_falls_back_to_fixed_cpu_environment(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            cpu_python = root / ".venvs" / MODULE.OBJECT_DETECTION_CPU_ENVIRONMENT / "Scripts" / "python.exe"
            gpu_python = root / ".venvs" / MODULE.OBJECT_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
            cpu_python.parent.mkdir(parents=True)
            gpu_python.parent.mkdir(parents=True)
            cpu_python.write_bytes(b"fixed-cpu-interpreter")
            gpu_python.write_bytes(b"fixed-gpu-interpreter")
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": false, "torch": "2.11.0+cu128"}\n', "")):
                execution = MODULE.object_detection_execution_environment(root)
            self.assertEqual(execution["device"], "cpu")
            self.assertEqual(execution["environment"], MODULE.OBJECT_DETECTION_CPU_ENVIRONMENT)

    def test_change_detection_execution_uses_fixed_gpu_environment_when_probe_succeeds(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            gpu_python = root / ".venvs" / MODULE.CHANGE_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
            gpu_python.parent.mkdir(parents=True)
            gpu_python.write_bytes(b"fixed-interpreter")
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": true, "torch": "2.11.0+cu128"}\n', "")):
                execution = MODULE.change_detection_execution_environment(root)
            self.assertEqual(execution["device"], "cuda")
            self.assertEqual(execution["environment"], MODULE.CHANGE_DETECTION_GPU_ENVIRONMENT)

    def test_change_detection_execution_falls_back_to_fixed_cpu_environment(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            cpu_python = root / ".venvs" / MODULE.CHANGE_DETECTION_CPU_ENVIRONMENT / "Scripts" / "python.exe"
            gpu_python = root / ".venvs" / MODULE.CHANGE_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
            cpu_python.parent.mkdir(parents=True)
            gpu_python.parent.mkdir(parents=True)
            cpu_python.write_bytes(b"fixed-cpu-interpreter")
            gpu_python.write_bytes(b"fixed-gpu-interpreter")
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": false, "torch": "2.11.0+cu128"}\n', "")):
                execution = MODULE.change_detection_execution_environment(root)
            self.assertEqual(execution["device"], "cpu")
            self.assertEqual(execution["environment"], MODULE.CHANGE_DETECTION_CPU_ENVIRONMENT)

    def test_semantic_validation_run_is_discovered(self) -> None:
        runs = MODULE.semantic_runs(ROOT)
        self.assertTrue(any(item["id"] == "validation-20260817" for item in runs))