"""Pieces of the Inter-2-Audio playground: audio, chat layout, parsing, the result summary. Kept apart from app.py (which loads the model and the UI) so they can be tested without a GPU. The task presets (prompts + JSON schemas) live in tasks.py. Decoding defaults: audio before text, greedy, repetition penalty 1.0. """ from __future__ import annotations import ast import html import json import re from typing import Any import numpy as np TARGET_SR = 16_000 ENCODER_WINDOW_S = 30.0 DEFAULT_REPETITION_PENALTY = 1.0 DEFAULT_MAX_NEW_TOKENS = 768 def build_messages(user_prompt: str, system_prompt: str = "", audio_first: bool = True) -> list[dict[str, Any]]: """One user turn holding the audio placeholder and the prompt, plus a system turn only if given.""" text = {"type": "text", "text": user_prompt} audio = {"type": "audio", "audio": "clip"} # placeholder; the waveform goes to the processor separately messages = [{"role": "system", "content": [{"type": "text", "text": system_prompt}]}] if system_prompt else [] return messages + [{"role": "user", "content": [audio, text] if audio_first else [text, audio]}] class AudioInputError(ValueError): """A problem with the uploaded audio; its message is safe to show to the user.""" def load_audio(path: str, truncate: bool = True) -> tuple[np.ndarray, float]: """Mono float32 at 16 kHz, cut to the encoder window. Returns (waveform, original duration in s).""" import librosa try: wav, _ = librosa.load(path, sr=TARGET_SR, mono=True) except Exception as exc: raise AudioInputError("Could not read this audio file. Try a WAV, MP3 or FLAC file.") from exc duration = len(wav) / TARGET_SR if duration == 0: raise AudioInputError("The audio file is empty.") if duration > ENCODER_WINDOW_S + 1e-3: if not truncate: raise AudioInputError(f"The clip is {duration:.1f}s; the model hears at most {ENCODER_WINDOW_S:.0f}s.") wav = wav[: int(ENCODER_WINDOW_S * TARGET_SR)] return wav.astype(np.float32), duration def parse_json(text: str) -> Any | None: """The model's JSON answer, tolerating a code fence, surrounding prose or Python-style literals.""" text = text.strip() fence = re.match(r"```(?:json)?\s*(.*?)\s*```", text, re.DOTALL) if fence: text = fence.group(1).strip() match = re.search(r"[\[{].*[\]}]", text, re.DOTALL) for candidate in (text, match.group(0) if match else None): if candidate is None: continue try: return json.loads(candidate) except json.JSONDecodeError: pass try: return ast.literal_eval(candidate) except (ValueError, SyntaxError): pass return None def _esc(value: Any) -> str: return html.escape(str(value)) def _badge(level: Any) -> str: if not level: return "" return f'{_esc(level)}' def _tile(caption: str, value: Any, level: Any = None) -> str: return ( f'
{_esc(payload["evidence"])}
' if payload.get("evidence") else "" meta_html = f'' if meta else "" return ( f'{_esc(emotion["rationale"])}
' if "signals" in payload: rows = [ f'Could not parse JSON from the output. See the raw text.
' elif isinstance(payload, dict) and "primary_emotion" in payload: body = _primary_emotion(payload) elif isinstance(payload, dict) and payload.keys() & {"emotions", "engagement_status", "signals", "delivery"}: body = _combined(payload) elif isinstance(payload, dict): body = 'Parsed JSON with keys: ' + ", ".join(_esc(k) for k in payload) + "
" else: body = f'Parsed JSON ({type(payload).__name__}).
' notes_html = "".join(f'{n}
' for n in notes) return f'