"""Определение доступности уровней AI-модуля (`min`/`medium`/`max`) инстанса. Матрица уровней (модели, требования RAM/GPU/VRAM, пути моделей на дисковых томах) — константа `TIERS` (`services/ai_tiers.py`), единственный источник истины — ADR-004 (`docs/architecture/adr/004-ai-tier-matrix.md`). Детект читает обнаруженное `install.sh` железо (`HW_*` в `.env`, `core.config.Settings`) и факт наличия файлов моделей на дисковых томах — без зависимости от torch/nvidia-smi внутри процесса backend/воркеров (переменные пишет установщик, а не рантайм-детект GPU). """ from pathlib import Path from pydantic import BaseModel from core.config import Settings, get_settings from core.plugins.config import AiLevel, InstanceConfig from services.ai_tiers import TIERS, WHISPER_MODELS_ROOT, TierSpec _PRESET_BY_LEVEL: dict[AiLevel, int] = {"min": 3, "medium": 4, "max": 5} """Номер пресета инсталлятора, соответствующего уровню (ADR-004, таблица требований железа) — используется в тексте причины недоступности.""" class AiLevelStatus(BaseModel): """Доступность одного уровня AI с человекочитаемой причиной отказа.""" level: AiLevel available: bool reason: str | None = None def detect_ai_levels(cfg: InstanceConfig) -> list[AiLevelStatus]: """Вернуть статусы всех уровней AI по обнаруженному железу и скачанным моделям. `cfg` пока не влияет на результат (доступность уровня зависит только от железа и файлов моделей на диске, не от текущих настроек инстанса), но остаётся параметром сигнатуры — используется и `api/admin.py`, и `services/instance_settings.py::update`, где эффективная конфигурация уже под рукой. """ settings = get_settings() statuses: list[AiLevelStatus] = [] for level in ("min", "medium", "max"): reasons = _unavailability_reasons(level, TIERS[level], settings) statuses.append( AiLevelStatus(level=level, available=not reasons, reason="; ".join(reasons) or None) ) return statuses def _unavailability_reasons(level: AiLevel, spec: TierSpec, settings: Settings) -> list[str]: """Собрать причины недоступности уровня `level` (пустой список — уровень доступен).""" reasons: list[str] = [] if settings.hw_ram_mb is not None and settings.hw_ram_mb < spec.min_ram_mb: reasons.append(f"недостаточно RAM: нужно {spec.min_ram_mb // 1024} ГБ") if spec.requires_gpu: required_vram_gb = (spec.min_vram_mb or 0) // 1024 if not settings.hw_gpu_name: reasons.append(f"требуется GPU NVIDIA ≥{required_vram_gb} ГБ VRAM, не обнаружен") elif spec.min_vram_mb is not None and (settings.hw_vram_mb or 0) < spec.min_vram_mb: found_vram_gb = (settings.hw_vram_mb or 0) // 1024 reasons.append( f"требуется GPU NVIDIA ≥{required_vram_gb} ГБ VRAM, " f"обнаружено только {found_vram_gb} ГБ" ) preset = _PRESET_BY_LEVEL[level] for path in spec.model_files: if not _model_downloaded(path): reasons.append( f"{_describe_model_file(path)} не скачана — запустите install.sh " f"с пресетом {preset}" ) return reasons def _model_downloaded(path: str) -> bool: """Проверить, скачана ли модель по пути на томе. Файл — непустой; каталог (например, кэш huggingface_hub с хэшированными поддиректориями снапшотов) — непустой каталог, без проверки конкретных вложенных файлов. """ p = Path(path) if p.is_dir(): return any(p.iterdir()) return p.is_file() and p.stat().st_size > 0 def _describe_model_file(path: str) -> str: """Человекочитаемое имя модели для причины недоступности («модель транскрибации small»).""" name = Path(path).name kind = "транскрибации" if path.startswith(f"{WHISPER_MODELS_ROOT}/") else "суммаризации" return f"модель {kind} {name}"