Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
| # Video Analysis API v4.2 | |
| # Whisper transcription + emotion2vec voice emotion + HSEmotion facial emotions | |
| import os | |
| import base64 | |
| import math | |
| import tempfile | |
| import subprocess | |
| import shutil | |
| import threading | |
| from concurrent.futures import ThreadPoolExecutor | |
| from fastapi import FastAPI, UploadFile, File, Query | |
| from fastapi.middleware.cors import CORSMiddleware | |
| import torch | |
| import service_auth | |
| import numpy as np | |
| print("=" * 50, flush=True) | |
| print(f"[STARTUP] CUDA available: {torch.cuda.is_available()}", flush=True) | |
| if torch.cuda.is_available(): | |
| print(f"[STARTUP] GPU: {torch.cuda.get_device_name(0)}", flush=True) | |
| print("=" * 50, flush=True) | |
| # No interactive docs: the API is for the EchoDepth app only (service_auth.py) | |
| app = FastAPI(title="EchoDepth Video Analysis API", docs_url=None, redoc_url=None, openapi_url=None) | |
| # Pass check first, CORS added after so it wraps it: a refusal still reaches the browser readably | |
| app.middleware("http")(service_auth.require_pass) | |
| app.add_middleware(CORSMiddleware, allow_origins=service_auth.ALLOWED_ORIGINS, allow_credentials=False, allow_methods=["GET", "POST"], allow_headers=["*"]) | |
| print(f"[STARTUP] Service passes: {'enforced' if service_auth.enforcing() else 'OFF - set VIDEO_SERVICE_SECRET'}", flush=True) | |
| # Lazy load models. Loads take the lock, so the start-up warm-up thread and a first request | |
| # arriving meanwhile never load the same model twice. | |
| _LOAD = threading.RLock() | |
| # One analysis at a time. Requests are answered from a thread pool now (the endpoints are plain | |
| # def), so /health and the job polls stay responsive during an analysis; this keeps two films from | |
| # sharing the GPU and from mixing their per-request entries in MODEL_ERRORS. | |
| _ANALYSE = threading.Lock() | |
| # Entries in MODEL_ERRORS that describe one request rather than a model that failed to load. | |
| # Cleared at the start of each analysis: before, one failed CEI pass was reported on every | |
| # analysis after it until the service restarted. | |
| _PER_REQUEST_ERRORS = ("cei", "stills", "proxy", "decode") | |
| def _loaded_once(fn): | |
| """Run a model getter under the load lock.""" | |
| def wrapper(*args, **kwargs): | |
| with _LOAD: | |
| return fn(*args, **kwargs) | |
| wrapper.__name__ = fn.__name__ | |
| wrapper.__doc__ = fn.__doc__ | |
| return wrapper | |
| whisper_model = None | |
| emotion2vec_model = None | |
| face_emotion_model = None | |
| # Why a model did not load, keyed by analyser. An optional model that fails to import used to be | |
| # swallowed here and the request still answered 200 OK, so a service missing funasr returned a | |
| # complete-looking analysis with the voice channel silently absent — and downstream, congruence | |
| # defaulted rather than reporting itself unmeasured. Every response now says what actually ran. | |
| MODEL_ERRORS: dict = {} | |
| # emotion2vec labels (9 classes) | |
| EMOTION2VEC_LABELS = ['angry', 'disgust', 'fearful', 'happy', 'neutral', 'other', 'sad', 'surprised', 'unknown'] | |
| # Normalize emotion labels from emotion2vec (handles Chinese/mixed labels) | |
| EMOTION_LABEL_MAP = { | |
| '中立/neutral': 'neutral', '中立': 'neutral', 'neutral': 'neutral', 'neutra': 'neutral', | |
| '开心/happy': 'happy', '开心': 'happy', 'happy': 'happy', | |
| '生气/angry': 'angry', '生气': 'angry', 'angry': 'angry', | |
| '难过/sad': 'sad', '难过': 'sad', 'sad': 'sad', | |
| '恐惧/fear': 'fear', '恐惧': 'fear', 'fear': 'fear', 'fearful': 'fear', | |
| '惊讶/surprise': 'surprise', '惊讶': 'surprise', 'surprise': 'surprise', 'surprised': 'surprise', | |
| '厌恶/disgust': 'disgust', '厌恶': 'disgust', 'disgust': 'disgust', | |
| '其他/other': 'other', '其他': 'other', 'other': 'other', | |
| '<unk>': 'unknown', 'unknown': 'unknown', 'unk': 'unknown', | |
| } | |
| def normalize_emotion_label(label): | |
| """Normalize emotion label to clean English. | |
| emotion2vec_plus_large emits '吃惊/surprised', '恐惧/fearful' and '厌恶/disgusted', none of which | |
| were in the map, so three of its nine classes reached the app untranslated. Anything the map | |
| misses now falls back to the English half after the slash, mapped again. | |
| """ | |
| label_lower = label.lower().strip() | |
| if label_lower in EMOTION_LABEL_MAP: | |
| return EMOTION_LABEL_MAP[label_lower] | |
| english = label_lower.split('/')[-1].strip() | |
| return EMOTION_LABEL_MAP.get(english, {'disgusted': 'disgust'}.get(english, english)) | |
| whisper_fast_model = None | |
| def get_whisper_fast(): | |
| """Whisper small, for short spoken answers people correct before sending (Reflections). | |
| Several times quicker than medium on this CPU. Leave cpu_threads at its default: os.cpu_count() | |
| reports the host's cores, not the Space's two, and oversubscribing them made it six times slower. | |
| The brand-name misses that ruled small out for | |
| ads do not matter here: every transcript is shown to its speaker to correct first. | |
| """ | |
| global whisper_fast_model | |
| if whisper_fast_model is None: | |
| from faster_whisper import WhisperModel | |
| device = "cuda" if torch.cuda.is_available() else "cpu" | |
| compute = "float16" if device == "cuda" else "int8" | |
| print(f"[whisper] Loading small (fast profile) on {device}", flush=True) | |
| whisper_fast_model = WhisperModel("small", device=device, compute_type=compute) | |
| return whisper_fast_model | |
| def get_whisper(): | |
| global whisper_model | |
| if whisper_model is None: | |
| from faster_whisper import WhisperModel | |
| device = "cuda" if torch.cuda.is_available() else "cpu" | |
| compute = "float16" if device == "cuda" else "int8" | |
| # "medium" everywhere. "small" dropped a sung brand name at the end of a 30-second film | |
| # ("making smart moves with..." for "with Go.Compare") whatever it was prompted with; | |
| # medium with the brand as a hint hears it. The memorability words channel was fitted to | |
| # "small" transcripts of the training ads (2026-09-21); a medium re-run is the next cut. | |
| model_size = "medium" | |
| print(f"[whisper] Loading {model_size} on {device}", flush=True) | |
| whisper_model = WhisperModel(model_size, device=device, compute_type=compute) | |
| return whisper_model | |
| def get_emotion2vec(): | |
| """Load emotion2vec model for voice emotion recognition""" | |
| global emotion2vec_model | |
| if emotion2vec_model is None: | |
| try: | |
| from funasr import AutoModel | |
| print("[voice] Loading emotion2vec model...", flush=True) | |
| # Use HuggingFace model name (emotion2vec/) not ModelScope (iic/) | |
| emotion2vec_model = AutoModel( | |
| model="emotion2vec/emotion2vec_plus_large", | |
| hub="hf", | |
| device="cuda" if torch.cuda.is_available() else "cpu" | |
| ) | |
| print("[voice] emotion2vec loaded successfully", flush=True) | |
| except Exception as e: | |
| print(f"[voice] Failed to load emotion2vec: {e}", flush=True) | |
| MODEL_ERRORS["voice_emotion"] = f"{type(e).__name__}: {e}" | |
| emotion2vec_model = "failed" | |
| return emotion2vec_model if emotion2vec_model != "failed" else None | |
| def get_face_emotion(): | |
| """Load HSEmotion ONNX model for facial emotion recognition""" | |
| global face_emotion_model | |
| if face_emotion_model is None: | |
| try: | |
| import onnxruntime as ort | |
| from hsemotion_onnx.facial_emotions import HSEmotionRecognizer, get_model_path | |
| print("[face] Loading HSEmotion ONNX...", flush=True) | |
| face_emotion_model = HSEmotionRecognizer(model_name='enet_b0_8_best_afew') | |
| # The library opens its session on the CPU provider only. With onnxruntime-gpu in the | |
| # image, reopen it on CUDA; the model and its outputs are the same, only where it runs. | |
| # Checked against torch too: onnxruntime-gpu lists the CUDA provider on a machine with | |
| # no GPU (the CPU fallback Space runs this same image). | |
| if torch.cuda.is_available() and "CUDAExecutionProvider" in ort.get_available_providers(): | |
| face_emotion_model.ort_session = ort.InferenceSession( | |
| get_model_path('enet_b0_8_best_afew'), providers=["CUDAExecutionProvider", "CPUExecutionProvider"]) | |
| print(f"[face] HSEmotion ONNX loaded on {face_emotion_model.ort_session.get_providers()[0]}", flush=True) | |
| except Exception as e: | |
| import traceback | |
| print(f"[face] Failed to load HSEmotion ONNX: {e}", flush=True) | |
| print(f"[face] Traceback: {traceback.format_exc()}", flush=True) | |
| MODEL_ERRORS["face_emotion"] = f"{type(e).__name__}: {e}" | |
| face_emotion_model = "failed" | |
| return face_emotion_model if face_emotion_model != "failed" else None | |
| ocr_reader = None | |
| def get_ocr(): | |
| """EasyOCR reader, loaded once. Optional like the emotion models: a service without it still | |
| answers, and says so, rather than pretending the film carried no on-screen copy.""" | |
| global ocr_reader | |
| if ocr_reader is None: | |
| try: | |
| import easyocr | |
| print("[ocr] Loading EasyOCR...", flush=True) | |
| # On the GPU when there is one. It was pinned to the CPU when the Space had six | |
| # gigabytes of VRAM, where adding it beside Whisper, emotion2vec and HSEmotion killed | |
| # the service mid-request; the GPUs it runs on now have 24 GB or more, and OCR on full | |
| # frames was the slowest pass on the CPU. | |
| ocr_reader = easyocr.Reader(["en"], gpu=torch.cuda.is_available(), verbose=False) | |
| print(f"[ocr] EasyOCR loaded on {'cuda' if torch.cuda.is_available() else 'cpu'}", flush=True) | |
| except Exception as e: | |
| print(f"[ocr] Failed to load EasyOCR: {e}", flush=True) | |
| MODEL_ERRORS["screen_text"] = f"{type(e).__name__}: {e}" | |
| ocr_reader = "failed" | |
| return ocr_reader if ocr_reader != "failed" else None | |
| class _TextPass: | |
| """Words on screen, at fps frames a second of the film. See analyze_screen_text.""" | |
| def __init__(self, video_fps, fps=1.0, min_conf=0.4): | |
| self.reader = get_ocr() | |
| self.interval = max(1, int(video_fps / fps)) | |
| self.video_fps = video_fps | |
| self.min_conf = min_conf | |
| self.out = [] | |
| def wants(self, frame_num): | |
| return self.reader is not None and frame_num % self.interval == 0 | |
| def add(self, frame, frame_num): | |
| t = frame_num / self.video_fps | |
| try: | |
| found = self.reader.readtext(frame) | |
| except Exception as e: | |
| print(f"[ocr] frame {t:.1f}s failed: {e}", flush=True) | |
| found = [] | |
| words = [ | |
| {"text": txt.strip(), "confidence": round(float(conf), 3)} | |
| for _box, txt, conf in found | |
| if conf >= self.min_conf and txt.strip() | |
| ] | |
| if words: | |
| self.out.append({"time": round(t, 2), "words": words, | |
| "text": " ".join(w["text"] for w in words)}) | |
| def finish(self): | |
| return self.out | |
| def analyze_screen_text(video_path, fps=1.0, min_conf=0.4): | |
| """Read the words on screen, sampled through the film. | |
| An advert with no voiceover is not an advert with no message: the message is in the supers and | |
| the end frame. Reading only the audio channel and concluding "nothing was said" is how a | |
| perfectly clear piece of copy came back unscored. | |
| """ | |
| return scan_frames(video_path, screen_text=True, faces=False, ocr_fps=fps, ocr_min_conf=min_conf)[0] | |
| def brand_presence(text_timeline, transcript, duration, brand): | |
| """When the brand is actually present, and where. | |
| Ehrenberg-Bass: an advert nobody attributes is an advert that sells someone else's product. | |
| This is the measurable half of that — first appearance, share of the film it is on screen for, | |
| whether it is there in the opening seconds, and whether it closes the film. | |
| """ | |
| if not brand or duration <= 0: | |
| return None | |
| # Comma-separated aliases, because a brand is rarely one string on screen: this campaign | |
| # carries WRU, Wales and Cymru, and matching only the legal name would have reported the | |
| # brand absent from a film that shows it four seconds in. | |
| aliases = [a.strip().lower() for a in brand.split(",") if a.strip()] | |
| if not aliases: | |
| return None | |
| tokens = [] | |
| for a in aliases: | |
| parts = [t for t in a.replace("&", " ").split() if len(t) >= 3] | |
| tokens.extend(parts or [a]) | |
| if not tokens: | |
| return None | |
| def _close(a, b, budget): | |
| """Levenshtein within a small budget — enough for one or two misread characters.""" | |
| if abs(len(a) - len(b)) > budget: | |
| return False | |
| prev = list(range(len(b) + 1)) | |
| for i, ca in enumerate(a, 1): | |
| cur = [i] | |
| for j, cb in enumerate(b, 1): | |
| cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb))) | |
| if min(cur) > budget: | |
| return False | |
| prev = cur | |
| return prev[-1] <= budget | |
| def hit(text): | |
| low = (text or "").lower() | |
| if any(t in low for t in tokens): | |
| return True | |
| # Brand names are set in display type, which is where OCR slips: this campaign's CYMRU | |
| # came back as GYMRU and GIMRU. An exact match would have reported the brand absent from | |
| # frames it is plainly on, which is the opposite of the finding. | |
| words = [w.strip(".,:;!?/()-") for w in low.split()] | |
| for t in tokens: | |
| budget = 1 if len(t) <= 5 else 2 | |
| if any(_close(t, w, budget) for w in words if abs(len(w) - len(t)) <= budget): | |
| return True | |
| return False | |
| frames = [f for f in text_timeline if hit(f.get("text"))] | |
| sampled = len(text_timeline) or 1 | |
| first = min((f["time"] for f in frames), default=None) | |
| last = max((f["time"] for f in frames), default=None) | |
| return { | |
| "brand": brand, | |
| "on_screen_frames": len(frames), | |
| "sampled_frames": len(text_timeline), | |
| "share_of_frames": round(100 * len(frames) / sampled, 1) if text_timeline else None, | |
| "first_seen_s": first, | |
| "last_seen_s": last, | |
| # The two placements that carry attribution: the hook, and the end frame nobody skips past. | |
| "in_opening_3s": bool(first is not None and first <= 3), | |
| "in_closing_3s": bool(last is not None and last >= duration - 3), | |
| "spoken": hit(transcript), | |
| } | |
| text_emotion_model = None | |
| # GoEmotions label -> (valence, arousal, dominance), 0-1. Same table the Stage Read pipeline uses, | |
| # so a line of ad copy and a line of a speech are scored on one scale. | |
| TEXT_VAD = { | |
| "admiration": (0.80, 0.50, 0.55), "amusement": (0.80, 0.60, 0.60), "anger": (0.15, 0.85, 0.75), | |
| "annoyance": (0.25, 0.60, 0.55), "approval": (0.75, 0.40, 0.60), "caring": (0.75, 0.40, 0.55), | |
| "confusion": (0.35, 0.50, 0.30), "curiosity": (0.65, 0.55, 0.50), "desire": (0.70, 0.65, 0.55), | |
| "disappointment": (0.20, 0.40, 0.30), "disapproval": (0.25, 0.50, 0.55), "disgust": (0.15, 0.60, 0.55), | |
| "embarrassment": (0.20, 0.55, 0.20), "excitement": (0.85, 0.85, 0.65), "fear": (0.15, 0.80, 0.20), | |
| "gratitude": (0.85, 0.45, 0.50), "grief": (0.10, 0.45, 0.20), "joy": (0.90, 0.70, 0.65), | |
| "love": (0.90, 0.60, 0.60), "nervousness": (0.25, 0.75, 0.25), "optimism": (0.80, 0.55, 0.65), | |
| "pride": (0.80, 0.60, 0.75), "realization": (0.55, 0.45, 0.50), "relief": (0.75, 0.30, 0.55), | |
| "remorse": (0.15, 0.40, 0.20), "sadness": (0.10, 0.30, 0.20), "surprise": (0.60, 0.80, 0.40), | |
| "neutral": (0.50, 0.30, 0.50), | |
| } | |
| def get_text_emotion(): | |
| global text_emotion_model | |
| if text_emotion_model is None: | |
| try: | |
| from transformers import pipeline as hf_pipeline | |
| print("[text] Loading GoEmotions...", flush=True) | |
| text_emotion_model = hf_pipeline("text-classification", | |
| model="SamLowe/roberta-base-go_emotions", | |
| top_k=None, device=0 if torch.cuda.is_available() else -1) | |
| print("[text] GoEmotions loaded", flush=True) | |
| except Exception as e: | |
| print(f"[text] Failed to load GoEmotions: {e}", flush=True) | |
| MODEL_ERRORS["copy_emotion"] = f"{type(e).__name__}: {e}" | |
| text_emotion_model = "failed" | |
| return text_emotion_model if text_emotion_model != "failed" else None | |
| def _normalise_case(text): | |
| """Set-in-caps copy to sentence case before scoring. | |
| The model reads capitals as shouting: the identical sentence scores 95 in capitals and 13 in | |
| sentence case. End frames are nearly always set in caps and a rewritten line arrives in | |
| sentence case, so without this every original beats every alternative on typesetting alone — | |
| which is the opposite of the comparison the report claims to be making. | |
| """ | |
| letters = [c for c in text if c.isalpha()] | |
| if letters and sum(c.isupper() for c in letters) / len(letters) > 0.7: | |
| lowered = text.lower() | |
| return lowered[:1].upper() + lowered[1:] | |
| return text | |
| def score_lines(lines): | |
| """Emotional content of each line of copy. | |
| `feel` is 1 minus the neutral score: how much emotional work the words are doing at all. It is | |
| the number that matters for advertising, because the failure mode is not being disliked, it is | |
| being ignored — and a line that reads as pure neutral has nothing for anyone to react to. | |
| """ | |
| model = get_text_emotion() | |
| if not model or not lines: | |
| return [] | |
| out = [] | |
| for line in lines: | |
| text = (line or "").strip() | |
| if len(text) < 3: | |
| continue | |
| try: | |
| scores = {d["label"]: float(d["score"]) for d in model(_normalise_case(text)[:500])[0]} | |
| except Exception as e: | |
| print(f"[text] scoring failed: {e}", flush=True) | |
| continue | |
| total = sum(scores.values()) or 1e-9 | |
| vad = [sum(TEXT_VAD.get(k, (0.5, 0.3, 0.5))[i] * v for k, v in scores.items()) / total for i in range(3)] | |
| top = sorted(((k, v) for k, v in scores.items() if k != "neutral"), key=lambda kv: -kv[1])[:3] | |
| out.append({ | |
| "text": text, | |
| "feel": round(1 - scores.get("neutral", 0.0), 3), | |
| "valence": round(vad[0], 3), | |
| "arousal": round(vad[1], 3), | |
| "dominance": round(vad[2], 3), | |
| "top": [{"emotion": k, "score": round(v, 3)} for k, v in top], | |
| }) | |
| return out | |
| YUNET_MODEL = os.path.join(os.path.dirname(os.path.abspath(__file__)), "stage_read", "models", "face_detection_yunet_2023mar.onnx") | |
| # Frames wider than this are downscaled before detection: YuNet is a small network and 1080p buys | |
| # nothing over 960 wide for faces that are in shot at all, at four times the cost. | |
| YUNET_MAX_WIDTH = 960 | |
| _MEMORABILITY = None | |
| try: | |
| import memorability as memorability_mod | |
| except Exception as e: # the service still starts without it; the pass then reports why | |
| memorability_mod = None | |
| MODEL_ERRORS["memorability"] = f"{type(e).__name__}: {e}" | |
| def get_memorability(): | |
| """CLIP, PerceptCLIP-Memorability and the fitted LAMBDA model, loaded on first use (about 3 GB of weights).""" | |
| global _MEMORABILITY | |
| if _MEMORABILITY is None: | |
| if memorability_mod is None: | |
| raise RuntimeError(MODEL_ERRORS.get("memorability", "memorability module missing")) | |
| yunet = os.path.join(os.path.dirname(os.path.abspath(__file__)), "stage_read", "models", "face_detection_yunet_2023mar.onnx") | |
| predictor = memorability_mod.Predictor() | |
| text_model = memorability_mod.TextModel() if predictor.needs_text() else None | |
| _MEMORABILITY = (memorability_mod.FrameModels(None, yunet), predictor, text_model) | |
| return _MEMORABILITY | |
| def get_face_detector(): | |
| """The detector behind the four-frames-a-second emotion pass. | |
| YuNet where its model is present (the Dockerfile fetches it for Stage Read), the Haar cascade | |
| otherwise. The cascade found no face on 43% of frames of a film that has a person in shot for | |
| most of it: it wants a frontal face of a decent size and loses profiles, small faces and motion | |
| blur, and the second-by-second chart came out as a comb. YuNet handles all three. | |
| This pass is not part of the Creative Effectiveness Index. The index's own two-a-second pass in | |
| cei_measures.py keeps the cascade, because the corpus was measured with it and must stay so. | |
| """ | |
| import cv2 | |
| if os.path.exists(YUNET_MODEL): | |
| try: | |
| det = cv2.FaceDetectorYN.create(YUNET_MODEL, "", (320, 320), score_threshold=0.6, nms_threshold=0.3, top_k=50) | |
| return ("yunet", det) | |
| except Exception as e: | |
| print(f"[face] YuNet unavailable ({e}); using the Haar cascade", flush=True) | |
| return ("haar", cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')) | |
| def detect_faces(detector, frame, gray): | |
| """(x, y, w, h) boxes in the frame's own pixels, from whichever detector get_face_detector chose.""" | |
| import cv2 | |
| kind, det = detector | |
| if kind == "haar": | |
| return [tuple(int(v) for v in f) for f in det.detectMultiScale(gray, 1.1, 5, minSize=(60, 60))] | |
| h, w = frame.shape[:2] | |
| scale = min(1.0, YUNET_MAX_WIDTH / max(w, 1)) | |
| small = cv2.resize(frame, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_AREA) if scale < 1 else frame | |
| det.setInputSize((small.shape[1], small.shape[0])) | |
| _, faces = det.detect(small) | |
| out = [] | |
| if faces is not None: | |
| for f in faces: | |
| x, y, fw, fh = (float(v) / scale for v in f[:4]) | |
| if fw >= 24 and fh >= 24: | |
| out.append((max(0, int(x)), max(0, int(y)), int(fw), int(fh))) | |
| return out | |
| def extract_audio(video_path, audio_path): | |
| """Returns None on success, or a message explaining the failure. | |
| The bare `except: return False` this replaced turned every cause — a corrupt upload, a missing | |
| ffmpeg, a process that could not start at all — into the same unactionable "Failed to extract | |
| audio" at the API boundary. | |
| """ | |
| from ffmpeg_util import run_ffmpeg, FFmpegError | |
| try: | |
| run_ffmpeg(["-i", video_path, "-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", audio_path]) | |
| return None | |
| except FFmpegError as e: | |
| print(f"[audio] {e}", flush=True) | |
| return str(e) | |
| except Exception as e: | |
| print(f"[audio] {type(e).__name__}: {e}", flush=True) | |
| return f"{type(e).__name__}: {e}" | |
| def transcribe(audio_path, brand=None): | |
| """ | |
| With a brand, its name goes to Whisper as the initial prompt, which biases recognition toward | |
| the term; a brand name that is sung, or unusual, is otherwise the word most likely to be lost. | |
| """ | |
| model = get_whisper() | |
| prompt = f"{brand.split(',')[0].strip()}." if brand and brand.strip() else None | |
| segments, info = model.transcribe(audio_path, beam_size=5, language="en", initial_prompt=prompt) | |
| return [{"start": s.start, "end": s.end, "text": s.text.strip(), "confidence": round(math.exp(s.avg_logprob), 4)} for s in segments], info.language | |
| def _voice_result(result): | |
| """emotion2vec's output for one clip, as the response carries it.""" | |
| if result and len(result) > 0: | |
| scores = result[0].get("scores", []) | |
| labels = result[0].get("labels", EMOTION2VEC_LABELS) | |
| if scores: | |
| all_emotions = {normalize_emotion_label(labels[i]): round(float(scores[i]), 4) for i in range(len(scores))} | |
| top_idx = int(np.argmax(scores)) | |
| return { | |
| "emotion": normalize_emotion_label(labels[top_idx]), | |
| "confidence": round(float(scores[top_idx]), 4), | |
| "all_emotions": all_emotions, | |
| } | |
| return None | |
| def _voice_clip(model, y, sr): | |
| """One clip through emotion2vec, from memory. | |
| The clip used to be written to a temporary WAV and read back. The source is 16-bit PCM, so the | |
| samples are exact multiples of 1/32768 and that round trip changed nothing; passing the array | |
| skips a file write and read per segment. A funasr that refuses arrays gets the file instead. | |
| """ | |
| import soundfile as sf | |
| try: | |
| return _voice_result(model.generate(y, fs=sr, granularity="utterance", extract_embedding=False)) | |
| except Exception as e: | |
| print(f"[voice] array input refused ({type(e).__name__}); using a temporary file", flush=True) | |
| with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f: | |
| sf.write(f.name, y, sr) | |
| temp_path = f.name | |
| try: | |
| return _voice_result(model.generate(temp_path, granularity="utterance", extract_embedding=False)) | |
| finally: | |
| os.unlink(temp_path) | |
| def _segment_samples(y, sr, start, end): | |
| """The samples librosa.load(offset=start, duration=end-start) returns, cut from the whole file.""" | |
| i0 = int(start * sr) | |
| return y[i0:i0 + int((end - start) * sr)] | |
| def analyze_voice_segments(audio_path, segments): | |
| """Voice emotion for each segment, reading the audio once. | |
| Each segment used to reload the file from its own offset and go through a temporary WAV. Same | |
| samples, same model, same output; one read instead of one per sentence. | |
| """ | |
| import librosa | |
| model = get_emotion2vec() | |
| if not model or not segments: | |
| return [None] * len(segments) | |
| y_all, sr = librosa.load(audio_path, sr=16000) | |
| out = [] | |
| for seg in segments: | |
| y = _segment_samples(y_all, sr, seg["start"], seg["end"]) | |
| if len(y) < 1600: # Too short | |
| out.append(None) | |
| continue | |
| try: | |
| out.append(_voice_clip(model, y, sr)) | |
| except Exception as e: | |
| print(f"[voice] emotion2vec error: {e}", flush=True) | |
| out.append(None) | |
| return out | |
| def analyze_voice_emotion(audio_path, start, end): | |
| """Voice emotion for one stretch of the audio (the transcription jobs read it one at a time).""" | |
| return analyze_voice_segments(audio_path, [{"start": start, "end": end}])[0] | |
| # HSEmotion-ONNX labels (correct order: Anger, Contempt, Disgust, Fear, Happiness, Neutral, Sadness, Surprise) | |
| FACE_EMOTION_LABELS = ['angry', 'contempt', 'disgust', 'fear', 'happy', 'neutral', 'sad', 'surprise'] | |
| # Face crops sent to HSEmotion in one call. One call per face was most of the cost of the pass on a | |
| # GPU; the scores are the model's own either way (a batch is the same per-image forward pass). | |
| FACE_BATCH = 32 | |
| class _FacePass: | |
| """Faces and their emotions, at fps frames a second of the film. See analyze_faces.""" | |
| def __init__(self, video_fps, fps=0.5, use_emotions=True): | |
| self.detector = get_face_detector() | |
| self.emotion_model = get_face_emotion() if use_emotions else None | |
| self.interval = max(1, int(video_fps / fps)) | |
| self.video_fps = video_fps | |
| self.timeline = [] | |
| self.pending = [] # (timeline index, face index, crop) awaiting HSEmotion | |
| def wants(self, frame_num): | |
| return frame_num % self.interval == 0 | |
| def add(self, frame, frame_num): | |
| import cv2 | |
| time = frame_num / self.video_fps | |
| gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) | |
| faces = detect_faces(self.detector, frame, gray) | |
| emotions = {} | |
| # Every face's box and reading, in the frame's own pixels, so the player can draw the | |
| # measurement on the person it was taken from. | |
| per_face = [{"box": [int(x), int(y), int(w), int(h)], "emotions": {}} for (x, y, w, h) in faces] | |
| self.timeline.append({"time": time, "faces": len(faces), "emotions": emotions, | |
| "size": [int(frame.shape[1]), int(frame.shape[0])], "per_face": per_face}) | |
| if len(faces) > 0 and self.emotion_model: | |
| for fi, (x, y, w, h) in enumerate(faces): | |
| # Expand face region slightly for better detection | |
| pad = int(w * 0.1) | |
| x1 = max(0, x - pad) | |
| y1 = max(0, y - pad) | |
| x2 = min(frame.shape[1], x + w + pad) | |
| y2 = min(frame.shape[0], y + h + pad) | |
| crop = frame[y1:y2, x1:x2] | |
| if crop.size: | |
| self.pending.append((len(self.timeline) - 1, fi, crop.copy())) | |
| if len(self.pending) >= FACE_BATCH: | |
| self._flush() | |
| elif len(faces) > 0: | |
| # Fallback: basic heuristic if HSEmotion not available | |
| x, y, w, h = faces[0] | |
| face_roi = gray[y:y+h, x:x+w] | |
| brightness = face_roi.mean() / 255 | |
| emotions["neutral"] = round(0.5 + brightness * 0.2, 3) | |
| def _flush(self): | |
| if not self.pending: | |
| return | |
| batch, self.pending = self.pending, [] | |
| try: | |
| _, scores = self.emotion_model.predict_multi_emotions([c for _, _, c in batch], logits=False) | |
| except Exception as e: | |
| print(f"[face] HSEmotion batch error ({e}); scoring faces one at a time", flush=True) | |
| scores = [] | |
| for _, _, crop in batch: | |
| try: | |
| _, sc = self.emotion_model.predict_emotions(crop, logits=False) | |
| scores.append(sc[0] if len(sc.shape) > 1 else sc) | |
| except Exception as e2: | |
| print(f"[face] HSEmotion error for face: {e2}", flush=True) | |
| scores.append(None) | |
| for (ti, fi, _), sc in zip(batch, scores): | |
| if sc is None or len(sc) == 0: | |
| continue | |
| face_emo = {FACE_EMOTION_LABELS[i]: float(sc[i]) for i in range(min(len(sc), len(FACE_EMOTION_LABELS)))} | |
| self.timeline[ti]["per_face"][fi]["emotions"] = {k: round(v, 4) for k, v in face_emo.items()} | |
| self.timeline[ti].setdefault("_raw", []).append(face_emo) | |
| def finish(self): | |
| if self.emotion_model: | |
| self._flush() | |
| # Aggregate: average emotions across all faces in each frame | |
| for entry in self.timeline: | |
| raw = entry.pop("_raw", None) | |
| if raw: | |
| for label in FACE_EMOTION_LABELS: | |
| entry["emotions"][label] = round(sum(f.get(label, 0) for f in raw) / len(raw), 4) | |
| return self.timeline | |
| def scan_frames(video_path, screen_text=True, faces=True, face_fps=4, use_emotions=True, ocr_fps=1.0, ocr_min_conf=0.4): | |
| """The on-screen text pass and the face pass from one read of the film. | |
| Each pass used to open the file and decode every frame itself. They read the same frames they | |
| always did (each keeps its own sampling interval), from a single decode. Returns | |
| (text_timeline, face_timeline). | |
| """ | |
| import cv2 | |
| cap = cv2.VideoCapture(video_path) | |
| if not cap.isOpened(): | |
| return [], [] | |
| video_fps = cap.get(cv2.CAP_PROP_FPS) or 30 | |
| text_pass = _TextPass(video_fps, ocr_fps, ocr_min_conf) if screen_text else None | |
| face_pass = _FacePass(video_fps, face_fps, use_emotions) if faces else None | |
| frame_num = 0 | |
| try: | |
| while True: | |
| ret, frame = cap.read() | |
| if not ret: | |
| break | |
| if text_pass and text_pass.wants(frame_num): | |
| text_pass.add(frame, frame_num) | |
| if face_pass and face_pass.wants(frame_num): | |
| face_pass.add(frame, frame_num) | |
| frame_num += 1 | |
| finally: | |
| cap.release() | |
| return (text_pass.finish() if text_pass else []), (face_pass.finish() if face_pass else []) | |
| def analyze_faces(video_path, fps=0.5, use_emotions=True): | |
| """Analyze faces in video frames with optional HSEmotion""" | |
| return scan_frames(video_path, screen_text=False, faces=True, face_fps=fps, use_emotions=use_emotions)[1] | |
| def video_duration(video_path): | |
| """Running time from the container.""" | |
| import cv2 | |
| cap = cv2.VideoCapture(str(video_path)) | |
| try: | |
| fps = cap.get(cv2.CAP_PROP_FPS) or 0 | |
| frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0) | |
| return round(frames / fps, 3) if fps > 0 and frames > 0 else None | |
| finally: | |
| cap.release() | |
| def cut_stills(video_path, duration, width=320, max_frames=240): | |
| """One small JPEG per second of the film, as data URLs. | |
| The report puts a frame beside every moment it names, and it cuts those frames in the browser | |
| from the file it already has open. That fails outright on a master the browser cannot decode - | |
| ProRes, which is what an agency has to hand, or HEVC - and the moments print as "no still". This | |
| strip is the fallback: the service has the film decoded already, so it costs one seek per | |
| second. At 320 wide and quality 72 a frame is about 12 KB, so a 30-second film adds about | |
| 500 KB to the response. Sampled mid-second (0.5, 1.5, ...) so a black slate at 0 is never the | |
| frame chosen. Longer than four minutes, the step widens so the strip stays under max_frames. | |
| """ | |
| import cv2 # imported per function throughout this module, so a missing wheel fails one pass rather than the service | |
| if not duration or duration <= 0: | |
| return [] | |
| step = max(1.0, float(math.ceil(duration / max_frames))) | |
| cap = cv2.VideoCapture(video_path) | |
| out = [] | |
| try: | |
| t = 0.5 | |
| while t < duration: | |
| cap.set(cv2.CAP_PROP_POS_MSEC, t * 1000.0) | |
| ok, frame = cap.read() | |
| if ok and frame is not None and frame.size: | |
| h, w = frame.shape[:2] | |
| scale = width / max(w, 1) | |
| small = cv2.resize(frame, (width, max(1, int(round(h * scale)))), interpolation=cv2.INTER_AREA) | |
| ok2, buf = cv2.imencode(".jpg", small, [cv2.IMWRITE_JPEG_QUALITY, 72]) | |
| if ok2: | |
| out.append({"time": round(t, 2), | |
| "jpeg": "data:image/jpeg;base64," + base64.b64encode(buf.tobytes()).decode("ascii")}) | |
| t += step | |
| finally: | |
| cap.release() | |
| return out | |
| def decodable_copy(video_path): | |
| """The path the frame passes should read: the upload itself, unless OpenCV cannot decode it. | |
| OpenCV's Linux wheel ships an ffmpeg with no software AV1 decoder, and YouTube serves a third | |
| of its downloads as AV1. Every frame pass then returned nothing at all, quietly - "Your | |
| platform doesn't support hardware accelerated AV1 decoding" in the log, and a film reported as | |
| measured with no faces in it. Probed by reading one frame. On failure the system ffmpeg, which | |
| does decode AV1, writes an H.264 copy for the frame passes to read; audio still comes from the | |
| original, and the copy is the caller's to delete. Returns (path_to_read, copy_to_delete). | |
| """ | |
| import cv2 | |
| cap = cv2.VideoCapture(video_path) | |
| try: | |
| ok, frame = cap.read() | |
| finally: | |
| cap.release() | |
| if ok and frame is not None and frame.size: | |
| return video_path, None | |
| from ffmpeg_util import run_ffmpeg, FFmpegError | |
| copy = video_path + ".h264.mp4" | |
| print("[process] OpenCV could not decode a frame; re-encoding to H.264 for the frame passes", flush=True) | |
| try: | |
| # crf 16 is visually lossless for a face pass that downsamples anyway; veryfast because the | |
| # point is to get frames at all, not to archive them. | |
| encode_h264(["-i", video_path, "-an"], 16, ["-pix_fmt", "yuv420p", copy]) | |
| except FFmpegError as e: | |
| print(f"[process] re-encode failed: {e}", flush=True) | |
| MODEL_ERRORS["decode"] = f"FFmpegError: {e}" | |
| return video_path, None | |
| return copy, copy | |
| _NVENC = None | |
| def nvenc_available(): | |
| """Whether ffmpeg can encode H.264 on the GPU here: listed, and a one-frame test encode works. | |
| Checked once. The encoder needs the driver's video capability in the container | |
| (NVIDIA_DRIVER_CAPABILITIES includes video, set in the Dockerfile); without it ffmpeg lists | |
| h264_nvenc and then fails to open it, which is why this tries an encode rather than trusting the list. | |
| """ | |
| global _NVENC | |
| if _NVENC is None: | |
| _NVENC = False | |
| if torch.cuda.is_available(): | |
| try: | |
| from ffmpeg_util import run_ffmpeg | |
| run_ffmpeg(["-f", "lavfi", "-i", "color=c=black:s=256x256:d=0.1", "-frames:v", "1", | |
| "-c:v", "h264_nvenc", "-f", "null", "-"], timeout=60) | |
| _NVENC = True | |
| except Exception as e: | |
| print(f"[encode] GPU H.264 encoder unavailable, using libx264: {e}", flush=True) | |
| print(f"[encode] H.264 encoder: {'h264_nvenc' if _NVENC else 'libx264'}", flush=True) | |
| return _NVENC | |
| def h264_args(crf, preset="veryfast"): | |
| """Encoder arguments for an H.264 copy at about the quality of libx264 at this crf.""" | |
| if nvenc_available(): | |
| return ["-c:v", "h264_nvenc", "-preset", "p4", "-rc", "vbr", "-cq", str(crf), "-b:v", "0"] | |
| return ["-c:v", "libx264", "-preset", preset, "-crf", str(crf)] | |
| def encode_h264(args_before, crf, args_after, timeout=None): | |
| """run_ffmpeg with an H.264 video encoder, on the GPU when it can; falls back to libx264 if the GPU encode fails.""" | |
| from ffmpeg_util import run_ffmpeg | |
| try: | |
| return run_ffmpeg(args_before + h264_args(crf) + args_after, **({"timeout": timeout} if timeout else {})) | |
| except Exception: | |
| if not nvenc_available(): | |
| raise | |
| print("[encode] GPU encode failed; retrying with libx264", flush=True) | |
| return run_ffmpeg(args_before + ["-c:v", "libx264", "-preset", "veryfast", "-crf", str(crf)] + args_after, | |
| **({"timeout": timeout} if timeout else {})) | |
| PLAYABLE_CODECS = {"h264", "vp8", "vp9", "av1"} | |
| # A proxy bigger than this is left out of the response rather than making it unwieldy. | |
| PROXY_MAX_BYTES = 40 * 1024 * 1024 | |
| def probe_codec(video_path): | |
| """codec_name and pix_fmt of the first video stream, from the system ffprobe; (None, None) if it cannot say.""" | |
| try: | |
| from ffmpeg_util import ffmpeg_exe | |
| exe = shutil.which("ffprobe") or os.path.join(os.path.dirname(ffmpeg_exe()), "ffprobe") | |
| r = subprocess.run([exe, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=codec_name,pix_fmt", | |
| "-of", "csv=p=0", video_path], capture_output=True, text=True, timeout=60) | |
| parts = (r.stdout.strip().splitlines() or [""])[0].split(",") | |
| return (parts[0] or None, parts[1] if len(parts) > 1 and parts[1] else None) | |
| except Exception: | |
| return (None, None) | |
| def make_proxy(video_path, force=False): | |
| """A browser-playable copy of a master the browser cannot decode, or None when the upload already is one. | |
| Agencies send ProRes. A browser plays its audio track over a black frame, the thumbnail comes out | |
| black, and the report cannot cut a still from it. This writes a 720p H.264 with AAC audio, the | |
| one format every browser plays, and hands it back base64 in the response so the app can play it, | |
| cut frames from it and store it as the playable asset beside the original. A master that is | |
| already H.264, VP8, VP9 or AV1 needs none of this and gets None. | |
| """ | |
| from ffmpeg_util import run_ffmpeg | |
| codec, pix = probe_codec(video_path) | |
| if not force and codec in PLAYABLE_CODECS and (pix or "yuv420p") == "yuv420p": | |
| return None | |
| out = video_path + ".proxy.mp4" | |
| print(f"[proxy] {codec or 'unknown'} {pix or ''} is not browser-playable; writing a 720p H.264 proxy", flush=True) | |
| # Video and audio only, and no timecode: a ProRes master carries a timecode, and the mp4 muxer | |
| # writes it back as a data track that shows up as an unknown stream some browsers refuse the | |
| # whole file for. Mapping the streams is not enough, because the muxer takes it from metadata. | |
| encode_h264(["-i", video_path, "-map", "0:v:0", "-map", "0:a?", "-dn", "-sn", "-map_metadata", "-1", | |
| "-write_tmcd", "0", "-vf", "scale=-2:min(720\\,ih)"], 23, | |
| ["-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", out], | |
| timeout=900) | |
| try: | |
| size = os.path.getsize(out) | |
| if size > PROXY_MAX_BYTES: | |
| MODEL_ERRORS["proxy"] = f"proxy is {size // 1_000_000} MB, over the {PROXY_MAX_BYTES // 1_000_000} MB the response carries" | |
| return None | |
| with open(out, "rb") as fh: | |
| data = base64.b64encode(fh.read()).decode("ascii") | |
| finally: | |
| if os.path.exists(out): | |
| os.unlink(out) | |
| print(f"[proxy] {size // 1000} KB", flush=True) | |
| return {"mp4_base64": data, "bytes": size, "source_codec": f"{codec or 'unknown'}{(' ' + pix) if pix else ''}"} | |
| def process_video(video_path, emotions=True, vocal_emotion=True, screen_text=True, brand=None, cei=False, stills=False, proxy=False, memorability=False): | |
| with _ANALYSE: | |
| for key in _PER_REQUEST_ERRORS: | |
| MODEL_ERRORS.pop(key, None) | |
| if memorability_mod is not None: | |
| MODEL_ERRORS.pop("memorability", None) | |
| return _process_video(video_path, emotions, vocal_emotion, screen_text, brand, cei, stills, proxy, memorability) | |
| def _process_video(video_path, emotions, vocal_emotion, screen_text, brand, cei, stills, proxy, memorability): | |
| audio_path = video_path + ".wav" | |
| audio_error = extract_audio(video_path, audio_path) | |
| if audio_error: | |
| return {"success": False, "error": f"Failed to extract audio. {audio_error}"} | |
| # See decodable_copy. Everything that reads frames reads frames_path; duration and audio come | |
| # from the upload itself. | |
| frames_path, frames_copy = decodable_copy(video_path) | |
| try: | |
| # From the container, read once. See below for why not from the transcript. | |
| container_duration = video_duration(video_path) | |
| # The passes run side by side rather than one after another. The sound (Whisper, then | |
| # emotion2vec per segment) needs nothing from the pictures; the text-and-faces read, the | |
| # Creative Effectiveness pass and memorability each read the film themselves. Only | |
| # memorability waits, for the transcript its words channel scores. Each pass is the same | |
| # code on the same input as before, so a film measures the same; it just takes the time of | |
| # the longest branch rather than the sum of all of them. | |
| def sound(): | |
| print("[process] Transcribing...", flush=True) | |
| segs, lang = transcribe(audio_path, brand) | |
| voices = [] | |
| if vocal_emotion and segs: | |
| print("[process] Analyzing voice emotions (emotion2vec)...", flush=True) | |
| voices = analyze_voice_segments(audio_path, segs) | |
| return segs, lang, voices | |
| def pictures(): | |
| if screen_text: | |
| print("[process] Reading on-screen text...", flush=True) | |
| print(f"[process] Analyzing faces (emotions={emotions})...", flush=True) | |
| return scan_frames(frames_path, screen_text=screen_text, faces=True, face_fps=4, use_emotions=emotions) | |
| def effectiveness(): | |
| # Creative Effectiveness Index inputs. A separate pass on purpose: it samples at 2fps | |
| # through DeepFace because the 138-video corpus the index compares against was scored | |
| # that way, and a measure taken with this service's 4fps HSEmotion pass would not sit | |
| # on the same scale as the thresholds. | |
| try: | |
| import cei_measures | |
| # DeepFace pulls in TensorFlow on first use. Imported under the load lock, as every | |
| # model is: two threads importing torch, transformers and TensorFlow at once is what | |
| # left "KeyError: 'torch.distributed.fsdp'" and dropped whole passes. | |
| with _LOAD: | |
| cei_measures._deepface() | |
| print("[process] Measuring for the Creative Effectiveness Index...", flush=True) | |
| return cei_measures.measure(frames_path, audio_path) | |
| except Exception as e: | |
| import traceback | |
| print(f"[cei] failed: {e}", flush=True) | |
| print(traceback.format_exc(), flush=True) | |
| MODEL_ERRORS["cei"] = f"{type(e).__name__}: {e}" | |
| return None | |
| def recall(sound_future): | |
| # Memorability: predicted brand recall a day or more after viewing, from the pictures | |
| # and the cutting alone. A model fitted to LAMBDA, 2,183 real ads whose viewers were | |
| # asked 24 to 72 hours later which brands they remembered (see memorability.py). Its | |
| # own pass because it reads the film at 1fps through CLIP, which none of the others do. | |
| try: | |
| segs = sound_future.result()[0] | |
| print("[process] Reading memorability...", flush=True) | |
| models, predictor, text_model = get_memorability() | |
| out = memorability_mod.read_memorability( | |
| frames_path, models, predictor, container_duration, | |
| transcript=" ".join(seg.get("text", "") for seg in segs) if segs else "", | |
| brand=(brand or "").split(",")[0].strip() or None, text_model=text_model) | |
| if out.get("error"): | |
| MODEL_ERRORS["memorability"] = out["error"] | |
| return out | |
| except Exception as e: | |
| import traceback | |
| print(f"[memorability] failed: {e}", flush=True) | |
| print(traceback.format_exc(), flush=True) | |
| MODEL_ERRORS["memorability"] = f"{type(e).__name__}: {e}" | |
| return None | |
| def playable(): | |
| # A playable copy for the browser; see make_proxy. Never fatal. Alongside the rest | |
| # because a ProRes master's re-encode is minutes on its own. | |
| try: | |
| return make_proxy(video_path) | |
| except Exception as e: | |
| print(f"[proxy] failed: {e}", flush=True) | |
| MODEL_ERRORS["proxy"] = f"{type(e).__name__}: {e}" | |
| return None | |
| with ThreadPoolExecutor(max_workers=5, thread_name_prefix="analyse") as pool: | |
| sound_f = pool.submit(sound) | |
| pictures_f = pool.submit(pictures) | |
| cei_f = pool.submit(effectiveness) if cei else None | |
| recall_f = pool.submit(recall, sound_f) if memorability else None | |
| proxy_f = pool.submit(playable) if proxy else None | |
| segments, language, voices = sound_f.result() | |
| text_timeline, face_timeline = pictures_f.result() | |
| cei_metrics = cei_f.result() if cei_f else None | |
| memorability_out = recall_f.result() if recall_f else None | |
| proxy_out = proxy_f.result() if proxy_f else None | |
| # A film with no speech used to return before any of the frame passes: a music-only | |
| # advert, which is most of the effectiveness corpus, came back as a 0-second film with | |
| # nothing measured and no error. Every pass tolerates an empty transcript, so a silent | |
| # film goes through like any other and its voice measures are simply absent. | |
| if not segments: | |
| print("[process] No speech; measured the pictures anyway", flush=True) | |
| # The emotional content of the copy, scored once per distinct line. | |
| copy_emotion = [] | |
| if screen_text and text_timeline: | |
| seen, distinct = set(), [] | |
| for f in text_timeline: | |
| t = (f.get("text") or "").strip() | |
| key = t.lower() | |
| if t and key not in seen: | |
| seen.add(key) | |
| distinct.append(t) | |
| print(f"[process] Scoring {len(distinct)} lines of on-screen copy...", flush=True) | |
| copy_emotion = score_lines(distinct) | |
| timeline = [] | |
| for i, seg in enumerate(segments): | |
| entry = { | |
| "start_time": seg["start"], | |
| "end_time": seg["end"], | |
| "text": seg["text"], | |
| } | |
| # Voice emotion (emotion2vec) | |
| voice_emo = voices[i] if i < len(voices) else None | |
| if voice_emo: | |
| entry["voice_emotion"] = voice_emo | |
| # Find closest face frame | |
| seg_mid = (seg["start"] + seg["end"]) / 2 | |
| closest = min(face_timeline, key=lambda f: abs(f["time"] - seg_mid), default=None) | |
| if closest and abs(closest["time"] - seg_mid) < 2: | |
| entry["faces_detected"] = closest["faces"] | |
| if closest["emotions"]: | |
| entry["facial_emotions"] = closest["emotions"] | |
| timeline.append(entry) | |
| # From the container, not from the last speech segment. Deriving it from the transcript | |
| # made a 20-second advert report as 16 seconds whenever nobody spoke near the end, and | |
| # every percentage-of-runtime figure moved with it between runs on the same file. | |
| duration = container_duration or (max(t["end_time"] for t in timeline) if timeline else 0) | |
| transcript = " ".join(t["text"] for t in timeline) | |
| features_enabled = ["transcription"] | |
| if vocal_emotion: | |
| features_enabled.append("vocal_emotion") | |
| if emotions: | |
| features_enabled.append("facial_emotions") | |
| # Stills for the report's moment cards; see cut_stills. Never fatal: a film with no | |
| # pictures against its moments is still a measured film. | |
| stills_out = None | |
| if stills: | |
| try: | |
| print("[process] Cutting stills...", flush=True) | |
| stills_out = cut_stills(frames_path, duration) | |
| print(f"[process] {len(stills_out)} stills", flush=True) | |
| except Exception as e: | |
| print(f"[stills] failed: {e}", flush=True) | |
| MODEL_ERRORS["stills"] = f"{type(e).__name__}: {e}" | |
| return { | |
| "success": True, | |
| "timeline": timeline, | |
| # Facial emotion at the sampling rate, not collapsed onto speech segments. The timeline | |
| # above has one entry per thing that was said, so an advert with music and no voiceover | |
| # produced a single data point from sixty-odd measured frames — no curve, no peak, and | |
| # a report with nothing in it. The visual channel is the whole content of that advert. | |
| "face_timeline": face_timeline, | |
| "text_timeline": text_timeline, | |
| "screen_text_full": " ".join(f["text"] for f in text_timeline), | |
| "copy_emotion": copy_emotion, | |
| "cei": cei_metrics, | |
| "memorability": memorability_out, | |
| "stills": stills_out, | |
| "proxy": proxy_out, | |
| "brand_presence": brand_presence(text_timeline, transcript, duration, brand), | |
| "transcript_full": transcript, | |
| "duration": duration, | |
| "metadata": { | |
| "language": language, | |
| "segments": len(timeline), | |
| # What was asked for. `analysers` below is what actually ran — the two differ | |
| # whenever an optional model is missing, and only the second one is a fact. | |
| "features_enabled": features_enabled, | |
| "analysers": { | |
| "transcription": {"ran": True}, | |
| "voice_emotion": { | |
| "ran": vocal_emotion and "voice_emotion" not in MODEL_ERRORS, | |
| "requested": bool(vocal_emotion), | |
| "error": MODEL_ERRORS.get("voice_emotion"), | |
| }, | |
| "face_emotion": { | |
| "ran": emotions and "face_emotion" not in MODEL_ERRORS, | |
| "requested": bool(emotions), | |
| "error": MODEL_ERRORS.get("face_emotion"), | |
| }, | |
| "screen_text": { | |
| "ran": screen_text and "screen_text" not in MODEL_ERRORS, | |
| "requested": bool(screen_text), | |
| "error": MODEL_ERRORS.get("screen_text"), | |
| }, | |
| "copy_emotion": { | |
| "ran": bool(copy_emotion), | |
| "requested": bool(screen_text), | |
| "error": MODEL_ERRORS.get("copy_emotion"), | |
| }, | |
| "cei": { | |
| "ran": cei_metrics is not None, | |
| "requested": bool(cei), | |
| "error": MODEL_ERRORS.get("cei"), | |
| }, | |
| "memorability": { | |
| "ran": memorability_out is not None and "memorability" not in MODEL_ERRORS, | |
| "requested": bool(memorability), | |
| "error": MODEL_ERRORS.get("memorability"), | |
| }, | |
| }, | |
| "degraded": sorted(MODEL_ERRORS.keys()), | |
| } | |
| } | |
| finally: | |
| if os.path.exists(audio_path): | |
| os.unlink(audio_path) | |
| if frames_copy and os.path.exists(frames_copy): | |
| os.unlink(frames_copy) | |
| def root(): | |
| return {"service": "EchoDepth Video Analysis", "version": "4.2-emotion2vec", "gpu": torch.cuda.is_available()} | |
| def health(): | |
| return {"status": "ok", "gpu": torch.cuda.is_available(), "auth": "enforced" if service_auth.enforcing() else "off"} | |
| def models_status(): | |
| """Check which models are loaded/available""" | |
| import traceback | |
| status = { | |
| "whisper": "not_loaded", | |
| "hsemotion": "not_loaded", | |
| "emotion2vec": "not_loaded", | |
| "errors": [], | |
| "gpu": torch.cuda.is_available(), | |
| "gpu_name": torch.cuda.get_device_name(0) if torch.cuda.is_available() else None | |
| } | |
| # Check whisper | |
| try: | |
| model = get_whisper() | |
| status["whisper"] = "loaded" if model else "failed" | |
| except Exception as e: | |
| status["whisper"] = "error" | |
| status["errors"].append(f"whisper: {str(e)}") | |
| # Check HSEmotion | |
| try: | |
| model = get_face_emotion() | |
| status["hsemotion"] = "loaded" if model else "failed" | |
| if not model: | |
| status["errors"].append("hsemotion: model returned None (check logs)") | |
| except Exception as e: | |
| status["hsemotion"] = "error" | |
| status["errors"].append(f"hsemotion: {str(e)}\n{traceback.format_exc()}") | |
| # Check emotion2vec | |
| try: | |
| model = get_emotion2vec() | |
| status["emotion2vec"] = "loaded" if model else "failed" | |
| except Exception as e: | |
| status["emotion2vec"] = "error" | |
| status["errors"].append(f"emotion2vec: {str(e)}") | |
| # Where each pass runs: the check that the GPU is being used, not just present | |
| face = face_emotion_model if face_emotion_model not in (None, "failed") else None | |
| status["devices"] = { | |
| "whisper": "cuda" if torch.cuda.is_available() else "cpu", | |
| "face_emotion": face.ort_session.get_providers()[0] if face else None, | |
| "screen_text": ("cuda" if torch.cuda.is_available() else "cpu") if ocr_reader not in (None, "failed") else None, | |
| "h264_encoder": "h264_nvenc" if nvenc_available() else "libx264", | |
| "cei_deepface": "cpu (deliberately; see cei_measures._deepface)", | |
| } | |
| return status | |
| def score_text(payload: dict): | |
| """Score arbitrary lines of copy on the same scale as the copy read off the film. | |
| Exists so a rewritten call to action is measured by the model that judged the original, rather | |
| than against a second opinion that cannot be compared with the first. | |
| """ | |
| lines = payload.get("lines") or [] | |
| if not isinstance(lines, list): | |
| return {"success": False, "error": "lines must be an array of strings"} | |
| scored = score_lines([str(l) for l in lines[:40]]) | |
| return {"success": True, "scored": scored, "error": MODEL_ERRORS.get("copy_emotion")} | |
| def memorability_words(payload: dict): | |
| """ | |
| Score different words against the same pictures. `channels` is the per-channel prediction the | |
| film's own read returned; only the words channel is recomputed from `transcript`. For scoring a | |
| rewritten script against the film as shot. | |
| """ | |
| transcript = (payload.get("transcript") or "").strip() | |
| channels = payload.get("channels") or {} | |
| if not channels: | |
| return {"success": False, "error": "channels from the film's memorability read are required"} | |
| try: | |
| _, predictor, text_model = get_memorability() | |
| if text_model is None or "words" not in predictor.channels: | |
| return {"success": False, "error": "this model has no words channel"} | |
| return {"success": True, **memorability_mod.score_words(predictor, text_model, transcript, payload.get("brand"), channels)} | |
| except Exception as e: | |
| return {"success": False, "error": f"{type(e).__name__}: {e}"} | |
| def proxy_only(file: UploadFile = File(...)): | |
| """A playable 720p H.264 copy of an upload, and nothing else. | |
| For attaching a video to an analysis that was saved without one: the film was measured, the | |
| stored copy failed or was never made, and re-running eleven minutes of analysis to get a | |
| player back is the wrong price. Always transcodes, so the result is small and plays anywhere | |
| whatever came in. | |
| """ | |
| print(f"[API] Proxy only: {file.filename}", flush=True) | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp: | |
| shutil.copyfileobj(file.file, tmp) | |
| video_path = tmp.name | |
| try: | |
| MODEL_ERRORS.pop("proxy", None) | |
| out = make_proxy(video_path, force=True) | |
| return {"success": out is not None, "proxy": out, "error": MODEL_ERRORS.get("proxy")} | |
| except Exception as e: | |
| return {"success": False, "proxy": None, "error": f"{type(e).__name__}: {e}"} | |
| finally: | |
| if os.path.exists(video_path): | |
| os.unlink(video_path) | |
| # A plain def, like the other upload routes: FastAPI runs it on its thread pool. As an async def it | |
| # ran the whole analysis on the event loop, so /health, the transcription and Stage Read polls and | |
| # every other request waited until the film was done. | |
| def analyze( | |
| file: UploadFile = File(...), | |
| emotions: bool = Query(True, description="Enable facial emotion detection (HSEmotion)"), | |
| vocal_emotion: bool = Query(True, description="Enable voice emotion detection (emotion2vec)"), | |
| screen_text: bool = Query(True, description="Read on-screen copy (EasyOCR)"), | |
| brand: str = Query(None, description="Brand name to look for on screen and in speech"), | |
| cei: bool = Query(False, description="Measure for the Creative Effectiveness Index (DeepFace, 2fps)"), | |
| stills: bool = Query(False, description="Return one 320px JPEG per second, for reports on a master the browser cannot decode"), | |
| proxy: bool = Query(False, description="Return a 720p H.264 copy when the upload is not browser-playable (ProRes, HEVC)"), | |
| memorability: bool = Query(False, description="Predict brand recall a day later from the pictures (CLIP at 1fps, model fitted to LAMBDA)") | |
| ): | |
| print(f"[API] Received: {file.filename} (emotions={emotions}, vocal_emotion={vocal_emotion}, screen_text={screen_text}, brand={brand}, cei={cei}, stills={stills}, proxy={proxy}, memorability={memorability})", flush=True) | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp: | |
| shutil.copyfileobj(file.file, tmp) | |
| video_path = tmp.name | |
| try: | |
| return process_video(video_path, emotions=emotions, vocal_emotion=vocal_emotion, screen_text=screen_text, brand=brand, cei=cei, stills=stills, proxy=proxy, memorability=memorability) | |
| finally: | |
| if os.path.exists(video_path): | |
| os.unlink(video_path) | |
| # ------------------------------------------------------------------------------------------------ | |
| # Transcription jobs: Voice Analysis, Practice and script audio. | |
| # | |
| # These used to go to AssemblyAI, which trains on customer audio unless opted out and keeps | |
| # transcripts until they are deleted. They now run here, behind the app's signed pass: Whisper for | |
| # the words, emotion2vec for the voice, GoEmotions for what the words mean. | |
| # | |
| # A job, not a request: on CPU hardware Whisper medium runs at around real time, so a few minutes | |
| # of audio outlasts any sensible HTTP timeout. The caller posts the file and polls. | |
| # | |
| # RETENTION. The upload and its audio are deleted as soon as the job finishes, pass or fail. The | |
| # result is held in memory for TRANSCRIBE_TTL_S so the caller can collect it, then dropped. Nothing | |
| # is written anywhere else. | |
| # ------------------------------------------------------------------------------------------------ | |
| import re as _re | |
| import threading as _threading | |
| import time as _time | |
| import uuid as _uuid | |
| from fastapi import HTTPException as _HTTPException | |
| TRANSCRIBE_JOBS: dict = {} | |
| TRANSCRIBE_TTL_S = 3600 | |
| _TRANSCRIBE_RUN = _threading.Lock() # one at a time: the CPU is shared with /analyze | |
| _JOB_ID = _re.compile(r"^[0-9a-f]{32}$") | |
| def _text_sentiment(score): | |
| """AssemblyAI-style POSITIVE / NEGATIVE / NEUTRAL from GoEmotions, with a strength 0-1. | |
| Valence is the emotion-weighted pleasantness (0.5 is neither); feel is how much emotion the | |
| words carry at all. A line with little feeling is NEUTRAL whatever its lean. | |
| """ | |
| if not score: | |
| return "NEUTRAL", 0.0 | |
| valence, feel = score["valence"], score["feel"] | |
| lean = (valence - 0.5) * 2 # -1 .. 1 | |
| if feel < 0.25 or abs(lean) < 0.15: | |
| return "NEUTRAL", round(1 - feel, 3) | |
| return ("POSITIVE" if lean > 0 else "NEGATIVE"), round(min(1.0, abs(lean) * 0.5 + feel * 0.5), 3) | |
| def _sentences(audio_path): | |
| """Whisper's words regrouped into sentences, each with its own start and end. | |
| Whisper segments by pauses, so one segment can hold a compliment and a complaint, which then | |
| cancel out to NEUTRAL. The voice page charts sentiment sentence by sentence, as AssemblyAI | |
| returned it, so the words are split at sentence ends (or a long pause) instead. | |
| """ | |
| model = get_whisper() | |
| # vad_filter skips the silent stretches of a recording rather than decoding them. Spoken audio | |
| # only: /analyze keeps it off, because a sung brand line is the word most likely to be lost and | |
| # the voice activity detector is built for speech. | |
| segments, info = model.transcribe(audio_path, beam_size=5, language="en", word_timestamps=True, vad_filter=True) | |
| out, words, confs = [], [], [] | |
| def flush(): | |
| if words: | |
| text = "".join(w.word for w in words).strip() | |
| if text: | |
| out.append({"start": words[0].start, "end": words[-1].end, "text": text, | |
| "confidence": round(sum(confs) / len(confs), 4) if confs else None}) | |
| words.clear(); confs.clear() | |
| for seg in segments: | |
| for w in (seg.words or []): | |
| if words and w.start - words[-1].end > 1.5: | |
| flush() | |
| words.append(w) | |
| confs.append(float(w.probability)) | |
| if w.word.strip().endswith((".", "?", "!")): | |
| flush() | |
| flush() | |
| return out, info.language | |
| def _sweep_transcribe_jobs(): | |
| cutoff = _time.time() - TRANSCRIBE_TTL_S | |
| for job_id in [k for k, j in TRANSCRIBE_JOBS.items() if j.get("finished_at") and j["finished_at"] < cutoff]: | |
| TRANSCRIBE_JOBS.pop(job_id, None) | |
| def _fast_segments(audio_path): | |
| """Whisper small, greedy, silence skipped, no word timings: the fast profile's words.""" | |
| segments, info = get_whisper_fast().transcribe( | |
| audio_path, beam_size=1, language="en", vad_filter=True, condition_on_previous_text=False, | |
| ) | |
| return [{"start": s.start, "end": s.end, "text": s.text.strip(), "confidence": None} | |
| for s in segments if s.text.strip()], info.language | |
| def _run_transcription(job_id, src_path, vocal, fast=False): | |
| """fast: the Reflections profile. Whisper small and greedy, no per-sentence text sentiment, | |
| and one voice-emotion read over the whole answer instead of one per sentence.""" | |
| job = TRANSCRIBE_JOBS[job_id] | |
| audio_path = src_path + ".wav" | |
| try: | |
| with _TRANSCRIBE_RUN: | |
| job["status"] = "running" | |
| err = extract_audio(src_path, audio_path) | |
| if err: | |
| raise RuntimeError(f"Could not read the audio: {err}") | |
| import soundfile as sf | |
| duration = float(sf.info(audio_path).duration) | |
| if fast: | |
| # The words are what the speaker waits for, so the job is done as soon as they are; | |
| # the voice read follows in the same job, and voice_pending says when it has landed. | |
| segments, language = _fast_segments(audio_path) | |
| out = [{ | |
| "start": round(seg["start"], 3), "end": round(seg["end"], 3), "text": seg["text"], | |
| "sentiment": "NEUTRAL", "sentiment_confidence": 0.0, "asr_confidence": None, | |
| "voice_emotion": None, | |
| } for seg in segments] | |
| result = { | |
| "text": " ".join(s["text"] for s in out).strip(), | |
| "language": language, | |
| "duration": round(duration, 3), | |
| "confidence": None, | |
| "segments": out, | |
| "engine": "whisper-small-fast+emotion2vec", | |
| "voice_pending": bool(vocal and segments), | |
| } | |
| job.update(status="done", result=result) | |
| if result["voice_pending"]: | |
| try: | |
| voice = analyze_voice_emotion(audio_path, 0, duration) | |
| for seg in out: | |
| seg["voice_emotion"] = voice | |
| finally: | |
| result["voice_pending"] = False | |
| return | |
| segments, language = _sentences(audio_path) | |
| out = [] | |
| for seg in segments: | |
| scored = score_lines([seg["text"]]) | |
| sentiment, strength = _text_sentiment(scored[0] if scored else None) | |
| voice = analyze_voice_emotion(audio_path, seg["start"], seg["end"]) if vocal else None | |
| out.append({ | |
| "start": round(seg["start"], 3), "end": round(seg["end"], 3), "text": seg["text"], | |
| "sentiment": sentiment, "sentiment_confidence": strength, | |
| "asr_confidence": seg.get("confidence"), | |
| "voice_emotion": voice, | |
| }) | |
| confs = [s["asr_confidence"] for s in out if s["asr_confidence"] is not None] | |
| job.update(status="done", result={ | |
| "text": " ".join(s["text"] for s in out).strip(), | |
| "language": language, | |
| "duration": round(duration, 3), | |
| "confidence": round(sum(confs) / len(confs), 4) if confs else None, | |
| "segments": out, | |
| "engine": "whisper-medium+emotion2vec+goemotions", | |
| }) | |
| except Exception as e: | |
| print(f"[transcribe] job failed: {type(e).__name__}: {e}", flush=True) | |
| job.update(status="error", error=f"{e}") | |
| finally: | |
| job["finished_at"] = _time.time() | |
| for p in (src_path, audio_path): | |
| try: | |
| if os.path.exists(p): | |
| os.unlink(p) | |
| except OSError: | |
| pass | |
| def _warm_fast_profile(): | |
| """Load the models as the Space starts, so the first job after a wake does not wait for them. | |
| The fast transcription profile first, because a speaker is waiting on it; then everything | |
| /analyze uses. The weights are in the image (prefetch_models.py), so this is loading, not | |
| downloading. In the background: the service answers meanwhile, and the load lock means a | |
| request that arrives first simply waits for the model it needs. | |
| """ | |
| try: | |
| get_whisper_fast() | |
| get_emotion2vec() | |
| print("[whisper] fast profile ready", flush=True) | |
| except Exception as e: | |
| print(f"[whisper] fast profile warm-up failed: {e}", flush=True) | |
| for name, load in (("whisper medium", get_whisper), ("HSEmotion", get_face_emotion), ("EasyOCR", get_ocr), | |
| ("GoEmotions", get_text_emotion), ("memorability", get_memorability)): | |
| try: | |
| load() | |
| print(f"[warm] {name} ready", flush=True) | |
| except Exception as e: | |
| print(f"[warm] {name} failed: {e}", flush=True) | |
| try: | |
| import cei_measures | |
| with _LOAD: | |
| cei_measures.warm() | |
| print("[warm] DeepFace ready", flush=True) | |
| except Exception as e: | |
| print(f"[warm] DeepFace failed: {e}", flush=True) | |
| nvenc_available() | |
| if os.environ.get("ECHODEPTH_NO_WARMUP") != "1": | |
| _threading.Thread(target=_warm_fast_profile, daemon=True).start() | |
| def transcribe_submit(file: UploadFile = File(...), vocal_emotion: bool = Query(True), | |
| profile: str = Query("full", description="full, or fast for short answers the speaker corrects")): | |
| """Start a transcription job. Returns {job_id}; poll GET /transcribe/{job_id}.""" | |
| _sweep_transcribe_jobs() | |
| suffix = os.path.splitext(file.filename or "")[1] or ".bin" | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp: | |
| shutil.copyfileobj(file.file, tmp) | |
| src_path = tmp.name | |
| job_id = _uuid.uuid4().hex | |
| TRANSCRIBE_JOBS[job_id] = {"status": "queued", "created_at": _time.time()} | |
| _threading.Thread(target=_run_transcription, args=(job_id, src_path, vocal_emotion, profile == "fast"), daemon=True).start() | |
| return {"job_id": job_id, "status": "queued"} | |
| def transcribe_status(job_id: str): | |
| if not _JOB_ID.match(job_id or ""): | |
| raise _HTTPException(400, "bad job id") | |
| job = TRANSCRIBE_JOBS.get(job_id) | |
| if not job: | |
| raise _HTTPException(404, "unknown or expired job") | |
| out = {"status": job["status"]} | |
| if job["status"] == "done": | |
| out["result"] = job["result"] | |
| if job["status"] == "error": | |
| out["error"] = job.get("error") | |
| return out | |
| # Stage Read (multi-speaker stage measurement) — see docs/stage-read-spec.md. Optional: the service | |
| # still starts if its extra dependencies are missing. | |
| try: | |
| from stage_read.api import router as stage_read_router | |
| app.include_router(stage_read_router) | |
| print("[STARTUP] Stage Read router mounted", flush=True) | |
| except Exception as _e: | |
| print(f"[STARTUP] Stage Read unavailable: {_e}", flush=True) | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run(app, host="0.0.0.0", port=7860) | |