# -*- coding: utf-8 -*- """ app.py — CrimeSceneSolver (Scene Solver v3) — Hugging Face Space ================================================================ Faithful port of the notebook's `demo_v2` application: the four-tab frontend (Home / Analyse / Sherl0ck Agent / Contact), the nine-stage SceneSolver pipeline, the Sherl0ck agentic escalation layer (Cohere reasoning + LFM2.5-Audio voice), evidence dispatch, incoming-call behaviour and the forensic PDF/JSON outputs. Deployment notes: * `import spaces` must run before scenesolver_core (ZeroGPU CUDA model). * GPU-heavy entry points are decorated with @spaces.GPU (dynamic duration). * All secrets (COHERE_API_KEY, TELEGRAM_*, DISCORD_WEBHOOK_URL, SMTP credentials) come from environment variables / Space secrets only. """ import spaces # noqa: F401 — must precede torch device selection (ZeroGPU) from scenesolver_core import ( # noqa: E402 AUDIOPATHS, Config, SceneSolverPipeline, frame_to_timestamp, get_pipeline, log, ) import gradio as gr import numpy as np import json, os, traceback, smtplib, ssl, re, base64 from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from PIL import Image # ───────────────────────────────────────────────────────────────────────────── # SHARED CONSTANTS (used by both the web app and the agent layer) # ───────────────────────────────────────────────────────────────────────────── ADMIN_EMAIL = os.environ.get("ADMIN_EMAIL", "anumulasamarth008@gmail.com") SMTP_SERVER = "smtp.gmail.com" SMTP_PORT = 465 SENDER_EMAIL = os.environ.get("SENDER_EMAIL", "") SENDER_PASS = os.environ.get("SENDER_PASS", "") # Design tokens — pitch black / white / orange-saffron C_BG = "#000000" C_ORANGE = "#FF8C00" C_RED = "#FF4500" C_GREEN = "#4CAF50" # Shared HTML blocks NAV_HTML = """
CrimeSceneSolver Scene Solver v3
""" FOOTER_HTML = """

DISCLAIMER — This report is produced by machine learning models and should be treated as an investigative aid, not a definitive conclusion. Human expert review is strongly recommended before any operational action is taken.

CrimeSceneSolver  ·  Scene Solver v3  ·  Confidential  ·  ML outputs require expert review
""" def sec_label(text: str) -> str: """Shared section-label helper used across the web app and agent layer.""" return (f"
{text}
") def risk_color(r: float) -> str: if r >= 0.70: return C_RED if r >= 0.40: return C_ORANGE return C_GREEN # ───────────────────────────────────────────────────────────────────────────── # EMAIL # ───────────────────────────────────────────────────────────────────────────── def send_contact_email(name: str, email: str, phone: str, message: str) -> bool: if not SENDER_EMAIL or not SENDER_PASS: with open("/tmp/css_contact.log", "a") as f: f.write(f"[CONTACT] {name} | {email} | {phone}\n") return False try: msg = MIMEMultipart("alternative") msg["Subject"] = f"[CrimeSceneSolver] Contact: {name}" msg["From"] = SENDER_EMAIL msg["To"] = ADMIN_EMAIL msg.attach(MIMEText(f"Name:{name}\nEmail:{email}\nPhone:{phone}\nMessage:\n{message}\n", "plain")) ctx = ssl.create_default_context() with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT, context=ctx) as srv: srv.login(SENDER_EMAIL, SENDER_PASS) srv.sendmail(SENDER_EMAIL, ADMIN_EMAIL, msg.as_string()) return True except Exception: return False # ───────────────────────────────────────────────────────────────────────────── # PIPELINE WRAPPER # ───────────────────────────────────────────────────────────────────────────── def run_pipeline(video_path: str, enable_llava: bool = True) -> dict: pipeline = get_pipeline() try: results = pipeline.analyze( video_path=video_path, base_output_dir="/tmp/css_webapp", enable_llava=bool(enable_llava), enable_rl=True, enable_tracking=True, ) finally: if enable_llava: pipeline.release_stage8() s0 = results.get("stage0") or {} s1 = results.get("stage1") or {} s2 = results.get("stage2") or {} s3 = results.get("stage3") or {} s4 = results.get("stage4") or {} s5 = results.get("stage5") or {} s6 = results.get("stage6") or {} s7 = results.get("stage7") or {} s8 = results.get("stage8") or {} fps = float(s0.get("fps", 25.0)) def ts(fi): return frame_to_timestamp(int(fi), fps) risk_score = float(s7.get("final_risk", s6.get("fused_risk", 0.0))) rl_status = s7.get("status", "") _rl_map = {"CRITICAL":"CRITICAL","HIGH_RISK":"HIGH","MODERATE":"MODERATE","LOW_RISK":"LOW"} risk_label = (_rl_map.get(rl_status, "") or results.get("risk_label_final") or s6.get("risk_level", "LOW")) status = results.get("status", "UNKNOWN") narrative = (s8 or {}).get("unified_narrative") or (s8 or {}).get("description") or "" report = { "overall_status": status, "final_risk_score": f"{risk_score:.1%}", "risk_level": risk_label, "dynamic_threshold": f"{results.get('threshold',0):.6f}", "ae_max_error": f"{s0.get('max_error',0):.6f}", "ae_avg_error": f"{s0.get('avg_error',0):.6f}", "peak_timestamp": ts(s0.get("peak_frame_idx",0)), "anomaly_window": f"{ts(s0.get('anomaly_start_frame',0))} \u2192 {ts(s0.get('anomaly_end_frame',0))}", "stage1_tf_binary": f"{s1.get('classification','?')} ({s1.get('confidence',0):.1%})", "stage2_anomaly_type": f"{s2.get('class','N/A')} ({s2.get('confidence',0):.1%})", "stage2_top3": s2.get("top3", []), "stage2_all_probs": s2.get("all_probabilities", {}), "stage3_weapon": ("YES — " + str(s3.get("dominant_class","?")) if s3.get("weapon_flag") else "None detected"), "stage4_audio": f"{s4.get('status','N/A')} (score: {s4.get('audio_score',0):.4f})", "stage5_motion": f"{s5.get('motion_intensity',0):.3f} | {', '.join(s5.get('behaviors',[]) or ['STATIC'])}", "stage6_fusion_risk": f"{s6.get('fused_risk',0):.1%} [{s6.get('risk_level','?')}]", "stage6_reasoning": s6.get("reasoning_trace", []), "stage7_rl_final": f"{s7.get('final_risk',0):.1%} [{rl_status}]", "stage7_interpretation": s7.get("interpretation",""), "llava_narrative": narrative, # raw values for charts "_stage1_conf": float(s1.get("confidence", 0)), "_stage2_conf": float(s2.get("confidence", 0)), "_stage2_top3_raw": s2.get("top3", []), "_stage2_all_probs": s2.get("all_probabilities", {}), "_risk_score": risk_score, "_fused_risk": float(s6.get("fused_risk", 0)), "_audio_score": float(s4.get("audio_score", 0)), "_motion": float(s5.get("motion_intensity", 0)), } frames_l = s0.get("frames", []) scores_l = s0.get("scores", []) samp_idx = s0.get("sampled_indices", []) top_frame = None if frames_l and scores_l: psi = int(s0.get("peak_score_idx", 0)) pfi = min(samp_idx[psi] if psi < len(samp_idx) else 0, len(frames_l) - 1) top_frame = frames_l[pfi] def _serial(o): if isinstance(o, np.ndarray): return o.tolist() if isinstance(o, np.integer): return int(o) if isinstance(o, np.floating): return float(o) return str(o) jd = { "status": status, "risk_score": round(risk_score, 4), "risk_level": risk_label, "threshold": round(float(results.get("threshold", 0)), 6), "stage0": {"max_error": round(float(s0.get("max_error",0)),6), "avg_error": round(float(s0.get("avg_error",0)),6), "peak_timestamp": ts(s0.get("peak_frame_idx",0)), "anomaly_start": ts(s0.get("anomaly_start_frame",0)), "anomaly_end": ts(s0.get("anomaly_end_frame",0))}, "stage1": {"classification": s1.get("classification","?"), "confidence": round(float(s1.get("confidence",0)),4)}, "stage2": {"class": s2.get("class","N/A"), "confidence": round(float(s2.get("confidence",0)),4), "top3": s2.get("top3",[])}, "stage3": {"weapon_flag": bool(s3.get("weapon_flag",False)), "dominant_class": s3.get("dominant_class","None"), "weapon_count": int(s3.get("weapon_count",0))}, "stage4": {"status": s4.get("status","N/A"), "audio_score": round(float(s4.get("audio_score",0)),4)}, "stage5": {"motion_intensity": round(float(s5.get("motion_intensity",0)),4), "behaviors": s5.get("behaviors",[])}, "stage6": {"fused_risk": round(float(s6.get("fused_risk",0)),4), "risk_level": s6.get("risk_level","?"), "reasoning": s6.get("reasoning_trace",[])}, "stage7": {"final_risk": round(risk_score,4), "status": rl_status, "interpretation": s7.get("interpretation","")}, "llava_narrative": narrative, } pdf = (results.get("pdf_path_local") or results.get("pdf_path") or "/tmp/forensic_report_v3.pdf") if not os.path.exists(pdf): pdf = None graph_path = results.get("graph_path", "/tmp/deviation_graph_v3.png") if not os.path.exists(graph_path): graph_path = None return {"risk_score": risk_score, "risk_label": risk_label, "status": status, "report": report, "top_frame": top_frame, "json_data": jd, "pdf_path": pdf, "graph_path": graph_path} # ───────────────────────────────────────────────────────────────────────────── # HTML HELPERS # ───────────────────────────────────────────────────────────────────────────── def img_to_data_uri(path: str) -> str: """Encode an image file as a base64 data-URI for inline HTML embedding.""" if not path or not os.path.exists(path): return "" ext = os.path.splitext(path)[1].lower().lstrip(".") mime = {"png":"image/png","jpg":"image/jpeg","jpeg":"image/jpeg"}.get(ext,"image/png") with open(path, "rb") as f: encoded = base64.b64encode(f.read()).decode("utf-8") return f"data:{mime};base64,{encoded}" def spinner_html() -> str: return """
Processing… this may take several minutes
""" def no_result_html() -> str: return ("
" "Upload footage and press ANALYSE to begin.
") # ───────────────────────────────────────────────────────────────────────────── # REPORT HTML BUILDER # ───────────────────────────────────────────────────────────────────────────── def build_report_html(report: dict, risk_score: float, risk_label: str, status: str, graph_path: str = None) -> str: col = risk_color(risk_score) def numfmt(v: str) -> str: return re.sub(r'([\d]+\.?[\d]*%?)', r"\1", v) def row(label: str, value: str, mono: bool = False) -> str: vd = numfmt(str(value)) vfont = "font-family:'Roboto Mono',monospace;font-size:.8rem;" if mono else "font-size:.82rem;" return (f"" f"{label}" f"{vd}" f"") def section_hdr(title: str) -> str: return (f"
{title}
") def section(title: str, rows_html: str, extra: str = "") -> str: return (f"
" + section_hdr(title) + f"{rows_html}
" + extra + "
") # Risk badge is_crit = (status == "CRITICAL") badge_text = f"{'ANOMALY DETECTED' if is_crit else 'NO ANOMALY DETECTED'} ({status})" badge = (f"
" f"
" f"{risk_score:.1%}
" f"
{badge_text}  |  {risk_label} RISK
" f"
") # Raw chart values stage1_conf = float(report.get("_stage1_conf", 0)) stage2_conf = float(report.get("_stage2_conf", 0)) top3_raw = report.get("_stage2_top3_raw", []) or [] all_probs = report.get("_stage2_all_probs", {}) or {} fused_risk = float(report.get("_fused_risk", risk_score)) audio_score = float(report.get("_audio_score", 0)) motion = float(report.get("_motion", 0)) PIE_COLORS = ["#c0392b","#2471a3","#d35400","#6c3483","#117a65","#1a5276","#7d6608"] # ── Donut pie (SVG) ────────────────────────────────────────────────────── def donut_pie_svg(probs: dict) -> str: if not probs: return "" import math items = sorted(probs.items(), key=lambda x: x[1], reverse=True) labels = [i[0] for i in items] sizes = [max(float(i[1]), 1e-6) for i in items] total = sum(sizes) cx = cy = R = 90; R = 70; r = 36 wedges = "" start = -90 for idx, (label, val) in enumerate(zip(labels, sizes)): sweep = (val / total) * 360 end = start + sweep x1o, y1o = cx + R * math.cos(math.radians(start)), cy + R * math.sin(math.radians(start)) x2o, y2o = cx + R * math.cos(math.radians(end)), cy + R * math.sin(math.radians(end)) x1i, y1i = cx + r * math.cos(math.radians(end)), cy + r * math.sin(math.radians(end)) x2i, y2i = cx + r * math.cos(math.radians(start)), cy + r * math.sin(math.radians(start)) lg = 1 if sweep > 180 else 0 fill = PIE_COLORS[idx % len(PIE_COLORS)] mid_a = start + sweep / 2 lx = cx + (R + r) / 2 * math.cos(math.radians(mid_a)) ly = cy + (R + r) / 2 * math.sin(math.radians(mid_a)) num_lbl = (f"{idx+1}" if sweep > 15 else "") wedges += (f"{num_lbl}") start = end legend = "".join( f"
" f"
" f"" f"{i+1}. {cls} " f"({v/total*100:.1f}%)
" for i, (cls, v) in enumerate(zip(labels, sizes)) ) svg = (f"{wedges}" f"" f"ANOMALY" f"CLASSES") return (f"
" f"
{svg}
{legend}
") # ── Horizontal bar chart ────────────────────────────────────────────────── def conf_bar_chart(top3: list) -> str: if not top3: return "" BAR_COLORS = ["#c0392b","#d35400","#5d6d7e"] rows = "".join( f"
" f"
" f"{cls}" f"{prob*100:.1f}%" f"
" f"
" f"
" for idx, (cls, prob) in enumerate(top3[:3]) ) return (f"
" f"
" f"Top-3 Anomaly Predictions
{rows}
") # ── Signal gauge bars ───────────────────────────────────────────────────── def risk_gauge(val: float, label: str, color: str) -> str: pct = min(100, int(val * 100)) return (f"
" f"
" f"{label}" f"{val:.1%}" f"
" f"
" f"
") # ── Small donut pies for Stage 1 / Stage 2 ─────────────────────────────── def small_donut(pct: float, color: str, label: str, size: int = 84) -> str: import math cx = cy = size // 2 R = cx - 8 r = R - 12 sweep = pct * 360 end_a = -90 + sweep x1, y1 = cx + R*math.cos(math.radians(-90)), cy + R*math.sin(math.radians(-90)) x2, y2 = cx + R*math.cos(math.radians(end_a)), cy + R*math.sin(math.radians(end_a)) xi1, yi1 = cx + r*math.cos(math.radians(end_a)), cy + r*math.sin(math.radians(end_a)) xi2, yi2 = cx + r*math.cos(math.radians(-90)), cy + r*math.sin(math.radians(-90)) lg = 1 if sweep > 180 else 0 if pct >= 0.999: arc = (f"" f"") else: arc = (f"" f"" f"") return (f"
" f"" f"{arc}" f"" f"{pct:.0%}" f"
" f"{label}
") charts_cls = ( f"
" + small_donut(stage1_conf, '#FF8C00', 'TF BINARY · Stage 1') + small_donut(stage2_conf, '#c0392b', 'ANOMALY · Stage 2') + f"
{conf_bar_chart(top3_raw)}
" + (f"
{donut_pie_svg(all_probs)}
" if all_probs else "") + "
" ) charts_sensing = ( f"
" f"
Signal Intensities
" + risk_gauge(fused_risk, "Fusion Risk (Stage 6)", risk_color(fused_risk)) + risk_gauge(audio_score, "Audio Score (Stage 4)", "#2471a3") + risk_gauge(min(motion, 1.0), "Motion Intensity (Stage 5)", "#117a65") + "
" ) # ── Table rows ──────────────────────────────────────────────────────────── ae_rows = (row("Overall Status", report.get("overall_status","")) + row("Dynamic Threshold", report.get("dynamic_threshold",""), mono=True) + row("AE Max Error", report.get("ae_max_error",""), mono=True) + row("AE Avg Error", report.get("ae_avg_error",""), mono=True) + row("Peak Timestamp", report.get("peak_timestamp",""), mono=True) + row("Anomaly Window", report.get("anomaly_window",""), mono=True)) cl_rows = (row("TF Binary (Stage 1)", report.get("stage1_tf_binary","")) + row("Anomaly Type (Stage 2)", report.get("stage2_anomaly_type",""))) for i, (cls, prob) in enumerate(report.get("stage2_top3",[]), 1): cl_rows += row(f" Top-{i}", f"{cls} ({prob:.2%})", mono=True) det_rows = (row("Weapon Detection (Stage 3)", report.get("stage3_weapon","")) + row("Audio Status (Stage 4)", report.get("stage4_audio","")) + row("Motion (Stage 5)", report.get("stage5_motion",""))) fus_rows = row("Fusion Risk (Stage 6)", report.get("stage6_fusion_risk","")) for i, line in enumerate(report.get("stage6_reasoning",[]), 1): fus_rows += row(f" Step {i}", line) rl_rows = (row("RL3033 Final Risk (Stage 7)", report.get("stage7_rl_final",""), mono=True) + row("Interpretation", report.get("stage7_interpretation",""))) narr = report.get("llava_narrative","") narr_html = ( f"
" + section_hdr("AI Visual Narrative (LLaVA — Stage 8)") + f"
{narr}
" + "
" ) if narr else "" ae_graph_html = "" if graph_path: data_uri = img_to_data_uri(graph_path) if data_uri: ae_graph_html = ( f"
" + section_hdr("AE Deviation Graph — Temporal Reconstruction Error") + f"
" f"AE Deviation Graph" f"
" f"Red = reconstruction error  ·  Green dashed = dynamic threshold " f" ·  Star = peak anomaly frame" f"
" ) return ( f"
" + badge + ae_graph_html + section("Autoencoder — Stage 0", ae_rows) + section("Classification — Stages 1 & 2", cl_rows, charts_cls) + section("Detection & Sensing — Stages 3-5", det_rows, charts_sensing) + section("Fusion Layer — Stage 6", fus_rows) + section("RL3033 Temporal Reasoning — Stage 7", rl_rows) + narr_html + "
" ) def process_video(video): if video is None: return (no_result_html(), None, None, None) try: out = run_pipeline(video) r = out["risk_score"] report_html = build_report_html(out["report"], r, out["risk_label"], out["status"], graph_path=out.get("graph_path")) fi = Image.fromarray(out["top_frame"]) if out["top_frame"] is not None else None return (report_html, fi, json.dumps(out["json_data"], indent=2, default=str), out["pdf_path"]) except Exception: tb = traceback.format_exc() return (f"
Analysis failed:\n\n{tb}
", None, None, None) def handle_contact(name: str, email: str, phone: str, msg: str) -> str: """Shared contact form handler — used in both the base app and agent layer tabs.""" if not name.strip() or not email.strip() or not msg.strip(): return ("
" "Name, email, and message are required.
") if not re.match(r"^[^@]+@[^@]+\.[^@]+$", email.strip()): return ("
" "Please enter a valid email address.
") send_contact_email(name.strip(), email.strip(), phone.strip(), msg.strip()) return ("
" "✓ Message received. We will respond at the provided email.
") # ───────────────────────────────────────────────────────────────────────────── # CSS — pitch black / white / orange palette # ───────────────────────────────────────────────────────────────────────────── APP_CSS = """ @import url('https://fonts.googleapis.com/css2?family=Roboto+Mono:wght@300;400;500;600;700&display=swap'); :root { --bg: #000000; --surface:#080808; --border: #1e1e1e; --white: #FFFFFF; --dim: #888888; --orange: #FF8C00; --red: #FF4500; --mono: 'Roboto Mono', monospace; --serif: Charter, 'Bitstream Charter', Georgia, serif; } *, *::before, *::after { box-sizing:border-box; margin:0; padding:0; } html, body, .gradio-container, .gradio-container > * { background:var(--bg) !important; color:var(--white) !important; font-family:var(--serif) !important; } .gradio-container { zoom: 1.3 !important; } * { -webkit-font-smoothing:antialiased; } ::-webkit-scrollbar { width:4px; height:4px; } ::-webkit-scrollbar-track { background:var(--bg); } ::-webkit-scrollbar-thumb { background:#333; border-radius:2px; } @keyframes arc-spin { from { transform:rotate(0deg); } to { transform:rotate(360deg); } } /* TABS */ .tab-nav { background:var(--bg) !important; border-bottom:1px solid var(--border) !important; padding:0 24px !important; display:flex !important; gap:0 !important; align-items:center !important; justify-content:flex-start !important; } .tab-nav button { font-family:var(--mono) !important; font-size:.72rem !important; letter-spacing:.13em !important; color:#999 !important; background:transparent !important; border:none !important; border-bottom:2px solid transparent !important; padding:14px 0 !important; min-width:160px !important; width:160px !important; text-align:center !important; text-transform:uppercase !important; transition:color .18s, border-color .18s !important; font-weight:600 !important; white-space:nowrap !important; flex:0 0 160px !important; } .tab-nav button:hover { color:#ddd !important; } .tab-nav button.selected { color:#FFFFFF !important; border-bottom-color:var(--orange) !important; font-weight:700 !important; } /* GRADIO INTERNALS */ .gradio-container .block, .gradio-container .panel, .gradio-container .form, .gradio-container fieldset, .gradio-container .wrap, .gradio-container .container, .gradio-container > div { background:var(--bg) !important; border:none !important; } /* INPUTS */ input, textarea, select { background:#0d0d0d !important; border:1px solid #2a2a2a !important; border-radius:3px !important; color:#f0f0f0 !important; font-family:var(--serif) !important; font-size:.88rem !important; padding:12px 14px !important; transition:border-color .18s !important; width:100% !important; } input:focus, textarea:focus { border-color:var(--orange) !important; outline:none !important; box-shadow:0 0 0 1px var(--orange) !important; } label, .label-wrap span { font-family:var(--mono) !important; font-size:.65rem !important; letter-spacing:.1em !important; text-transform:uppercase !important; color:#ccc !important; font-weight:600 !important; } /* BUTTONS */ .gr-button { background:var(--white) !important; color:var(--bg) !important; border-radius:3px !important; font-family:var(--mono) !important; font-size:.72rem !important; letter-spacing:.12em !important; font-weight:700 !important; } .gr-button:hover { background:var(--orange) !important; } #ana-btn, #ana-btn button { width:auto !important; max-width:260px !important; min-height:44px !important; height:44px !important; padding:0 30px !important; } /* VIDEO UPLOAD */ #vid-up { min-height:150px !important; max-height:200px !important; } #vid-up .upload-container, #vid-up .wrap { background:#060606 !important; border:1px dashed #2a2a2a !important; border-radius:3px !important; min-height:150px !important; max-height:200px !important; } #vid-up .wrap:hover { border-color:var(--orange) !important; } /* OUTPUT AREAS */ #report-html { background:var(--bg) !important; } #kf-img, #kf-img img { border-radius:3px !important; border:1px solid var(--border) !important; width:100% !important; } #json-out, #json-out textarea { background:#060606 !important; color:#bbb !important; font-family:var(--mono) !important; font-size:.75rem !important; line-height:1.7 !important; border:1px solid var(--border) !important; border-radius:3px !important; } #pdf-out { border:1px solid var(--border) !important; border-radius:3px !important; } #pdf-out .file-preview { background:#060606 !important; } .rule { border:none; border-top:1px solid #111; margin:28px 0; } .contact-card { background:#080808; border:1px solid #1e1e1e; border-radius:4px; padding:28px 28px 20px; } .contact-info-block { background:#060606; border:1px solid #222; border-radius:3px; padding:18px 20px; margin-top:16px; } footer { display:none !important; } """ HOME_HTML = """
Forensic AI Platform  ·  Scene Solver v3

Uncover the Fact.

CrimeSceneSolver is an end-to-end forensic intelligence platform. Upload CCTV footage and a nine-stage AI pipeline analyses it, then Sherl𝟬ck, the agentic escalation layer, briefs you by voice or text, dispatches evidence to all channels, and generates a full PDF dossier.

Nine-Stage Forensic Pipeline
STAGE 0
Autoencoder (AE)
Primary anomaly signal. Reconstruction error over a dynamic threshold triggers the pipeline.
STAGE 1
TimeSformer Binary
Semantic gate: classifies the clip as Normal or Suspicious. Normal overrides the AE signal to SAFE.
STAGE 2
TimeSformer Multi-class
Fine-grained anomaly type across 7 categories using AE-weighted temporal sampling.
STAGE 3
YOLO Weapon Detection
Hard safety trigger. Melee and firearm detection within the anomaly window. Forces 0.90 risk floor.
STAGE 4
Audio Feature Extraction
ZCR, MFCC variance, spectral flux, RMS, and centroid scored against global thresholds.
STAGE 5
Tracking & Motion
YOLO tracking with Farneback optical flow fallback. Detects FAST_MOVEMENT, CLUSTERING, STATIC.
STAGE 6
Fusion Layer
Explainable rule-based reasoning combining all stage outputs into a single risk score with a full trace.
STAGE 7
RL3033 Agent
PPO reinforcement learning agent. Final risk blends TFB, TFM, Fusion, and RL scores (70/20/5/5%).
STAGE 8
LLaVA Narrative
Four keyframes described in natural language and synthesised into a coherent scene narrative.
Sherl𝟬ck — Agentic Escalation Layer
Voice Call Interface
When a CRITICAL or HIGH risk is detected, Sherl𝟬ck initiates an incoming call. Accept it to speak directly with the agent, Whisper transcribes your speech, Cohere generates an analytical response, and XTTS v2 delivers it as a voice reply. Server-side VAD handles turn detection automatically.
Text Chat Interface
A persistent text chat powered by Cohere command-a-05-2026 with the full pipeline JSON pre-loaded as context. Ask about any stage output, request reasoning traces, or say "send report" to dispatch evidence.
Evidence Dispatch
One click dispatches the full forensic dossier simultaneously to Telegram, Discord, and Email, including the peak anomaly frame and the 11-page PDF report. CRITICAL alerts auto-dispatch without user action.
11-Page PDF Dossier
Auto-generated PDF covering every stage: AE temporal graph, classification charts, weapon detections with bounding boxes, audio feature breakdown, motion tracking, fusion reasoning trace, RL3033 observation space, anomalous frames, and LLaVA narrative.
End-to-End Workflow
01
Upload Footage
Drop CCTV footage on the Analyse tab and press Analyse Footage.
02
Pipeline Runs
All nine stages execute sequentially. Report, frame, JSON, and PDF are produced.
03
Sherl0ck Calls
On HIGH or CRITICAL risk, an incoming call overlay appears. Accept to begin the briefing.
04
Dispatch & Close
Dispatch evidence to all channels, download the PDF dossier, and end the session.
Pricing
BASIC
Free
9-stage pipeline
5 analyses / month
PDF dossier
JSON output
PRO
₹49/mo
Unlimited analyses
Sherl0ck voice & text agent
Auto-dispatch (Telegram, Discord, Email)
LLaVA narratives
Priority support
ENTERPRISE
Custom
Custom model fine-tuning
On-premise deployment
Dedicated SLA
API access
From the Field
“We closed a complex six-month investigation in under two weeks after deploying SceneSolver. The nine-stage pipeline, Sherl0ck voice briefings, and automated evidence dispatch changed how our forensics team operates entirely.”
— Detective Lara Monroe, Forensics Division
""" # ══════════════════════════════════════════════════════════════════════════════ # AGENTIC ESCALATION LAYER : Sherl0ck Agent # Run AFTER Cell 2 (pipeline), Cell 5 (web app), and Cell 6 (dependencies). # Reuses shared constants/helpers from Cell 5: NAV_HTML, FOOTER_HTML, APP_CSS, # sec_label, risk_color, img_to_data_uri, spinner_html, no_result_html, # run_pipeline, build_report_html, process_video, handle_contact, HOME_HTML. # # VOICE SUBSYSTEM (this revision): # Continuous hands-free voice after Accept Call. # Browser MediaStream + Web-Audio VAD + MediaRecorder # → same-origin POST /ssvoice/turn (binary WebM bytes) # → FFmpeg → mono 24 kHz WAV # → LFM2.5-Audio-1.5B ASR # → Cohere / Sherl0ck reasoning # → LFM2.5-Audio-1.5B US male TTS # → browser playback + live waveforms + barge-in # → automatic next turn until End Call. # # PRIMARY BUG FIXED: # Previous Gradio textbox bridge delivered size=0 to Python while the browser # held a real nonzero Blob. Transport is now raw HTTP binary on the same # Gradio FastAPI app — BROWSER BYTES == SERVER BYTES. # ══════════════════════════════════════════════════════════════════════════════ import os, re, json, smtplib, ssl, base64, traceback, threading, subprocess, sys, io, wave import struct, math, time, uuid, tempfile, shutil from pathlib import Path as _Path from datetime import datetime, timezone from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.base import MIMEBase from email import encoders from typing import Optional, Dict, Any, List, Tuple import requests import numpy as np import gradio as gr try: import liquid_audio # noqa: F401 — ensure dependency cell installed it except Exception: pass # Deployment note: liquid-audio is installed from requirements.txt (no runtime pip). # ── Suppress non-critical log noise ─────────────────────────────────────────── import logging as _logging, warnings as _warnings _warnings.filterwarnings("ignore", category=DeprecationWarning) _warnings.filterwarnings("ignore", category=UserWarning, module="gradio") _warnings.filterwarnings("ignore", category=UserWarning, module="transformers") for _noisy in ("gradio.routes", "gradio.queueing", "starlette", "uvicorn.error", "httpx", "httpcore", "urllib3", "transformers.modeling_utils", "transformers.generation.configuration_utils"): _logging.getLogger(_noisy).setLevel(_logging.ERROR) os.environ.setdefault("TRANSFORMERS_VERBOSITY", "error") # ── Config ──────────────────────────────────────────────────────────────────── TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "") TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "") DISCORD_WEBHOOK_URL = os.environ.get("DISCORD_WEBHOOK_URL", "") COHERE_API_KEY = os.environ.get("COHERE_API_KEY", "") ALERT_EMAIL_FROM = os.environ.get("ALERT_EMAIL_FROM", "") ALERT_EMAIL_PASS = os.environ.get("ALERT_EMAIL_PASS", "") ALERT_EMAIL_TO = os.environ.get("ALERT_EMAIL_TO", "") ALERT_SMTP_SERVER = "smtp.gmail.com" ALERT_SMTP_PORT = 465 CALLER_TUNE_PATH = os.environ.get("CALLER_TUNE_PATH", AUDIOPATHS.get("CALLER_TUNE", "") if isinstance(AUDIOPATHS, dict) else "") THRESH_CRITICAL = 0.90 THRESH_HIGH = 0.70 COHERE_MODEL = "command-a-03-2025" # ── Optional deps ───────────────────────────────────────────────────────────── try: import cohere as _cohere COHERE_AVAILABLE = True print("[AgentLayer] Cohere OK") except ImportError: COHERE_AVAILABLE = False print("[AgentLayer] Cohere unavailable") # Cohere is only usable when the (env-provided) API key is present — the public # repository contains no credentials; Sherl0ck falls back to the notebook's # offline replies when the key is absent. COHERE_READY = COHERE_AVAILABLE and bool(COHERE_API_KEY.strip()) if COHERE_AVAILABLE and not COHERE_READY: print("[AgentLayer] COHERE_API_KEY not set - Sherl0ck will use offline fallback replies.") try: import torch except ImportError: torch = None print("[AgentLayer] torch unavailable") try: import torchaudio except ImportError: torchaudio = None print("[AgentLayer] torchaudio unavailable") HF_LFM_AUDIO_REPO = "LiquidAI/LFM2.5-Audio-1.5B" LFM_AUDIO_AVAILABLE = False _lfm_processor = None _lfm_model = None _lfm_lock = threading.Lock() try: from liquid_audio import LFM2AudioModel, LFM2AudioProcessor, ChatState, LFMModality LFM_AUDIO_AVAILABLE = True print("[AgentLayer] liquid-audio import OK") except ImportError as e: LFM_AUDIO_AVAILABLE = False print(f"[AgentLayer] liquid-audio unavailable: {e}") # ══════════════════════════════════════════════════════════════════════════════ # SVG ICONS # ══════════════════════════════════════════════════════════════════════════════ def _icon(body: str, size: int = 18, col: str = "currentColor") -> str: return (f'{body}') ICON_SHIELD = _icon('', 18) ICON_DISPATCH = _icon('', 16) # ══════════════════════════════════════════════════════════════════════════════ # PIPELINE DATA HELPERS # ══════════════════════════════════════════════════════════════════════════════ def _has_pipeline_data(jd: dict) -> bool: return bool(jd) and isinstance(jd, dict) and len(jd) > 0 def _build_plain_context(jd: dict, risk_pct: int) -> str: lines = ["=== FORENSIC PIPELINE SUMMARY ==="] lines.append(f"Overall status : {jd.get('status', 'UNKNOWN')}") lines.append(f"Risk level label : {jd.get('risk_level', 'UNKNOWN')}") lines.append(f"Final risk score : {risk_pct}%") s1 = jd.get("stage1") or {} if s1: lines.append(f"Stage 1 TF-Binary: {s1.get('classification','?')} @ " f"{round(float(s1.get('confidence',0))*100)}% confidence") s2 = jd.get("stage2") or {} cls2 = s2.get("class", "") if s2 and cls2 and cls2 != "Skipped": lines.append(f"Stage 2 Anomaly : {cls2} @ {round(float(s2.get('confidence',0))*100)}% confidence") else: lines.append("Stage 2 Anomaly : Skipped / Normal scene") s3 = jd.get("stage3") or {} if s3: lines.append(f"Stage 3 Weapon : detected={'YES' if s3.get('weapon_flag') else 'NO'}, " f"class={s3.get('dominant_class','?')}, count={s3.get('weapon_count',0)}") s4 = jd.get("stage4") or {} if s4: lines.append(f"Stage 4 Audio : {s4.get('status','?')}, " f"score={round(float(s4.get('audio_score',0))*100)}%") s5 = jd.get("stage5") or {} if s5: bhv = ", ".join(s5.get("behaviors") or ["STATIC"]) lines.append(f"Stage 5 Motion : intensity={round(float(s5.get('motion_intensity',0))*100)}%, behaviors=[{bhv}]") s6 = jd.get("stage6") or {} if s6: lines.append(f"Stage 6 Fusion : fused_risk={round(float(s6.get('fused_risk',0))*100)}%, level={s6.get('risk_level','?')}") if s6.get("reasoning"): lines.append(f" Fusion reasoning: {s6['reasoning']}") s7 = jd.get("stage7") or {} if s7: lines.append(f"Stage 7 RL3033 : final_risk={round(float(s7.get('final_risk',0))*100)}%, status={s7.get('status','?')}") if s7.get("interpretation"): lines.append(f" Interpretation: {s7['interpretation']}") s0 = jd.get("stage0") or {} if s0: lines.append(f"Anomaly window : {s0.get('anomaly_start','?')} -> {s0.get('anomaly_end','?')}, peak={s0.get('peak_timestamp','?')}") narr = jd.get("llava_narrative", "") if narr: lines.append(f"LLaVA narrative : {str(narr)[:600]}") lines.append("=== END SUMMARY ===") return "\n".join(lines) def _build_priming_messages(payload: dict, level: str) -> list: risk_pct = round(float(payload.get("risk", 0.0)) * 100) jd = payload.get("json_data") or {} if not _has_pipeline_data(jd): return [ {"role": "user", "content": "Confirm your data status."}, {"role": "assistant", "content": "No pipeline data is currently loaded. I cannot answer analysis questions yet."}, ] plain = _build_plain_context(jd, risk_pct) return [ {"role": "user", "content": ( f"Loading the forensic pipeline context for this session. " f"Alert level: {level}. Risk score: {risk_pct}%. " f"Here is the full summary — confirm you have received it:\n\n{plain}" )}, {"role": "assistant", "content": ( f"Confirmed. I have received and processed the forensic pipeline context. " f"This session is locked to alert level {level} with a final risk score of {risk_pct}%. " f"I am ready to answer your questions about the findings." )}, ] def _build_text_system_prompt(payload: dict, level: str) -> str: risk_pct = round(float(payload.get("risk", 0.0)) * 100) jd = payload.get("json_data") or {} has_data = _has_pipeline_data(jd) try: full_json = json.dumps(jd, indent=2, default=str) except Exception: full_json = str(jd) plain_ctx = _build_plain_context(jd, risk_pct) if has_data else "(no pipeline data)" no_data_block = "" if not has_data: no_data_block = ( "\n\nDATA STATUS: No pipeline data is loaded yet.\n" "Tell the analyst: 'No forensic analysis has been loaded yet. " "Please run the pipeline on the Analyse tab first.'\nDo NOT fabricate findings.\n" ) return ( f"You are Sherl0ck, a senior forensic surveillance analyst AI integrated into the " f"SceneSolver platform. Communicate with precision, depth, and professional clarity.\n\n" f"SESSION CONSTANTS — NEVER CHANGE:\n" f" Alert Level : {level}\n" f" Risk Score : {risk_pct}% (ALWAYS use {risk_pct}%, never any other value)\n" f"{no_data_block}\n" f"COMMUNICATION RULES:\n" f"1. You are Sherl0ck. You may confirm this identity if asked.\n" f"2. Respond with full analytical depth. Be thorough when warranted, concise otherwise.\n" f"3. No markdown headers. No leading dashes or bullets. Use numbered lists for sequences, " f" plain prose for explanations. Bold key terms with **term** only when essential.\n" f"4. All risk scores as whole-number percentages (e.g. {risk_pct}%).\n" f"5. Risk score and alert level are FROZEN. Never alter them.\n" f"6. If the analyst says 'send report', 'dispatch', 'forward evidence', etc. — confirm " f" report dispatched to all channels with delivery status.\n" f"7. Answer ONLY from the pipeline data below. Do not fabricate.\n" f"8. The investigative-aid disclaimer was stated at session open. Do not repeat it.\n\n" f"PRE-PARSED CONTEXT (primary reference):\n{plain_ctx}\n\n" f"FULL PIPELINE JSON:\n{full_json}\n" ) def _build_voice_system_prompt(payload: dict, level: str) -> str: risk_pct = round(float(payload.get("risk", 0.0)) * 100) jd = payload.get("json_data") or {} plain_ctx = _build_plain_context(jd, risk_pct) if _has_pipeline_data(jd) else "(no data)" try: full_json = json.dumps(jd, indent=2, default=str) except Exception: full_json = str(jd) return ( f"You are Sherl0ck, a forensic surveillance AI agent speaking aloud to an analyst.\n\n" f"PERSONA: male, mature, calm, confident, warm, steady, reassuring, precise, experienced. " f"If the analyst sounds frightened, acknowledge briefly, reduce fear, establish control, " f"explain what can be determined. Do not amplify fear, become theatrical, or sound patronizing.\n\n" f"RULES:\n" f"1. You are Sherl0ck. You may confirm this if asked.\n" f"2. Plain words only. No punctuation except percent signs. No markdown, no bullets.\n" f"3. Short sentences. 2-4 sentences max per response.\n" f"4. All scores as whole-number percentages e.g. {risk_pct}%.\n" f"5. Risk score FIXED at {risk_pct}%. Alert level FIXED at {level}.\n" f"6. No disclaimer repetition.\n" f"7. If user says send report — confirm dispatch to all channels.\n" f"8. Answer ONLY from pipeline data.\n\n" f"CONTEXT SUMMARY:\n{plain_ctx}\n\nFULL JSON:\n{full_json}\n" ) # ══════════════════════════════════════════════════════════════════════════════ # CONVERSATION STATE # ══════════════════════════════════════════════════════════════════════════════ _text_conversation_history: list = [] _text_active_payload: dict = {} _voice_conversation_history: list = [] _voice_active_payload: dict = {} # ══════════════════════════════════════════════════════════════════════════════ # TEXT AGENT # ══════════════════════════════════════════════════════════════════════════════ def text_agent_init(payload: dict, level: str) -> str: global _text_conversation_history, _text_active_payload _text_conversation_history = [] _text_active_payload = payload risk_pct = round(float(payload.get("risk", 0.0)) * 100) jd = payload.get("json_data") or {} has_data = _has_pipeline_data(jd) print(f"[AgentLayer][Text] init: has_data={has_data} jd_keys={list(jd.keys())[:8]}") if not has_data: reply = ( "No forensic analysis has been loaded yet. Please run the pipeline on the " "**Analyse** tab first — upload your footage and press Analyse. " "Once complete, click **Start Chat** to begin your briefing." ) _text_conversation_history.append({"role": "assistant", "content": reply}) return reply fallback = ( f"**Alert Level: {level} | Risk Score: {risk_pct}%**\n\n" "Forensic pipeline analysis complete. I am ready to answer your questions.\n\n" "_This is an investigative aid — not a definitive conclusion. " "Human expert review is required before any operational action._" ) if not COHERE_READY: _text_conversation_history.append({"role": "assistant", "content": fallback}) return fallback try: co = _cohere.ClientV2(COHERE_API_KEY) msgs = [{"role": "system", "content": _build_text_system_prompt(payload, level)}] msgs.extend(_build_priming_messages(payload, level)) msgs.append({"role": "user", "content": ( f"The analysis is complete. Alert level is {level}, risk score is {risk_pct}%. " f"Provide a structured opening briefing: headline finding, key contributing factors " f"with actual values, high-confidence detections above 70%, temporal context, and " f"the LLaVA narrative in your own words. Plain prose, no markdown headers. " f"End with the standard investigative-aid disclaimer." )}) resp = co.chat(model=COHERE_MODEL, messages=msgs) reply = resp.message.content[0].text.strip() except Exception as e: print(f"[AgentLayer][Text] Cohere init FAIL: {e}") reply = fallback _text_conversation_history.append({"role": "assistant", "content": reply}) return reply def _build_dispatch_reply(payload: dict, level: str) -> str: results = dispatch_all_channels(payload, level) ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") risk_pct = round(float(payload.get("risk", 0.0)) * 100) lines = [ f"**Forensic dossier dispatched.** Timestamp: {ts}", f"Alert Level: **{level}** | Risk Score: **{risk_pct}%**", "", ] all_ok = True for ch, ok in results.items(): lines.append(f"{'OK' if ok else 'FAIL'} **{ch.capitalize()}**: {'Delivered' if ok else 'Failed'}") if not ok: all_ok = False lines += ["", "**Dossier contents transmitted:**", str(payload.get("summary", ""))] if payload.get("image"): lines.append("Peak anomaly frame: attached (JPEG)") if payload.get("pdf"): lines.append("Forensic PDF report: attached") lines.append("") lines.append("All channels confirmed delivery." if all_ok else "One or more channels failed — verify manually.") return "\n".join(lines) _DISPATCH_HOOKS = ( "send report", "send pdf", "share evidence", "dispatch report", "forward report", "share report", "forward evidence", "send dossier", "transmit report", "transmit dossier", "send findings", "escalate", "send alert", ) def text_agent_respond(user_text: str) -> str: global _text_conversation_history, _text_active_payload if not user_text or not user_text.strip(): return "Please enter a message." _text_conversation_history.append({"role": "user", "content": user_text.strip()}) if any(h in user_text.lower() for h in _DISPATCH_HOOKS): level = determine_alert_level(float(_text_active_payload.get("risk", 0.0))) reply = _build_dispatch_reply(_text_active_payload, level) _text_conversation_history.append({"role": "assistant", "content": reply}) return reply if not COHERE_READY: reply = f"AI offline. You said: {user_text.strip()}" _text_conversation_history.append({"role": "assistant", "content": reply}) return reply try: co = _cohere.ClientV2(COHERE_API_KEY) level = determine_alert_level(float(_text_active_payload.get("risk", 0.0))) msgs = [{"role": "system", "content": _build_text_system_prompt(_text_active_payload, level)}] msgs.extend(_build_priming_messages(_text_active_payload, level)) real_history = _text_conversation_history[:-1] for m in real_history[-14:]: msgs.append({"role": m["role"], "content": m["content"]}) msgs.append({"role": "user", "content": user_text.strip()}) resp = co.chat(model=COHERE_MODEL, messages=msgs) reply = resp.message.content[0].text.strip() except Exception as e: print(f"[AgentLayer][Text] Cohere respond FAIL: {e}") reply = "I encountered a processing error. Please repeat your question." _text_conversation_history.append({"role": "assistant", "content": reply}) return reply def handle_text_input(user_text: str): if not user_text or not user_text.strip(): return gr.update(), "" text_agent_respond(user_text) return _render_chat_html(_text_conversation_history), "" # ══════════════════════════════════════════════════════════════════════════════ # VOICE AGENT (Cohere reasoning; LFM is speech bridge) # ══════════════════════════════════════════════════════════════════════════════ def _sanitise_for_tts(text: str) -> str: text = re.sub(r'\bSherl[0o]ck\b', 'Sherlock', text, flags=re.IGNORECASE) text = re.sub(r'[^\w\s%]', ' ', text) return re.sub(r' {2,}', ' ', text).strip() def voice_agent_init(payload: dict, level: str) -> str: global _voice_conversation_history, _voice_active_payload _voice_conversation_history = [] _voice_active_payload = payload risk_pct = round(float(payload.get("risk", 0.0)) * 100) suffix = ( " This is an investigative AI assessment not a definitive conclusion" " Speak your question and pause briefly I will respond automatically" " Say send report to dispatch evidence to all channels" ) fallback = f"Alert level {level} risk score {risk_pct}%" + suffix if not COHERE_READY: _voice_conversation_history.append({"role": "CHATBOT", "content": fallback}) return fallback try: co = _cohere.ClientV2(COHERE_API_KEY) msgs = [{"role": "system", "content": _build_voice_system_prompt(payload, level)}] msgs.extend(_build_priming_messages(payload, level)) msgs.append({"role": "user", "content": ( f"State alert level is {level} and risk score is {risk_pct}% in one short sentence. " "No punctuation except percent. Under 80 characters." )}) resp = co.chat(model=COHERE_MODEL, messages=msgs) reply = _sanitise_for_tts(resp.message.content[0].text.strip()) + suffix except Exception as e: print(f"[AgentLayer][Voice] Cohere init FAIL: {e}") reply = fallback reply = _sanitise_for_tts(reply) _voice_conversation_history.append({"role": "CHATBOT", "content": reply}) return reply def voice_agent_respond(user_text: str): global _voice_conversation_history, _voice_active_payload if not user_text or not user_text.strip(): return "Please say that again", None _voice_conversation_history.append({"role": "USER", "content": user_text.strip()}) if any(h in user_text.lower() for h in ("send report","send pdf","share evidence","dispatch report","forward report")): level = determine_alert_level(float(_voice_active_payload.get("risk", 0.0))) results = dispatch_all_channels(_voice_active_payload, level) ok = all(results.values()) reply = _sanitise_for_tts( f"Report dispatched to all channels " f"{'Delivery confirmed' if ok else 'Some channels may have failed'}" ) _voice_conversation_history.append({"role": "CHATBOT", "content": reply}) return reply, "report_sent" if not COHERE_READY: reply = _sanitise_for_tts(f"AI offline You said {user_text.strip()[:60]}") _voice_conversation_history.append({"role": "CHATBOT", "content": reply}) return reply, None try: co = _cohere.ClientV2(COHERE_API_KEY) level = determine_alert_level(float(_voice_active_payload.get("risk", 0.0))) msgs = [{"role": "system", "content": _build_voice_system_prompt(_voice_active_payload, level)}] msgs.extend(_build_priming_messages(_voice_active_payload, level)) tail = _voice_conversation_history[-12:] start_idx = 0 for i, m in enumerate(tail): if m.get("role", "USER").upper() in ("USER", "user"): start_idx = i break for m in tail[start_idx:]: role = m.get("role", "USER").upper() msgs.append({ "role": "user" if role == "USER" else "assistant", "content": m.get("content", "") }) if not msgs or msgs[-1]["role"] != "user": msgs.append({"role": "user", "content": user_text.strip()}) resp = co.chat(model=COHERE_MODEL, messages=msgs) reply = _sanitise_for_tts(resp.message.content[0].text.strip()) except Exception as e: print(f"[AgentLayer][Voice] Cohere respond FAIL: {e}") reply = "Processing error Please repeat your question" _voice_conversation_history.append({"role": "CHATBOT", "content": reply}) return reply, None # ══════════════════════════════════════════════════════════════════════════════ # NOTIFICATION / DISPATCH # ══════════════════════════════════════════════════════════════════════════════ def _fmt_msg(payload: dict, level: str) -> str: ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") risk = float(payload.get("risk", 0.0)) jd = payload.get("json_data") or {} summ = str(payload.get("summary", "No summary available.")) sep, line = "=" * 56, "-" * 56 header = (f"{sep}\n SCENESOLVER | FORENSIC ALERT DOSSIER\n" f" Level : {level}\n Risk Score : {round(risk*100)}%\n" f" Timestamp : {ts}\n{sep}") body = ["", "DETECTION SUMMARY", line] body += [f" {r.strip()}" for r in summ.split("\n") if r.strip()] if _has_pipeline_data(jd): body += ["", "STAGE-BY-STAGE BREAKDOWN", line] body += [f" {r}" for r in _build_plain_context(jd, round(risk*100)).split("\n")] footer = ( f"\n{line}\n ACTION REQUIRED\n Automated escalation triggered.\n" f"{line}\n DISCLAIMER: ML investigative aid. Human review required.\n{sep}" if level == "CRITICAL" else f"\n{line}\n DISCLAIMER: Investigative aid only. Human review required.\n{sep}" ) return "\n".join([header] + body + [footer]) def _fmt_discord(payload: dict, level: str) -> str: ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") risk = float(payload.get("risk", 0.0)) jd = payload.get("json_data") or {} summ = str(payload.get("summary", "No summary available.")) col = ":red_circle:" if level == "CRITICAL" else ":orange_circle:" lines = [ f"**SCENESOLVER FORENSIC ALERT DOSSIER** {col}", f"**Level:** `{level}` | **Risk:** `{round(risk*100)}%` | **Time:** `{ts}`", "```", ] lines += [f" {r.strip()}" for r in summ.split("\n") if r.strip()] if _has_pipeline_data(jd): lines.append("") lines += [f" {r}" for r in _build_plain_context(jd, round(risk*100)).split("\n")] lines += ["```"] if level == "CRITICAL": lines.append("**ACTION REQUIRED** — automated escalation triggered.") lines.append("*Investigative aid only — human review required before any action.*") return "\n".join(lines) def build_alert_payload(pipeline_output: dict) -> dict: risk = float(pipeline_output.get("risk_score", 0.0)) jd = pipeline_output.get("json_data") or {} parts = [ f"Status : {jd.get('status','UNKNOWN')}", f"Risk Level : {jd.get('risk_level','?')}", f"Risk Score : {round(risk*100)}%", ] for key, label in [("stage1","TF Binary"),("stage2","Anomaly Type"),("stage3","Weapon"), ("stage4","Audio"),("stage5","Motion"),("stage6","Fusion"),("stage7","RL3033")]: s = jd.get(key) or {} if not s: continue if key == "stage1": parts.append(f"{label:<12s}: {s.get('classification','?')} ({round(float(s.get('confidence',0))*100)}%)") elif key == "stage2" and s.get("class") not in ("Skipped", None): parts.append(f"{label:<12s}: {s.get('class','?')} ({round(float(s.get('confidence',0))*100)}%)") elif key == "stage3" and s.get("weapon_flag"): parts.append(f"{label:<12s}: {s.get('dominant_class','?')} (count={s.get('weapon_count',0)})") elif key == "stage4": parts.append(f"{label:<12s}: {s.get('status','?')} score={round(float(s.get('audio_score',0))*100)}%") elif key == "stage5": bhv = ", ".join(s.get("behaviors") or ["STATIC"]) parts.append(f"{label:<12s}: intensity={round(float(s.get('motion_intensity',0))*100)}% [{bhv}]") elif key == "stage6": parts.append(f"{label:<12s}: {round(float(s.get('fused_risk',0))*100)}% [{s.get('risk_level','?')}]") elif key == "stage7": parts.append(f"{label:<12s}: {round(float(s.get('final_risk',0))*100)}% [{s.get('status','?')}]") narr = jd.get("llava_narrative","") if narr: parts.append(f"Narrative : {str(narr)[:300]}") top_frame_path = None frame = pipeline_output.get("top_frame") if frame is not None: try: from PIL import Image as _I top_frame_path = "/tmp/agent_top_frame.jpg" _I.fromarray(frame).save(top_frame_path) except Exception: pass pdf_path = pipeline_output.get("pdf_path") if pdf_path and not os.path.exists(str(pdf_path)): pdf_path = None print(f"[AgentLayer] Payload risk={risk:.3f} jd_keys={list(jd.keys())[:8]}") return {"risk": risk, "summary": "\n".join(parts), "image": top_frame_path, "pdf": pdf_path, "json_data": jd} def determine_alert_level(risk: float) -> str: if risk >= THRESH_CRITICAL: return "CRITICAL" if risk >= THRESH_HIGH: return "HIGH" return "LOW" def send_telegram_text(text: str) -> bool: if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID: return False url, ok = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage", True for chunk in [text[i:i+4000] for i in range(0, len(text), 4000)]: try: r = requests.post(url, data={"chat_id": TELEGRAM_CHAT_ID, "text": chunk, "parse_mode": ""}, timeout=15) if r.status_code != 200: ok = False except Exception: ok = False return ok def send_telegram_image(path: str, caption: str = "") -> bool: if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID: return False if not path or not os.path.exists(path): return False try: with open(path, "rb") as f: r = requests.post(f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendPhoto", data={"chat_id": TELEGRAM_CHAT_ID, "caption": caption[:1000]}, files={"photo": f}, timeout=30) return r.status_code == 200 except Exception: return False def send_telegram_file(path: str, caption: str = "") -> bool: if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID: return False if not path or not os.path.exists(path): return False try: with open(path, "rb") as f: r = requests.post(f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendDocument", data={"chat_id": TELEGRAM_CHAT_ID, "caption": caption[:1000]}, files={"document": f}, timeout=60) return r.status_code == 200 except Exception: return False def send_telegram_alert(payload: dict, level: str) -> bool: ok = send_telegram_text(_fmt_msg(payload, level)) if payload.get("image"): send_telegram_image(payload["image"], f"Peak anomaly frame [{level}]") if payload.get("pdf"): send_telegram_file(payload["pdf"], f"Forensic report [{level}]") return ok def send_discord_text(content: str) -> bool: if not DISCORD_WEBHOOK_URL or DISCORD_WEBHOOK_URL.startswith("YOUR_"): return False ok = True for chunk in [content[i:i+1900] for i in range(0, len(content), 1900)]: try: r = requests.post(DISCORD_WEBHOOK_URL, json={"content": chunk}, timeout=15) if r.status_code not in (200, 204): ok = False except Exception: ok = False return ok def send_discord_file(path: str, fname: str = None) -> bool: if not path or not os.path.exists(path): return False if not DISCORD_WEBHOOK_URL or DISCORD_WEBHOOK_URL.startswith("YOUR_"): return False try: with open(path, "rb") as f: r = requests.post(DISCORD_WEBHOOK_URL, files={"file": (fname or os.path.basename(path), f)}, timeout=60) return r.status_code in (200, 204) except Exception: return False def send_discord_alert(payload: dict, level: str) -> bool: ok = send_discord_text(_fmt_discord(payload, level)) if payload.get("image"): send_discord_file(payload["image"], "peak_frame.jpg") if payload.get("pdf"): send_discord_file(payload["pdf"], "forensic_report.pdf") return ok def send_email_alert(payload: dict, level: str) -> bool: if not ALERT_EMAIL_FROM or not ALERT_EMAIL_PASS: return False try: msg = MIMEMultipart() risk = float(payload.get("risk", 0.0)) msg["Subject"] = f"SceneSolver Forensic Alert [{level}] Risk {round(risk*100)}%" msg["From"], msg["To"] = ALERT_EMAIL_FROM, ALERT_EMAIL_TO msg.attach(MIMEText(_fmt_msg(payload, level), "plain", "utf-8")) for fp, fn in [(payload.get("image"), "peak_frame.jpg"), (payload.get("pdf"), "forensic_report.pdf")]: if fp and os.path.exists(str(fp)): with open(fp, "rb") as f: p = MIMEBase("application", "octet-stream"); p.set_payload(f.read()) encoders.encode_base64(p) p.add_header("Content-Disposition", f"attachment; filename={fn}") msg.attach(p) ctx = ssl.create_default_context() with smtplib.SMTP_SSL(ALERT_SMTP_SERVER, ALERT_SMTP_PORT, context=ctx) as srv: srv.login(ALERT_EMAIL_FROM, ALERT_EMAIL_PASS) srv.sendmail(ALERT_EMAIL_FROM, ALERT_EMAIL_TO, msg.as_string()) return True except Exception as e: print(f"[Email] FAIL: {e}"); return False def dispatch_all_channels(payload: dict, level: str) -> dict: print(f"[AgentLayer] Dispatching [{level}]...") return { "telegram": send_telegram_alert(payload, level), "discord": send_discord_alert(payload, level), "email": send_email_alert(payload, level), } # ══════════════════════════════════════════════════════════════════════════════ # LFM2.5-Audio-1.5B — ASR + TTS # ══════════════════════════════════════════════════════════════════════════════ _SPECIAL_TOKEN_RE = re.compile(r'<\|[a-z_]+\|>') def _strip_special_tokens(text: str) -> str: return _SPECIAL_TOKEN_RE.sub('', text or '').strip() def _looks_like_english(text: str) -> bool: if not text or not text.strip(): return False stripped = text.strip() if re.search(r'[\u3040-\u30ff\u4e00-\u9fff\uac00-\ud7af\u0e00-\u0e7f\u0400-\u04FF]', stripped): return False non_ascii = sum(1 for ch in stripped if ord(ch) > 127) if non_ascii / max(len(stripped), 1) > 0.15: return False return True def _load_lfm_audio(): global _lfm_processor, _lfm_model if not LFM_AUDIO_AVAILABLE: return None, None if _lfm_model is not None and _lfm_processor is not None: return _lfm_processor, _lfm_model with _lfm_lock: if _lfm_model is None: print("[AgentLayer] Loading LFM2.5-Audio-1.5B (first use, ~3.65GB)...") _lfm_processor = LFM2AudioProcessor.from_pretrained(HF_LFM_AUDIO_REPO).eval() _lfm_model = LFM2AudioModel.from_pretrained(HF_LFM_AUDIO_REPO).eval() if torch is not None and torch.cuda.is_available(): try: _lfm_model = _lfm_model.to("cuda") print("[AgentLayer] LFM2.5-Audio moved to CUDA") except Exception as e: print(f"[AgentLayer] CUDA move skipped: {e}") print("[AgentLayer] LFM2.5-Audio-1.5B loaded") return _lfm_processor, _lfm_model @spaces.GPU(duration=90) def _lfm_transcribe(audio_path: str) -> str: if not audio_path or not os.path.exists(str(audio_path)): return "" processor, model = _load_lfm_audio() if model is None or torchaudio is None or torch is None: print("[LFM-ASR] model or torch/torchaudio unavailable") return "" try: wav, sr = torchaudio.load(audio_path) with _lfm_lock, torch.inference_mode(): chat = ChatState(processor) chat.new_turn("system") chat.add_text("Perform ASR.") chat.end_turn() chat.new_turn("user") chat.add_audio(wav, sr) chat.end_turn() chat.new_turn("assistant") chunks = [] for t in model.generate_sequential(**chat, max_new_tokens=256): if t.numel() == 1: chunks.append(processor.text.decode(t)) text = "".join(chunks) text = _strip_special_tokens(text) if not _looks_like_english(text): print(f"[LFM-ASR] Rejected non-English/garbled output: '{text}'") return "" print(f"[LFM-ASR] {text}") return text except Exception as e: print(f"[LFM-ASR] FAIL: {e}") return "" _LFM_VOICE_SYSTEM_PROMPT = ( "Perform TTS. Use the US male voice. " "Speak calmly, confidently, warmly, and reassuringly. " "Use a steady, mature male delivery." ) @spaces.GPU(duration=120) def lfm_text_to_speech(text: str, out_path: str = "/tmp/sherl0ck_reply.wav"): if not text or not text.strip(): return None if torchaudio is None or torch is None: print("[LFM-TTS] torch/torchaudio unavailable") return None text = _sanitise_for_tts(text) processor, model = _load_lfm_audio() if model is None: print("[LFM-TTS] model unavailable") return None try: with _lfm_lock, torch.inference_mode(): chat = ChatState(processor) chat.new_turn("system") chat.add_text(_LFM_VOICE_SYSTEM_PROMPT) chat.end_turn() chat.new_turn("user") chat.add_text(text) chat.end_turn() chat.new_turn("assistant") audio_out = [] for token in model.generate_sequential( **chat, max_new_tokens=512, audio_temperature=0.8, audio_top_k=64, ): if token.numel() > 1: audio_out.append(token) if not audio_out: print("[LFM-TTS] no audio tokens generated") return None codes_list = audio_out[:-1] if len(audio_out) > 1 else audio_out if not codes_list: print("[LFM-TTS] no audio codes after trimming EOA") return None audio_codes = torch.stack(codes_list, 1).unsqueeze(0) waveform = processor.decode(audio_codes) wf = waveform.detach().float().cpu() if wf.numel() < 240: print(f"[LFM-TTS] waveform too short numel={wf.numel()}") return None if not torch.isfinite(wf).all(): print("[LFM-TTS] waveform has non-finite values") return None if wf.dim() == 1: wf = wf.unsqueeze(0) torchaudio.save(out_path, wf, 24000) try: if not os.path.exists(out_path) or os.path.getsize(out_path) < 44: print("[LFM-TTS] saved file missing/too small") return None with open(out_path, "rb") as _bf: _hdr = _bf.read(12) if _hdr[0:4] != b"RIFF" or _hdr[8:12] != b"WAVE": print("[LFM-TTS] saved file not RIFF/WAVE") return None print(f"[LFM-TTS] saved ok bytes={os.path.getsize(out_path)}") except Exception as _ve: print(f"[LFM-TTS] saved file check fail: {_ve}") return None return out_path except Exception as e: print(f"[LFM-TTS] FAIL: {e}") return None def _caller_tune_uri() -> str: path = (CALLER_TUNE_PATH or "").strip() if not path or not os.path.exists(path): return "" try: ext = os.path.splitext(path)[1].lower() mime = "audio/mpeg" if ext == ".mp3" else "audio/wav" with open(path, "rb") as f: data = base64.b64encode(f.read()).decode() return f"data:{mime};base64,{data}" except Exception as e: print(f"[CallerTune] FAIL: {e}"); return "" def _make_synth_ring_bytes() -> bytes: sr = 22050 duration = 2.4 freq = 480 n = int(sr * duration) buf = io.BytesIO() with wave.open(buf, 'wb') as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(sr) for i in range(n): t = i / sr in_burst = (t < 0.40) or (0.45 <= t < 0.85) amp = int(28000 * math.sin(2 * math.pi * freq * t)) if in_burst else 0 wf.writeframes(struct.pack(' bool: try: cmd = [ "ffmpeg", "-y", "-loglevel", "error", "-i", src_path, "-ac", "1", "-ar", str(sr), "-c:a", "pcm_s16le", dst_path, ] r = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if r.returncode != 0: print(f"[VoiceTransport] ffmpeg FAIL rc={r.returncode} stderr={(r.stderr or '')[:400]}") return False if not os.path.exists(dst_path) or os.path.getsize(dst_path) < 44: print("[VoiceTransport] ffmpeg produced empty/missing wav") return False return True except Exception as e: print(f"[VoiceTransport] ffmpeg exception: {e}") return False def _wav_duration_sec(path: str) -> float: try: with wave.open(path, "rb") as wf: return wf.getnframes() / float(wf.getframerate() or 1) except Exception: return 0.0 def _validate_wav_file(path: str, expect_sr: int = 24000) -> tuple: """Return (ok, meta_or_error). Rejects non-RIFF / empty / absurd files.""" try: if not path or not os.path.exists(path): return False, "missing" sz = os.path.getsize(path) if sz < 44: return False, "too_small" with open(path, "rb") as f: hdr = f.read(12) if len(hdr) < 12 or hdr[0:4] != b"RIFF" or hdr[8:12] != b"WAVE": return False, "not_riff_wave" with wave.open(path, "rb") as wf: ch = wf.getnchannels() sw = wf.getsampwidth() sr = wf.getframerate() nframes = wf.getnframes() dur = nframes / float(sr or 1) if ch < 1 or sw < 1 or sr < 8000 or nframes < 1: return False, f"bad_params ch={ch} sw={sw} sr={sr} n={nframes}" # Allow nearby sample rates; ffmpeg should have forced 24k if abs(int(sr) - int(expect_sr)) > 100 and int(sr) not in (16000, 22050, 24000, 44100, 48000): return False, f"unexpected_sr={sr}" return True, {"duration": float(dur), "sr": int(sr), "channels": int(ch), "sampwidth": int(sw), "nframes": int(nframes), "bytes": int(sz)} except Exception as e: return False, f"exc:{e}" # Serialize voice turns so opening / barge-in / double-submit cannot overlap LFM+Cohere _voice_turn_lock = threading.Lock() _VOICE_MAX_BODY_BYTES = 3 * 1024 * 1024 # 3 MB raw WebM upper bound _processed_turns: set = set() _processed_turns_lock = threading.Lock() def process_voice_bytes(raw: bytes, session_id: str = "", turn_id: str = "", is_opening: bool = False) -> dict: t0 = time.time() result = { "ok": False, "status": "error", "transcript": "", "reply": "", "audio_b64": "", "audio_mime": "audio/wav", "turn_id": turn_id, "session_id": session_id, "action": None, "error": "", "timings_ms": {}, } nbytes = len(raw) if raw else 0 print(f"[VoiceTransport] RECEIVED bytes={nbytes} " f"session={session_id[:8] if session_id else '-'} turn={turn_id}") # Reject oversized bodies if nbytes > _VOICE_MAX_BODY_BYTES: result["error"] = "body_too_large" result["status"] = "body_too_large" print(f"[VoiceTransport] REJECT body_too_large bytes={nbytes}") return result # Duplicate turn guard (same session+turn) turn_key = f"{session_id}:{turn_id}:{int(bool(is_opening))}" with _processed_turns_lock: if turn_id and turn_key in _processed_turns and not is_opening: result["error"] = "duplicate_turn" result["status"] = "duplicate_turn" print(f"[VoiceTransport] REJECT duplicate_turn {turn_key}") return result if turn_id: _processed_turns.add(turn_key) # bound memory if len(_processed_turns) > 500: _processed_turns.clear() # Must hold voice_live for non-opening (session active) if not is_opening and not _AGENT_STATE.get("voice_live"): result["error"] = "call_not_active" result["status"] = "call_not_active" print("[VoiceTransport] REJECT call_not_active") return result # Serialize entire turn (LFM lock alone is not enough across ASR+Cohere+TTS) if not _voice_turn_lock.acquire(blocking=True, timeout=120): result["error"] = "turn_busy" result["status"] = "turn_busy" print("[VoiceTransport] REJECT turn_busy") return result try: return _process_voice_bytes_locked(raw, session_id, turn_id, is_opening, result, t0) finally: _voice_turn_lock.release() def _process_voice_bytes_locked(raw, session_id, turn_id, is_opening, result, t0): if is_opening: try: payload = _AGENT_STATE.get("payload", {}) or _voice_active_payload level = _AGENT_STATE.get("level", "LOW") print("[Voice] OPENING START") opening = voice_agent_init(payload, level) print(f"[Voice] OPENING TEXT chars={len(opening)}") out_path = f"/tmp/sherl0ck_opening_{int(time.time()*1000)}.wav" print("[Voice] TTS_START (opening)") tts_path = _tts_blocking(opening, out_path) print(f"[Voice] TTS_DONE path={tts_path}") if tts_path and os.path.exists(tts_path): vok, vmeta = _validate_wav_file(tts_path, expect_sr=24000) if not vok: print(f"[Voice] OPENING TTS_INVALID_WAV {vmeta}") result.update({"ok": True, "status": "ok_no_audio", "reply": opening, "error": f"tts_invalid_wav:{vmeta}"}) else: with open(tts_path, "rb") as f: ab = f.read() result.update({ "ok": True, "status": "ok", "transcript": "", "reply": opening, "audio_b64": base64.b64encode(ab).decode("ascii"), "audio_mime": "audio/wav", }) print(f"[Voice] AUDIO_RESPONSE bytes={len(ab)} (opening)") else: print("[Voice] TTS_FAILURE (opening)") result.update({"ok": True, "status": "ok_no_audio", "reply": opening, "error": "tts_failed"}) result["timings_ms"]["total"] = int((time.time() - t0) * 1000) return result except Exception as e: print(f"[Voice] OPENING FAIL: {e}") traceback.print_exc() result["error"] = str(e) return result if not raw or len(raw) < 64: print("[VoiceTransport] Empty or tiny packet — rejecting") result["error"] = "empty_packet" result["status"] = "empty" return result stamp = int(time.time() * 1000) webm_path = f"/tmp/ssvoice_in_{stamp}.webm" wav_path = f"/tmp/ssvoice_in_{stamp}.wav" try: with open(webm_path, "wb") as f: f.write(raw) print(f"[VoiceTransport] WEBM bytes={os.path.getsize(webm_path)} path={webm_path}") except Exception as e: result["error"] = f"write_webm_failed:{e}" return result print("[Voice] FFMPEG_START") ok_conv = _ffmpeg_to_wav(webm_path, wav_path, sr=24000) if not ok_conv: # HARD FAILURE — never rename/copy WebM to .wav result["error"] = "ffmpeg_failed" result["status"] = "convert_error" print("[Voice] FFMPEG_FAILURE") try: if os.path.exists(webm_path): os.remove(webm_path) except Exception: pass return result print("[Voice] FFMPEG_SUCCESS") wav_sz = os.path.getsize(wav_path) # Validate real WAV structure before LFM wav_ok, wav_meta = _validate_wav_file(wav_path, expect_sr=24000) if not wav_ok: result["error"] = f"invalid_wav:{wav_meta}" result["status"] = "invalid_wav" print(f"[Voice] WAV_INVALID {wav_meta}") for p in (webm_path, wav_path): try: if os.path.exists(p): os.remove(p) except Exception: pass return result dur = float(wav_meta.get("duration", 0.0)) print(f"[VoiceTransport] WAV bytes={wav_sz} duration={dur:.3f}s sr={wav_meta.get('sr')} ch={wav_meta.get('channels')}") result["timings_ms"]["convert"] = int((time.time() - t0) * 1000) if dur < 0.15 or wav_sz < 1000: print("[VoiceTransport] utterance too short — skip ASR") result["status"] = "too_short" result["error"] = "utterance_too_short" result["ok"] = True for p in (webm_path, wav_path): try: if os.path.exists(p): os.remove(p) except Exception: pass return result print("[Voice] ASR_START") t_asr = time.time() transcript = _lfm_transcribe(wav_path) result["timings_ms"]["asr"] = int((time.time() - t_asr) * 1000) print(f"[Voice] ASR_DONE transcript={transcript!r}") result["transcript"] = transcript or "" if not transcript or len(transcript.strip()) < 2: result["status"] = "no_speech" result["error"] = "no_speech" result["ok"] = True print("[Voice] No usable speech — returning to LISTENING") return result print("[Voice] COHERE_START") t_co = time.time() reply, action = voice_agent_respond(transcript) result["timings_ms"]["cohere"] = int((time.time() - t_co) * 1000) print(f"[Voice] COHERE_DONE reply={reply[:120]!r} action={action}") result["reply"] = reply result["action"] = action print("[Voice] TTS_START") t_tts = time.time() out_path = f"/tmp/sherl0ck_reply_{stamp}.wav" tts_path = _tts_blocking(reply, out_path) result["timings_ms"]["tts"] = int((time.time() - t_tts) * 1000) print(f"[Voice] TTS_DONE path={tts_path}") if tts_path and os.path.exists(tts_path): vok, vmeta = _validate_wav_file(tts_path, expect_sr=24000) if not vok: print(f"[Voice] TTS_INVALID_WAV {vmeta}") result["ok"] = True result["status"] = "ok_no_audio" result["error"] = f"tts_invalid_wav:{vmeta}" else: with open(tts_path, "rb") as f: ab = f.read() result["audio_b64"] = base64.b64encode(ab).decode("ascii") result["audio_mime"] = "audio/wav" print(f"[Voice] AUDIO_RESPONSE bytes={len(ab)} turn={turn_id}") result["ok"] = True result["status"] = "ok" else: result["ok"] = True result["status"] = "ok_no_audio" result["error"] = "tts_failed" print("[Voice] TTS_FAILURE") result["timings_ms"]["total"] = int((time.time() - t0) * 1000) for p in (webm_path, wav_path): try: if os.path.exists(p): os.remove(p) except Exception: pass return result _SSVOICE_ROUTES_REGISTERED = False def _register_ssvoice_routes(fastapi_app): global _SSVOICE_ROUTES_REGISTERED if _SSVOICE_ROUTES_REGISTERED: return try: from fastapi import Request from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware except ImportError: print("[VoiceTransport] fastapi import failed — routes not registered") return try: fastapi_app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) except Exception as e: print(f"[VoiceTransport] CORS middleware note: {e}") @fastapi_app.post("/ssvoice/turn") async def ssvoice_turn(request: Request): import anyio try: session_id = request.headers.get("x-ss-session", "") or "" turn_id = request.headers.get("x-ss-turn", "") or "" is_opening = request.headers.get("x-ss-opening", "0") == "1" # Reject absurd Content-Length early when present cl = request.headers.get("content-length") if cl is not None: try: if int(cl) > _VOICE_MAX_BODY_BYTES: return JSONResponse({"ok": False, "status": "body_too_large", "error": "body_too_large"}, status_code=413) except Exception: pass raw = await request.body() print(f"[VoiceTransport] REQUEST_ENTERED " f"session_id={session_id} body_bytes={len(raw)} " f"opening={is_opening} turn={turn_id}") if len(raw) > _VOICE_MAX_BODY_BYTES: return JSONResponse({"ok": False, "status": "body_too_large", "error": "body_too_large"}, status_code=413) active = _AGENT_STATE.get("voice_session_id", "") if active and session_id and session_id != active and not is_opening: print(f"[VoiceTransport] stale session {session_id[:8]} != {active[:8]}") return JSONResponse({"ok": False, "status": "stale_session", "error": "stale_session", "turn_id": turn_id}, status_code=409) # Run blocking ASR/LLM/TTS off the event loop out = await anyio.to_thread.run_sync( process_voice_bytes, raw, session_id, turn_id, is_opening ) return JSONResponse(out) except Exception as e: print(f"[VoiceTransport] /ssvoice/turn EXCEPTION: {e}") traceback.print_exc() return JSONResponse({"ok": False, "status": "error", "error": str(e)}, status_code=500) @fastapi_app.get("/ssvoice/health") async def ssvoice_health(): return JSONResponse({ "ok": True, "lfm": LFM_AUDIO_AVAILABLE, "cohere": COHERE_AVAILABLE, "voice_live": bool(_AGENT_STATE.get("voice_live")), "session": (_AGENT_STATE.get("voice_session_id") or "")[:8], }) @fastapi_app.post("/ssvoice/end") async def ssvoice_end(request: Request): sid = request.headers.get("x-ss-session", "") print(f"[VoiceTransport] END session={sid[:8] if sid else '-'}") _AGENT_STATE["voice_live"] = False if sid and sid == _AGENT_STATE.get("voice_session_id"): _AGENT_STATE["voice_session_id"] = "" return JSONResponse({"ok": True}) @fastapi_app.post("/ssvoice/accept") async def ssvoice_accept(request: Request): """Mint/return the authoritative voice session. Browser calls this from the Accept gesture.""" global _voice_active_payload try: payload = _AGENT_STATE.get("payload", {}) or {} level = _AGENT_STATE.get("level", "LOW") # Reuse existing live session if already minted by Gradio accept_call sid = _AGENT_STATE.get("voice_session_id") or "" if not sid or not _AGENT_STATE.get("voice_live"): sid = uuid.uuid4().hex _AGENT_STATE["voice_session_id"] = sid _AGENT_STATE["voice_turn"] = 0 _AGENT_STATE["voice_live"] = True _AGENT_STATE["ring_stopped"] = True _voice_active_payload = payload print(f"[VoiceTransport] ACCEPT session_id={sid} level={level} " f"has_payload={bool(payload)}") return JSONResponse({ "ok": True, "session_id": sid, "level": level, "risk": float(payload.get("risk", 0.0) or 0.0), }) except Exception as e: print(f"[VoiceTransport] ACCEPT FAIL: {e}") traceback.print_exc() return JSONResponse({"ok": False, "error": str(e)}, status_code=500) _SSVOICE_ROUTES_REGISTERED = True print("[VoiceTransport] Routes registered: POST /ssvoice/turn GET /ssvoice/health " "POST /ssvoice/end POST /ssvoice/accept") AGENT_CSS = APP_CSS + """ @keyframes phone-shake { 0%,100% { transform:rotate(0deg) scale(1); } 10%,30%,50%,70% { transform:rotate(-14deg) scale(1.08); } 20%,40%,60%,80% { transform:rotate(14deg) scale(1.08); } 90% { transform:rotate(-6deg) scale(1); } } @keyframes ring-pulse { 0%,100% { box-shadow:0 0 0 0 rgba(255,69,0,0); } 50% { box-shadow:0 0 0 18px rgba(255,69,0,0); } } @keyframes fade-in-down { from { opacity:0; transform:translateY(-18px) scale(.97); } to { opacity:1; transform:translateY(0) scale(1); } } @keyframes pulse-dot { 0%,100% { opacity:1; transform:scale(1); } 50% { opacity:.25; transform:scale(.65); } } @keyframes scan-line { 0% { transform:translateY(-100%); } 100% { transform:translateY(100vh); } } @keyframes progress-bar { 0% { transform:scaleX(0); } 100% { transform:scaleX(1); } } @keyframes glow-border { 0%,100% { border-color:rgba(255,69,0,.5); } 50% { border-color:rgba(255,69,0,1); box-shadow:0 0 20px rgba(255,69,0,.18); } } @keyframes status-flash { 0%,100% { opacity:1; } 50% { opacity:.55; } } #ss-overlay-root { position:fixed !important; top:0 !important; left:0 !important; width:0 !important; height:0 !important; overflow:visible !important; pointer-events:none !important; z-index:2147483647 !important; } #ss-call-card { position:fixed !important; top:20px !important; right:20px !important; width:360px !important; pointer-events:all !important; background:rgba(6,3,1,.97) !important; backdrop-filter:blur(16px) !important; border:1px solid rgba(255,69,0,.7) !important; border-radius:4px !important; overflow:hidden !important; animation:fade-in-down .3s cubic-bezier(.2,.8,.3,1), ring-pulse 2.4s ease-in-out infinite !important; box-shadow:0 24px 64px rgba(0,0,0,.95) !important; } #ss-call-card .cc-scan { position:absolute !important; top:0 !important; left:0 !important; width:100% !important; height:2px !important; background:linear-gradient(90deg,transparent,rgba(255,69,0,.5),transparent) !important; animation:scan-line 3s linear infinite !important; pointer-events:none !important; } #ss-call-card .cc-header { display:flex !important; align-items:center !important; justify-content:space-between !important; padding:6px 14px !important; background:rgba(255,69,0,.07) !important; border-bottom:1px solid rgba(255,69,0,.18) !important; } #ss-call-card .cc-header-label { font-family:var(--mono) !important; font-size:.5rem !important; letter-spacing:.2em !important; text-transform:uppercase !important; color:rgba(255,69,0,.85) !important; font-weight:700 !important; } #ss-call-card .cc-header-dot { width:6px !important; height:6px !important; border-radius:50% !important; background:#FF4500 !important; animation:pulse-dot 1s ease-in-out infinite !important; } #ss-call-card .cc-body { padding:18px 18px 14px !important; display:flex !important; align-items:center !important; gap:16px !important; } #ss-call-card .cc-phone-icon { width:52px !important; height:52px !important; border-radius:50% !important; background:rgba(255,69,0,.1) !important; border:1px solid rgba(255,69,0,.3) !important; display:flex !important; align-items:center !important; justify-content:center !important; animation:phone-shake 1.2s ease-in-out infinite !important; flex-shrink:0 !important; } #ss-call-card .cc-phone-icon.stopped { animation:none !important; } #ss-call-card .cc-info-title { font-family:var(--serif) !important; font-size:.95rem !important; font-weight:700 !important; color:#fff !important; line-height:1.3 !important; } #ss-call-card .cc-info-sub { font-family:var(--mono) !important; font-size:.6rem !important; color:rgba(255,255,255,.45) !important; margin-top:4px !important; } #ss-call-card .cc-risk-badge { font-family:var(--mono) !important; font-size:.65rem !important; font-weight:700 !important; color:#FF4500 !important; background:rgba(255,69,0,.09) !important; border:1px solid rgba(255,69,0,.3) !important; border-radius:3px !important; padding:2px 8px !important; margin-top:5px !important; display:inline-block !important; animation:status-flash 2s ease-in-out infinite !important; } #ss-call-card .cc-actions { display:flex !important; gap:8px !important; padding:0 18px 18px !important; } .cc-btn-accept { flex:1 !important; background:linear-gradient(135deg,#2e7d32,#388e3c) !important; color:#fff !important; font-family:var(--mono) !important; font-size:.6rem !important; letter-spacing:.1em !important; font-weight:700 !important; text-transform:uppercase !important; border:none !important; border-radius:3px !important; padding:11px 0 !important; cursor:pointer !important; } .cc-btn-accept:hover { background:linear-gradient(135deg,#388e3c,#43a047) !important; } .cc-btn-dismiss { background:rgba(255,255,255,.05) !important; color:rgba(255,255,255,.4) !important; font-family:var(--mono) !important; font-size:.6rem !important; letter-spacing:.1em !important; font-weight:600 !important; text-transform:uppercase !important; border:1px solid rgba(255,255,255,.1) !important; border-radius:3px !important; padding:11px 14px !important; cursor:pointer !important; } .cc-btn-dismiss:hover { background:rgba(255,255,255,.09) !important; } #ss-call-card .cc-progress { height:2px !important; background:rgba(255,69,0,.1) !important; } #ss-call-card .cc-progress-fill { height:100% !important; width:100% !important; background:linear-gradient(90deg,#FF4500,#FF8C00) !important; transform-origin:left !important; animation:progress-bar 4s linear infinite !important; } .agent-page { max-width:860px; margin:0 auto; padding:0 28px 80px; font-family:var(--serif); } .agent-sec-label { font-family:var(--mono); font-size:.62rem; letter-spacing:.14em; text-transform:uppercase; color:#ddd; margin-bottom:10px; font-weight:800; border-left:3px solid var(--orange); padding-left:10px; } .agent-status-card { background:var(--surface); border:1px solid var(--border); border-radius:3px; padding:20px 22px; display:flex; align-items:flex-start; justify-content:space-between; gap:16px; position:relative; overflow:hidden; } .agent-status-card.critical { border-color:rgba(255,69,0,.4); background:rgba(255,69,0,.04); animation:glow-border 2s ease-in-out infinite; } .agent-status-card.high { border-color:rgba(255,140,0,.35); background:rgba(255,140,0,.03); } .agent-dep-chips { display:flex; flex-wrap:wrap; gap:5px; margin-top:10px; } .agent-dep-chip { font-family:var(--mono); font-size:.5rem; letter-spacing:.08em; background:rgba(255,255,255,.05); border:1px solid rgba(255,255,255,.1); border-radius:3px; padding:3px 9px; display:flex; align-items:center; gap:5px; } .agent-call-strip { border-radius:3px; padding:14px 18px; display:flex; align-items:center; gap:14px; background:rgba(255,69,0,.06); border:1px solid rgba(255,69,0,.28); margin:12px 0 0; cursor:pointer; user-select:none; } .strip-phone { animation:phone-shake 1.2s ease-in-out infinite; flex-shrink:0; } .agent-dual-panel { display:grid; grid-template-columns:1fr 1fr; gap:20px; margin-top:8px; align-items:start; } @media (max-width:700px) { .agent-dual-panel { grid-template-columns:1fr; } } .agent-panel { background:var(--surface); border:1px solid var(--border); border-radius:3px; padding:18px 18px 16px; } .agent-panel.voice-panel { border-color:rgba(76,175,80,.15); background:rgba(76,175,80,.02); } .agent-panel.text-panel { border-color:rgba(100,149,237,.18); background:rgba(100,149,237,.02); } .panel-header { font-family:var(--mono); font-size:.52rem; letter-spacing:.16em; text-transform:uppercase; font-weight:700; margin-bottom:10px; display:flex; align-items:center; gap:8px; } .panel-header.voice-header { color:rgba(76,175,80,.85); } .panel-header.text-header { color:rgba(100,149,237,.85); } .experimental-badge { font-family:var(--mono); font-size:.42rem; letter-spacing:.1em; text-transform:uppercase; font-weight:700; color:rgba(255,193,7,.9); background:rgba(255,193,7,.08); border:1px solid rgba(255,193,7,.25); border-radius:3px; padding:2px 7px; margin-left:4px; } .panel-desc { font-family:var(--mono); font-size:.52rem; letter-spacing:.05em; color:rgba(255,255,255,.25); margin-bottom:12px; line-height:1.85; } .chat-display { background:#050505; border:1px solid var(--border); border-radius:3px; padding:16px 18px; min-height:280px; max-height:440px; overflow-y:auto; display:flex; flex-direction:column; gap:16px; } .chat-display .msg-agent { display:flex; flex-direction:column; gap:5px; align-self:flex-start; max-width:96%; } .chat-display .msg-agent .msg-lbl { font-family:var(--mono); font-size:.55rem; letter-spacing:.1em; text-transform:uppercase; color:var(--orange); font-weight:600; display:flex; align-items:center; gap:6px; } .chat-display .msg-agent .msg-lbl .lbl-dot { width:6px; height:6px; border-radius:50%; background:var(--orange); flex-shrink:0; display:inline-block; } .chat-display .msg-agent .msg-lbl .brand-name { display:inline-flex; align-items:baseline; } .chat-display .msg-agent .msg-lbl .brand-zero { font-family:'Times New Roman',Times,serif; font-weight:900; font-size:1.1em; } .chat-display .msg-agent .msg-bubble { background:#080808; border:1px solid var(--border); border-left:3px solid rgba(255,140,0,.5); border-radius:0 3px 3px 3px; padding:14px 16px; font-family:var(--serif); font-size:.84rem; line-height:1.8; color:#e0ddd8; } .chat-display .msg-you { display:flex; flex-direction:column; gap:5px; align-self:flex-end; max-width:78%; } .chat-display .msg-you .msg-lbl { font-family:var(--mono); font-size:.55rem; letter-spacing:.1em; text-transform:uppercase; color:rgba(100,149,237,.7); font-weight:600; text-align:right; display:flex; align-items:center; justify-content:flex-end; gap:6px; } .chat-display .msg-you .msg-lbl .lbl-dot { width:6px; height:6px; border-radius:50%; background:rgba(100,149,237,.7); flex-shrink:0; display:inline-block; } .chat-display .msg-you .msg-bubble { background:#0d0d0d; border:1px solid #2a2a2a; border-right:3px solid rgba(100,149,237,.4); border-radius:3px 0 3px 3px; padding:12px 16px; font-family:var(--serif); font-size:.84rem; line-height:1.75; color:rgba(220,230,255,.85); } .chat-display .msg-system { align-self:center; font-family:var(--mono); font-size:.6rem; font-weight:600; letter-spacing:.06em; color:rgba(76,175,80,.6); background:rgba(76,175,80,.05); border:1px solid rgba(76,175,80,.15); border-radius:3px; padding:6px 12px; text-align:center; } .chat-display .msg-empty { flex:1; display:flex; flex-direction:column; align-items:center; justify-content:center; font-family:var(--serif); font-size:.88rem; color:#333; text-align:center; padding:48px 24px; } #text-chat-input textarea, #text-chat-input input { background:#0d0d0d !important; color:#e0ddd8 !important; font-family:var(--serif) !important; font-size:.86rem !important; border:1px solid #2a2a2a !important; border-radius:3px !important; padding:12px 14px !important; min-height:52px !important; height:52px !important; resize:none !important; overflow:hidden !important; } #voice-conv-box textarea { background:#060606 !important; color:rgba(200,200,200,.7) !important; font-family:var(--mono) !important; font-size:.65rem !important; line-height:1.9 !important; border:1px solid rgba(76,175,80,.12) !important; border-radius:3px !important; } #dispatch-status textarea { background:#060606 !important; color:rgba(255,255,255,.35) !important; font-family:var(--mono) !important; font-size:.63rem !important; border:1px solid var(--border) !important; border-radius:3px !important; } .agent-btn, .agent-btn button { height:40px !important; min-height:40px !important; padding:0 20px !important; font-family:var(--mono) !important; font-size:.65rem !important; letter-spacing:.12em !important; font-weight:700 !important; text-transform:uppercase !important; border-radius:3px !important; border:none !important; cursor:pointer !important; background:#ffffff !important; color:#000000 !important; } .agent-btn:hover, .agent-btn button:hover { background:var(--orange) !important; } .btn-accept, .btn-accept button { background:linear-gradient(135deg,#1b5e20,#2e7d32) !important; color:#fff !important; } .btn-end, .btn-end button { background:linear-gradient(135deg,#7f0000,#b71c1c) !important; color:#fff !important; } .btn-dispatch, .btn-dispatch button { background:#ffffff !important; color:#000 !important; width:100% !important; height:44px !important; } .btn-send, .btn-send button { background:#ffffff !important; color:#000 !important; min-height:44px !important; width:100% !important; } .btn-start, .btn-start button { background:#ffffff !important; color:#000 !important; width:100% !important; height:40px !important; } .live-indicator { display:inline-flex; align-items:center; gap:8px; background:rgba(76,175,80,.08); border:1px solid rgba(76,175,80,.25); border-radius:3px; padding:5px 12px; margin:10px 0 6px; } .live-dot { display:inline-block; width:6px; height:6px; border-radius:50%; background:#4CAF50; animation:pulse-dot 1.2s ease-in-out infinite; } .live-label { font-family:var(--mono); font-size:.55rem; letter-spacing:.1em; color:rgba(76,175,80,.9); font-weight:700; text-transform:uppercase; } #ss-voice-ui { margin-top: 8px; border: 1px solid rgba(76,175,80,.18); border-radius: 3px; background: #050505; padding: 12px; } #ss-wave-canvas { width: 100%; height: 72px; display: block; background: #0a0a0a; border-radius: 2px; border: 1px solid #1a1a1a; } #ss-voice-status-line { font-family: var(--mono); font-size: .56rem; letter-spacing: .1em; text-transform: uppercase; color: rgba(76,175,80,.85); margin-top: 8px; min-height: 16px; } #ss-voice-status-line.mode-listening { color: rgba(76,175,80,.9); } #ss-voice-status-line.mode-recording { color: #FF4500; } #ss-voice-status-line.mode-processing { color: #FF8C00; } #ss-voice-status-line.mode-speaking { color: #64B5F6; } #ss-voice-status-line.mode-error { color: #FF4500; } #ss-playback-audio { position: fixed !important; left: -9999px !important; width: 1px !important; height: 1px !important; opacity: 0 !important; pointer-events: none !important; } #ss-ringtone-out { position:fixed !important; bottom:-200px !important; left:-200px !important; width:1px !important; height:1px !important; overflow:hidden !important; pointer-events:none !important; opacity:0 !important; } """ SSVOICE_CONTROLLER_HTML = r""" """ # ══════════════════════════════════════════════════════════════════════════════ # GLOBAL CALL OVERLAY JS + HTML HELPERS # ══════════════════════════════════════════════════════════════════════════════ def _global_call_js(level: str, risk: float, tune_uri: str) -> str: col = "#FF4500" if level == "CRITICAL" else "#FF8C00" show = "true" if level in ("CRITICAL", "HIGH") else "false" sub = f"Risk Score: {round(risk * 100)}%" return f""" """ def _stop_call_js() -> str: return """ """ _tune_check = (CALLER_TUNE_PATH or "").strip() print(f"[AgentLayer] Caller tune: {'found at ' + _tune_check if _tune_check and os.path.exists(_tune_check) else 'NOT FOUND - synth ring will be used'}") def _agent_sec(text: str, icon: str = "") -> str: ic = (f"" f"{icon.replace('stroke=\"currentColor\"','stroke=\"rgba(255,140,0,.6)\"')}") if icon else "" return f"
{ic}{text}
" def _call_strip_html(level: str, risk: float, visible: bool) -> str: if not visible or level == "LOW": return "" col = "#FF4500" if level == "CRITICAL" else "#FF8C00" label = "CRITICAL ANOMALY DETECTED" if level == "CRITICAL" else "HIGH RISK ANOMALY" risk_pct = round(risk * 100) return ( f"
" f"
" f"" f"" f"
" f"
" f"
{label}
" f"
" f"Incoming call · Risk: " f"{risk_pct}%" f" · Click to open Agent tab" f"
" f"
" f"Open Agent →" f"
" ) def _alert_status_html(level: str, risk: float) -> str: col = {"CRITICAL": "#FF4500", "HIGH": "#FF8C00", "LOW": "#4CAF50"}.get(level, "#888") label = {"CRITICAL": "CRITICAL ANOMALY", "HIGH": "HIGH RISK", "LOW": "NO ALERT"}.get(level, level) card_cls = {"CRITICAL": "critical", "HIGH": "high", "LOW": ""}.get(level, "") risk_pct = round(risk * 100) icon_svg = ( f'' + ( '' '' if level != "LOW" else '' ) + '' ) def _chip(lbl, ok): c, col2 = ("#4CAF50", "rgba(76,175,80,.9)") if ok else ("#444", "rgba(255,255,255,.2)") return (f'' f'' f'{lbl}') sanim = ' style="animation:status-flash 2s ease-in-out infinite;"' if level == "CRITICAL" else "" chips = ( _chip("LFM2.5-Audio", LFM_AUDIO_AVAILABLE) + _chip("Cohere", COHERE_READY) + _chip("cmd-a-03-2025", COHERE_READY) + _chip("HTTP /ssvoice", True) ) return ( f"
" f"
" f"
Alert Status
" f"
{icon_svg}" f"
{label}
" f"
{chips}
" f"
" f"
" f"{risk_pct}%
" ) def _clean_llm_markdown(text: str) -> str: text = text.replace('\r\n', '\n').replace('\r', '\n') cleaned = [] for line in text.split('\n'): m = re.match(r'^#{1,6}\s+(.+)$', line.strip()) if m: cleaned.append(f'**{m.group(1)}**'); continue s = re.sub(r'^[\s]*[-\*]\s+(?!\*)', '', line) if s != line: cleaned.append(s); continue s2 = re.sub(r'^[\s]*[\u2022\u00b7]\s*', '', line) if s2 != line: cleaned.append(s2); continue cleaned.append(line) return re.sub(r'\n{3,}', '\n\n', '\n'.join(cleaned)).strip() _SHERL0CK_LABEL_HTML = 'Sherl0ck' _CHAT_SCROLL_OBSERVER_JS = """ """ def _render_bubble_content(raw: str) -> str: raw = raw.replace("&", "&").replace("<", "<").replace(">", ">") raw = re.sub(r'\*\*(.+?)\*\*', r'\1', raw) raw = re.sub(r'(\d+\.?\d*%?)', r"\1", raw) parts = [] for para in re.split(r'\n\n+', raw): para = para.strip() if not para: continue lines = [l.strip() for l in para.split('\n') if l.strip()] if lines and all(re.match(r'^\d+[\.\)]\s+', l) for l in lines): lh = '' for l in lines: m2 = re.match(r'^(\d+)[\.\)]\s+(.+)$', l) if m2: lh += (f'
{m2.group(1)}.' f'{m2.group(2)}
') parts.append(lh) else: parts.append(f'
{para.replace(chr(10),"
")}
') return ''.join(parts) or f'
{raw}
' def _render_chat_html(history: list) -> str: if not history: return ('
' '
Run an analysis then click Start Chat.
') parts = ['
'] for m in history: role, content = m.get("role", "user"), m.get("content", "") if role == "assistant": rendered = _render_bubble_content(_clean_llm_markdown(content)) parts.append( '
' f'
{_SHERL0CK_LABEL_HTML}
' f'
{rendered}
' ) elif role == "user": safe = content.replace("&","&").replace("<","<").replace(">",">").replace("\n","
") parts.append( '
You
' f'
{safe}
' ) else: safe = content.replace("&","&").replace("<","<").replace(">",">") parts.append(f'
{safe}
') parts.append('
') return ''.join(parts) def _analyze_gpu_duration(video=None, include_llava=False): """Dynamic ZeroGPU reservation: Stage 8 (LLaVA) adds minutes of VLM time.""" return 420 if include_llava else 240 @spaces.GPU(duration=_analyze_gpu_duration) def process_video_with_agent(video, include_llava=False): empty = (no_result_html(), None, None, None, _alert_status_html("LOW", 0.0), "", "", gr.update(visible=False)) if video is None: return empty try: raw = run_pipeline(video, enable_llava=bool(include_llava)) except Exception: tb = traceback.format_exc() print(f"[AgentLayer] Pipeline error:\n{tb}") return (f"
{tb}
", None, None, None, _alert_status_html("LOW", 0.0), "", "", gr.update(visible=False)) r = raw["risk_score"] report_html = build_report_html(raw["report"], r, raw["risk_label"], raw["status"], graph_path=raw.get("graph_path")) from PIL import Image as _PIL fi = _PIL.fromarray(raw["top_frame"]) if raw.get("top_frame") is not None else None jd_str = json.dumps(raw["json_data"], indent=2, default=str) pdf_out = raw.get("pdf_path") payload = build_alert_payload(raw) level = determine_alert_level(r) _AGENT_STATE.update({"payload": payload, "level": level, "voice_live": False, "text_live": False}) print(f"[AgentLayer] State updated: level={level} risk={r:.3f}") if level == "CRITICAL": print("[AgentLayer] CRITICAL - auto-dispatching.") dispatch_all_channels(payload, level) tune = _caller_tune_uri() status_html = _alert_status_html(level, r) strip_html = _call_strip_html(level, r, visible=(level in ("CRITICAL", "HIGH"))) banner_js = _global_call_js(level, r, tune) if level in ("CRITICAL", "HIGH"): def _prewarm(): try: _load_lfm_audio() except Exception: pass _tts_executor.submit(_prewarm) return (report_html, fi, jd_str, pdf_out, status_html, strip_html, banner_js, gr.update(visible=(level in ("CRITICAL", "HIGH")))) def process_video_with_agent_for_agent_tab(video, include_llava=False): result = list(process_video_with_agent(video, include_llava)) result[5] = "" return tuple(result) def accept_call(): """Server-side Accept: mint/reuse session, stop ring, show voice UI. CRITICAL: Browser microphone + SSVoice.start MUST already have been started from the trusted user gesture via SSVoice.accept() (Gradio js= / overlay). Do NOT inject here — Gradio HTML updates often do not re-execute scripts, which was the live failure mode. """ global _voice_active_payload payload = _AGENT_STATE.get("payload", {}) level = _AGENT_STATE.get("level", "LOW") # Prefer session already minted by browser POST /ssvoice/accept sid = _AGENT_STATE.get("voice_session_id") or "" if not sid: sid = uuid.uuid4().hex _AGENT_STATE["voice_session_id"] = sid _AGENT_STATE["voice_turn"] = 0 _AGENT_STATE["voice_live"] = True _AGENT_STATE["ring_stopped"] = True _voice_active_payload = payload print(f"[AgentLayer] accept_call session={sid} level={level} " f"(browser should already own this session)") # UI only — no boot script that starts voice return ( "Call connected. Speak naturally — Sherl0ck is listening.\n", gr.update(visible=False, interactive=False, value="Accept Call"), gr.update(visible=True), gr.update(visible=True), _stop_call_js(), "", # ringtone placeholder (browser-native ring already stopped) ) def end_call(conv: str): _AGENT_STATE["voice_live"] = False sid = _AGENT_STATE.get("voice_session_id", "") _AGENT_STATE["voice_session_id"] = "" bye = "Sherlock signing off Investigation session ended Stay vigilant and safe" updated = (conv or "") + f"[Sherl0ck] {bye}\n" end_js = f""" """ print(f"[AgentLayer] end_call session={sid[:8] if sid else '-'}") return ( updated, gr.update(visible=True, interactive=True, value="Accept Call"), gr.update(visible=False), gr.update(visible=False), end_js, ) def init_text_chat(): payload = _AGENT_STATE.get("payload", {}) level = _AGENT_STATE.get("level", "LOW") jd = payload.get("json_data") or {} print(f"[AgentLayer][Text] init_text_chat: payload_keys={list(payload.keys())} jd_keys={list(jd.keys())[:8]}") if not payload or not _has_pipeline_data(jd): return _render_chat_html([{"role": "system", "content": "No analysis loaded. Run the pipeline on the Analyse tab first."}]) _AGENT_STATE["text_live"] = True text_agent_init(payload, level) return _render_chat_html(_text_conversation_history) def manual_dispatch(): payload = _AGENT_STATE.get("payload", {}) level = _AGENT_STATE.get("level", "LOW") if not payload: return "No analysis loaded Run pipeline on the Analyse tab first" results = dispatch_all_channels(payload, level) ts = datetime.now(timezone.utc).strftime("%H:%M UTC") lines = [f"Dispatch [{level}] {ts}"] lines += [f" {ch.capitalize():<12s} {'OK' if ok else 'FAIL'}" for ch, ok in results.items()] return "\n".join(lines) def _refresh_agent_ui(): payload = _AGENT_STATE.get("payload", {}) level = _AGENT_STATE.get("level", "LOW") risk = float(payload.get("risk", 0.0)) tune = _caller_tune_uri() incoming = level in ("CRITICAL", "HIGH") return ( _alert_status_html(level, risk), "", _global_call_js(level, risk, tune), gr.update(visible=incoming, interactive=incoming, value="Accept Call"), ) # ══════════════════════════════════════════════════════════════════════════════ # GRADIO 5 RUNTIME THEME + OVERRIDES (Hugging Face Space port) # The notebook CSS was written against the Gradio 4 DOM (.tab-nav, .gr-button). # The Space runs Gradio 5.49.1, whose tabs are button[role="tab"] with a # .selected class, whose buttons are .primary/.secondary and whose checkbox is # a styled input[type="checkbox"]. The theme below pins every Gradio design # token to the notebook palette (pitch black / white / orange-saffron) and the # CSS afterwards forces the remaining runtime surfaces. Status colors used # inside analysis results are untouched. # ══════════════════════════════════════════════════════════════════════════════ SS_THEME = ( gr.themes.Base( primary_hue=gr.themes.colors.orange, secondary_hue=gr.themes.colors.neutral, neutral_hue=gr.themes.colors.neutral, font=[gr.themes.GoogleFont("Roboto Mono"), "ui-monospace", "SFMono-Regular", "monospace"], font_mono=[gr.themes.GoogleFont("Roboto Mono"), "ui-monospace", "SFMono-Regular", "monospace"], ).set( body_background_fill="#000000", body_background_fill_dark="#000000", body_text_color="#FFFFFF", body_text_color_dark="#FFFFFF", body_text_color_subdued="#888888", body_text_color_subdued_dark="#888888", block_background_fill="#000000", block_background_fill_dark="#000000", block_border_color="#1e1e1e", block_border_color_dark="#1e1e1e", block_border_width="1px", block_label_background_fill="#000000", block_label_background_fill_dark="#000000", block_label_text_color="#cccccc", block_label_text_color_dark="#cccccc", block_title_text_color="#f2f2f2", block_title_text_color_dark="#f2f2f2", panel_background_fill="#000000", panel_background_fill_dark="#000000", panel_border_color="#1e1e1e", panel_border_color_dark="#1e1e1e", border_color_primary="#1e1e1e", border_color_primary_dark="#1e1e1e", border_color_accent="#FF8C00", border_color_accent_dark="#FF8C00", input_background_fill="#0d0d0d", input_background_fill_dark="#0d0d0d", input_border_color="#2a2a2a", input_border_color_dark="#2a2a2a", input_border_color_focus="#FF8C00", input_border_color_focus_dark="#FF8C00", input_placeholder_color="#555555", input_placeholder_color_dark="#555555", checkbox_background_color="#0d0d0d", checkbox_background_color_dark="#0d0d0d", checkbox_background_color_selected="#FF8C00", checkbox_background_color_selected_dark="#FF8C00", checkbox_border_color="#2a2a2a", checkbox_border_color_dark="#2a2a2a", checkbox_border_color_selected="#FF8C00", checkbox_border_color_selected_dark="#FF8C00", checkbox_label_text_color="#f0f0f0", checkbox_label_text_color_dark="#f0f0f0", button_primary_background_fill="#FF8C00", button_primary_background_fill_dark="#FF8C00", button_primary_background_fill_hover="#ffa033", button_primary_background_fill_hover_dark="#ffa033", button_primary_text_color="#000000", button_primary_text_color_dark="#000000", button_primary_border_color="#FF8C00", button_primary_border_color_dark="#FF8C00", button_secondary_background_fill="#ffffff", button_secondary_background_fill_dark="#ffffff", button_secondary_background_fill_hover="#FF8C00", button_secondary_background_fill_hover_dark="#FF8C00", button_secondary_text_color="#000000", button_secondary_text_color_dark="#000000", button_secondary_border_color="#ffffff", button_secondary_border_color_dark="#ffffff", table_border_color="#1e1e1e", table_border_color_dark="#1e1e1e", table_even_background_fill="#050505", table_even_background_fill_dark="#050505", table_odd_background_fill="#000000", table_odd_background_fill_dark="#000000", loader_color="#FF8C00", loader_color_dark="#FF8C00", color_accent_soft="rgba(255,140,0,0.15)", link_text_color="#FF8C00", link_text_color_dark="#FF8C00", link_text_color_visited="#FF8C00", link_text_color_visited_dark="#FF8C00", ) ) GRADIO5_RUNTIME_CSS = """ /* ══════════ Gradio 5 runtime overrides — notebook palette enforcement ══════════ */ html, body { background:#000000 !important; } .gradio-container, .gradio-container .contain, .gradio-container main { background:#000000 !important; color:#ffffff !important; } .gradio-container a { color:#FF8C00 !important; } /* Tabs — Gradio 5 renders button[role=tab] + .selected (no .tab-nav) */ .gradio-container [role="tablist"], .gradio-container .tab-nav { background:#000000 !important; border-bottom:1px solid #1e1e1e !important; gap:0 !important; padding:0 24px !important; box-shadow:none !important; } .gradio-container button[role="tab"], .gradio-container .tab-nav button { font-family:'Roboto Mono',ui-monospace,monospace !important; font-size:.72rem !important; letter-spacing:.13em !important; color:#999999 !important; background:transparent !important; border:none !important; border-bottom:2px solid transparent !important; padding:14px 0 !important; margin-right:26px !important; text-transform:uppercase !important; font-weight:600 !important; white-space:nowrap !important; box-shadow:none !important; border-radius:0 !important; } .gradio-container button[role="tab"]:hover, .gradio-container .tab-nav button:hover { color:#dddddd !important; } .gradio-container button[role="tab"].selected, .gradio-container button[role="tab"][aria-selected="true"], .gradio-container .tab-nav button.selected { color:#ffffff !important; background:transparent !important; border-bottom:2px solid #FF8C00 !important; font-weight:700 !important; } /* Blocks / panels / forms — never gray */ .gradio-container .block, .gradio-container .form, .gradio-container .panel, .gradio-container fieldset, .gradio-container .gap, .gradio-container section { background:#000000 !important; box-shadow:none !important; } .gradio-container .block { border-color:#1e1e1e !important; } .gradio-container .label-wrap, .gradio-container .label-wrap span, .gradio-container span[data-testid="block-info"], .gradio-container .label-text { color:#cccccc !important; background:transparent !important; font-family:'Roboto Mono',ui-monospace,monospace !important; letter-spacing:.08em !important; } .gradio-container .block-label, .gradio-container .block-title { color:#cccccc !important; background:#000000 !important; } /* Inputs */ .gradio-container input:not([type="checkbox"]):not([type="radio"]), .gradio-container textarea, .gradio-container select { background:#0d0d0d !important; color:#f0f0f0 !important; border:1px solid #2a2a2a !important; border-radius:3px !important; } .gradio-container input:not([type="checkbox"]):focus, .gradio-container textarea:focus, .gradio-container select:focus { border-color:#FF8C00 !important; box-shadow:0 0 0 1px #FF8C00 !important; outline:none !important; } .gradio-container input::placeholder, .gradio-container textarea::placeholder { color:#555555 !important; } /* Checkbox — orange accent, never blue (Gradio 5 styles the raw input) */ .gradio-container input[type="checkbox"] { background:#0d0d0d !important; border:1px solid #2a2a2a !important; accent-color:#FF8C00 !important; } .gradio-container input[type="checkbox"]:checked { background:#FF8C00 !important; border-color:#FF8C00 !important; } .gradio-container input[type="checkbox"]:focus, .gradio-container input[type="checkbox"]:focus-visible { outline:1px solid #FF8C00 !important; box-shadow:0 0 0 1px #FF8C00 !important; } /* Upload / file surfaces — never gray */ .gradio-container .wrap.default, .gradio-container .wrap.default:hover, .gradio-container .wrap-inner, .gradio-container .upload-container, .gradio-container .file-preview, .gradio-container .file-preview > div, .gradio-container [data-testid="file-preview"] { background:#060606 !important; color:#888888 !important; } .gradio-container .wrap.default { border:1px dashed #2a2a2a !important; } .gradio-container .wrap.default:hover, .gradio-container .wrap-inner:hover { border-color:#FF8C00 !important; } .gradio-container video { background:#000000 !important; } /* Buttons — Gradio 5 uses .primary / .secondary */ .gradio-container button.primary { background:#FF8C00 !important; color:#000000 !important; border-color:#FF8C00 !important; } .gradio-container button.primary:hover { background:#ffa033 !important; border-color:#ffa033 !important; } .gradio-container button.secondary { background:#ffffff !important; color:#000000 !important; border:1px solid #ffffff !important; } .gradio-container button.secondary:hover { background:#FF8C00 !important; border-color:#FF8C00 !important; color:#000000 !important; } .gradio-container button:focus, .gradio-container button:focus-visible { outline:1px solid #FF8C00 !important; box-shadow:none !important; } /* Status / progress / misc */ .gradio-container .progress-text, .gradio-container .progress-level-inner { color:#FF8C00 !important; } .gradio-container .progress-level > div { background:#FF8C00 !important; } .gradio-container .meta-text, .gradio-container .meta-text-center { color:#666666 !important; background:#000000 !important; } .gradio-container .prose, .gradio-container .prose p, .gradio-container .prose li { color:#dddddd !important; } .gradio-container ::selection { background:rgba(255,140,0,.35) !important; } .gradio-container *:focus-visible { outline:1px solid #FF8C00 !important; } .gradio-container .toast-body { background:#0d0d0d !important; color:#f0f0f0 !important; border:1px solid #2a2a2a !important; } footer { display:none !important; } """ AGENT_CSS = AGENT_CSS + GRADIO5_RUNTIME_CSS # ══════════════════════════════════════════════════════════════════════════════ # BUILD DEMO — Agent-enabled app # ══════════════════════════════════════════════════════════════════════════════ _T_HOME = "\u00a0\u00a0\u00a0 Home \u00a0\u00a0\u00a0" _T_ANALYSE = "\u00a0\u00a0 Analyse \u00a0\u00a0" _T_AGENT = "\u00a0 Sherl\u09e6ck Agent \u00a0" _T_CONTACT = "\u00a0\u00a0 Contact \u00a0\u00a0" # Fix tab label to match original branding _T_AGENT = "\u00a0 Sherl\u09e6ck Agent \u00a0".replace("\u09e6", "0") # Use proper zero in brand (original used mathematical/special zero sometimes) _T_AGENT = "\u00a0 Sherl\u0030ck Agent \u00a0" with gr.Blocks(title="CrimeSceneSolver", css=AGENT_CSS, theme=SS_THEME) as demo_v2: gr.HTML(NAV_HTML) gr.HTML(SSVOICE_CONTROLLER_HTML) gr.HTML('
') gr.HTML(_CHAT_SCROLL_OBSERVER_JS) # Browser-only ringtone — avoid Gradio StaticAudio/HLS (share-link breakage) ringtone_out = gr.HTML(value="", visible=False, elem_id="ss-ringtone-placeholder") with gr.Tabs(): # ── HOME ──────────────────────────────────────────────────────────── with gr.TabItem(_T_HOME): gr.HTML(HOME_HTML) gr.HTML(FOOTER_HTML) # ── ANALYSE ──────────────────────────────────────────────────────── with gr.TabItem(_T_ANALYSE): _analyse_strip_tmp = gr.HTML("") _analyse_global_js = gr.HTML("") gr.HTML("""
Forensic Analysis

Case File Analysis

Upload CCTV footage to run the full nine-stage forensic pipeline.

""") with gr.Column(): gr.HTML('
') gr.HTML(sec_label("Upload Footage")) vid_in2 = gr.Video(label="", elem_id="vid-up", sources=["upload"], show_label=False, height=175) gr.HTML('
') llava_chk = gr.Checkbox( value=False, label="Enable LLaVA contextual explanation (Stage 8)", info="Adds the final multimodal narrative stage and may take substantially longer.", elem_id="llava-chk", ) gr.HTML('
') gr.HTML('
') ana_btn2 = gr.Button("Analyse Footage", elem_id="ana-btn", elem_classes="gr-button") gr.HTML('
') spinner2 = gr.HTML("") gr.HTML('
') gr.HTML(sec_label("Alert Status")) alert_status_out2 = gr.HTML(value=_alert_status_html("LOW", 0.0)) gr.HTML('
') gr.HTML(sec_label("Risk Assessment & Report")) report_out2 = gr.HTML( value=("
" "Upload footage and press Analyse.
"), elem_id="report-html") gr.HTML('
') gr.HTML(sec_label("Peak Anomaly Frame")) frame_out2 = gr.Image(label="", elem_id="kf-img", show_label=False) gr.HTML('
') gr.HTML(sec_label("Download Forensic PDF Report")) pdf_out2 = gr.File(label="", elem_id="pdf-out", show_label=False, file_types=[".pdf"], interactive=False) gr.HTML('
') gr.HTML(sec_label("Structured JSON Output")) json_out2 = gr.Textbox(label="", elem_id="json-out", show_label=False, lines=20, placeholder="JSON output...") gr.HTML('
') def _show_s(): return gr.update(value=spinner_html()) def _hide_s(): return gr.update(value="") gr.HTML(FOOTER_HTML) # ── SHERL0CK AGENT ──────────────────────────────────────────────── with gr.TabItem(_T_AGENT): gr.HTML("""
Agentic Escalation Layer

Sherl0ck Agent

Run an analysis on the Analyse tab first. After Accept Call, speak naturally — no push-to-talk. Text chat remains independent.

""") with gr.Column(): gr.HTML('
') gr.HTML(_agent_sec("System Status", ICON_SHIELD)) agent_alert_html = gr.HTML(value=_alert_status_html("LOW", 0.0)) gr.HTML('
') call_strip_html = gr.HTML(value="") agent_global_js = gr.HTML("") gr.HTML(_agent_sec("Evidence Dispatch", ICON_DISPATCH)) with gr.Row(equal_height=True): dispatch_btn = gr.Button("Send Report to All Channels", elem_classes=["agent-btn", "btn-dispatch"]) dispatch_status = gr.Textbox(label="", elem_id="dispatch-status", show_label=False, lines=3, placeholder="Dispatch log...", interactive=False) with gr.Row(equal_height=False, elem_classes=["agent-dual-panel"]): # TEXT CHAT COLUMN with gr.Column(elem_classes=["agent-panel", "text-panel"]): gr.HTML("""
Text Chat with Sherl0ck Full depth
Ask Sherl0ck anything about the analysis. Say send report to dispatch evidence to all channels.
""") text_chat_display = gr.HTML( value=('
' '
' 'Run an analysis then click Start Chat.' '
'), elem_id="text-chat-display", visible=True ) start_chat_btn = gr.Button("Start Chat", elem_classes=["agent-btn", "btn-start"]) with gr.Row(equal_height=False, elem_id="chat-input-row"): text_in = gr.Textbox( label="",show_label=False,elem_id="text-chat-input", placeholder="Type your question for Sherl0ck...", lines=1,max_lines=1,submit_btn=False,scale=4 ) send_text_btn = gr.Button("Send", elem_id="send-text-btn", elem_classes=["agent-btn", "btn-send"], scale=1) # VOICE CALL COLUMN with gr.Column(elem_classes=["agent-panel", "voice-panel"]): gr.HTML("""
Voice Call with Sherl0ck LFM2.5-Audio Hands-free VAD
Accept the call once. Microphone stays live with automatic speech detection, barge-in, and continuous turns. Only End Call stops it.
""") with gr.Row(equal_height=True): accept_btn = gr.Button("Accept Call", elem_classes=["agent-btn", "btn-accept"], visible=False, scale=1) end_btn = gr.Button("End Call", elem_classes=["agent-btn", "btn-end"], visible=False, scale=1) voice_conv_box = gr.Textbox( label="", elem_id="voice-conv-box", show_label=False, lines=8, placeholder="Accept the call to begin. Voice transcript appears here.", interactive=False) with gr.Column(visible=False) as voice_controls_col: gr.HTML("""
Sherl0ck - Call Active
Waiting…
""") gr.HTML('
') # agent-page _SCROLL_CHAT_JS = """ () => { let box = document.getElementById("ss-chat-box"); if (box) { box.scrollTop = box.scrollHeight; } } """ start_chat_btn.click(fn=init_text_chat, inputs=None, outputs=[text_chat_display], scroll_to_output=False ).then(fn=None, inputs=None, outputs=None, js=_SCROLL_CHAT_JS) send_text_btn.click( fn=handle_text_input,inputs=[text_in],outputs=[text_chat_display, text_in], scroll_to_output=False ).then(fn=None, inputs=None, outputs=None, js=_SCROLL_CHAT_JS) text_in.submit( fn=handle_text_input,inputs=[text_in],outputs=[text_chat_display, text_in], scroll_to_output=False ).then(fn=None, inputs=None, outputs=None, js=_SCROLL_CHAT_JS) dispatch_btn.click(fn=manual_dispatch, inputs=None, outputs=[dispatch_status]) gr.HTML(FOOTER_HTML) # ── CONTACT ───────────────────────────────────────────────────────── with gr.TabItem(_T_CONTACT): gr.HTML("""
Get In Touch

Contact Us

Reach out with questions, partnership inquiries, or case-specific requests.

""") with gr.Column(): gr.HTML('
') gr.HTML('
') ct_n2 = gr.Textbox(label="Name", placeholder="Your full name") gr.HTML('
') ct_e2 = gr.Textbox(label="Email", placeholder="your@email.com") gr.HTML('
') ct_p2 = gr.Textbox(label="Phone Number", placeholder="+91 XXXXXXXXXX") gr.HTML('
') ct_m2 = gr.Textbox(label="Message", placeholder="Your message...", lines=5) gr.HTML('
') gr.HTML('
') ct_b2 = gr.Button("Send Message", elem_classes="gr-button", elem_id="send-btn") gr.HTML('
') ct_r2 = gr.HTML("") gr.HTML("""
Direct Contact
""") gr.HTML(""" """) gr.HTML('
') ct_b2.click(handle_contact, [ct_n2, ct_e2, ct_p2, ct_m2], ct_r2) gr.HTML(FOOTER_HTML) # Wire analyse pipeline def _maybe_ring(): """Ringtone is browser-native via _global_call_js; keep Gradio output clear.""" level = _AGENT_STATE.get("level", "LOW") if level in ("CRITICAL", "HIGH"): _AGENT_STATE["ring_stopped"] = False print("[Ringtone] browser-native ring armed (no Gradio stream)") return "" _pipe_event = ( ana_btn2.click(fn=_show_s, inputs=None, outputs=spinner2, queue=True ).then(fn=process_video_with_agent, inputs=[vid_in2, llava_chk], outputs=[report_out2, frame_out2, json_out2, pdf_out2, alert_status_out2, _analyse_strip_tmp, _analyse_global_js, accept_btn] ).then(fn=_hide_s, inputs=None, outputs=spinner2, queue=True) ) _pipe_event.then(fn=_refresh_agent_ui, inputs=None, outputs=[agent_alert_html, call_strip_html, agent_global_js, accept_btn]) _ring_event = _pipe_event.then(fn=_maybe_ring, inputs=None, outputs=ringtone_out) _ACCEPT_JS = """ async () => { console.log('[SSVoice] GRADIO_ACCEPT_JS'); try { if (window.SSVoice && window.SSVoice.accept) { await window.SSVoice.accept('gradio-btn'); } else if (window.SSVoice && window.SSVoice.prepare) { await window.SSVoice.prepare(); } } catch (e) { console.log('[SSVoice] GRADIO_ACCEPT_JS_FAIL', e); } return []; } """ accept_btn.click( fn=accept_call, inputs=None, outputs=[voice_conv_box, accept_btn, end_btn, voice_controls_col, agent_global_js, ringtone_out], js=_ACCEPT_JS, cancels=[_ring_event]) end_btn.click(fn=end_call, inputs=[voice_conv_box], outputs=[voice_conv_box, accept_btn, end_btn, voice_controls_col, agent_global_js]) # ══════════════════════════════════════════════════════════════════════════════ # ZERO-GPU ASSET PREFETCH # Background warm-up of the HF cache (mirrors the notebook's startup downloads # without blocking Space startup). Every download is also resolved lazily on # first use, so a missed prefetch never breaks an analysis. # ══════════════════════════════════════════════════════════════════════════════ def _prefetch_assets(): for key in ("AE", "TS_BINARY", "TS_MULTI", "YOLO", "RL_AGENT", "AAE", "LLAVA"): try: Config.PATHS[key] except Exception as e: print(f"[Prefetch] {key} failed: {e}") try: from huggingface_hub import snapshot_download as _snap _snap("LiquidAI/LFM2.5-Audio-1.5B") except Exception as e: print(f"[Prefetch] LFM2.5-Audio failed: {e}") print("[Prefetch] SceneSolver asset cache warm.") if os.environ.get("SCENESOLVER_NO_PREFETCH", "").strip() not in ("1", "true", "True"): threading.Thread(target=_prefetch_assets, daemon=True, name="AssetPrefetch").start() else: print("[Prefetch] Skipped (SCENESOLVER_NO_PREFETCH set - assets resolve lazily).") # ══════════════════════════════════════════════════════════════════════════════ # LAUNCH — Hugging Face Space entry point (Gradio SDK, ZeroGPU hardware) # `demo` is the conventional Blocks name Spaces look for. No share=True, no # server_port — the Spaces runtime configures the public URL and port. # ══════════════════════════════════════════════════════════════════════════════ demo = demo_v2 print(f"[AgentLayer] Gradio version: {getattr(gr, '__version__', '?')}") demo.queue(default_concurrency_limit=1) _launched_app = None try: _launched_app, _local_url, _share_url = demo.launch(prevent_thread_lock=True) except Exception: traceback.print_exc() demo.launch() # blocking fallback (voice HTTP routes unavailable) if _launched_app is not None: try: _register_ssvoice_routes(_launched_app) print("[AgentLayer] /ssvoice routes attached to the live Gradio FastAPI app") except Exception as e: print(f"[AgentLayer] Voice route registration failed: {e}") traceback.print_exc() print("[AgentLayer] demo_v2 is live with continuous HTTP voice transport…") try: demo.block_thread() except (KeyboardInterrupt, OSError): pass