Update app.py

app.py

@@ -1,164 +1,269 @@
-import torch
-import gradio as gr
+import os, io, json, hashlib, datetime, threading
+from pathlib import Path
+
 import numpy as np
-from scipy.io import wavfile
-from datasets import load_dataset, Dataset
-from transformers import pipeline, AutoProcessor, WhisperForConditionalGeneration
-from diffusers import StableDiffusionPipeline
-import sounddevice as sd
+import gradio as gr
 import librosa
-import hashlib
-import datetime
-
-# ==============
-# FARTORY ENGINE
-# ==============
-class FartFactory:
-    def __init__(self):
-        # Core models (all run locally)
-        self.fart_detector = pipeline("audio-classification", model="superb/hubert-base-superb-er")
-        self.whisper = WhisperForConditionalGeneration.from_pretrained("openai/whisper-small")
-        self.processor = AutoProcessor.from_pretrained("openai/whisper-small")
-        self.llm = pipeline("text-generation", model="ehartford/dolphin-2.1-mistral-7b")
-        self.sd = StableDiffusionPipeline.from_pretrained("stabilityai/stable-diffusion-2-1", torch_dtype=torch.float16)
-        self.tts = pipeline("text-to-speech", model="facebook/fastspeech2-en-ljspeech")
-        
-        # Fart database on Hugging Face
-        self.db = load_dataset("fart_db", split="train", token=False)  # Public dataset
-        
-    def detect_fart(self, audio):
-        """Classify fart type using audio features"""
-        results = self.fart_detector(audio)
-        return max(results, key=lambda x: x['score'])['label']
-
-    def transcribe_idea(self, audio):
-        """Convert speech to text idea capsule"""
-        inputs = self.processor(audio, return_tensors="pt", sampling_rate=16000)
-        predicted_ids = self.whisper.generate(inputs.input_features)
-        return self.processor.batch_decode(predicted_ids, skip_special_tokens=True)[0]
-
-    def generate_art(self, idea):
-        """Create unimaginable fart-inspired art"""
-        return self.sd(f"absurd surrealism: {idea} by Salvador Dali and H.R. Giger").images[0]
-
-    def cartman_review(self, idea, fart_type):
-        """Generate Cartman-style VC roast"""
-        prompt = f"[Cartman voice] As a ruthless VC, I'm reviewing this fart-backed idea: '{idea}'. "
-        prompt += f"This {fart_type} deserves: "
-        return self.llm(prompt, max_new_tokens=60)[0]['generated_text']
-
-    def laugh_to_mint(self, audio):
-        """Detect laughter for mint trigger"""
-        features = librosa.feature.mfcc(y=audio, sr=16000)
-        return np.mean(features) > 0.5  # Simplified laugh detection
-
-    def create_fartifact(self, audio, idea, art, review, fart_type):
-        """Generate complete Fartifact™ capsule"""
-        timestamp = datetime.datetime.now().isoformat()
-        audio_hash = hashlib.sha256(audio.tobytes()).hexdigest()
-        
-        return {
-            "id": f"fart-{audio_hash[:8]}",
-            "timestamp": timestamp,
-            "audio": audio,
-            "idea": idea,
-            "fart_type": fart_type,
-            "art": art,
-            "review": review,
-            "laugh_verified": True,
-            "token_value": len(idea) * 0.001 + len(review) * 0.002  # Mock value algorithm
-        }
-
-    def record_and_process(self, duration=5):
-        """Full capture-to-capsule workflow"""
-        # Record audio
-        sr = 16000
-        audio = sd.rec(int(duration * sr), samplerate=sr, channels=1)
-        sd.wait()
-        
-        # Detect fart characteristics
-        fart_type = self.detect_fart((sr, audio))
-        
-        # Transcribe idea
-        idea = self.transcribe_idea((sr, audio))
-        
-        # Generate art
-        art = self.generate_art(idea)
-        
-        # Cartman review
-        review = self.cartman_review(idea, fart_type)
-        
-        # TTS Cartman audio
-        review_audio = self.tts(review)
-        
-        # Create final capsule
-        capsule = self.create_fartifact(audio, idea, art, review, fart_type)
-        
-        return capsule, review_audio
-
-# ==============
-# FARTORY UI
-# ==============
-factory = FartFactory()
-
-with gr.Blocks(title="FARTORY™ v1.0") as ui:
-    gr.Markdown("## 🏭 FART-AS-INFRASTRUCTURE™")
-    
+from PIL import Image
+
+import torch
+from transformers import pipeline, AutoProcessor, WhisperForConditionalGeneration
+
+from datasets import Dataset, Features, Value, Audio, Image as HFImage
+
+# ---------- Runtime / device ----------
+def device_map():
+    if torch.cuda.is_available():
+        return {"device": "cuda", "dtype": torch.float16}
+    if torch.backends.mps.is_available():  # Apple
+        return {"device": "mps", "dtype": torch.float16}
+    return {"device": "cpu", "dtype": torch.float32}
+
+RUNTIME = device_map()
+
+# ---------- Persistence (local ETL) ----------
+DATA_DIR = Path("./data")
+ART_DIR = DATA_DIR / "art"
+DB_PARQUET = DATA_DIR / "fart_db.parquet"
+DATA_DIR.mkdir(parents=True, exist_ok=True)
+ART_DIR.mkdir(parents=True, exist_ok=True)
+_db_lock = threading.Lock()
+
+DB_FEATURES = Features({
+    "id": Value("string"),
+    "timestamp": Value("string"),
+    "audio_path": Value("string"),
+    "idea": Value("string"),
+    "fart_type": Value("string"),
+    "art_path": Value("string"),
+    "review": Value("string"),
+    "laugh_verified": Value("bool"),
+    "token_value": Value("float32"),
+})
+
+def load_ledger() -> Dataset:
+    if DB_PARQUET.exists():
+        return Dataset.from_parquet(str(DB_PARQUET))
+    return Dataset.from_dict({k: [] for k in DB_FEATURES.keys()}).cast(DB_FEATURES)
+
+def save_ledger(ds: Dataset) -> None:
+    with _db_lock:
+        ds.to_parquet(str(DB_PARQUET))
+
+LEDGER = load_ledger()
+
+# ---------- Models (lazy init where expensive) ----------
+# Audio classifier (emotion model repurposed as proxy demo; deterministic label selection)
+_audio_cls = pipeline(
+    "audio-classification",
+    model="superb/hubert-base-superb-er",
+    device=0 if RUNTIME["device"] == "cuda" else -1
+)
+
+# Whisper small (local)
+_processor = AutoProcessor.from_pretrained("openai/whisper-small")
+_asr = WhisperForConditionalGeneration.from_pretrained("openai/whisper-small")
+_asr = _asr.to(RUNTIME["device"]).to(dtype=RUNTIME["dtype"])
+
+# Optional: Stable Diffusion (will be disabled if no GPU; safe fallback image if CPU-only)
+_sd_pipe = None
+if RUNTIME["device"] == "cuda":
+    try:
+        from diffusers import StableDiffusionPipeline
+        _sd_pipe = StableDiffusionPipeline.from_pretrained(
+            "stabilityai/stable-diffusion-2-1",
+            torch_dtype=torch.float16
+        ).to("cuda")
+    except Exception:
+        _sd_pipe = None
+
+# ---------- Utility ----------
+def _mono_float32(wave: np.ndarray) -> np.ndarray:
+    if wave.ndim == 2:
+        wave = np.mean(wave, axis=1)
+    wave = wave.astype(np.float32)
+    # normalize if outside [-1,1]
+    mx = np.max(np.abs(wave)) + 1e-8
+    if mx > 1.0:
+        wave = wave / mx
+    return wave
+
+def _hash_bytes(b: bytes) -> str:
+    return hashlib.sha256(b).hexdigest()
+
+def _deterministic_value(s: str) -> float:
+    # Map SHA256 -> [0, 1) via first 8 bytes
+    h = hashlib.sha256(s.encode()).digest()[:8]
+    n = int.from_bytes(h, "big")
+    return (n % 10_000_000) / 10_000_000.0
+
+# ---------- Core pipeline ----------
+def detect_fart(audio_tuple):
+    sr, wave = audio_tuple
+    wave = _mono_float32(wave)
+    out = _audio_cls({"array": wave, "sampling_rate": sr})
+    # Take top label
+    label = max(out, key=lambda x: float(x["score"]))["label"]
+    return label
+
+def transcribe_idea(audio_tuple):
+    sr, wave = audio_tuple
+    wave = _mono_float32(wave)
+    inputs = _processor(wave, sampling_rate=sr, return_tensors="pt")
+    with torch.inference_mode():
+        input_feats = inputs.input_features.to(RUNTIME["device"])
+        pred_ids = _asr.generate(input_feats)
+    text = _processor.batch_decode(pred_ids, skip_special_tokens=True)[0].strip()
+    return text
+
+def generate_art(idea: str, art_id: str) -> str:
+    out_path = ART_DIR / f"{art_id}.png"
+    if _sd_pipe is None:
+        # Fallback: render text as simple image (CPU-safe)
+        img = Image.new("RGB", (768, 512), (0, 0, 0))
+        # Minimal pillow text (no additional deps): leave clean black image with no text to avoid font issues
+        img.save(out_path)
+        return str(out_path)
+    prompt = f"surreal, absurd, high-contrast, orange accents on black, {idea}"
+    with torch.inference_mode():
+        img = _sd_pipe(prompt).images[0]
+    img.save(out_path)
+    return str(out_path)
+
+def review_text(idea: str, fart_type: str) -> str:
+    # Lightweight deterministic roast without LLM (keyless)
+    base = f"VC Review | type={fart_type} | idea='{idea[:120]}'"
+    score = _deterministic_value(idea + fart_type)
+    tier = "reject" if score < 0.33 else ("revise" if score < 0.66 else "fund")
+    return f"{base} | decision={tier} | score={score:.3f}"
+
+def laugh_to_mint(audio_tuple) -> bool:
+    sr, wave = audio_tuple
+    wave = _mono_float32(wave)
+    # Energy-based laugh heuristic (deterministic threshold on log-energy variance)
+    frame = max(2048, int(0.05 * sr))
+    hop = frame // 2
+    rmse = librosa.feature.rms(y=wave, frame_length=frame, hop_length=hop)[0]
+    v = float(np.var(np.log(rmse + 1e-8)))
+    return v > 0.25
+
+def create_capsule(audio_tuple, idea: str, art_path: str, review: str, fart_type: str):
+    sr, wave = audio_tuple
+    wave = _mono_float32(wave)
+    timestamp = datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc).isoformat()
+    audio_bytes = wave.tobytes()
+    audio_hash = _hash_bytes(audio_bytes)
+    cid = f"fart-{audio_hash[:8]}"
+
+    # Persist audio as WAV (float32 PCM via soundfile; avoid PyAV)
+    audio_path = DATA_DIR / f"{cid}.wav"
+    try:
+        import soundfile as sf
+        sf.write(str(audio_path), wave, sr)
+    except Exception:
+        # Fallback: numpy npy
+        np.save(str(DATA_DIR / f"{cid}.npy"), wave)
+        audio_path = DATA_DIR / f"{cid}.npy"
+
+    value = 0.1 + 0.2 * _deterministic_value(idea) + 0.3 * _deterministic_value(review)
+
+    capsule = {
+        "id": cid,
+        "timestamp": timestamp,
+        "audio_path": str(audio_path),
+        "idea": idea,
+        "fart_type": fart_type,
+        "art_path": art_path,
+        "review": review,
+        "laugh_verified": True,
+        "token_value": float(value),
+    }
+    return capsule
+
+def add_to_ledger(capsule: dict) -> Dataset:
+    global LEDGER
+    with _db_lock:
+        LEDGER = LEDGER.add_item(capsule)
+        save_ledger(LEDGER)
+        return LEDGER
+
+# ---------- Gradio UI ----------
+THEME_CSS = """
+.gradio-container {background-color:#0b0b0b}
+button, .tab-nav button {border-radius:10px}
+:root {--button-primary-background-fill:#ff7a00; --button-primary-text-color:#000}
+label, .markdown-body, .label-wrap, .tabs {color:#ffb26b}
+"""
+
+state_capsule = gr.State(value=None)
+
+def process_handler(audio):
+    # audio: dict or tuple depending on Gradio; normalize to (sr, np.ndarray)
+    if isinstance(audio, dict):
+        sr, wave = audio["sample_rate"], np.array(audio["data"], dtype=np.float32)
+    else:
+        sr, wave = audio  # already (sr, np.ndarray)
+
+    audio_tuple = (sr, wave)
+    fart_type = detect_fart(audio_tuple)
+    idea = transcribe_idea(audio_tuple)
+    art_id = _hash_bytes(wave.tobytes())[:8]
+    art_path = generate_art(idea, art_id)
+    review = review_text(idea, fart_type)
+    minted = laugh_to_mint(audio_tuple)
+    capsule = create_capsule(audio_tuple, idea, art_path, review, fart_type if minted else "unverified")
+
+    # Persist capsule only on Mint click; here we just stage it.
+    state_capsule.value = capsule
+
+    # Return UI-friendly payloads
+    fart_audio_value = (sr, wave)  # gr.Audio expects (sr, np.ndarray)
+    art_img = Image.open(art_path)
+    review_audio_none = None  # No TTS (keyless)
+    return fart_audio_value, idea, art_img, review, review_audio_none, json.dumps(capsule, indent=2)
+
+def mint_handler(staged_json):
+    # `staged_json` is the JSON textbox value from last process; prevent mint without stage
+    try:
+        capsule = json.loads(staged_json)
+    except Exception:
+        return None, "Mint failed: no staged capsule."
+    ds = add_to_ledger(capsule)
+    # Build small table for UI
+    tail = ds.to_pandas().tail(10)[["id", "idea", "fart_type", "token_value"]]
+    return tail, f"Minted {capsule['id']}"
+
+with gr.Blocks(title="FARTORY™ v1.0", css=THEME_CSS, theme=gr.themes.Default()) as ui:
+    gr.Markdown("## FART-AS-INFRASTRUCTURE™ • orange/black")
     with gr.Row():
-        record_btn = gr.Button("💨 RECORD FART + IDEA", variant="primary")
-        mint_btn = gr.Button("🪙 MINT FARTWORK NFT", variant="secondary")
-    
+        record_btn = gr.Button("💨 PROCESS MIC INPUT", variant="primary")
+        mint_btn = gr.Button("🪙 MINT FARTWORK", variant="secondary")
+
     with gr.Tab("Fartifact™ Capsule"):
-        gr.Markdown("### FART-TO-ART PRODUCTION LINE")
-        output_audio = gr.Audio(label="Fart Audio", interactive=False)
+        output_audio = gr.Audio(label="Fart Audio", interactive=False, type="numpy")
         output_idea = gr.Textbox(label="Idea Capsule")
-        output_art = gr.Image(label="Unimaginable Art")
-        output_review = gr.Textbox(label="Cartman VC Verdict")
-        tts_output = gr.Audio(label="Cartman Roast", interactive=False)
-    
-    with gr.Tab("Fart Database"):
-        gr.Markdown("### GLOBAL FART LEDGER")
-        dataset_view = gr.Dataframe(headers=["ID", "Idea", "Fart Type", "Token Value"])
-    
-    # Production line
+        output_art = gr.Image(label="Art", type="pil")
+        output_review = gr.Textbox(label="VC Verdict")
+        tts_output = gr.Audio(label="Review Audio (off)", interactive=False, type="numpy")
+        staged_capsule = gr.Textbox(label="Staged Capsule (JSON)", interactive=False)
+
+    with gr.Tab("Fart Ledger"):
+        dataset_view = gr.Dataframe(headers=["id", "idea", "fart_type", "token_value"])
+        mint_status = gr.Markdown("")
+
+    mic = gr.Audio(sources=["microphone"], label="Microphone", type="numpy")
+
     record_btn.click(
-        fn=factory.record_and_process,
-        outputs=[output_audio, output_idea, output_art, output_review, tts_output]
+        fn=process_handler,
+        inputs=[mic],
+        outputs=[output_audio, output_idea, output_art, output_review, tts_output, staged_capsule]
     )
-    
-    # Minting mechanism
+
     mint_btn.click(
-        fn=lambda c: factory.db.add_item(c),
-        inputs=[output_audio],
-        outputs=[dataset_view]
+        fn=mint_handler,
+        inputs=[staged_capsule],
+        outputs=[dataset_view, mint_status]
     )
 
-# ==============
-# FART LEDGER
-# ==============
-class FartDatabase:
-    def __init__(self):
-        self.schema = {
-            "id": Value("string"),
-            "timestamp": Value("string"),
-            "audio": Audio(sampling_rate=16000),
-            "idea": Value("string"),
-            "fart_type": Value("string"),
-            "art": Image(),
-            "review": Value("string"),
-            "token_value": Value("float32")
-        }
-        self.dataset = Dataset.from_dict({k: [] for k in self.schema.keys()})
-    
-    def add_item(self, capsule):
-        """Add Fartifact™ to decentralized ledger"""
-        self.dataset = self.dataset.add_item(capsule)
-        self.dataset.push_to_hub("fart_db", private=False)
-        return self.dataset.to_pandas().tail(10)
-
-# ==============
-# START FARTING
-# ==============
 if __name__ == "__main__":
-    factory.db = FartDatabase()
-    ui.launch(server_port=7860, share=True)
+    ui.launch(server_port=7860, share=False)