# -*- coding: utf-8 -*- """フォルダ内の動画をまとめて end-to-end 推論し、結果を保存するテスト用スクリプト。 pipeline_full.py（グレード算出まで含む完全版）をフォルダ全体に対して回し、 1動画=1結果を JSONL / CSV / 個別JSON に保存する。モデルは最初に一度だけ読み込む。 使い方（VPSで .venv 有効化後）: source ../.venv/bin/activate # パッケージ直下の .venv export RAMEN_SUITE_DIR="$(cd .. && pwd)/upstream_suite" cd app python batch_test.py --input /path/to/videos --slot 開店前 --out results # 濃度(brix)や時間帯を動画ごとに変えたい場合は CSV を渡す: # CSVヘッダ例: filename,brix,slot,store python batch_test.py --input /path/to/videos --csv meta.csv --out results 出力（--out 既定 = ./results/）: results/results.jsonl … 1行1動画の全結果（中間結果込み） results/results.csv … スコア/グレードの一覧表（Excelで開ける） results/details/<動画名>.json … 動画ごとの詳細JSON """ from __future__ import annotations import argparse import csv import json import shutil import sys import time import traceback from pathlib import Path APP_DIR = Path(__file__).resolve().parent # パッケージ直下（app/ の1つ上）。実行時のカレントディレクトリに依存させないための基準点。 PKG_ROOT = APP_DIR.parent # app/ を明示的に通す（どのフォルダから・どの呼び出し方でも app/ 内を読めるように） sys.path.insert(0, str(APP_DIR)) from pipeline_full import FullPipeline # noqa: E402 sys.path設定後にimportする必要がある VIDEO_EXTENSIONS = {".mp4", ".mov", ".m4v", ".avi", ".mkv"} def resolve_input_dir(value: str) -> Path: """入力フォルダを解決する。絶対パスはそのまま。相対パスは カレントディレクトリ基準 → パッケージ基準 の順に探す（どこから実行しても動くように）。""" p = Path(value).expanduser() if p.is_absolute(): return p.resolve() candidates = [(Path.cwd() / p).resolve(), (PKG_ROOT / p).resolve()] for c in candidates: if c.is_dir(): return c tried = "\n".join(f" - {c}" for c in candidates) raise SystemExit(f"入力フォルダが見つかりません（次を探しました）:\n{tried}") def resolve_out_dir(value: str | None) -> Path: """出力フォルダを解決する。未指定なら パッケージ直下/results（実行場所に依存しない）。""" if value is None: return PKG_ROOT / "results" p = Path(value).expanduser() return p.resolve() if p.is_absolute() else (Path.cwd() / p).resolve() def describe_error(exc: BaseException) -> str: """例外を「何が無い/何が起きた」まで分かる1行にする。 repr()はFileNotFoundErrorのファイル名を含まないため使わない。""" msg = f"{type(exc).__name__}: {exc}" fname = getattr(exc, "filename", None) if fname and str(fname) not in msg: msg += f" (対象: {fname})" return msg def preflight() -> list[str]: """実行前に必須コマンド・ファイルを検査し、不足を日本語の指示付きで返す。""" problems: list[str] = [] # ① ffmpeg / ffprobe（フレーム抽出に必須。無いと動画処理中に FileNotFoundError になる） for exe in ("ffmpeg", "ffprobe"): if shutil.which(exe) is None: problems.append( f"コマンド '{exe}' がPATHにありません。→ sudo apt-get install -y ffmpeg " f"（または bash setup.sh を再実行）") # ② LLM評価プロンプト定義 import batch_process as B if not B.POLICY_PATH.exists(): problems.append(f"プロンプト定義がありません: {B.POLICY_PATH} → prompt/ フォルダごと転送してください") # ③ グレード算出モデル for pkl in ("human_total_score.pkl", "human_soup_score.pkl"): p = APP_DIR / "models" / pkl if not p.exists(): problems.append(f"採点モデルがありません: {p} → app/models/ を転送してください") # ④ 上流画像AIの学習済み重み（初回動画の処理時に読み込まれるため、事前に検査する） import density_resolver as dr for label, ck in (("スープ抽出(RF-DETR)", dr.CROP_CKPT), ("濃度分類", dr.DENSITY_CKPT), ("湯気分類", dr.STEAM_CKPT)): if not Path(ck).exists(): problems.append( f"上流モデル({label})がありません: {ck} → upstream_suite/ をパッケージ直下に配置。" f"別の場所に置いた場合は 環境変数 RAMEN_SUITE_DIR でその場所を指定（現在: " f"{dr.SUITE}）") # ⑤ APIキー（ゲートを通過した動画のLLM評価で必要。無くても起動はできるため警告扱い） import os key_file = PKG_ROOT / "secrets" / "byteplus.key" if not os.environ.get("ARK_API_KEY") and not (key_file.exists() and key_file.stat().st_size > 0): problems.append( f"[警告] BytePlus APIキーが未設定です（LLM評価の段階で失敗します）。" f"→ {key_file} にキーを1行保存 か export ARK_API_KEY=xxxx") return problems def slot_recognized(slot: str | None) -> bool: """時間帯がモデルの学習カテゴリに解釈されるか（scoring_engine の判定規則と同一）。 未知の文字列でもエラーにはならないが特徴量が全て0になり精度が黙って落ちるため、警告に使う。""" if not slot: return False return ("開店" in slot) or any(k in slot for k in ("17", "18", "19", "夜", "夕")) def _jsonable(obj): """numpy型(np.bool_/np.int64/np.float64等)をjson.dumpsが扱える標準型へ変換する。""" if hasattr(obj, "item"): # numpyスカラは .item() でPython標準型に落ちる try: return obj.item() except Exception: pass return str(obj) # CSVサマリに出す列（読みやすい順） CSV_COLUMNS = [ "video", "status", "total_score", "total_grade", "soup_score", "soup_grade", "readable", "llm_used", "density_value", "density_source", "note", "error", ] def load_meta_csv(csv_path: Path) -> dict[str, dict]: """filename -> {brix, slot, store} の辞書を作る（列は無くてもよい）。""" meta: dict[str, dict] = {} with csv_path.open(encoding="utf-8-sig", newline="") as fh: for row in csv.DictReader(fh): name = (row.get("filename") or row.get("video") or "").strip() if not name: continue meta[name.lower()] = { "brix": (row.get("brix") or "").strip() or None, "slot": (row.get("slot") or "").strip() or None, "store": (row.get("store") or "").strip() or None, } return meta def main() -> None: ap = argparse.ArgumentParser(description="フォルダ内動画の一括テスト推論") ap.add_argument("--input", required=True, help="動画が入ったフォルダ") ap.add_argument("--out", default=None, help="結果の保存先フォルダ（既定: パッケージ直下の results/）") ap.add_argument("--slot", default=None, help="全動画共通の時間帯（CSV未指定時に使用）") ap.add_argument("--brix", default=None, help="全動画共通の濃度（CSV未指定時に使用）") ap.add_argument("--store", default=None, help="全動画共通の店名（任意）") ap.add_argument("--csv", default=None, help="動画ごとのメタ情報CSV（filename,brix,slot,store）") args = ap.parse_args() input_dir = resolve_input_dir(args.input) videos = sorted(p for p in input_dir.iterdir() if p.suffix.lower() in VIDEO_EXTENSIONS) if not videos: raise SystemExit(f"動画が1本もありません（対応拡張子 {sorted(VIDEO_EXTENSIONS)}）: {input_dir}") meta_csv = load_meta_csv(Path(args.csv).expanduser().resolve()) if args.csv else {} out_dir = resolve_out_dir(args.out) details_dir = out_dir / "details" details_dir.mkdir(parents=True, exist_ok=True) jsonl_path = out_dir / "results.jsonl" csv_path = out_dir / "results.csv" print(f"入力: {input_dir} 動画 {len(videos)} 本") print(f"出力: {out_dir}") # 実行前チェック: 不足があれば動画を処理する前に、何をどう直すかまで表示して止まる issues = preflight() fatal = [x for x in issues if not x.startswith("[警告]")] for x in issues: print(("✗ " if not x.startswith("[警告]") else "⚠ ") + x) if fatal: raise SystemExit(f"\n環境に不足が {len(fatal)} 件あります。上記を解消してから再実行してください。") if not issues: print("事前チェック: すべてOK（ffmpeg・モデル・プロンプト・キー）") print("モデル読込中（初回のみ1〜2分）...", flush=True) pipe = FullPipeline() # 上流モデルもここで読み込む（1本目の処理中に読み込むと、ライブラリの警告ログが進捗表示に割り込むため） pipe.resolver._ensure_models() rows: list[dict] = [] warned_slots: set[str] = set() with jsonl_path.open("w", encoding="utf-8") as jf: for i, video in enumerate(videos, 1): m = meta_csv.get(video.name.lower(), {}) # CSVの値が空欄なら --brix/--slot/--store の共通値へフォールバック brix = m.get("brix") or args.brix slot = m.get("slot") or args.slot store = m.get("store") or args.store key = slot or "(未指定)" if not slot_recognized(slot) and key not in warned_slots: warned_slots.add(key) print(f"⚠ 時間帯 '{key}' は学習時のカテゴリ（開店前 / 17時〜19時）に解釈されません。" f"エラーにはなりませんが精度が落ちます") print(f"[{i}/{len(videos)}] {video.name} ...", end="", flush=True) t0 = time.time() try: result = pipe.run(video, brix=brix, slot=slot, store=store) err = "" except Exception as exc: # 1本失敗しても続行 err = describe_error(exc) result = {"status": "error", "error": err, "traceback": traceback.format_exc()} record = {"video": video.name, **result} jf.write(json.dumps(record, ensure_ascii=False, default=_jsonable) + "\n") jf.flush() (details_dir / f"{video.stem}.json").write_text( json.dumps(record, ensure_ascii=False, indent=2, default=_jsonable), encoding="utf-8") up = result.get("_upstream", {}) or {} rows.append({ "video": video.name, "status": result.get("status", ""), "total_score": result.get("total_score", ""), "total_grade": result.get("total_grade", ""), "soup_score": result.get("soup_score", ""), "soup_grade": result.get("soup_grade", ""), "readable": up.get("readable", ""), "llm_used": up.get("llm_used", ""), "density_value": up.get("density_value", ""), "density_source": up.get("density_source", ""), "note": up.get("note", ""), "error": err, }) dt = time.time() - t0 print(f" {result.get('status','?')} " f"(total={result.get('total_score','-')}/{result.get('total_grade','-')}, " f"soup={result.get('soup_score','-')}/{result.get('soup_grade','-')}) {dt:.0f}s") with csv_path.open("w", encoding="utf-8-sig", newline="") as cf: w = csv.DictWriter(cf, fieldnames=CSV_COLUMNS) w.writeheader() w.writerows(rows) scored = sum(1 for r in rows if r["status"] == "scored") need = sum(1 for r in rows if r["status"] == "要確認") errs = sum(1 for r in rows if r["status"] == "error") print("\n===== 完了 =====") print(f"採点: {scored} / 要確認: {need} / エラー: {errs} （計 {len(rows)} 本）") print(f"一覧CSV : {csv_path}") print(f"全結果 : {jsonl_path}") print(f"個別JSON: {details_dir}") if __name__ == "__main__": sys.exit(main())