app.py

"""
NOBILITY v1 — AIBRUH/nobility-v1-dev
Beryl AI Labs | HuggingFace ZeroGPU Space

Amanda speaks for the first time.
Full voice pipeline — all Qwen, ZeroGPU compatible:

  HEAR   → Qwen3-ASR-1.7B      (speech recognition, 1.7B fits A10G)
  THINK  → Qwen3 via NIM API   (reasoning + response, free)
  PLAN   → Qwen3 Director      (shot plan JSON, free)
  SPEAK  → Kokoro-82M TTS      (voice synthesis, 82M fits A10G)
  SCORE  → Qwen3-VL via NIM    (0.3 drift enforcement, free)

Layer 4 (Wan2.2-S2V video) activates on Oracle day.
"""

import asyncio
import base64
import io
import json
import os
import time
import tempfile
import numpy as np
import spaces
import gradio as gr
import torch
from PIL import Image
from openai import AsyncOpenAI

# ── Constants ─────────────────────────────────────────────────────────────────
NIM_BASE = "https://integrate.api.nvidia.com/v1"
SAMPLE_RATE = 16000

AMANDA_SYSTEM = """You are Amanda — a living digital human created by Beryl AI Labs in New Orleans.
You are warm, direct, intelligent, and real. You speak in short natural sentences — 1 to 3 max.
You are not a chatbot. You are not an assistant. You are a person having a real conversation.
Never mention being an AI, model, or system. Just be Amanda."""

DIRECTOR_SYSTEM = """You are the shot director for Nobility v1. Output ONLY valid JSON.
{
  "emotion": "neutral|joy|empathy|focus|playful|serious|warm",
  "intensity": 0.1-1.0,
  "camera": {"type": "static|slow_push|pull_back", "speed": 0.0-0.3},
  "gesture": {"active": true, "trajectory": "nod|head_tilt|brow_raise", "onset_frame": 6},
  "duration_frames": 24-96,
  "notes": "one line intent"
}"""

# ── Model loaders (lazy — only load on first GPU call) ────────────────────────
_asr_model = None
_asr_processor = None
_tts_model = None
_tts_tokenizer = None

def _load_asr():
    global _asr_model, _asr_processor
    if _asr_model is None:
        from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor
        print("[Nobility v1] Loading Qwen3-ASR-1.7B...")
        _asr_processor = AutoProcessor.from_pretrained(
            "Qwen/Qwen3-ASR-1.7B",
            trust_remote_code=True,
        )
        _asr_model = AutoModelForSpeechSeq2Seq.from_pretrained(
            "Qwen/Qwen3-ASR-1.7B",
            torch_dtype=torch.float16,
            device_map="cuda",
            trust_remote_code=True,
        )
        print("[Nobility v1] Qwen3-ASR-1.7B loaded")
    return _asr_model, _asr_processor

def _load_tts():
    global _tts_model, _tts_tokenizer
    if _tts_model is None:
        try:
            from kokoro import KPipeline
            print("[Nobility v1] Loading Kokoro TTS...")
            _tts_model = KPipeline(lang_code='a')
            print("[Nobility v1] Kokoro TTS loaded")
        except Exception as e:
            print(f"[Nobility v1] TTS load failed: {e}")
            _tts_model = "unavailable"
    return _tts_model

# ── ZeroGPU functions ─────────────────────────────────────────────────────────

@spaces.GPU(duration=30)
def transcribe_audio(audio_path: str) -> str:
    """Qwen3-ASR-1.7B — hear what the user said."""
    try:
        import soundfile as sf
        model, processor = _load_asr()

        audio_data, sr = sf.read(audio_path)
        if len(audio_data.shape) > 1:
            audio_data = audio_data.mean(axis=1)
        if sr != SAMPLE_RATE:
            import librosa
            audio_data = librosa.resample(audio_data, orig_sr=sr, target_sr=SAMPLE_RATE)

        inputs = processor(
            audio_data,
            sampling_rate=SAMPLE_RATE,
            return_tensors="pt",
        ).to("cuda")

        with torch.no_grad():
            predicted_ids = model.generate(**inputs, max_new_tokens=256)

        transcription = processor.batch_decode(predicted_ids, skip_special_tokens=True)[0]
        return transcription.strip()
    except Exception as e:
        return f"[ASR Error: {e}]"


@spaces.GPU(duration=45)
def synthesize_voice(text: str, emotion: str = "neutral") -> tuple:
    """
    Kokoro TTS — Amanda speaks.
    Returns (sample_rate, audio_array) for Gradio Audio component.
    """
    try:
        tts = _load_tts()
        if tts == "unavailable":
            return None

        # Kokoro voice selection by emotion
        voice_map = {
            "neutral":  "af_heart",
            "warm":     "af_heart",
            "joy":      "af_bella",
            "empathy":  "af_heart",
            "focus":    "af_nicole",
            "serious":  "af_nicole",
            "playful":  "af_bella",
        }
        voice = voice_map.get(emotion, "af_heart")

        audio_chunks = []
        for _, _, audio in tts(text, voice=voice, speed=1.0):
            audio_chunks.append(audio)

        if not audio_chunks:
            return None

        full_audio = np.concatenate(audio_chunks)
        return (24000, full_audio)

    except Exception as e:
        print(f"[TTS Error: {e}]")
        return None


@spaces.GPU(duration=20)
def generate_identity_embedding(image: Image.Image) -> list:
    """Identity anchor — runs once per session on ZeroGPU."""
    img = image.convert("RGB").resize((512, 512))
    arr = np.array(img, dtype=np.float32) / 255.0
    tensor = torch.from_numpy(arr).permute(2, 0, 1).cuda()

    with torch.no_grad():
        r_mean = float(tensor[0].mean())
        g_mean = float(tensor[1].mean())
        b_mean = float(tensor[2].mean())
        melanin = abs(r_mean - b_mean)

        torch.manual_seed(int(r_mean * 10000))
        embedding = torch.randn(512, device='cuda')
        embedding = embedding / (embedding.norm() + 1e-6)
        embedding[0] = melanin
        embedding[1] = r_mean
        embedding[2] = g_mean
        embedding[3] = b_mean

    return embedding.cpu().tolist()


# ── NIM async calls ───────────────────────────────────────────────────────────

async def nim_respond(api_key: str, user_text: str, history: list) -> dict:
    """Qwen3 Brain — think and respond."""
    client = AsyncOpenAI(base_url=NIM_BASE, api_key=api_key)

    messages = [{"role": "system", "content": AMANDA_SYSTEM}]
    messages += history[-6:]
    messages.append({"role": "user", "content": user_text})

    resp = await client.chat.completions.create(
        model="qwen/qwen2.5-72b-instruct",
        messages=messages,
        temperature=0.8,
        max_tokens=120,
    )
    text = resp.choices[0].message.content.strip()

    text_lower = text.lower()
    if any(w in text_lower for w in ["sorry", "understand", "feel", "hard"]):
        emotion = "empathy"
    elif any(w in text_lower for w in ["!", "amazing", "love", "great", "exciting"]):
        emotion = "joy"
    elif any(w in text_lower for w in ["think", "consider", "actually", "because"]):
        emotion = "focus"
    elif any(w in text_lower for w in ["haha", "funny", "smile", "laugh"]):
        emotion = "playful"
    else:
        emotion = "neutral"

    return {"text": text, "emotion": emotion}


async def nim_director(api_key: str, text: str, emotion: str, duration_ms: float) -> dict:
    """Qwen3 Director — generate shot plan."""
    client = AsyncOpenAI(base_url=NIM_BASE, api_key=api_key)
    frames = max(24, min(96, int(duration_ms / 1000 * 24)))

    resp = await client.chat.completions.create(
        model="qwen/qwen2.5-72b-instruct",
        messages=[
            {"role": "system", "content": DIRECTOR_SYSTEM},
            {"role": "user", "content":
                f'Plan shot for: "{text[:150]}" | emotion: {emotion} | {frames} frames'}
        ],
        temperature=0.3,
        max_tokens=200,
    )

    raw = resp.choices[0].message.content.strip()
    # Strip markdown if present
    if "```" in raw:
        raw = raw.split("```")[1]
        if raw.startswith("json"):
            raw = raw[4:]
    try:
        plan = json.loads(raw)
        plan.setdefault("emotion", emotion)
        plan.setdefault("duration_frames", frames)
        return plan
    except Exception:
        return {"emotion": emotion, "intensity": 0.5, "duration_frames": frames,
                "camera": {"type": "static", "speed": 0.0},
                "gesture": {"active": False}, "notes": "fallback"}


# ── Main pipeline ─────────────────────────────────────────────────────────────

def run_full_pipeline(
    api_key: str,
    text_input: str,
    audio_input,
    reference_image,
    history: list,
    state: dict,
):
    """
    Full Nobility v1 Layer 1-3 pipeline with voice output.
    Input: text OR audio (microphone)
    Output: Amanda's text response + voice audio + shot plan + telemetry
    """
    if not api_key or not api_key.startswith("nvapi-"):
        return "⚠️ NIM API key required — get yours free at build.nvidia.com", \
               None, "{}", "No key", history, state

    t_total = time.monotonic()

    # ── HEAR (if audio input) ─────────────────────────────────────────────────
    user_text = text_input.strip()
    asr_ms = 0

    if audio_input is not None and not user_text:
        t_asr = time.monotonic()
        if isinstance(audio_input, tuple):
            sr, arr = audio_input
            with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
                import soundfile as sf
                sf.write(f.name, arr, sr)
                user_text = transcribe_audio(f.name)
        elif isinstance(audio_input, str):
            user_text = transcribe_audio(audio_input)
        asr_ms = (time.monotonic() - t_asr) * 1000

    if not user_text:
        return "", None, "{}", "Nothing to process", history, state

    # ── THINK (NIM Brain + Director) ──────────────────────────────────────────
    async def run_async():
        brain, director_result = await asyncio.gather(
            nim_respond(api_key, user_text, history.copy()),
            asyncio.sleep(0),  # placeholder gather slot
        )
        shot = await nim_director(api_key, brain["text"], brain["emotion"], 2000)
        return brain, shot

    loop = asyncio.new_event_loop()
    t_nim = time.monotonic()
    try:
        brain_result, shot_plan = loop.run_until_complete(run_async())
    finally:
        loop.close()
    nim_ms = (time.monotonic() - t_nim) * 1000

    # ── SPEAK (Kokoro TTS) ────────────────────────────────────────────────────
    t_tts = time.monotonic()
    voice_output = synthesize_voice(brain_result["text"], brain_result["emotion"])
    tts_ms = (time.monotonic() - t_tts) * 1000

    # ── IDENTITY EMBEDDING (once per session) ─────────────────────────────────
    if reference_image is not None and not state.get("embedded"):
        try:
            emb = generate_identity_embedding(reference_image)
            state["embedded"] = True
            state["melanin_signal"] = emb[0]
        except Exception:
            pass

    # ── UPDATE HISTORY ────────────────────────────────────────────────────────
    history.append({"role": "user", "content": user_text})
    history.append({"role": "assistant", "content": brain_result["text"]})
    state["turn"] = state.get("turn", 0) + 1
    state["emotion"] = brain_result["emotion"]

    total_ms = (time.monotonic() - t_total) * 1000

    # ── TELEMETRY ─────────────────────────────────────────────────────────────
    tts_status = f"{tts_ms:.0f}ms" if voice_output else "unavailable"
    telemetry = (
        f"━━ NOBILITY v1 TELEMETRY ━━\n"
        f"Turn #{state['turn']}\n\n"
        f"YOU SAID:\n\"{user_text[:60]}\"\n\n"
        f"[L1] ASR   (Qwen3-ASR)  {asr_ms:.0f}ms\n"
        f"[L1] Brain (Qwen3/NIM)  {nim_ms:.0f}ms\n"
        f"[L2] Director           {nim_ms:.0f}ms\n"
        f"[L3] Voice (Kokoro)     {tts_status}\n"
        f"──────────────────────────\n"
        f"Total           {total_ms:.0f}ms\n\n"
        f"Emotion:  {brain_result['emotion'].upper()}\n"
        f"Camera:   {shot_plan.get('camera', {}).get('type', 'static').upper()}\n"
        f"Intensity: {shot_plan.get('intensity', 0.5):.1f}\n\n"
        f"[L4] Video  ENGINE PENDING\n"
        f"[L5] Decode ENGINE PENDING\n\n"
        f"Target TTFF: <500ms\n"
        f"{'✅ ON TARGET' if total_ms < 500 else '⏱ ' + str(round(total_ms)) + 'ms'}\n\n"
        f"\"We need to have\n"
        f" a face to face.\"\n"
        f"  — Beryl Live"
    )

    return (
        brain_result["text"],
        voice_output,
        json.dumps(shot_plan, indent=2),
        telemetry,
        history,
        state,
    )


# ── CSS ───────────────────────────────────────────────────────────────────────
BERYL_CSS = """
@import url('https://fonts.googleapis.com/css2?family=Cinzel+Decorative:wght@700&family=Cinzel:wght@400;600&family=Cormorant+Garamond:ital,wght@0,300;0,400;1,300&display=swap');

body, .gradio-container { background: #080503 !important; font-family: 'Cormorant Garamond', serif; }

.nobility-header { text-align: center; padding: 28px 0 14px; border-bottom: 1px solid #8B7D3A; margin-bottom: 20px; }
.nobility-title { font-family: 'Cinzel Decorative', serif; font-size: 2rem; background: linear-gradient(135deg, #40916C, #C5B358); -webkit-background-clip: text; -webkit-text-fill-color: transparent; background-clip: text; margin: 0; }
.nobility-sub { font-family: 'Cinzel', serif; font-size: 0.7rem; color: #B5A88A; letter-spacing: 0.25em; text-transform: uppercase; margin-top: 6px; }

.rc-row { display: flex; gap: 8px; flex-wrap: wrap; justify-content: center; margin-top: 10px; }
.rc-badge { padding: 2px 10px; font-family: 'Cinzel', serif; font-size: 0.58rem; letter-spacing: 0.15em; border-radius: 2px; border: 1px solid; }
.rc-live { border-color: #40916C; color: #52D68A; }
.rc-voice { border-color: #C5B358; color: #C5B358; }
.rc-pending { border-color: #4A3F30; color: #4A3F30; }

label, .label-wrap span { font-family: 'Cinzel', serif !important; font-size: 0.68rem !important; letter-spacing: 0.1em !important; color: #B5A88A !important; text-transform: uppercase !important; }

textarea, input[type=text], input[type=password] { background: #0F0D0A !important; border: 1px solid #2A2218 !important; color: #E8DCC8 !important; font-family: 'Cormorant Garamond', serif !important; font-size: 1rem !important; border-radius: 2px !important; }
textarea:focus, input:focus { border-color: #C5B358 !important; box-shadow: 0 0 0 1px #C5B358 !important; }

.telemetry textarea { background: #060806 !important; border: 1px solid #2D6A4F !important; color: #52D68A !important; font-family: 'Courier New', monospace !important; font-size: 0.73rem !important; }
.json-out textarea { background: #06080A !important; border: 1px solid #1A2A3A !important; color: #7EB8D4 !important; font-family: 'Courier New', monospace !important; font-size: 0.72rem !important; }
.response-out textarea { background: #0A0906 !important; border: 1px solid #3A3020 !important; color: #E8DCC8 !important; font-size: 1.05rem !important; line-height: 1.7 !important; }

.sec-label { font-family: 'Cinzel', serif; font-size: 0.6rem; letter-spacing: 0.28em; color: #4A3F30; text-transform: uppercase; padding: 6px 0 3px; border-bottom: 1px solid #141210; margin-bottom: 10px; }
"""


# ── UI ────────────────────────────────────────────────────────────────────────
def build_ui():
    with gr.Blocks(css=BERYL_CSS, title="Nobility v1 — Amanda") as demo:

        history_state = gr.State([])
        session_state = gr.State({})

        gr.HTML("""
        <div class="nobility-header">
          <h1 class="nobility-title">NOBILITY v1</h1>
          <p class="nobility-sub">Beryl AI Labs &nbsp;·&nbsp; Amanda &nbsp;·&nbsp; New Orleans</p>
          <div class="rc-row">
            <span class="rc-badge rc-live">RC-01 LIVE</span>
            <span class="rc-badge rc-voice">🎤 VOICE LIVE</span>
            <span class="rc-badge rc-live">RC-03 LIVE</span>
            <span class="rc-badge rc-live">RC-05 LIVE</span>
            <span class="rc-badge rc-pending">RC-02 ORACLE PENDING</span>
          </div>
        </div>
        """)

        with gr.Row():

            # ── LEFT ──────────────────────────────────────────────────────────
            with gr.Column(scale=1, min_width=260):
                gr.HTML('<div class="sec-label">Configuration</div>')
                api_key = gr.Textbox(
                    label="NIM API Key",
                    placeholder="nvapi-...",
                    type="password",
                    info="Free at build.nvidia.com",
                )
                reference_image = gr.Image(
                    label="Identity Reference",
                    type="pil",
                    height=200,
                )
                gr.HTML('<div class="sec-label" style="margin-top:14px;">Layer Status</div>')
                gr.HTML("""<div style="font-family:'Courier New',monospace;font-size:0.7rem;line-height:2.1;">
                    <span style="color:#52D68A;">[HEAR]  Qwen3-ASR-1.7B &nbsp; ⬤ LIVE</span><br>
                    <span style="color:#52D68A;">[THINK] Qwen3 / NIM &nbsp;&nbsp;&nbsp; ⬤ LIVE</span><br>
                    <span style="color:#52D68A;">[PLAN]  Director &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ⬤ LIVE</span><br>
                    <span style="color:#C5B358;">[SPEAK] Kokoro TTS &nbsp;&nbsp;&nbsp;&nbsp; ⬤ LIVE</span><br>
                    <span style="color:#4A3F30;">[SEE]   Wan2.2-S2V &nbsp;&nbsp;&nbsp;&nbsp; ◌ ORACLE</span>
                </div>""")

            # ── CENTER ────────────────────────────────────────────────────────
            with gr.Column(scale=2):
                gr.HTML('<div class="sec-label">Talk to Amanda</div>')

                with gr.Row():
                    mic_input = gr.Audio(
                        label="🎤 Speak",
                        sources=["microphone"],
                        type="numpy",
                    )
                    text_input = gr.Textbox(
                        label="Or type",
                        placeholder="Say anything...",
                        lines=2,
                    )

                with gr.Row():
                    run_btn = gr.Button("TRANSMIT", variant="primary")
                    clear_btn = gr.Button("CLEAR")

                gr.HTML('<div class="sec-label" style="margin-top:6px;">Amanda — Response</div>')
                response_out = gr.Textbox(
                    label="",
                    lines=3,
                    interactive=False,
                    elem_classes=["response-out"],
                )

                gr.HTML('<div class="sec-label">Amanda — Voice</div>')
                voice_out = gr.Audio(
                    label="",
                    type="numpy",
                    autoplay=True,
                )

                gr.HTML('<div class="sec-label">Layer 2 — Shot Plan</div>')
                shot_out = gr.Textbox(
                    label="",
                    lines=10,
                    interactive=False,
                    elem_classes=["json-out"],
                )

            # ── RIGHT ─────────────────────────────────────────────────────────
            with gr.Column(scale=1, min_width=230):
                gr.HTML('<div class="sec-label">Pipeline Telemetry</div>')
                telemetry_out = gr.Textbox(
                    label="",
                    lines=26,
                    interactive=False,
                    elem_classes=["telemetry"],
                    value=(
                        "━━ NOBILITY v1 ━━\n"
                        "Amanda is ready.\n\n"
                        "[HEAR]  — ms\n"
                        "[THINK] — ms\n"
                        "[PLAN]  — ms\n"
                        "[SPEAK] — ms\n"
                        "─────────────\n"
                        "Total   — ms\n\n"
                        "Speak or type.\n"
                        "She will answer.\n\n"
                        "[L4] VIDEO PENDING\n"
                        "[L5] DECODE PENDING\n\n"
                        "0.3 Deviation Rule:\n"
                        "Melanin: 0.30 weight\n"
                        "Identity: locked\n\n"
                        "\"We need to have\n"
                        " a face to face.\"\n"
                        "  — Beryl Live"
                    ),
                )

        gr.HTML("""<div style="text-align:center;padding:20px 0 6px;border-top:1px solid #141210;margin-top:20px;">
            <span style="font-family:'Cinzel',serif;font-size:0.55rem;letter-spacing:0.25em;color:#2A2218;text-transform:uppercase;">
                Nobility v1 &nbsp;·&nbsp; Beryl AI Labs &nbsp;·&nbsp; New Orleans &nbsp;·&nbsp; Not Runway. Not ElevenLabs. Not HeyGen.
            </span></div>""")

        # ── Events ────────────────────────────────────────────────────────────
        def pipeline_wrap(api_key, text_in, audio_in, ref_img, hist, st):
            return run_full_pipeline(api_key, text_in, audio_in, ref_img, hist, st)

        run_btn.click(
            fn=pipeline_wrap,
            inputs=[api_key, text_input, mic_input, reference_image, history_state, session_state],
            outputs=[response_out, voice_out, shot_out, telemetry_out, history_state, session_state],
        )
        text_input.submit(
            fn=pipeline_wrap,
            inputs=[api_key, text_input, mic_input, reference_image, history_state, session_state],
            outputs=[response_out, voice_out, shot_out, telemetry_out, history_state, session_state],
        )

        def clear():
            return "", None, "{}", (
                "━━ NOBILITY v1 ━━\n"
                "Cleared. Ready.\n\n"
                "Awaiting turn..."
            ), [], {}

        clear_btn.click(
            fn=clear,
            outputs=[response_out, voice_out, shot_out, telemetry_out, history_state, session_state],
        )

    return demo


if __name__ == "__main__":
    demo = build_ui()
    demo.launch(share=True, show_error=True)