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.runtime_model_provenance_service import RuntimeModelProvenanceService 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, "accelerator_ready": None, "model_path_set": None, "model_file_exists": None, "model_provenance_manifest_path": None, "model_provenance_valid": 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 if not assume_dependencies: try: adapter = yolo_adapter_class(resolved_settings) validate_runtime = getattr(adapter, "validate_runtime", None) if validate_runtime is not None: validate_runtime() except AppError as exc: result["checks"]["accelerator_ready"] = False result["status"] = "accelerator_unavailable" result["message"] = exc.message result["error_code"] = exc.code result["details"] = exc.details return result result["checks"]["accelerator_ready"] = True 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 result["checks"]["model_provenance_manifest_path"] = str( RuntimeModelProvenanceService.manifest_path_for_model(model_path) ) try: RuntimeModelProvenanceService.validate_for_runtime( model_path=model_path, model_id=resolved_settings.yolo_model_id, task_type="object_detection", expected_model_version=resolved_settings.yolo_model_version, allowed_frameworks=("ultralytics/pytorch", "ultralytics", "pytorch"), ) except AppError as exc: result["checks"]["model_provenance_valid"] = False result["status"] = "contract_incomplete" result["message"] = ( "Configured YOLO weights are not runnable until their immutable runtime provenance sidecar validates: " f"{exc.message}" ) result["error_code"] = exc.code result["details"] = exc.details return result result["checks"]["model_provenance_valid"] = True 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, resolved_settings) tile_paths = [ DetectionService._resolve_tile_path(tile, Path(tile_manifest_path).expanduser(), resolved_settings) 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, "configured_device": settings.yolo_device, "cuda_required": settings.yolo_require_cuda, } @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