# -*- coding: utf-8 -*- """Build image-only density feature datasets from ramen soup videos. This script uses brix / human density values only as labels in the output. They are not used as model inputs. The intended production path is: video -> multi-frame image features -> trained density estimator """ from __future__ import annotations import argparse import csv import json import re import shutil import subprocess import tempfile from dataclasses import dataclass from pathlib import Path from typing import Any import cv2 import numpy as np import yaml ROOT = Path(__file__).resolve().parents[1] VIDEO_EXTENSIONS = {".mp4", ".mov", ".m4v", ".avi", ".mkv"} DEFAULT_RATIOS = [0.2, 0.35, 0.5, 0.65, 0.8] @dataclass class VideoItem: video_path: Path row_index: int metadata: dict[str, Any] def main() -> int: parser = argparse.ArgumentParser(description="Build image-only density features from videos.") parser.add_argument("--csv", required=True, help="Source CSV containing ID and label columns.") parser.add_argument("--video-dir", required=True, help="Directory containing videos named by ID.") parser.add_argument("--output", default="local_batch_api/density_features/density_features.csv") parser.add_argument("--frames-dir", default="local_batch_api/density_features/frames") parser.add_argument("--config", default="local_batch_api/config.test10.yaml") parser.add_argument("--limit", type=int, default=0) parser.add_argument("--ratios", default=",".join(str(v) for v in DEFAULT_RATIOS)) parser.add_argument("--save-frames", action="store_true") args = parser.parse_args() config = load_config(Path(args.config)) ffmpeg = resolve_binary(config.get("processing", {}).get("ffmpeg_path"), "ffmpeg") ffprobe = resolve_binary(config.get("processing", {}).get("ffprobe_path"), "ffprobe") csv_path = Path(args.csv).expanduser().resolve() video_dir = Path(args.video_dir).expanduser().resolve() output_path = resolve_path(args.output) frames_dir = resolve_path(args.frames_dir) ratios = [float(value.strip()) for value in args.ratios.split(",") if value.strip()] items = discover_items(csv_path, video_dir) if args.limit > 0: items = items[: args.limit] output_path.parent.mkdir(parents=True, exist_ok=True) if args.save_frames: frames_dir.mkdir(parents=True, exist_ok=True) rows: list[dict[str, Any]] = [] for item in items: print(f"FEATURES {item.video_path.name}") frame_records = extract_feature_frames( item=item, ffmpeg=ffmpeg, ffprobe=ffprobe, ratios=ratios, frames_dir=frames_dir, save_frames=args.save_frames, ) if not frame_records: rows.append(error_row(item, "no readable frames")) continue rows.append(aggregate_item_features(item, frame_records)) write_csv(output_path, rows) jsonl_path = output_path.with_suffix(".jsonl") with jsonl_path.open("w", encoding="utf-8") as fh: for row in rows: fh.write(json.dumps(row, ensure_ascii=False) + "\n") print(f"done rows={len(rows)} output={output_path}") return 0 def load_config(path: Path) -> dict[str, Any]: source = path if path.exists() else ROOT / path if not source.exists(): return {} with source.open("r", encoding="utf-8") as fh: return yaml.safe_load(fh) or {} def resolve_path(value: str) -> Path: path = Path(value) return path if path.is_absolute() else ROOT / path def discover_items(csv_path: Path, video_dir: Path) -> list[VideoItem]: videos = {path.name.lower(): path for path in video_dir.iterdir() if path.suffix.lower() in VIDEO_EXTENSIONS} items: list[VideoItem] = [] with csv_path.open("r", encoding="utf-8-sig", newline="") as fh: for idx, row in enumerate(csv.DictReader(fh)): video_id = (row.get("ID") or row.get("id") or "").strip() if not video_id: continue video_path = resolve_video_by_id(video_dir, video_id, videos) if video_path.exists(): items.append(VideoItem(video_path=video_path, row_index=idx, metadata=dict(row))) else: print(f"WARN missing_video row={idx + 1} id={video_id}") return items def resolve_video_by_id(video_dir: Path, video_id: str, videos: dict[str, Path]) -> Path: for extension in sorted(VIDEO_EXTENSIONS): candidate = videos.get(f"{video_id}{extension}".lower()) if candidate: return candidate return video_dir / f"{video_id}.MOV" def extract_feature_frames( item: VideoItem, ffmpeg: str, ffprobe: str, ratios: list[float], frames_dir: Path, save_frames: bool, ) -> list[dict[str, Any]]: duration = probe_float(ffprobe, item.video_path, "duration") fps = probe_fps(ffprobe, item.video_path) records: list[dict[str, Any]] = [] for ratio in ratios: seek_seconds = max(0.0, duration * ratio) if duration > 0 else 0.0 image = extract_frame(ffmpeg, item.video_path, seek_seconds) if image is None: continue features = image_features(image) frame_path = "" if save_frames: frame_path = str(frames_dir / f"{item.video_path.stem}_r{int(ratio * 100):02d}.jpg") write_jpeg(Path(frame_path), image, 92) records.append( { "ratio": ratio, "seek_seconds": seek_seconds, "frame_no": int(round(seek_seconds * fps)) if fps > 0 else 0, "frame_path": frame_path, **features, } ) return records def extract_frame(ffmpeg: str, video_path: Path, seek_seconds: float) -> np.ndarray | None: with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp: tmp_path = Path(tmp.name) try: result = subprocess.run( [ ffmpeg, "-hide_banner", "-loglevel", "error", "-ss", f"{seek_seconds:.3f}", "-i", str(video_path), "-frames:v", "1", "-q:v", "2", "-y", str(tmp_path), ], capture_output=True, text=True, timeout=60, ) if result.returncode != 0 or not tmp_path.exists(): return None data = np.frombuffer(tmp_path.read_bytes(), dtype=np.uint8) return cv2.imdecode(data, cv2.IMREAD_COLOR) finally: try: tmp_path.unlink() except FileNotFoundError: pass def image_features(image: np.ndarray) -> dict[str, float]: resized = resize_for_features(image) return { **prefixed_features("full", region_features(resized)), **prefixed_features("liquid_roi", region_features(extract_liquid_roi(resized))), **prefixed_features("center_crop", region_features(extract_center_crop(resized))), } def region_features(image: np.ndarray) -> dict[str, float]: gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) hsv = cv2.cvtColor(image, cv2.COLOR_BGR2HSV) lab = cv2.cvtColor(image, cv2.COLOR_BGR2LAB) rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) h = hsv[:, :, 0] s = hsv[:, :, 1] / 255.0 v = hsv[:, :, 2] / 255.0 l_channel = lab[:, :, 0] / 255.0 a_channel = lab[:, :, 1].astype(np.float32) - 128.0 b_channel = lab[:, :, 2].astype(np.float32) - 128.0 brown_mask = ((h >= 8) & (h <= 35) & (s > 0.18) & (v > 0.22) & (v < 0.92)) cream_mask = ((h >= 10) & (h <= 42) & (s > 0.08) & (s <= 0.36) & (v > 0.45)) white_mask = ((s < 0.16) & (v > 0.72)) dark_mask = v < 0.18 valid_mask = ~dark_mask edges = cv2.Canny(gray, 60, 160) lap = cv2.Laplacian(gray, cv2.CV_64F) valid_count = max(int(valid_mask.sum()), 1) brown_ratio = float((brown_mask & valid_mask).sum() / valid_count) cream_ratio = float((cream_mask & valid_mask).sum() / valid_count) white_ratio = float((white_mask & valid_mask).sum() / valid_count) edge_density = float((edges > 0).mean()) sharpness = float(lap.var()) contrast = float(gray.std() / 255.0) brightness = float(v[valid_mask].mean()) if valid_mask.any() else float(v.mean()) saturation = float(s[valid_mask].mean()) if valid_mask.any() else float(s.mean()) cloudiness_index = float(np.clip(cream_ratio + white_ratio * 0.45 - brown_ratio * 0.25 - edge_density * 0.8, 0.0, 1.0)) transparency_proxy = float(np.clip(edge_density * 4.0 + brown_ratio * 0.5 - white_ratio * 0.3, 0.0, 1.0)) steam_haze_proxy = float(np.clip(white_ratio * 0.75 + max(0.0, 0.22 - contrast) * 2.0, 0.0, 1.0)) return { "brightness": brightness, "saturation": saturation, "contrast": contrast, "sharpness": sharpness, "edge_density": edge_density, "brown_ratio": brown_ratio, "cream_ratio": cream_ratio, "white_ratio": white_ratio, "lab_l_mean": float(l_channel[valid_mask].mean()) if valid_mask.any() else float(l_channel.mean()), "lab_a_mean": float(a_channel[valid_mask].mean()) if valid_mask.any() else float(a_channel.mean()), "lab_b_mean": float(b_channel[valid_mask].mean()) if valid_mask.any() else float(b_channel.mean()), "cloudiness_index": cloudiness_index, "transparency_proxy": transparency_proxy, "steam_haze_proxy": steam_haze_proxy, **classic_color_features(rgb, hsv, lab), } def classic_color_features(rgb: np.ndarray, hsv: np.ndarray, lab: np.ndarray) -> dict[str, float]: flat_rgb = rgb.reshape(-1, 3) brightness = flat_rgb.max(axis=1) foreground = brightness >= 12 if int(foreground.sum()) < 10: foreground = np.ones_like(foreground, dtype=bool) flat_hsv = hsv.reshape(-1, 3) flat_lab = lab.reshape(-1, 3) channels = { "R": flat_rgb[foreground, 0], "G": flat_rgb[foreground, 1], "B": flat_rgb[foreground, 2], "H": flat_hsv[foreground, 0], "S": flat_hsv[foreground, 1], "V": flat_hsv[foreground, 2], "L": flat_lab[foreground, 0], "a": flat_lab[foreground, 1], "bb": flat_lab[foreground, 2], } features: dict[str, float] = { "classic_fg_ratio": float(foreground.mean()), } for name, values in channels.items(): features.update(channel_stats(values, f"classic_{name}")) for name in ("L", "S", "a", "bb"): features.update(histogram_features(channels[name], f"classic_{name}", bins=8)) features["classic_RG_ratio"] = features["classic_R_mean"] / (features["classic_G_mean"] + 1e-6) features["classic_RB_ratio"] = features["classic_R_mean"] / (features["classic_B_mean"] + 1e-6) features["classic_sat_over_val"] = features["classic_S_mean"] / (features["classic_V_mean"] + 1e-6) return features def channel_stats(values: np.ndarray, prefix: str) -> dict[str, float]: if values.size == 0: return {f"{prefix}_{key}": 0.0 for key in ("mean", "std", "p10", "p25", "p50", "p75", "p90")} p10, p25, p50, p75, p90 = np.percentile(values, [10, 25, 50, 75, 90]) return { f"{prefix}_mean": float(values.mean()), f"{prefix}_std": float(values.std()), f"{prefix}_p10": float(p10), f"{prefix}_p25": float(p25), f"{prefix}_p50": float(p50), f"{prefix}_p75": float(p75), f"{prefix}_p90": float(p90), } def histogram_features(values: np.ndarray, prefix: str, bins: int) -> dict[str, float]: if values.size == 0: return {f"{prefix}_h{i}": 0.0 for i in range(bins)} hist, _ = np.histogram(values, bins=bins, range=(0, 255)) hist = hist.astype(np.float64) / max(1, int(hist.sum())) return {f"{prefix}_h{i}": float(hist[i]) for i in range(bins)} def prefixed_features(prefix: str, values: dict[str, float]) -> dict[str, float]: return {f"{prefix}_{key}": value for key, value in values.items()} def extract_liquid_roi(image: np.ndarray) -> np.ndarray: height, width = image.shape[:2] x1 = int(width * 0.12) x2 = int(width * 0.88) y1 = int(height * 0.34) y2 = int(height * 0.94) roi = image[y1:y2, x1:x2] if roi.size == 0: return image mask = np.zeros(roi.shape[:2], dtype=np.uint8) center = (mask.shape[1] // 2, int(mask.shape[0] * 0.52)) axes = (int(mask.shape[1] * 0.47), int(mask.shape[0] * 0.42)) cv2.ellipse(mask, center, axes, 0, 0, 360, 255, -1) background = np.full_like(roi, fill_value=(0, 0, 0)) return np.where(mask[:, :, None] > 0, roi, background) def extract_center_crop(image: np.ndarray) -> np.ndarray: height, width = image.shape[:2] x1 = int(width * 0.18) x2 = int(width * 0.82) y1 = int(height * 0.40) y2 = int(height * 0.88) crop = image[y1:y2, x1:x2] return crop if crop.size else image def resize_for_features(image: np.ndarray) -> np.ndarray: height, width = image.shape[:2] max_side = max(height, width) if max_side <= 960: return image scale = 960.0 / max_side return cv2.resize(image, (int(width * scale), int(height * scale)), interpolation=cv2.INTER_AREA) def aggregate_item_features(item: VideoItem, frames: list[dict[str, Any]]) -> dict[str, Any]: row: dict[str, Any] = base_label_row(item) row["frame_count"] = len(frames) row["frame_ratios"] = ",".join(str(frame["ratio"]) for frame in frames) for key in feature_keys(): values = np.array([float(frame[key]) for frame in frames], dtype=np.float64) row[f"{key}_mean"] = float(values.mean()) row[f"{key}_median"] = float(np.median(values)) row[f"{key}_min"] = float(values.min()) row[f"{key}_max"] = float(values.max()) row["frames_json"] = json.dumps(frames, ensure_ascii=False) row["error"] = "" return row def base_label_row(item: VideoItem) -> dict[str, Any]: metadata = item.metadata return { "id": metadata.get("ID") or metadata.get("id") or item.video_path.stem, "video_path": str(item.video_path), "csv_row_index": item.row_index, "store": metadata.get("提出者") or metadata.get("store") or "", "slot": metadata.get("時間帯") or metadata.get("slot") or "", "brix_raw_label": normalize_density_label(metadata.get("スープ濃度") or metadata.get("brix_raw") or ""), "human_soup_score": metadata.get("スープのみの点数") or metadata.get("soup") or "", "human_total_score": metadata.get("濃度込みの点数") or metadata.get("total") or "", "human_comment": metadata.get("評価備考") or metadata.get("comment") or "", } def error_row(item: VideoItem, error: str) -> dict[str, Any]: row = base_label_row(item) row["frame_count"] = 0 row["frame_ratios"] = "" row["frames_json"] = "[]" row["error"] = error return row def feature_keys() -> list[str]: base_keys = [ "brightness", "saturation", "contrast", "sharpness", "edge_density", "brown_ratio", "cream_ratio", "white_ratio", "lab_l_mean", "lab_a_mean", "lab_b_mean", "cloudiness_index", "transparency_proxy", "steam_haze_proxy", ] classic_channels = ["R", "G", "B", "H", "S", "V", "L", "a", "bb"] for channel in classic_channels: for stat in ("mean", "std", "p10", "p25", "p50", "p75", "p90"): base_keys.append(f"classic_{channel}_{stat}") for channel in ("L", "S", "a", "bb"): for index in range(8): base_keys.append(f"classic_{channel}_h{index}") base_keys.extend(["classic_fg_ratio", "classic_RG_ratio", "classic_RB_ratio", "classic_sat_over_val"]) return [f"{prefix}_{key}" for prefix in ("full", "liquid_roi", "center_crop") for key in base_keys] def normalize_density_label(value: Any) -> str: text = str(value).strip() match = re.search(r"\d+(?:\.\d+)?", text) return match.group(0) if match else text def write_csv(path: Path, rows: list[dict[str, Any]]) -> None: fields: list[str] = [] for row in rows: for key in row.keys(): if key not in fields: fields.append(key) with path.open("w", encoding="utf-8-sig", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=fields) writer.writeheader() writer.writerows(rows) def resolve_binary(configured: Any, name: str) -> str: if configured: path = Path(str(configured)).expanduser() if path.exists(): return str(path) found = shutil.which(name) if found: return found raise RuntimeError(f"{name} was not found. Install ffmpeg or set the path in config.") def probe_float(ffprobe: str, video_path: Path, field: str) -> float: result = subprocess.run( [ ffprobe, "-v", "error", "-show_entries", f"format={field}", "-of", "default=noprint_wrappers=1:nokey=1", str(video_path), ], capture_output=True, text=True, timeout=30, ) if result.returncode != 0: return 0.0 try: return float(result.stdout.strip()) except ValueError: return 0.0 def probe_fps(ffprobe: str, video_path: Path) -> float: result = subprocess.run( [ ffprobe, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=avg_frame_rate", "-of", "default=noprint_wrappers=1:nokey=1", str(video_path), ], capture_output=True, text=True, timeout=30, ) value = result.stdout.strip() if "/" in value: numerator, denominator = value.split("/", 1) try: return float(numerator) / max(float(denominator), 1.0) except ValueError: return 30.0 try: return float(value) except ValueError: return 30.0 def write_jpeg(path: Path, image: np.ndarray, quality: int) -> None: ok, encoded = cv2.imencode(".jpg", image, [int(cv2.IMWRITE_JPEG_QUALITY), quality]) if not ok: raise RuntimeError(f"Failed to encode JPEG: {path}") path.write_bytes(encoded.tobytes()) if __name__ == "__main__": raise SystemExit(main())