Upload app_hf.py

app_hf.py

@@ -0,0 +1,595 @@
+import os
+import json
+import zipfile
+import uuid
+from datetime import datetime
+
+import gradio as gr
+import numpy as np
+import matplotlib.pyplot as plt
+from PIL import Image
+from fpdf import FPDF
+import plotly.graph_objects as go
+
+
+# =========================
+# 1. 세션 ZIP 로딩
+# =========================
+def load_session(zip_path: str):
+    """
+    업로드된 ZIP을 고유 세션 폴더에 풀고
+    log.json, frames 디렉토리를 준비한다.
+    """
+    base_dir = "uploaded_sessions"
+    os.makedirs(base_dir, exist_ok=True)
+
+    session_id = datetime.now().strftime("%Y%m%d_%H%M%S_") + str(uuid.uuid4())[:8]
+    session_dir = os.path.join(base_dir, session_id)
+    os.makedirs(session_dir, exist_ok=True)
+
+    with zipfile.ZipFile(zip_path, "r") as zf:
+        zf.extractall(session_dir)
+
+    log_path = os.path.join(session_dir, "log.json")
+    if not os.path.exists(log_path):
+        raise FileNotFoundError("log.json not found inside ZIP. Check collector.py output format.")
+
+    with open(log_path, "r", encoding="utf-8") as f:
+        data = json.load(f)
+
+    samples = data.get("samples", [])
+    events = data.get("events", [])
+    meta = data.get("meta", {})
+
+    return samples, events, meta, session_dir
+
+
+# =========================
+# 2. 보조 함수들
+# =========================
+def moving_average(arr, window=5):
+    arr = np.array(arr, dtype=float)
+    if len(arr) == 0:
+        return arr
+    if len(arr) < window:
+        return arr
+    cumsum = np.cumsum(np.insert(arr, 0, 0))
+    ma = (cumsum[window:] - cumsum[:-window]) / float(window)
+    # 길이 맞추기: 앞부분은 그냥 원본 사용
+    head = arr[:window-1]
+    return np.concatenate([head, ma])
+
+
+def normalize(arr):
+    arr = np.array(arr, dtype=float)
+    if arr.size == 0:
+        return arr
+    min_v = np.min(arr)
+    max_v = np.max(arr)
+    if max_v - min_v < 1e-8:
+        return np.zeros_like(arr)
+    return (arr - min_v) / (max_v - min_v)
+
+
+# =========================
+# 3. 타임라인 분석
+# =========================
+def analyze_timeline(samples, alpha: float, ma_window: int = 5):
+    """
+    samples에서 time, p_surprise, bpm을 추출하고
+    BPM smoothing + baseline + combined_score 계산.
+    """
+    if len(samples) == 0:
+        raise ValueError("No samples found in log.json.")
+
+    times = np.array([s.get("time", 0.0) for s in samples], dtype=float)
+    ps = np.array([s.get("p_surprise", 0.0) for s in samples], dtype=float)
+
+    raw_bpm = []
+    for s in samples:
+        b = s.get("bpm", None)
+        raw_bpm.append(0.0 if b is None else float(b))
+    raw_bpm = np.array(raw_bpm, dtype=float)
+
+    # BPM smoothing (moving average)
+    bpm_smooth = moving_average(raw_bpm, window=ma_window)
+
+    # baseline (유효 BPM만 median)
+    valid_bpm = bpm_smooth[bpm_smooth > 0]
+    if len(valid_bpm) > 0:
+        baseline = float(np.median(valid_bpm))
+    else:
+        baseline = 0.0
+
+    # spike component: baseline 이상 부분만 사용
+    bpm_spike = np.where(bpm_smooth > baseline, bpm_smooth - baseline, 0.0)
+
+    # normalization
+    ps_norm = normalize(ps)
+    bpm_norm = normalize(bpm_spike)
+
+    beta = 1.0 - alpha
+    combined = alpha * ps_norm + beta * bpm_norm
+
+    return times, ps, bpm_smooth, combined, baseline
+
+
+# =========================
+# 4. Top3 + 간격 보정 + BPM 급상승 탐지
+# =========================
+def pick_top3_events(events, times, combined, min_gap_sec=1.0):
+    """
+    각 이벤트에 combined_score를 매기고,
+    최소 시간 간격 >= min_gap_sec 조건을 만족하는 Top3 선택.
+    """
+    if len(events) == 0:
+        return []
+
+    times = np.array(times, dtype=float)
+    enriched = []
+
+    for e in events:
+        t = float(e.get("time", 0.0))
+        idx = int(np.argmin(np.abs(times - t)))
+        score = float(combined[idx]) if 0 <= idx < len(combined) else 0.0
+
+        enriched.append({
+            "time": t,
+            "p_surprise": float(e.get("p_surprise", 0.0)),
+            "bpm": float(e["bpm"]) if e.get("bpm") is not None else None,
+            "frame_file": e.get("frame_file", ""),
+            "combined_score": score
+        })
+
+    # combined_score 기준 정렬
+    enriched.sort(key=lambda x: x["combined_score"], reverse=True)
+
+    selected = []
+    for cand in enriched:
+        if len(selected) == 0:
+            selected.append(cand)
+        else:
+            too_close = any(abs(cand["time"] - ev["time"]) < min_gap_sec for ev in selected)
+            if not too_close:
+                selected.append(cand)
+        if len(selected) >= 3:
+            break
+
+    return selected
+
+
+def detect_bpm_spikes(times, bpm, baseline, spike_delta=10.0, min_gap_sec=1.0):
+    """
+    BPM이 짧은 시간 안에 급상승하는 구간 탐지.
+    간단히: bpm[i] - bpm[i-1] >= spike_delta 이고,
+           이벤트 간 시간 간격 >= min_gap_sec.
+    """
+    spikes = []
+    last_spike_t = -1e9
+    for i in range(1, len(bpm)):
+        if bpm[i-1] <= 0 or bpm[i] <= 0:
+            continue
+        diff = bpm[i] - bpm[i-1]
+        if diff >= spike_delta and bpm[i] > baseline:
+            t = times[i]
+            if t - last_spike_t >= min_gap_sec:
+                spikes.append(t)
+                last_spike_t = t
+    return spikes
+
+
+# =========================
+# 5. Plotly 타임라인 + Heatmap
+# =========================
+def build_timeline_plot(times, ps, bpm, combined, top3, spike_times):
+    """
+    Plotly figure (p_surprise, BPM, combined + top3 marker + spike lines)
+    """
+    fig = go.Figure()
+
+    # p_surprise
+    fig.add_trace(
+        go.Scatter(
+            x=times,
+            y=ps,
+            mode="lines",
+            name="p_surprise",
+            line=dict(width=2)
+        )
+    )
+
+    # BPM (y2)
+    fig.add_trace(
+        go.Scatter(
+            x=times,
+            y=bpm,
+            mode="lines",
+            name="BPM (smoothed)",
+            line=dict(width=2, dash="dot"),
+            yaxis="y2"
+        )
+    )
+
+    # combined score
+    fig.add_trace(
+        go.Scatter(
+            x=times,
+            y=combined,
+            mode="lines",
+            name="combined_score",
+            line=dict(width=2)
+        )
+    )
+
+    # Top3 markers (combined score 위에 표시)
+    if top3:
+        top_times = [ev["time"] for ev in top3]
+        # 각 시간에서의 combined 값
+        y_vals = []
+        for t in top_times:
+            idx = int(np.argmin(np.abs(times - t)))
+            if 0 <= idx < len(combined):
+                y_vals.append(combined[idx])
+            else:
+                y_vals.append(None)
+
+        fig.add_trace(
+            go.Scatter(
+                x=top_times,
+                y=y_vals,
+                mode="markers+text",
+                name="Top3",
+                marker=dict(size=12, symbol="star", color="gold"),
+                text=[f"Top{i+1}" for i in range(len(top3))],
+                textposition="top center"
+            )
+        )
+
+    # 심박 급상승 구간: 세로선
+    shapes = []
+    for t in spike_times:
+        shapes.append(
+            dict(
+                type="line",
+                x0=t,
+                x1=t,
+                y0=0,
+                y1=1,
+                xref="x",
+                yref="paper",
+                line=dict(color="red", width=1, dash="dot")
+            )
+        )
+
+    fig.update_layout(
+        title="Surprise Timeline (Expression + Heart-rate)",
+        xaxis=dict(title="Time (s)"),
+        yaxis=dict(title="p_surprise / combined_score", side="left"),
+        yaxis2=dict(
+            title="BPM (smoothed)",
+            overlaying="y",
+            side="right"
+        ),
+        legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
+        shapes=shapes,
+        margin=dict(l=60, r=60, t=80, b=40)
+    )
+
+    return fig
+
+
+def build_heatmap(times, combined, bins=50):
+    """
+    시간에 따른 combined_score heatmap 생성.
+    """
+    if len(times) == 0:
+        fig = go.Figure()
+        fig.update_layout(title="Surprise Intensity Heatmap (no data)")
+        return fig
+
+    t_min, t_max = float(np.min(times)), float(np.max(times))
+    if t_max <= t_min:
+        t_max = t_min + 1e-6
+
+    edges = np.linspace(t_min, t_max, bins + 1)
+    centers = (edges[:-1] + edges[1:]) / 2.0
+
+    z_row = []
+    for i in range(bins):
+        mask = (times >= edges[i]) & (times < edges[i+1])
+        if np.any(mask):
+            z_row.append(float(np.max(combined[mask])))
+        else:
+            z_row.append(0.0)
+
+    z = np.array([z_row])
+
+    fig = go.Figure(data=go.Heatmap(
+        x=centers,
+        y=["surprise"],
+        z=z,
+        colorscale="YlOrRd",
+        colorbar=dict(title="Intensity")
+    ))
+
+    fig.update_layout(
+        title="Surprise Intensity Heatmap (combined_score over time)",
+        xaxis=dict(title="Time (s)"),
+        yaxis=dict(showticklabels=False),
+        margin=dict(l=60, r=60, t=80, b=40)
+    )
+
+    return fig
+
+
+# =========================
+# 6. PDF 생성 (matplotlib 이용)
+# =========================
+def create_pdf(session_dir, meta, top3, times, ps, bpm, combined, alpha, beta, baseline):
+    reports_dir = os.path.join(session_dir, "reports")
+    os.makedirs(reports_dir, exist_ok=True)
+
+    # matplotlib 타임라인 그림 생성
+    fig, axes = plt.subplots(3, 1, figsize=(7, 10), sharex=True)
+
+    axes[0].plot(times, ps, linewidth=1.2)
+    axes[0].set_ylabel("p_surprise")
+    axes[0].set_title("Surprise Probability")
+
+    axes[1].plot(times, bpm, linewidth=1.2)
+    axes[1].set_ylabel("BPM")
+    axes[1].set_title("Heart-rate (smoothed)")
+
+    axes[2].plot(times, combined, linewidth=1.2)
+    axes[2].set_ylabel("combined")
+    axes[2].set_xlabel("Time (s)")
+    axes[2].set_title("Combined Score")
+
+    for ax in axes:
+        ax.grid(True, alpha=0.3)
+
+    fig.tight_layout()
+
+    timeline_path = os.path.join(reports_dir, "timeline.png")
+    fig.savefig(timeline_path, bbox_inches="tight")
+    plt.close(fig)
+
+    # Top3 이미지 경로
+    img_paths = []
+    for ev in top3:
+        frame_rel = ev.get("frame_file", "")
+        frame_abs = os.path.join(session_dir, frame_rel)
+        if os.path.exists(frame_abs):
+            img_paths.append(frame_abs)
+        else:
+            img_paths.append(None)
+
+    pdf_path = os.path.join(reports_dir, "surprise_report.pdf")
+
+    pdf = FPDF()
+    pdf.add_page()
+
+    # Title
+    pdf.set_font("Arial", "B", 16)
+    pdf.cell(0, 10, "Surprise Analyzer Report", ln=1)
+
+    # Alpha/Beta/Baseline
+    pdf.set_font("Arial", "", 11)
+    pdf.cell(0, 8, f"Alpha (expression weight): {alpha:.2f}", ln=1)
+    pdf.cell(0, 8, f"Beta  (heart-rate weight): {beta:.2f}", ln=1)
+    pdf.cell(0, 8, f"Baseline BPM (median of valid BPM): {baseline:.2f}", ln=1)
+    pdf.ln(2)
+
+    # Meta 정보
+    if meta:
+        pdf.set_font("Arial", "B", 12)
+        pdf.cell(0, 8, "Session Meta", ln=1)
+        pdf.set_font("Arial", "", 11)
+        for k, v in meta.items():
+            pdf.cell(0, 6, f"- {k}: {v}", ln=1)
+        pdf.ln(2)
+
+    # Top3 summary
+    pdf.set_font("Arial", "B", 12)
+    pdf.cell(0, 8, "Top 3 Surprise Moments", ln=1)
+    pdf.set_font("Arial", "", 11)
+    if len(top3) == 0:
+        pdf.cell(0, 6, "No surprise events detected.", ln=1)
+    else:
+        for i, ev in enumerate(top3):
+            t = ev["time"]
+            ps_val = ev["p_surprise"]
+            bpm_val = ev["bpm"]
+            cs_val = ev["combined_score"]
+            pdf.multi_cell(
+                0, 6,
+                f"#{i+1} time = {t:.2f}s, "
+                f"p_surprise = {ps_val:.4f}, "
+                f"BPM = {bpm_val if bpm_val is not None else 'None'}, "
+                f"combined = {cs_val:.4f}"
+            )
+    pdf.ln(2)
+
+    # Timeline image
+    pdf.set_font("Arial", "B", 12)
+    pdf.cell(0, 8, "Surprise Timeline", ln=1)
+    pdf.image(timeline_path, w=180)
+    pdf.ln(4)
+
+    # Top3 images
+    pdf.set_font("Arial", "B", 12)
+    pdf.cell(0, 8, "Top 3 Frames", ln=1)
+    pdf.set_font("Arial", "", 11)
+
+    for i, p in enumerate(img_paths):
+        if p is None:
+            continue
+        pdf.cell(0, 6, f"Top {i+1}", ln=1)
+        pdf.image(p, w=80)
+        pdf.ln(3)
+
+    pdf.output(pdf_path)
+    return pdf_path
+
+
+# =========================
+# 7. Gradio 메인 처리 함수
+# =========================
+def process(zip_file, alpha):
+    """
+    Gradio에서 호출되는 메인 함수.
+    zip_file: 업로드된 ZIP 파일 경로
+    alpha: 슬라이더로 받은 α값
+    """
+    if zip_file is None:
+        return (
+            "Please upload a session ZIP created by collector.py.",
+            None, None, None,
+            None, None,
+            None
+        )
+
+    beta = 1.0 - alpha
+
+    # 1) 세션 로드
+    samples, events, meta, session_dir = load_session(zip_file)
+
+    # 2) 타임라인 분석 (BPM smoothing + combined)
+    times, ps, bpm_smooth, combined, baseline = analyze_timeline(samples, alpha, ma_window=5)
+
+    # 3) BPM 급상승 구간 탐지
+    spike_times = detect_bpm_spikes(times, bpm_smooth, baseline, spike_delta=10.0, min_gap_sec=1.0)
+
+    # 4) Top3 이벤트 선정 (최소 1초 간격)
+    top3 = pick_top3_events(events, times, combined, min_gap_sec=1.0)
+
+    # 5) Plotly 타임라인 + Heatmap 생성
+    timeline_fig = build_timeline_plot(times, ps, bpm_smooth, combined, top3, spike_times)
+    heatmap_fig = build_heatmap(times, combined, bins=50)
+
+    # 6) PDF 생성
+    pdf_path = create_pdf(session_dir, meta, top3, times, ps, bpm_smooth, combined, alpha, beta, baseline)
+
+    # 7) Top3 이미지 로딩
+    img1 = img2 = img3 = None
+    if len(top3) >= 1:
+        p1 = os.path.join(session_dir, top3[0]["frame_file"])
+        if os.path.exists(p1):
+            img1 = Image.open(p1)
+    if len(top3) >= 2:
+        p2 = os.path.join(session_dir, top3[1]["frame_file"])
+        if os.path.exists(p2):
+            img2 = Image.open(p2)
+    if len(top3) >= 3:
+        p3 = os.path.join(session_dir, top3[2]["frame_file"])
+        if os.path.exists(p3):
+            img3 = Image.open(p3)
+
+    # 8) Summary 텍스트
+    lines = [
+        f"Alpha (expression weight) = {alpha:.2f}",
+        f"Beta  (heart-rate weight) = {beta:.2f}",
+        f"Baseline BPM (median of valid BPM) = {baseline:.2f}",
+        f"Number of samples = {len(samples)}",
+        f"Number of raw events (expression-based) = {len(events)}",
+        f"Number of detected BPM spikes = {len(spike_times)}",
+        ""
+    ]
+
+    if spike_times:
+        lines.append("BPM spike times (s):")
+        lines.append(", ".join([f"{t:.2f}" for t in spike_times]))
+        lines.append("")
+
+    if len(top3) == 0:
+        lines.append("No surprise events detected above the current settings.")
+    else:
+        lines.append("Top 3 surprise moments (time, p_surprise, BPM, combined_score):")
+        for i, ev in enumerate(top3):
+            t = ev["time"]
+            ps_val = ev["p_surprise"]
+            bpm_val = ev["bpm"]
+            cs_val = ev["combined_score"]
+            lines.append(
+                f"#{i+1}  time = {t:.2f}s, "
+                f"p_surprise = {ps_val:.4f}, "
+                f"BPM = {bpm_val if bpm_val is not None else 'None'}, "
+                f"combined_score = {cs_val:.4f}"
+            )
+
+    summary = "\n".join(lines)
+
+    return summary, img1, img2, img3, timeline_fig, heatmap_fig, pdf_path
+
+
+# =========================
+# 8. Gradio UI
+# =========================
+with gr.Blocks(theme=gr.themes.Soft(primary_hue="indigo", neutral_hue="slate")) as demo:
+    gr.Markdown(
+        """
+        # 🎭 Surprise Analyzer (Expression + Heart-rate, v2)
+
+        이 웹앱은 **로컬 수집기(collector.py)** 로 녹화한 세션 ZIP을 업로드하면,  
+        표정 기반 놀람 확률(`p_surprise`)과 심박(`BPM`)을 결합해서 **Top 3 놀란 순간**을 찾아줍니다.
+
+        ### 🔁 전체 파이프라인
+        1. 로컬에서 웹캠 + 아두이노 심박 센서로 데이터 수집 (`collector.py` 실행)  
+        2. 생성된 `session_YYYYMMDD_XXXXXX.zip` 파일을 이곳에 업로드  
+        3. 아래 슬라이더로 **표정 vs 심박 가중치(α, β)** 조절  
+        4. **타임라인 그래프 + Heatmap + Top3 장면 + PDF 리포트** 확인
+
+        - ✅ BPM 그래프에 moving average smoothing 적용  
+        - ✅ p_surprise / BPM / combined_score를 **하나의 Plotly 타임라인**에 표시  
+        - ✅ Top3는 서로 **1초 이상 간격** 유지하도록 보정  
+        - ✅ combined_score 기반 **놀람 강도 Heatmap** 제공  
+        - ✅ **심박 급상승 구간 자동 탐지** 후 타임라인에 표시
+        ---
+        """
+    )
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            zip_input = gr.File(
+                label="📂 Upload session ZIP (collector.py output)",
+                type="filepath"
+            )
+            alpha_slider = gr.Slider(
+                minimum=0.0,
+                maximum=1.0,
+                value=0.6,
+                step=0.05,
+                label="α: Expression weight (β = 1 − α, heart-rate weight)"
+            )
+            analyze_btn = gr.Button("🚀 Run Analysis", variant="primary")
+
+        with gr.Column(scale=2):
+            summary_box = gr.Textbox(
+                label="📄 Analysis Summary",
+                lines=14
+            )
+
+    gr.Markdown("### 🏆 Top 3 Surprise Frames")
+
+    with gr.Row():
+        img1 = gr.Image(label="Top 1")
+        img2 = gr.Image(label="Top 2")
+        img3 = gr.Image(label="Top 3")
+
+    gr.Markdown("### 📊 Timelines & Heatmap")
+
+    timeline_plot = gr.Plot(label="Surprise / BPM / Combined Timeline")
+    heatmap_plot = gr.Plot(label="Surprise Intensity Heatmap")
+
+    gr.Markdown("### 📥 Download Report (PDF)")
+
+    pdf_file = gr.File(label="Download PDF Report")
+
+    analyze_btn.click(
+        fn=process,
+        inputs=[zip_input, alpha_slider],
+        outputs=[summary_box, img1, img2, img3, timeline_plot, heatmap_plot, pdf_file]
+    )
+
+if __name__ == "__main__":
+    demo.launch()