Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
| """ | |
| Memorability: predicted brand recall a day or more after viewing, from the film alone. | |
| What it is trained on. LAMBDA (behavior-in-the-wild/LAMBDA on Hugging Face, MIT): 2,183 real | |
| ads from 261 brands. 1,749 people watched ads on day one and were asked 24 to 72 hours later | |
| which brands they remembered seeing, then to describe the ad. The score for an ad is the share of | |
| its viewers who recalled the brand. Split-half agreement between two halves of the viewers is | |
| Spearman 0.77, so that is the ceiling any predictor can be judged against. | |
| What the film is reduced to, in channels that are each a ridge regression of their own and are | |
| then stacked (scripts/memorability/train.py chooses the weights on out-of-fold predictions): | |
| pictures one frame a second, 224px square, through CLIP ViT-L/14; the embeddings' mean and | |
| max over the film. What is in the picture, as the model that read the web sees it. | |
| words the Whisper transcript through a sentence embedding (all-MiniLM-L6-v2), plus how | |
| many words there are and whether the brand is said. A silent film is a legitimate | |
| state: about a third of the training ads have no speech. | |
| brand the brand's name through the same sentence embedding. What a language model knows | |
| about the name is the nearest thing to fame a film-only instrument can have. | |
| structure duration, cut rate (HSV histogram distance over 0.3, the paper's own rule), faces | |
| (YuNet), and statistics of PerceptCLIP-Memorability per frame (CLIP with a LoRA head | |
| fitted to LaMem, 60,000 photographs with recognition-memory scores). This channel has | |
| never earned a stacking weight; the per-frame curve is still returned for the strip. | |
| This file only applies the fitted model, in numpy, so the service needs no scikit-learn. | |
| What the number means, and does not. It is the model's estimate of the share of LAMBDA-style | |
| viewers who would recall the brand the next day, reported with the film's percentile among the | |
| 2,183 ads. LAMBDA viewers were asked to watch, so recall is far higher than in the wild; the | |
| percentile is the number to put weight on, the share is the scale it was fitted on. Each channel's | |
| own prediction is returned too, so a report can say what the pictures say, what the words say and | |
| what the name brings, and so a rewritten script can be scored against the same pictures. | |
| """ | |
| from __future__ import annotations | |
| import json, os, re, subprocess, importlib.util | |
| import numpy as np | |
| VERSION = "Memorability v0.2" | |
| FPS = 1 | |
| SIDE = 224 | |
| CUT_THRESHOLD = 0.30 | |
| TEXT_MODEL = "sentence-transformers/all-MiniLM-L6-v2" | |
| MODEL_JSON = os.path.join(os.path.dirname(os.path.abspath(__file__)), "memorability-model.json") | |
| # --------------------------------------------------------------------------------------------- | |
| # Frames | |
| # --------------------------------------------------------------------------------------------- | |
| def decode_frames(path: str, fps: int = FPS, side: int = SIDE, timeout: int = 600) -> np.ndarray | None: | |
| """Decode to (T, side, side, 3) RGB uint8: the short side resized to `side`, centre crop.""" | |
| vf = f"fps={fps},scale='if(gt(iw,ih),-2,{side})':'if(gt(iw,ih),{side},-2)',crop={side}:{side}" | |
| cmd = ["ffmpeg", "-loglevel", "error", "-i", path, "-vf", vf, "-f", "rawvideo", "-pix_fmt", "rgb24", "-"] | |
| try: | |
| out = subprocess.run(cmd, capture_output=True, timeout=timeout).stdout | |
| except subprocess.TimeoutExpired: | |
| return None | |
| n = len(out) // (side * side * 3) | |
| if n == 0: | |
| return None | |
| return np.frombuffer(out[: n * side * side * 3], dtype=np.uint8).reshape(n, side, side, 3) | |
| def _hsv_hist(frame_rgb: np.ndarray, cv2) -> np.ndarray: | |
| hsv = cv2.cvtColor(frame_rgb, cv2.COLOR_RGB2HSV) | |
| h = cv2.calcHist([hsv], [0, 1], None, [16, 8], [0, 180, 0, 256]).flatten() | |
| return h / max(1.0, h.sum()) | |
| def cuts(frames: np.ndarray, cv2) -> np.ndarray: | |
| """1 where a frame starts a new shot.""" | |
| out = np.zeros(len(frames), dtype=np.uint8) | |
| prev = None | |
| for i, f in enumerate(frames): | |
| h = _hsv_hist(f, cv2) | |
| if prev is not None and 0.5 * np.abs(h - prev).sum() > CUT_THRESHOLD: | |
| out[i] = 1 | |
| prev = h | |
| return out | |
| # --------------------------------------------------------------------------------------------- | |
| # Models that read the frames, and the one that reads the words | |
| # --------------------------------------------------------------------------------------------- | |
| class FrameModels: | |
| """CLIP ViT-L/14, PerceptCLIP-Memorability and YuNet, loaded once.""" | |
| def __init__(self, device: str | None = None, yunet_path: str | None = None): | |
| import torch | |
| from huggingface_hub import hf_hub_download | |
| from transformers import CLIPModel | |
| import cv2 | |
| self.torch, self.cv2 = torch, cv2 | |
| self.device = device or ("cuda" if torch.cuda.is_available() else "cpu") | |
| cp = hf_hub_download("PerceptCLIP/PerceptCLIP_Memorability", "modeling.py") | |
| spec = importlib.util.spec_from_file_location("perceptclip_modeling", cp) | |
| mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod) | |
| self.mem = mod.clip_lora_model().to(self.device) | |
| self.mem.load_state_dict(torch.load(hf_hub_download("PerceptCLIP/PerceptCLIP_Memorability", "perceptCLIP_Memorability.pth"), | |
| map_location=self.device)) | |
| self.mem.eval() | |
| self.clip = CLIPModel.from_pretrained("openai/clip-vit-large-patch14").to(self.device).eval() | |
| self.mean = torch.tensor((0.48145466, 0.4578275, 0.40821073), device=self.device).view(1, 3, 1, 1) | |
| self.std = torch.tensor((0.26862954, 0.26130258, 0.27577711), device=self.device).view(1, 3, 1, 1) | |
| self.det = cv2.FaceDetectorYN.create(yunet_path, "", (SIDE, SIDE), 0.6, 0.3, 5000) if yunet_path and os.path.exists(yunet_path) else None | |
| def run(self, frames: np.ndarray, batch: int = 16): | |
| torch = self.torch | |
| x_all = torch.from_numpy(np.ascontiguousarray(frames)).permute(0, 3, 1, 2).float().div_(255.0) | |
| clip_out, mem_out = [], [] | |
| with torch.no_grad(): | |
| for i in range(0, len(x_all), batch): | |
| x = (x_all[i:i + batch].to(self.device) - self.mean) / self.std | |
| mem_out.append(self.mem(x).squeeze(-1).float().cpu().numpy()) | |
| # Spelled out rather than get_image_features(): that helper returns a tensor in | |
| # transformers 4.x and an output object in 5.x, and the Space took 5.x. This is the | |
| # same arithmetic in both: the pooled vision output through the visual projection. | |
| pooled = self.clip.vision_model(pixel_values=x).pooler_output | |
| e = self.clip.visual_projection(pooled) | |
| e = e / e.norm(dim=-1, keepdim=True) | |
| clip_out.append(e.float().cpu().numpy()) | |
| faces = np.zeros(len(frames), dtype=np.int16) | |
| if self.det is not None: | |
| for i, f in enumerate(frames): | |
| _, det = self.det.detect(self.cv2.cvtColor(f, self.cv2.COLOR_RGB2BGR)) | |
| faces[i] = 0 if det is None else len(det) | |
| return np.concatenate(clip_out), np.concatenate(mem_out), faces | |
| class TextModel: | |
| """The sentence embedding for the words and brand channels, loaded once.""" | |
| def __init__(self, device: str | None = None): | |
| from sentence_transformers import SentenceTransformer | |
| self.model = SentenceTransformer(TEXT_MODEL, device=device) | |
| def embed(self, text: str) -> np.ndarray: | |
| v = self.model.encode([text or ""], normalize_embeddings=True, show_progress_bar=False)[0] | |
| return np.asarray(v, dtype=np.float32) | |
| # --------------------------------------------------------------------------------------------- | |
| # Per-film features | |
| # --------------------------------------------------------------------------------------------- | |
| def film_features(clip: np.ndarray, mem: np.ndarray, cut: np.ndarray, faces: np.ndarray, duration: float): | |
| """Returns (pictures vector 1536, structure vector, structure dict) for one film.""" | |
| clip = clip.astype(np.float32); mem = mem.astype(np.float32) | |
| cut = cut.astype(np.float32); faces = faces.astype(np.float32) | |
| T = len(mem); dur = float(duration) | |
| head = slice(0, min(3, T)); tail = slice(max(0, T - 3), T) | |
| sem = np.concatenate([clip.mean(0), clip.max(0)]) | |
| struct = { | |
| "duration": dur, "log_duration": float(np.log1p(dur)), | |
| "mem_mean": float(mem.mean()), "mem_max": float(mem.max()), "mem_min": float(mem.min()), "mem_std": float(mem.std()), | |
| "mem_head": float(mem[head].mean()), "mem_tail": float(mem[tail].mean()), | |
| "mem_top3": float(np.sort(mem)[-3:].mean()), | |
| "cut_rate": float(cut.sum() / max(1.0, dur)), "cuts": float(cut.sum()), | |
| "faces_share": float((faces > 0).mean()), "faces_mean": float(faces.mean()), "multi_share": float((faces >= 2).mean()), | |
| "faces_head": float((faces[head] > 0).mean()), "faces_tail": float((faces[tail] > 0).mean()), | |
| "sem_motion": float(1 - (clip[1:] * clip[:-1]).sum(1).mean()) if T > 1 else 0.0, | |
| } | |
| return sem, np.array(list(struct.values()), dtype=np.float32), struct | |
| STRUCT_NAMES = list(film_features(np.zeros((2, 768), np.float32), np.zeros(2, np.float32), np.zeros(2), np.zeros(2), 2.0)[2].keys()) | |
| TEXT_SCALAR_NAMES = ["log_words", "has_words", "brand_said"] | |
| def _norm_words(s: str) -> str: | |
| return re.sub(r"[^a-z0-9]+", " ", (s or "").lower()).strip() | |
| def brand_is_said(transcript: str, brand: str) -> bool: | |
| """ | |
| Whether the name is in the words, however it is spelt. "Go.Compare", "Go Compare" and | |
| "gocompare" are one name; Whisper writes whichever it likes, and the client types another. | |
| """ | |
| b = _norm_words(brand); t = _norm_words(transcript) | |
| if not b or not t: | |
| return False | |
| if b in t or b.replace(" ", "") in t.replace(" ", ""): | |
| return True | |
| # A multi-word name counts when its longest word is said ("Dick's Sporting Goods" as "Dick's"). | |
| longest = max(b.split(), key=len) | |
| return len(longest) >= 4 and re.search(rf"\b{re.escape(longest)}\b", t) is not None | |
| def text_features(text_model: TextModel, transcript: str, brand: str | None): | |
| """Returns (words vector 384 + scalars, brand vector 384).""" | |
| transcript = (transcript or "").strip() | |
| n_words = len(transcript.split()) | |
| words = np.concatenate([text_model.embed(transcript), | |
| np.array([np.log1p(n_words), 1.0 if n_words else 0.0, 1.0 if brand_is_said(transcript, brand or "") else 0.0], np.float32)]) | |
| brand_vec = text_model.embed(f"{brand.strip()} advert" if brand and brand.strip() else "an advert for an unknown brand") | |
| return words.astype(np.float32), brand_vec.astype(np.float32) | |
| # --------------------------------------------------------------------------------------------- | |
| # The fitted model, applied without scikit-learn | |
| # --------------------------------------------------------------------------------------------- | |
| class Predictor: | |
| def __init__(self, path: str = MODEL_JSON): | |
| with open(path, encoding="utf-8") as f: | |
| m = json.load(f) | |
| self.version = m.get("version", VERSION) | |
| self.channels = {c: {"coef": np.array(v["coef"], dtype=np.float64), "intercept": float(v["intercept"]), "weight": float(v["weight"])} | |
| for c, v in m["channels"].items()} | |
| self.stack_intercept = float(m["stack_intercept"]) | |
| self.struct_names = m["struct_names"] | |
| self.reference = np.array(sorted(m["reference_predictions"]), dtype=np.float64) | |
| self.breakpoints = {c: np.array(v, dtype=np.float64) for c, v in (m.get("channel_breakpoints") or {}).items()} | |
| self.reference_recall = np.array(sorted(m["reference_recall"]), dtype=np.float64) | |
| self.test_spearman = m.get("test_spearman") | |
| self.unseen_brand_spearman = m.get("unseen_brand_spearman") | |
| self.n_train = m.get("n_train") | |
| def channel_prediction(self, name: str, x: np.ndarray) -> float: | |
| c = self.channels[name] | |
| return float(c["intercept"] + x.astype(np.float64) @ c["coef"]) | |
| def needs_text(self) -> bool: | |
| return "words" in self.channels or "brand" in self.channels | |
| def channel_place(self, name: str, value: float) -> float | None: | |
| """Where a channel's reading falls among the reference ads, 0 to 100. High is better, as | |
| everywhere else: 70 means this channel reads the film above 70% of the ads the model was | |
| fitted on. Without breakpoints for the channel there is no place to report.""" | |
| b = self.breakpoints.get(name) | |
| if b is None or not len(b): | |
| return None | |
| return float(np.clip(np.searchsorted(b, value, side="right") - 1, 0, 100)) | |
| def combine(self, channel_preds: dict) -> dict: | |
| score = self.stack_intercept + sum(self.channels[c]["weight"] * p for c, p in channel_preds.items() if c in self.channels) | |
| score = float(np.clip(score, 0.0, 1.0)) | |
| pct = float(100.0 * np.searchsorted(self.reference, score, side="right") / max(1, len(self.reference))) | |
| return {"score": score, "percentile": pct} | |
| def predict(self, features: dict) -> dict: | |
| """features: {channel: vector}. Returns score, percentile and each channel's own prediction.""" | |
| preds = {c: self.channel_prediction(c, x) for c, x in features.items() if c in self.channels} | |
| out = self.combine(preds) | |
| out["channels"] = {c: round(p, 3) for c, p in preds.items()} | |
| out["channel_places"] = {c: v for c, v in ((c, self.channel_place(c, p)) for c, p in preds.items()) if v is not None} | |
| out["weights"] = {c: round(self.channels[c]["weight"], 3) for c in preds} | |
| return out | |
| # --------------------------------------------------------------------------------------------- | |
| # For the service | |
| # --------------------------------------------------------------------------------------------- | |
| def band_for(pct: float) -> str: | |
| return "Strong" if pct >= 75 else "Good" if pct >= 50 else "Watch" if pct >= 25 else "Act" | |
| def _finish(predictor: Predictor, p: dict) -> dict: | |
| return { | |
| "version": predictor.version, | |
| "score": round(p["score"], 3), | |
| "percentile": round(p["percentile"], 1), | |
| "band": band_for(p["percentile"]), | |
| "channels": p.get("channels", {}), | |
| # The same readings as `channels`, each placed out of 100 among the reference ads, so a | |
| # report can print them beside the film's own number without changing scale mid-sentence. | |
| "channel_places": p.get("channel_places", {}), | |
| "weights": p.get("weights", {}), | |
| "reference": {"n": int(len(predictor.reference)), "recall_median": round(float(np.median(predictor.reference_recall)), 3), | |
| # The median of the model's own predictions. A prediction is compared with | |
| # this, never with the median of what viewers actually recalled: a fitted | |
| # model pulls towards the middle, so the two medians are not the same ruler. | |
| "prediction_median": round(float(np.median(predictor.reference)), 3), | |
| "test_spearman": predictor.test_spearman, "unseen_brand_spearman": predictor.unseen_brand_spearman}, | |
| } | |
| def read_memorability(video_path: str, models: FrameModels, predictor: Predictor, duration: float | None = None, | |
| transcript: str | None = None, brand: str | None = None, text_model: TextModel | None = None) -> dict: | |
| frames = decode_frames(video_path) | |
| if frames is None or len(frames) < 2: | |
| return {"version": predictor.version, "error": "no frames decoded"} | |
| clip, mem, faces = models.run(frames) | |
| cut = cuts(frames, models.cv2) | |
| dur = float(duration) if duration else float(len(frames)) | |
| sem, sv, sd = film_features(clip, mem, cut, faces, dur) | |
| feats = {"pictures": sem, "structure": sv} | |
| if predictor.needs_text(): | |
| if text_model is None: | |
| return {"version": predictor.version, "error": "text model missing"} | |
| words, brand_vec = text_features(text_model, transcript or "", brand) | |
| feats["words"] = words; feats["brand"] = brand_vec | |
| out = _finish(predictor, predictor.predict(feats)) | |
| out.update({ | |
| "brand_said": brand_is_said(transcript or "", brand or ""), | |
| "words": len((transcript or "").split()), | |
| # Picture memorability second by second, for the strip and the moment cards. | |
| "per_second": [{"time": float(i), "mem": round(float(v), 3), "cut": int(cut[i]), "faces": int(faces[i])} for i, v in enumerate(mem)], | |
| "features": {k: round(v, 4) for k, v in sd.items()}, | |
| }) | |
| return out | |
| def score_words(predictor: Predictor, text_model: TextModel, transcript: str, brand: str | None, channels: dict) -> dict: | |
| """ | |
| Re-score a film with different words and the same pictures: `channels` is the film's earlier | |
| per-channel prediction (from read_memorability), of which only "words" is recomputed. For | |
| scoring a rewritten script against the film as shot. | |
| """ | |
| words, _ = text_features(text_model, transcript, brand) | |
| preds = {c: float(p) for c, p in channels.items() if c in predictor.channels and c != "words"} | |
| preds["words"] = predictor.channel_prediction("words", words) | |
| out = _finish(predictor, {**predictor.combine(preds), "channels": {c: round(p, 3) for c, p in preds.items()}, | |
| "channel_places": {c: v for c, v in ((c, predictor.channel_place(c, p)) for c, p in preds.items()) if v is not None}, | |
| "weights": {c: round(predictor.channels[c]["weight"], 3) for c in preds}}) | |
| out["brand_said"] = brand_is_said(transcript, brand or ""); out["words"] = len(transcript.split()) | |
| return out | |