#!/usr/bin/env python3 """ Detecta Layout A (tela cheia) vs Layout B (canto direito + compartilhamento) Versão 2: extrai frames como JPEG em arquivo (não pipe) para evitar timeout de I/O. Amostragem a cada 10s. Video 1280x720. Heuristica visual: - Layout A: Borrello full-screen, fundo escuro, rosto centralizado. Centro da imagem tem alta variacao de cor. Bordas sao escuras. - Layout B: Canto direito tem janelinha com Borrello (~380x320px na posicao x=900,y=200). Lado esquerdo tem slide/tela compartilhada — tipicamente fundo claro (branco/cinza). Divisao vertical visivel entre x=850-900. Basicamente: left_region (0-800,0-720) mean_lum > 120 = slide claro E center (300-900, 100-600) nao tem rosto dominante """ import subprocess import json import os import tempfile VIDEO = "/opt/mia/workspace/videos/borrello_lives/Feng Shui - Metafisica nos Imóveis.mp4" OUTPUT = "/opt/mia/workspace/videos/borrello_lives/cortes_feng_shui/layout_timeline.json" DURATION = 6664 STEP = 10 # segundos entre amostras TMPDIR = "/tmp/layout_frames" os.makedirs(TMPDIR, exist_ok=True) W, H = 320, 180 # escala reduzida para analise rapida def extract_frame(ts, out_path): """Extrai frame como JPEG escalado para 320x180.""" cmd = [ "ffmpeg", "-y", "-nostdin", "-ss", str(ts), "-i", VIDEO, "-vframes", "1", "-vf", f"scale={W}:{H}", "-q:v", "5", "-update", "1", out_path ] try: result = subprocess.run(cmd, capture_output=True, timeout=30) return result.returncode == 0 and os.path.exists(out_path) and os.path.getsize(out_path) > 0 except subprocess.TimeoutExpired: return False def analyze_jpeg(path): """ Abre JPEG via PIL e retorna metricas de luminancia por regiao. Sem PIL: usa python-pil ou cai em fallback com rawvideo de baixa res. """ try: from PIL import Image import numpy as np img = Image.open(path).convert("RGB") arr = np.array(img, dtype=float) lum = 0.299*arr[:,:,0] + 0.587*arr[:,:,1] + 0.114*arr[:,:,2] # Regioes (em coords 320x180): # left: x 0-200 (correspondente a 0-800 em 1280) # right: x 213-320 (correspondente a 850-1280 em 1280) # center: x 75-225, y 28-152 (correspondente a 300-900, 100-600 em orig) # top: y 0-45 (correspondente a 0-180 em 720) left = lum[:, 0:200] right = lum[:, 213:320] center = lum[28:152, 75:225] top = lum[0:45, 50:270] mid_left = lum[28:152, 175:190] # faixa vertical divisao esquerda mid_right = lum[28:152, 205:220] # faixa vertical divisao direita def stats(region): m = float(region.mean()) s = float(region.std()) return round(m, 1), round(s, 1) lm, ls = stats(left) rm, rs = stats(right) cm, cs = stats(center) tm, ts2 = stats(top) mlm, _ = stats(mid_left) mrm, _ = stats(mid_right) mid_diff = abs(mlm - mrm) # Score Layout B score_B = 0 reasons = [] # 1. Regiao esquerda muito clara = slide branco/colorido if lm > 130: score_B += 3 reasons.append(f"left_mean={lm}>130 (slide claro)") elif lm > 100: score_B += 1 reasons.append(f"left_mean={lm}>100 (slide moderado)") # 2. Divisao vertical brusca na regiao x=175-220 em coords reduzidas if mid_diff > 25: score_B += 2 reasons.append(f"mid_diff={mid_diff:.1f}>25 (separacao vertical)") # 3. Centro nao domina (stddev baixo no centro = rosto nao esta la) if cs < 35 and lm > 80: score_B += 1 reasons.append(f"center_std={cs}<35 (rosto nao centralizado)") # 4. Top muito claro = header de slide if tm > 160: score_B += 1 reasons.append(f"top_mean={tm}>160 (header slide claro)") layout = "B" if score_B >= 4 else "A" confidence = min(score_B / 6.0, 1.0) return layout, round(confidence, 2), { "left_mean": lm, "left_std": ls, "right_mean": rm, "right_std": rs, "center_std": cs, "top_mean": tm, "mid_diff": round(mid_diff, 1), "score_B": score_B, "reasons": reasons } except ImportError: # Sem PIL: fallback simples — le bytes JPEG e estima pela proporcao de bytes altos with open(path, "rb") as f: data = f.read() # JPEG maior = mais textura/variacao = Layout A tipicamente # Mas nao e confiavel. Retorna A com baixa confianca return "A", 0.3, {"error": "PIL indisponivel", "filesize": len(data)} samples = [] total_frames = DURATION // STEP print(f"Analisando {total_frames} frames (a cada {STEP}s)...", flush=True) for i, ts in enumerate(range(0, DURATION, STEP)): frame_path = os.path.join(TMPDIR, f"frame_{ts:05d}.jpg") ok = extract_frame(ts, frame_path) if not ok: layout, conf, meta = "A", 0.3, {"error": "frame_failed"} else: layout, conf, meta = analyze_jpeg(frame_path) # Limpa arquivo try: os.unlink(frame_path) except: pass samples.append({"ts": ts, "layout": layout, "confidence": conf, "meta": meta}) if i % 30 == 0: h = ts // 3600 m = (ts % 3600) // 60 s = ts % 60 score = meta.get("score_B", "?") print(f" [{i}/{total_frames}] {h:02d}:{m:02d}:{s:02d} → Layout {layout} score_B={score}", flush=True) # Consolida em intervalos contiguos intervals = [] if samples: cur = samples[0]["layout"] cur_start = 0 for i in range(1, len(samples)): if samples[i]["layout"] != cur: intervals.append({"start": cur_start, "end": samples[i]["ts"], "layout": cur, "duration": samples[i]["ts"] - cur_start}) cur = samples[i]["layout"] cur_start = samples[i]["ts"] intervals.append({"start": cur_start, "end": DURATION, "layout": cur, "duration": DURATION - cur_start}) # Suavizacao: absorve intervalos < 30s no vizinho anterior smoothed = [] for iv in intervals: if iv["duration"] < 30 and smoothed: smoothed[-1]["end"] = iv["end"] smoothed[-1]["duration"] = smoothed[-1]["end"] - smoothed[-1]["start"] else: smoothed.append(dict(iv)) result = {"samples": samples, "intervals_raw": intervals, "intervals": smoothed} with open(OUTPUT, "w") as f: json.dump(result, f, ensure_ascii=False, indent=2) print(f"\nSalvo: {OUTPUT}", flush=True) print(f"Intervalos suavizados ({len(smoothed)}):", flush=True) for iv in smoothed: h0, m0, s0 = iv["start"]//3600, (iv["start"]%3600)//60, iv["start"]%60 h1, m1, s1 = iv["end"]//3600, (iv["end"]%3600)//60, iv["end"]%60 dur = iv["duration"] print(f" [{h0:02d}:{m0:02d}:{s0:02d} - {h1:02d}:{m1:02d}:{s1:02d}] LAYOUT_{iv['layout']} {dur}s", flush=True) print("DONE", flush=True)