import os import logging import torch import gradio as gr from transformers import pipeline import numpy as np from pathlib import Path import base64 # Try soundfile first (official workaround for TorchCodec) try: import soundfile as sf AUDIO_LOADER = "soundfile" except ImportError: try: import librosa AUDIO_LOADER = "librosa" except ImportError: AUDIO_LOADER = None logging.warning("⚠️ No audio decoder available. Install soundfile or librosa.") # ───────────────────────────────────────────────────────────── # 🔧 CONFIGURATION # ───────────────────────────────────────────────────────────── MODEL_ID = os.getenv("MODEL_ID", "badrex/Ethio-ASR-multilingual-600M") SHORT_AUDIO_THRESHOLD_SEC = 180 TARGET_SAMPLE_RATE = 16000 CHUNK_LENGTH_S = 30 STRIDE_LENGTH_S = 5 MIN_CHUNK_DURATION = 2 # CPU optimization os.environ["OMP_NUM_THREADS"] = "4" os.environ["MKL_NUM_THREADS"] = "4" os.environ["OPENBLAS_NUM_THREADS"] = "4" torch.set_num_threads(4) torch.set_num_interop_threads(1) # Logging logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s") logger = logging.getLogger("ethio_asr") import base64 from pathlib import Path def get_logo_html(): """Return centered, slightly larger logo.""" try: logo_path = Path(__file__).parent / "logo.png" if not logo_path.exists(): print(f"[Logo] File not found: {logo_path}") return "" with open(logo_path, "rb") as f: encoded = base64.b64encode(f.read()).decode("utf-8") return f"""
Ethio CropAI Logo
""" except Exception as e: print(f"[Logo Error] {e}") return "" # ───────────────────────────────────────────────────────────── # 🤖 MODEL LOAD (Singleton) # ───────────────────────────────────────────────────────────── _pipe = None def get_pipeline(): global _pipe if _pipe is None: logger.info("⏳ Loading Ethio-ASR model...") try: _pipe = pipeline( "automatic-speech-recognition", model=MODEL_ID, device="cpu", dtype=torch.float32, chunk_length_s=CHUNK_LENGTH_S, stride_length_s=STRIDE_LENGTH_S, batch_size=1 ) logger.info("✅ Model loaded.") except Exception as e: logger.critical(f"❌ Model load failed: {e}") raise RuntimeError("Model init failed") from e return _pipe # ───────────────────────────────────────────────────────────── # 🎙️ AUDIO LOADING (Official TorchCodec Workaround) # ───────────────────────────────────────────────────────────── def load_audio_safe(filepath: str, target_sr: int = TARGET_SAMPLE_RATE) -> tuple[np.ndarray, int]: """ Load audio using soundfile or librosa (NOT torchaudio.load) to avoid TorchCodec. Returns: (mono_audio_array, sample_rate) Per official PyTorch docs: torchaudio.load() now requires TorchCodec [[5]][[6]]. Workaround: use soundfile/librosa which use system ffmpeg. """ if AUDIO_LOADER == "soundfile": # soundfile returns (data: np.ndarray, samplerate: int) data, sr = sf.read(filepath, dtype='float32') # Convert to mono if stereo if data.ndim > 1 and data.shape[1] > 1: data = data.mean(axis=1) return data.astype(np.float32), sr elif AUDIO_LOADER == "librosa": # librosa.load always returns mono by default with sr=None to preserve original data, sr = librosa.load(filepath, sr=None, mono=True) return data.astype(np.float32), sr else: # Fallback: try torchaudio with error handling (may still fail) try: import torchaudio waveform, sr = torchaudio.load(filepath) if waveform.dim() > 1 and waveform.shape[0] > 1: waveform = waveform.mean(dim=0, keepdim=True) return waveform.squeeze().numpy().astype(np.float32), sr except Exception as e: logger.error(f"All audio loaders failed: {e}") raise RuntimeError("Could not load audio. Please use .wav format.") from e def get_audio_duration(filepath: str) -> float: """Get duration using soundfile (reliable, no TorchCodec).""" try: import soundfile as sf info = sf.info(filepath) return info.duration if hasattr(info, 'duration') else -1 except: return -1 # Fail open def split_audio_into_chunks(audio_array: np.ndarray, sample_rate: int, chunk_duration: int = SHORT_AUDIO_THRESHOLD_SEC) -> list[np.ndarray]: """Split long audio into non-overlapping chunks.""" total_samples = len(audio_array) samples_per_chunk = chunk_duration * sample_rate if total_samples <= samples_per_chunk: return [audio_array] chunks = [] for start in range(0, total_samples, samples_per_chunk): end = min(start + samples_per_chunk, total_samples) chunk = audio_array[start:end] if len(chunk) / sample_rate >= MIN_CHUNK_DURATION: chunks.append(chunk) logger.info(f"✂️ Split into {len(chunks)} chunks") return chunks # ───────────────────────────────────────────────────────────── # 🎙️ TRANSCRIPTION LOGIC # ───────────────────────────────────────────────────────────── def transcribe_audio(audio_input) -> tuple[str, str]: # Normalize input path if isinstance(audio_input, dict) and "path" in audio_input: audio_path = audio_input["path"] elif isinstance(audio_input, str) and Path(audio_input).is_file(): audio_path = audio_input else: return "⚠️ No valid audio file provided.", "🔴 Invalid input" # Check duration (soundfile-based, no TorchCodec) duration = get_audio_duration(audio_path) is_long = duration > 0 and duration > SHORT_AUDIO_THRESHOLD_SEC logger.info(f"📊 Duration: {duration:.2f}s, long={is_long}") try: pipe = get_pipeline() # Load audio with safe loader (no torchaudio.load) logger.info(f"📥 Loading: {audio_path}") audio_array, sample_rate = load_audio_safe(audio_path) if audio_array.size == 0: return "❌ Empty audio file.", "🔴 Data error" # Resample if needed (using torch for consistency) if sample_rate != TARGET_SAMPLE_RATE: try: import torchaudio.functional as F import torch as th audio_tensor = th.from_numpy(audio_array).float() if audio_tensor.dim() == 1: audio_tensor = audio_tensor.unsqueeze(0) audio_resampled = F.resample(audio_tensor, orig_freq=sample_rate, new_freq=TARGET_SAMPLE_RATE) audio_array = audio_resampled.squeeze().numpy() sample_rate = TARGET_SAMPLE_RATE except Exception as e: logger.warning(f"Resampling failed: {e}; proceeding with original sample rate") # Route based on duration if not is_long: # Direct processing logger.info("🎯 Processing directly...") input_dict = {"raw": audio_array, "sampling_rate": sample_rate} result = pipe(input_dict) text = result.get("text", "").strip() else: # Chunked processing logger.info(f"🔄 Splitting {duration:.1f}s audio...") chunks = split_audio_into_chunks(audio_array, sample_rate, SHORT_AUDIO_THRESHOLD_SEC) chunk_texts = [] for i, chunk_array in enumerate(chunks): logger.info(f"🔍 Chunk {i+1}/{len(chunks)}") input_dict = {"raw": chunk_array, "sampling_rate": sample_rate} chunk_result = pipe(input_dict) chunk_text = chunk_result.get("text", "").strip() if chunk_text: chunk_texts.append(chunk_text) text = " ".join(chunk_texts).strip() if not text: return "🔇 No speech detected.", "🟡 Empty result" logger.info("✅ Transcription complete.") return text, "🟢 Success" except torch.OutOfMemoryError: return "❌ Memory limit. Try <2 min audio.", "🔴 OOM" except RuntimeError as e: if any(kw in str(e).lower() for kw in ["torchcodec", "ffmpeg", "decode", "format"]): return "❌ Audio format issue. Try .wav file.", "🔴 Format error" raise except Exception as e: logger.error(f"❌ Error: {e}", exc_info=True) msg = str(e)[:150] + ("..." if len(str(e)) > 150 else "") return f"❌ Error: {msg}", "🔴 Failed" # ───────────────────────────────────────────────────────────── # 🖥️ GRADIO UI (Gradio 6.x Verified) # ───────────────────────────────────────────────────────────── with gr.Blocks(title="Ethio CropAI ASR", analytics_enabled=False) as demo: logo_html = get_logo_html() gr.HTML(logo_html) gr.Markdown("# Ethio CropAI-ASR: Multilingual Speech-to-Text") gr.Markdown( " AI Voice Interface for Ethiopian Farmers by Ethio CropAI.\n" "Supported: `Amharic` | `Afaan Oromo` | `Tigrinya` | `Sidaama` | `Wolaytta`" ) with gr.Row(): with gr.Column(scale=1): audio_input = gr.Audio( type="filepath", label="🎙️ Upload or Record", sources=["upload", "microphone"], format="wav", buttons=["download", "share"] ) submit_btn = gr.Button("🔍 Transcribe", variant="primary", size="lg") status_md = gr.Markdown("🟢 Ready") with gr.Column(scale=2): output_text = gr.Textbox(label="📝 Result", lines=8, interactive=False) gr.Markdown( "💡 *Tips:*\n" "- ✅ Formats: `.wav`, `.mp3`, `.flac`, `.m4a` (via ffmpeg)\n" "- ⚡ ≤3 min: direct processing\n" "- 🔄 >3 min: auto-chunked\n" "- 🎯 16kHz mono = best accuracy" ) submit_btn.click(fn=transcribe_audio, inputs=[audio_input], outputs=[output_text, status_md]) with gr.Accordion("🔌 API Docs", open=False): gr.Markdown( "```python\n" "from gradio_client import Client\n" "c = Client('https://YOUR-SPACE.hf.space')\n" "text, status = c.predict(audio_input='audio.wav', fn_index=0)\n" "```\n" "🔗 Interactive: `https://YOUR-SPACE.hf.space/gradio/api/`" ) # ───────────────────────────────────────────────────────────── # 🚀 LAUNCH # ───────────────────────────────────────────────────────────── if __name__ == "__main__": demo.launch( share=False, show_error=True, server_port=int(os.getenv("PORT", 7860)), server_name="0.0.0.0", css=".gradio-container { max-width: 900px !important; margin: auto !important; }", theme=gr.themes.Soft(primary_hue="blue", neutral_hue="gray"), )