File size: 4,825 Bytes
d783468 fdb87aa d783468 fdb87aa d783468 dc624ae d783468 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | """Thorough remote real-life vs anime eval (multi-model) + optional local video check.
Writes JSON under scripts/out/ and prints a definitive GO/NO_GO conclusion.
"""
from __future__ import annotations
import json
import logging
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
REPO = ROOT.parent
sys.path.insert(0, str(ROOT))
from app.realism_eval import predict_real_life, run_realism_eval_multi_model # noqa: E402
from app.schemas import AppSettings # noqa: E402
from app.services import VIDEO_EXTENSIONS, extract_scores_with_experimental_media # noqa: E402
from app.storage import init_db # noqa: E402
from app.taxonomy import reload_taxonomy # noqa: E402
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)
logger = logging.getLogger("run_realism_debug_eval")
def _settings_from_db() -> AppSettings:
from app.api import _settings_from_db as load
return load()
def _probe_local_videos(settings: AppSettings, model: str, limit: int = 8) -> dict:
"""Score local videos with multi-frame pooling if present."""
roots = []
if settings.root_repo:
roots.append(Path(settings.root_repo).expanduser())
extra = os.environ.get("THR3SHR_VIDEO_PROBE_ROOT", "").strip()
if extra:
roots.append(Path(extra).expanduser())
videos: list[Path] = []
for root in roots:
if root is None or not root.is_dir():
continue
for path in root.iterdir():
if path.suffix.lower() in VIDEO_EXTENSIONS and path.is_file():
videos.append(path)
if videos:
break
videos = sorted(videos)[:limit]
items = []
for path in videos:
try:
scores = extract_scores_with_experimental_media(
path,
experimental_media_enabled=True,
tagger_model=model,
wd_general_threshold=settings.wd_general_threshold,
)
is_rl, folder, score, evidence = predict_real_life(
scores,
selected={"real_life", "photo", *(settings.selected_tags or [])},
)
items.append(
{
"path": str(path),
"predicted_real_life": is_rl,
"folder": folder,
"score": score,
"evidence": evidence,
}
)
except Exception as err:
items.append({"path": str(path), "error": str(err)})
return {
"model": model,
"count": len(items),
"predicted_real_life": sum(1 for i in items if i.get("predicted_real_life")),
"items": items,
}
def main() -> int:
init_db()
reload_taxonomy()
settings = _settings_from_db()
count = int(sys.argv[1]) if len(sys.argv) > 1 else 16
count = max(8, min(count, 40))
logger.info("starting multi-model realism eval count_per_class=%s", count)
multi = run_realism_eval_multi_model(count_per_class=count, settings=settings)
best = multi.get("best_model") or "wd_eva02_large"
logger.info("probing local videos with best_model=%s", best)
video_probe = _probe_local_videos(settings, best, limit=10)
out_dir = Path(__file__).resolve().parent / "out"
out_dir.mkdir(parents=True, exist_ok=True)
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
payload = {
"generated_at": datetime.now(timezone.utc).isoformat(),
"multi_model": multi,
"local_video_probe": video_probe,
}
out_path = out_dir / f"realism_debug_eval_{stamp}.json"
latest = out_dir / "realism_debug_eval_latest.json"
text = json.dumps(payload, indent=2)
out_path.write_text(text, encoding="utf-8")
latest.write_text(text, encoding="utf-8")
overall = multi.get("overall_conclusion") or {}
print("=== REALISM DEBUG EVAL ===")
print("decision:", overall.get("decision"))
print("best_model:", overall.get("best_model"))
print("summary:", overall.get("summary"))
print("video:", overall.get("video_and_gif"))
for report in multi.get("reports") or []:
m = report.get("metrics") or {}
print(
f"- {report.get('tagger_model')}: {report.get('conclusion', {}).get('decision')} "
f"P={m.get('precision')} R={m.get('recall')} F1={m.get('f1')} "
f"animeFP={m.get('anime_false_positive_rate')} n={report.get('count_evaluated')}"
)
print("local_video_predicted_real_life:", video_probe.get("predicted_real_life"), "/", video_probe.get("count"))
print("wrote", out_path)
return 0 if overall.get("decision") == "GO" else 2
if __name__ == "__main__":
raise SystemExit(main())
|