# -*- coding: utf-8 -*- """環境診断スクリプト — このパッケージが動く環境かを単体で検査する。 使い方（パッケージのどこからでも・引数不要）: python3 check_env.py .venv がある場合は自動で .venv の Python に切り替えて検査するため、 venv の有効化を忘れていても正しく診断できる。 終了コード: 0 = 実行可能（警告があっても動く） / 2 = 不足あり（実行しても必ず失敗する） """ from __future__ import annotations import importlib import os import shutil import subprocess import sys from pathlib import Path HERE = Path(__file__).resolve().parent # ------------------------------------------------------------------ # .venv があれば、その Python で自分自身を再実行する（venv有効化忘れ対策） # ------------------------------------------------------------------ def _venv_python() -> Path | None: for rel in (".venv/bin/python", ".venv/Scripts/python.exe"): p = HERE / rel if p.exists(): return p return None def _maybe_reexec_in_venv() -> None: venv_py = _venv_python() if venv_py is None: return if Path(sys.executable).resolve() == venv_py.resolve(): return print(f"※ .venv を検出したため {venv_py} で診断します") os.execv(str(venv_py), [str(venv_py), str(Path(__file__).resolve())] + sys.argv[1:]) # ------------------------------------------------------------------ # 検査項目 # ------------------------------------------------------------------ OK, WARN, NG = "✓", "⚠", "✗" # (import名, pip名, 用途) REQUIRED_PACKAGES = [ ("torch", "torch", "上流AI（湯気/スープ抽出/濃度）"), ("torchvision", "torchvision", "上流AI"), ("timm", "timm", "湯気分類(EfficientNet)"), ("rfdetr", "rfdetr", "スープ抽出(RF-DETR)"), ("albumentations", "albumentations", "画像前処理"), ("cv2", "opencv-python-headless", "画像/動画処理"), ("PIL", "pillow", "画像処理"), ("numpy", "numpy", "数値計算"), ("pandas", "pandas", "採点エンジン"), ("sklearn", "scikit-learn", "採点モデル(.pkl)読込"), ("joblib", "joblib", "採点モデル(.pkl)読込"), ("yaml", "pyyaml", "設定ファイル読込"), ("requests", "requests", "BytePlus API通信"), ] # 無くても推論は動くが、揃っているのが望ましいもの（警告どまり） OPTIONAL_PACKAGES = [ ("pycocotools", "pycocotools", "RF-DETRの周辺機能（推論のみなら必須ではない）"), ] SUITE = Path(os.environ.get("RAMEN_SUITE_DIR", str(HERE / "upstream_suite"))) REQUIRED_FILES = [ (HERE / "prompt" / "soup_eval_prompt_policy.md", "LLM評価プロンプト定義"), (HERE / "app" / "models" / "human_total_score.pkl", "採点モデル(総合)"), (HERE / "app" / "models" / "human_soup_score.pkl", "採点モデル(スープ)"), (HERE / "local_batch_api" / "config.test10.yaml", "パイプライン設定"), (HERE / "local_batch_api" / "calibration.json", "濃度キャリブレーション"), (SUITE / "03_soup-density-ai" / "models" / "crop" / "soup_valid_area_rfdetr_seg_v2" / "checkpoint_best_ema.pth", "上流モデル: スープ抽出(RF-DETR)"), (SUITE / "03_soup-density-ai" / "models" / "density" / "density_classifier_extracted_v1" / "model.pt", "上流モデル: 濃度分類"), (SUITE / "04_steam-classification" / "steam_detection_project" / "models" / "steam_classifier_best.pt", "上流モデル: 湯気分類"), ] def main() -> int: _maybe_reexec_in_venv() results: list[tuple[str, str, str]] = [] # (レベル, 項目, 対処) def add(level: str, item: str, fix: str = "") -> None: results.append((level, item, fix)) # 1. Python バージョン（3.10+ の構文を使用） v = sys.version_info if (v.major, v.minor) >= (3, 10): add(OK, f"Python {v.major}.{v.minor}.{v.micro}") else: add(NG, f"Python {v.major}.{v.minor} は古すぎます（3.10以上が必要）", "sudo apt-get install -y python3.10-venv などで新しいPythonを導入") # 2. 仮想環境 if _venv_python(): add(OK, "仮想環境 .venv") else: add(WARN, "仮想環境 .venv が未作成", "bash setup.sh を実行（システムPythonに直接入れている場合はこの警告は無視可）") # 3. OSコマンド for exe in ("ffmpeg", "ffprobe"): w = shutil.which(exe) if w: add(OK, f"コマンド {exe} ({w})") else: add(NG, f"コマンド {exe} がPATHにありません", "sudo apt-get install -y ffmpeg（bash setup.sh でも導入されます）") # 4. Pythonパッケージ for mod, pip_name, purpose in REQUIRED_PACKAGES: try: importlib.import_module(mod) add(OK, f"パッケージ {pip_name}") except ImportError: add(NG, f"パッケージ {pip_name} が未導入（{purpose}）", f"bash setup.sh を実行（個別なら .venv/bin/pip install {pip_name}）") except Exception as e: # 導入済みだが壊れている等 add(NG, f"パッケージ {pip_name} の読込に失敗: {type(e).__name__}: {e}", f".venv/bin/pip install --force-reinstall {pip_name}") for mod, pip_name, purpose in OPTIONAL_PACKAGES: try: importlib.import_module(mod) add(OK, f"パッケージ {pip_name}") except Exception: add(WARN, f"パッケージ {pip_name} が未導入（{purpose}）", f"必要なら .venv/bin/pip install {pip_name}") # 5. 必須ファイル for path, label in REQUIRED_FILES: if path.exists(): add(OK, f"{label}") else: fix = "パッケージ(zip)を展開し直して、フォルダ構成を変えずに配置してください" if "上流モデル" in label: fix += f"（RAMEN_SUITE_DIR で別置き指定も可。現在の探索先: {SUITE}）" add(NG, f"{label} がありません: {path}", fix) # 6. APIキー（無くても起動はするが、LLM評価で必ず失敗するため警告） key_file = HERE / "secrets" / "byteplus.key" if os.environ.get("ARK_API_KEY"): add(OK, "BytePlus APIキー（環境変数 ARK_API_KEY）") elif key_file.exists() and key_file.stat().st_size > 0: add(OK, "BytePlus APIキー（secrets/byteplus.key）") else: add(WARN, "BytePlus APIキーが未設定（LLM評価の段階で失敗します）", f"{key_file} にキーを1行保存 か export ARK_API_KEY=xxxx") # 7. BytePlusへの外向き通信（DNS解決のみ・軽量） try: import socket socket.getaddrinfo("ark.ap-southeast.bytepluses.com", 443) add(OK, "BytePlusエンドポイントの名前解決") except OSError: add(WARN, "BytePlusエンドポイントの名前解決に失敗（外向きHTTPSが塞がれている可能性）", "ファイアウォール/プロキシ設定で ark.ap-southeast.bytepluses.com:443 への接続を許可") # 8. ディスク空き（フレーム画像・一時動画用） try: free_gb = shutil.disk_usage(HERE).free / 1e9 if free_gb >= 10: add(OK, f"ディスク空き {free_gb:.0f}GB") else: add(WARN, f"ディスク空きが少ない: {free_gb:.1f}GB（10GB以上推奨）", "不要ファイルを削除") except OSError: pass # 9. swap（Linuxのみ・メモリ8GB環境のOOM対策） meminfo = Path("/proc/meminfo") if meminfo.exists(): try: info = {l.split(":")[0]: l.split(":")[1].strip() for l in meminfo.read_text().splitlines() if ":" in l} swap_kb = int(info.get("SwapTotal", "0 kB").split()[0]) mem_kb = int(info.get("MemTotal", "0 kB").split()[0]) if swap_kb == 0 and mem_kb < 12 * 1024 * 1024: add(WARN, f"swapが未設定（RAM {mem_kb//1024//1024}GB）。推論中にOOMで落ちる恐れ", "README_deploy.md 手順4.5 の swap設定を実施") else: add(OK, f"メモリ {mem_kb//1024//1024}GB / swap {swap_kb//1024//1024}GB") except (ValueError, OSError): pass # ---- 結果表示 ---- print() print("=" * 70) print(" ラーメンスープ評価AI 環境診断") print("=" * 70) ngs = [r for r in results if r[0] == NG] warns = [r for r in results if r[0] == WARN] for level, item, fix in results: print(f" {level} {item}") if fix and level != OK: print(f" → {fix}") print("-" * 70) if ngs: print(f" 判定: ✗ 実行不可 — 不足 {len(ngs)} 件を解消してください（多くは bash setup.sh で入ります）") return 2 if warns: print(f" 判定: ⚠ 実行可能（警告 {len(warns)} 件。内容によりLLM評価などで失敗する場合があります）") return 0 print(" 判定: ✓ すべてOK — このまま実行できます") return 0 if __name__ == "__main__": sys.exit(main())