# -*- coding: utf-8 -*- """濃度の解決（条件分岐）モジュール。 LLM送信前に「スープ濃度」を確定させる: - CSV等に濃度が与えられている場合 -> そのまま採用 (source=provided) - 与えられていない場合 -> 画像モデルで推定 (source=model) 湯気除外 -> スープ抽出 -> 濃度分類 の3段。 スープ表面が一度も見えない動画は誤った数値を出さず status=判定不可 とする。 精度最優先のため、信頼度ゲートを設ける: - high : 抽出信頼度>=conf_keep のフレームが multi_min 枚以上 かつ 投票占有率>=vote_min - medium : 抽出信頼度>=conf_keep のフレームが 1 枚以上 - 判定不可: 上記に満たない（スープ表面が確認できない） DensityResolver はモデルを一度だけ読み込む。resolve(video_path, metadata) を呼ぶ。 """ from __future__ import annotations import re import subprocess import sys import tempfile from collections import defaultdict from pathlib import Path from typing import Any import cv2 import numpy as np ROOT = Path(__file__).resolve().parents[1] SUITE = Path.home() / "Documents" / "開発しているもの" / "制作中" / "AIラーメン開発" / "ramen-soup-ai-suite" CROP_SRC = SUITE / "03_soup-density-ai" / "src" CROP_CKPT = SUITE / "03_soup-density-ai" / "models" / "crop" / "soup_valid_area_rfdetr_seg_v2" / "checkpoint_best_ema.pth" DENSITY_CKPT = SUITE / "03_soup-density-ai" / "models" / "density" / "density_classifier_extracted_v1" / "model.pt" STEAM_CKPT = SUITE / "04_steam-classification" / "steam_detection_project" / "models" / "steam_classifier_best.pt" DENSITY_COLUMNS = ("スープ濃度", "brix_raw", "brix", "density") def normalize_density(value: Any) -> str: """'12' '12度' などから数値文字列を取り出す。無ければ ''。""" if value is None: return "" m = re.search(r"\d+(?:\.\d+)?", str(value).strip()) return m.group(0) if m else "" def density_in_metadata(metadata: dict[str, Any]) -> str: for col in DENSITY_COLUMNS: v = normalize_density(metadata.get(col)) if v: return v return "" class DensityResolver: def __init__( self, conf_keep: float = 0.50, crop_threshold: float = 0.10, frames: int = 16, multi_min: int = 4, vote_min: float = 0.70, high_mean_conf: float = 0.72, crops_dir: Path | None = None, device_name: str = "auto", calibration_path: Path | None = None, ) -> None: self.conf_keep = conf_keep self.crop_threshold = crop_threshold self.frames = frames self.multi_min = multi_min self.vote_min = vote_min self.high_mean_conf = high_mean_conf self.crops_dir = crops_dir or (ROOT / "local_batch_api" / "resolver_crops") self.crops_dir.mkdir(parents=True, exist_ok=True) self._device_name = device_name self._loaded = False self._calibration = self._load_calibration(calibration_path) # ---- キャリブレーション補正 ---- # @staticmethod def _load_calibration(path: Path | None) -> dict[str, Any] | None: import json cal_path = path or (Path(__file__).resolve().parent / "calibration.json") if not cal_path.exists(): return None try: data = json.loads(cal_path.read_text(encoding="utf-8")) return data if data.get("enabled", True) and data.get("map") else None except Exception: return None def _calibrate(self, value: str | None) -> tuple[str | None, str | None]: """生の推定値に補正マップを適用。戻り値 (補正後, 生値)。マップが無ければそのまま。""" if value is None or not self._calibration: return value, value mapped = self._calibration["map"].get(str(value)) if mapped is None: return value, value return str(mapped), str(value) # ---- モデル遅延読み込み ---- # def _ensure_models(self) -> None: if self._loaded: return import torch self._torch = torch self.device = torch.device("cuda" if (self._device_name != "cpu" and torch.cuda.is_available()) else "cpu") # 湯気分類 try: import timm steam = timm.create_model("efficientnet_b0", pretrained=False, num_classes=2) except Exception: from torchvision import models import torch.nn as nn steam = models.efficientnet_b0(weights=None) steam.classifier[-1] = nn.Linear(steam.classifier[-1].in_features, 2) ck = torch.load(STEAM_CKPT, map_location=self.device) steam.load_state_dict(ck["model_state_dict"] if "model_state_dict" in ck else ck) steam.to(self.device).eval() self.steam_model = steam import albumentations as A from albumentations.pytorch import ToTensorV2 self.steam_tf = A.Compose([ A.Resize(height=224, width=224), A.Normalize(mean=(0.485, 0.456, 0.406), std=(0.229, 0.224, 0.225)), ToTensorV2(), ]) # スープ抽出 + 濃度 sys.path.insert(0, str(CROP_SRC)) from crop.extract_soup_area import build_predictor, extract_one_image from density.predict_density import predict_density self._extract_one_image = extract_one_image self._predict_density = predict_density self.crop_predictor = build_predictor("small", str(CROP_CKPT), self.crop_threshold) self._loaded = True # ---- 内部ヘルパ ---- # def _steam_prob(self, image_bgr) -> float: rgb = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2RGB) t = self.steam_tf(image=rgb)["image"].unsqueeze(0).to(self.device) with self._torch.no_grad(): return float(self._torch.softmax(self.steam_model(t), dim=1)[0, 1].item()) @staticmethod def _duration(video_path: Path) -> float: r = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", str(video_path)], capture_output=True, text=True, timeout=30) try: return float(r.stdout.strip()) except ValueError: return 0.0 @staticmethod def _frame(video_path: Path, seek: float): t = Path(tempfile.mktemp(suffix=".jpg")) subprocess.run(["ffmpeg", "-hide_banner", "-loglevel", "error", "-ss", f"{seek:.3f}", "-i", str(video_path), "-frames:v", "1", "-q:v", "2", "-y", str(t)], capture_output=True, timeout=60) if t.exists(): img = cv2.imdecode(np.frombuffer(t.read_bytes(), np.uint8), cv2.IMREAD_COLOR) t.unlink() return img return None @staticmethod def _write(path: Path, img) -> None: path.parent.mkdir(parents=True, exist_ok=True) ok, e = cv2.imencode(".jpg", img, [int(cv2.IMWRITE_JPEG_QUALITY), 92]) if ok: path.write_bytes(e.tobytes()) @staticmethod def _center_crop(img, ratio=0.6): h, w = img.shape[:2] cw, ch = int(w * ratio), int(h * ratio) x1, y1 = (w - cw) // 2, int(h * 0.30) return img[y1:min(h, y1 + ch), x1:x1 + cw] # ---- 本体 ---- # def resolve(self, video_path: str | Path, metadata: dict[str, Any] | None = None) -> dict[str, Any]: """濃度の有無に関わらず、まず『スープ可読性レイヤー』を常に通す。 可読性（スープ表面が読めるか = RF-DETR抽出信頼度）を判定し、LLMに見せる最良フレームを選ぶ。 その上で濃度を確定する: CSV提供値があればそれ、無ければ画像モデル推定。 可読性は濃度の出所と独立に算出される（湯気で表面が見えなければ assessable=False）。 """ metadata = metadata or {} provided = density_in_metadata(metadata) video_path = Path(video_path) if not video_path.exists(): return {"value": provided or None, "source": "provided" if provided else "model", "confidence": "given" if provided else "none", "status": "判定不可", "readable": False, "assessable": False, "best_frame_path": "", "note": "動画が見つからない", "frames_used": 0} self._ensure_models() dur = self._duration(video_path) ratios = list(np.linspace(0.05, 0.95, self.frames)) stem = video_path.stem candidates = [] # (crop_path, extract_conf) best_src = None # (source_frame_path, extract_conf) 最もスープが見える元フレーム least_steam = None # (steam_prob, source_frame_path) steam_probs = [] # 全フレームの湯気確率 for i, r in enumerate(ratios): img = self._frame(video_path, max(0.0, dur * r)) if img is None: continue fp = self.crops_dir / f"{stem}_src_f{i:02d}.jpg" self._write(fp, img) sp = self._steam_prob(img) steam_probs.append(sp) if least_steam is None or sp < least_steam[0]: least_steam = (sp, fp) res = self._extract_one_image( image_path=fp, output_dir=self.crops_dir, checkpoint_path=str(CROP_CKPT), model_size="small", threshold=self.crop_threshold, low_confidence_threshold=self.crop_threshold, predictor=self.crop_predictor, save_alpha=False, preserve_filename=False, output_kind="crop", skip_fallback_output=True) if int(res["extract_status"]) > 0: ec = float(res["extract_confidence"]) if best_src is None or ec > best_src[1]: best_src = (str(fp), ec) if ec >= self.conf_keep: candidates.append((res["soup_crop_path"], ec)) # LLMに見せるべき元フレーム: スープが最も見えるフレーム（無ければ湯気最薄） best_frame_path = best_src[0] if best_src else (str(least_steam[1]) if least_steam else "") steam_mean = float(np.mean(steam_probs)) if steam_probs else 1.0 steam_min = float(np.min(steam_probs)) if steam_probs else 1.0 # --- 可読性レイヤー: スープ可視フレームの濃度を加重投票 --- model_value = None model_conf = "none" readable = bool(candidates) readability_note = "スープ表面を確認できず（湯気/構図）" if candidates: score = defaultdict(float) for crop_path, ec in candidates: d = self._predict_density(crop_path, str(DENSITY_CKPT)) score[int(d["predicted_label"])] += ec * float(d["confidence"]) best = max(score.items(), key=lambda kv: kv[1]) model_value = str(best[0]) share = best[1] / sum(score.values()) n = len(candidates) mean_conf = float(np.mean([ec for _, ec in candidates])) # 精度最優先: 枚数・占有率・抽出信頼度の3条件を満たすときだけ high（=committal） model_conf = "high" if (n >= self.multi_min and share >= self.vote_min and mean_conf >= self.high_mean_conf) else "medium" readability_note = f"スープ可視{n}枚/占有率{share:.2f}/抽出信頼度平均{mean_conf:.2f}" assessable = readable and model_conf == "high" steam_fields = {"steam_mean": round(steam_mean, 4), "steam_min": round(steam_min, 4)} # キャリブレーション補正（生の推定値→補正値） cal_value, raw_value = self._calibrate(model_value) cal_fields = {"model_estimate_raw": raw_value} if (cal_value != raw_value) else {} # --- 濃度の確定（可読性とは独立）--- if provided: return {"value": provided, "source": "provided", "confidence": "given", "status": "ok", "readable": readable, "assessable": assessable, "best_frame_path": best_frame_path, "model_estimate": cal_value, "model_estimate_confidence": model_conf, "note": f"CSV提供値 / 可読性: {readability_note}", "frames_used": len(candidates), **steam_fields, **cal_fields} if cal_value is not None: return {"value": cal_value, "source": "model", "confidence": model_conf, "status": "ok", "readable": readable, "assessable": assessable, "best_frame_path": best_frame_path, "note": readability_note, "frames_used": len(candidates), **steam_fields, **cal_fields} return {"value": None, "source": "model", "confidence": "none", "status": "判定不可", "readable": False, "assessable": False, "best_frame_path": best_frame_path, "note": readability_note, "frames_used": 0, **steam_fields} if __name__ == "__main__": import argparse import csv import glob ap = argparse.ArgumentParser(description="濃度解決の動作確認（CSV濃度を伏せて画像推定を検証）") ap.add_argument("--video-dir", default=str(ROOT / "学習データ")) ap.add_argument("--ids", default="191-200") ap.add_argument("--label-csv", default=str(ROOT / "学習データ" / "*グリッド 10.csv")) ap.add_argument("--simulate-missing", action="store_true", help="CSV濃度を伏せてモデル推定経路を強制する") args = ap.parse_args() def parse_ids(s): s = s.strip() if "-" in s and "," not in s: lo, hi = s.split("-", 1) return [str(n) for n in range(int(lo), int(hi) + 1)] return [x.strip() for x in s.split(",") if x.strip()] human = {} fs = glob.glob(args.label_csv) if fs: for r in csv.DictReader(open(fs[0], encoding="utf-8-sig")): vid = (r.get("ID") or "").strip() if vid: human[vid] = normalize_density(r.get("スープ濃度")) resolver = DensityResolver() vdir = Path(args.video_dir) exts = [".mov", ".mp4", ".MOV", ".MP4"] print(f"{'ID':>4} {'人間':>4} {'解決値':>6} {'source':>9} {'信頼度':>7} {'可読':>5} {'評価可':>6} {'状態':>6} 備考") correct = 0 n = 0 for vid in parse_ids(args.ids): vpath = None for e in exts: c = vdir / f"{vid}{e}" if c.exists(): vpath = c break if vpath is None: continue meta = {} if args.simulate_missing else {"スープ濃度": human.get(vid, "")} res = resolver.resolve(vpath, meta) h = human.get(vid, "") val = res["value"] if res["value"] is not None else "-" mark = "" if res["status"] == "ok" and res["source"] == "model" and h and res["value"]: n += 1 if int(res["value"]) == int(h): correct += 1; mark = "✓" elif abs(int(res["value"]) - int(h)) == 1: mark = "±1" else: mark = f"差{abs(int(res['value'])-int(h))}" rd = "○" if res.get("readable") else "×" asb = "○" if res.get("assessable") else "×" print(f"{vid:>4} {h:>4} {val:>6} {res['source']:>9} {res['confidence']:>7} {rd:>5} {asb:>6} {res['status']:>6} {res['note']} {mark}") if n: print(f"\nモデル推定経路: 完全一致 {correct}/{n}")