Harden YOLO tile inference inputs
GeoIntel CI / docs-smoke (push) Has been cancelled
GeoIntel CI / contract-smoke (push) Has been cancelled

This commit is contained in:
Codex
2026-07-06 20:25:09 +02:00
parent d7f786729a
commit add4768a52
7 changed files with 137 additions and 12 deletions
+5
View File
@@ -275,6 +275,11 @@ python scripts/configure_yolo_model.py \
The smoke loads only the supplied local model file, does not run inference and
does not download weights.
Configured YOLO inference uses raster tile artifacts from the existing tile
manifest flow. Single-band or otherwise non-RGB tile images are converted to a
temporary RGB prediction image before inference; georeferencing still comes
from the persisted tile manifest transform/bounds metadata.
Optional tuning:
```bash
+53 -7
View File
@@ -1,7 +1,10 @@
from __future__ import annotations
from contextlib import contextmanager
from pathlib import Path
import tempfile
from typing import Any
from collections.abc import Iterator
from app.core.config import Settings
from app.core.errors import AppError
@@ -62,13 +65,24 @@ class YoloDetectionAdapter:
details={"tile_path": str(tile_path)},
status_code=422,
)
results = model.predict(
source=str(tile_path),
conf=float(confidence_threshold),
imgsz=int(self.settings.yolo_image_size),
device=self.settings.yolo_device,
verbose=False,
)
try:
with _prediction_source(tile_path) as prediction_source:
results = model.predict(
source=prediction_source,
conf=float(confidence_threshold),
imgsz=int(self.settings.yolo_image_size),
device=self.settings.yolo_device,
verbose=False,
)
except AppError:
raise
except Exception as exc:
raise AppError(
code="DETECTION_INFERENCE_FAILED",
message="Configured YOLO inference failed for a raster tile",
details={"tile_path": str(tile_path), "error": str(exc)},
status_code=503,
) from exc
detections: list[dict[str, Any]] = []
for result in results:
@@ -102,3 +116,35 @@ def _to_list(value: Any) -> list[Any]:
if hasattr(value, "tolist"):
return value.tolist()
return list(value)
@contextmanager
def _prediction_source(tile_path: Path) -> Iterator[str]:
temp_path: Path | None = None
try:
try:
from PIL import Image
except Exception:
yield str(tile_path)
return
try:
with Image.open(tile_path) as image:
if image.mode == "RGB" and len(image.getbands()) == 3:
yield str(tile_path)
return
rgb_image = image.convert("RGB")
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as handle:
temp_path = Path(handle.name)
rgb_image.save(temp_path)
yield str(temp_path)
return
except Exception:
if temp_path is not None:
raise
yield str(tile_path)
return
finally:
if temp_path is not None:
temp_path.unlink(missing_ok=True)
@@ -7,10 +7,12 @@ from uuid import uuid4
import pytest
from app.core.config import Settings
from app.core.errors import AppError
from app.models import AnalysisRun, Dataset, Detection, Job, Project
from app.services.detection_georeferencing import pixel_bbox_to_epsg4326_polygon
from app.services.detection_service import DetectionService
from app.services.model_registry_service import ModelRegistryService
from app.services.yolo_adapter import YoloDetectionAdapter
ROOT = Path(__file__).resolve().parents[2]
@@ -75,6 +77,33 @@ class MockYoloAdapter:
]
class RecordingPredictModel:
def __init__(self) -> None:
self.seen_sources: list[dict] = []
def predict(self, *, source, conf, imgsz, device, verbose):
from PIL import Image
with Image.open(source) as image:
self.seen_sources.append(
{
"path": str(source),
"mode": image.mode,
"bands": len(image.getbands()),
"conf": conf,
"imgsz": imgsz,
"device": device,
"verbose": verbose,
}
)
return []
class ExplodingPredictModel:
def predict(self, *, source, conf, imgsz, device, verbose):
raise RuntimeError("expected input[1, 1, 480, 640] to have 3 channels, but got 1 channels instead")
def _project_and_dataset(dataset_type: str = "raster"):
project_id = uuid4()
dataset_id = uuid4()
@@ -312,3 +341,35 @@ def test_yolo_run_persists_mocked_georeferenced_detections(tmp_path: Path) -> No
assert detections[0].properties_json == {"adapter": "mock", "tile_index": 0}
assert runs[0].status == "success"
assert jobs[0].status == "success"
def test_yolo_adapter_converts_single_band_tiles_to_rgb_before_prediction(tmp_path: Path) -> None:
Image = pytest.importorskip("PIL.Image")
tile_path = tmp_path / "single_band_tile.tif"
Image.new("L", (16, 16), 128).save(tile_path)
model = RecordingPredictModel()
settings = _settings(tmp_path, yolo_image_size=64, yolo_device="cpu")
detections = YoloDetectionAdapter(settings).predict_tile(model, tile_path, confidence_threshold=0.25)
assert detections == []
assert model.seen_sources[0]["mode"] == "RGB"
assert model.seen_sources[0]["bands"] == 3
assert model.seen_sources[0]["path"] != str(tile_path)
assert model.seen_sources[0]["conf"] == 0.25
assert model.seen_sources[0]["imgsz"] == 64
assert model.seen_sources[0]["device"] == "cpu"
assert model.seen_sources[0]["verbose"] is False
def test_yolo_adapter_wraps_prediction_runtime_errors(tmp_path: Path) -> None:
tile_path = tmp_path / "tile.tif"
tile_path.write_bytes(b"not an image but present")
settings = _settings(tmp_path)
with pytest.raises(AppError) as exc_info:
YoloDetectionAdapter(settings).predict_tile(ExplodingPredictModel(), tile_path, confidence_threshold=0.25)
assert exc_info.value.code == "DETECTION_INFERENCE_FAILED"
assert "Configured YOLO inference failed for a raster tile" in exc_info.value.message
assert exc_info.value.details["tile_path"] == str(tile_path)