165 lines
7.3 KiB
Python
165 lines
7.3 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from importlib import metadata
|
|
from pathlib import Path
|
|
from typing import Any, Type
|
|
|
|
from app.core.config import Settings, get_settings
|
|
from app.core.errors import AppError
|
|
from app.services.detection_service import DetectionService
|
|
from app.services.model_asset_catalog_service import ModelAssetCatalogService
|
|
from app.services.yolo_adapter import YoloDetectionAdapter
|
|
|
|
|
|
class YoloPreflightService:
|
|
@staticmethod
|
|
def run(
|
|
*,
|
|
settings: Settings | None = None,
|
|
tile_manifest_path: str | None = None,
|
|
yolo_adapter_class: Type[YoloDetectionAdapter] = YoloDetectionAdapter,
|
|
assume_dependencies: bool = False,
|
|
check_model_load: bool = False,
|
|
model_asset_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
resolved_settings = settings or get_settings()
|
|
selected_asset = None
|
|
if model_asset_id:
|
|
selected_asset = ModelAssetCatalogService.resolve_asset(model_asset_id, settings=resolved_settings)
|
|
resolved_settings = ModelAssetCatalogService.settings_for_asset(resolved_settings, selected_asset)
|
|
result: dict[str, Any] = {
|
|
"model_id": resolved_settings.yolo_model_id,
|
|
"model_asset_id": selected_asset.model_asset_id if selected_asset else None,
|
|
"model_path": resolved_settings.yolo_model_path,
|
|
"tile_manifest_path": tile_manifest_path,
|
|
"status": "not_configured",
|
|
"message": "",
|
|
"checks": {
|
|
"enabled": resolved_settings.yolo_enabled,
|
|
"dependencies_available": None,
|
|
"model_path_set": None,
|
|
"model_file_exists": None,
|
|
"model_load_requested": check_model_load,
|
|
"model_load_ok": None,
|
|
"manifest_path_set": None,
|
|
"manifest_valid": None,
|
|
"tile_paths_exist": None,
|
|
"tile_limit_ok": None,
|
|
},
|
|
"tile_count": 0,
|
|
"max_tiles": resolved_settings.yolo_max_tiles,
|
|
"will_download_models": False,
|
|
"will_run_inference": False,
|
|
"runtime": YoloPreflightService._runtime_details(
|
|
settings=resolved_settings,
|
|
assume_dependencies=assume_dependencies,
|
|
),
|
|
}
|
|
|
|
if not resolved_settings.yolo_enabled:
|
|
result["message"] = "YOLO is disabled. Set YOLO_ENABLED=true for configured local inference."
|
|
return result
|
|
|
|
dependencies_available = True if assume_dependencies else yolo_adapter_class.dependencies_available()
|
|
result["checks"]["dependencies_available"] = dependencies_available
|
|
if dependencies_available and not assume_dependencies:
|
|
result["runtime"]["cuda_available"] = YoloPreflightService._torch_cuda_available()
|
|
if not dependencies_available:
|
|
result["status"] = "dependency_unavailable"
|
|
result["message"] = "YOLO dependencies are not installed. Install backend optional extras with geointel-backend[ai]."
|
|
return result
|
|
|
|
result["checks"]["model_path_set"] = bool(resolved_settings.yolo_model_path)
|
|
if not resolved_settings.yolo_model_path:
|
|
result["message"] = "YOLO_MODEL_PATH is not set. GeoIntel will not download model weights automatically."
|
|
return result
|
|
|
|
model_path = Path(resolved_settings.yolo_model_path).expanduser()
|
|
model_exists = model_path.exists() and model_path.is_file()
|
|
result["checks"]["model_file_exists"] = model_exists
|
|
if not model_exists:
|
|
result["message"] = "YOLO_MODEL_PATH does not point to an existing local model file."
|
|
return result
|
|
|
|
if check_model_load:
|
|
try:
|
|
yolo_adapter_class(resolved_settings).load_model(model_path)
|
|
except AppError as exc:
|
|
result["status"] = "model_load_failed"
|
|
result["message"] = exc.message
|
|
result["error_code"] = exc.code
|
|
result["checks"]["model_load_ok"] = False
|
|
return result
|
|
except Exception as exc:
|
|
result["status"] = "model_load_failed"
|
|
result["message"] = "Configured YOLO model could not be loaded during compatibility smoke."
|
|
result["error_code"] = "DETECTION_MODEL_LOAD_FAILED"
|
|
result["details"] = {"error": str(exc)}
|
|
result["checks"]["model_load_ok"] = False
|
|
return result
|
|
result["checks"]["model_load_ok"] = True
|
|
|
|
result["checks"]["manifest_path_set"] = bool(tile_manifest_path)
|
|
if not tile_manifest_path:
|
|
result["status"] = "manifest_unavailable"
|
|
result["message"] = "Configured YOLO inference requires an existing raster tile manifest path."
|
|
return result
|
|
|
|
try:
|
|
manifest = DetectionService._load_tile_manifest(tile_manifest_path, resolved_settings.yolo_max_tiles)
|
|
tile_paths = [DetectionService._resolve_tile_path(tile, Path(tile_manifest_path).expanduser()) for tile in manifest["tiles"]]
|
|
except AppError as exc:
|
|
result["status"] = "manifest_invalid"
|
|
result["message"] = exc.message
|
|
result["error_code"] = exc.code
|
|
result["checks"]["manifest_valid"] = False
|
|
if exc.code != "DETECTION_TILE_LIMIT_EXCEEDED":
|
|
result["checks"]["tile_limit_ok"] = None
|
|
else:
|
|
result["checks"]["tile_limit_ok"] = False
|
|
return result
|
|
|
|
result["checks"]["manifest_valid"] = True
|
|
result["checks"]["tile_paths_exist"] = all(path.exists() and path.is_file() for path in tile_paths)
|
|
result["checks"]["tile_limit_ok"] = len(tile_paths) <= resolved_settings.yolo_max_tiles
|
|
result["tile_count"] = len(tile_paths)
|
|
result["status"] = "ready"
|
|
if check_model_load:
|
|
result["message"] = "Configured YOLO preflight passed. Local model load smoke passed and no inference was run."
|
|
else:
|
|
result["message"] = "Configured YOLO preflight passed. No model was loaded and no inference was run."
|
|
return result
|
|
|
|
@staticmethod
|
|
def _runtime_details(*, settings: Settings, assume_dependencies: bool) -> dict[str, Any]:
|
|
model_directory = None
|
|
if settings.yolo_model_path:
|
|
model_directory = str(Path(settings.yolo_model_path).expanduser().parent)
|
|
return {
|
|
"dependencies_assumed": assume_dependencies,
|
|
"model_directory": model_directory,
|
|
"yolo_config_dir": os.environ.get("YOLO_CONFIG_DIR"),
|
|
"torch_version": YoloPreflightService._package_version("torch"),
|
|
"ultralytics_version": YoloPreflightService._package_version("ultralytics"),
|
|
"cuda_available": None,
|
|
}
|
|
|
|
@staticmethod
|
|
def _package_version(package_name: str) -> str | None:
|
|
try:
|
|
return metadata.version(package_name)
|
|
except metadata.PackageNotFoundError:
|
|
return None
|
|
|
|
@staticmethod
|
|
def _torch_cuda_available() -> bool | None:
|
|
try:
|
|
import torch
|
|
except Exception:
|
|
return None
|
|
try:
|
|
return bool(torch.cuda.is_available())
|
|
except Exception:
|
|
return None
|