Update app.py

app.py

@@ -1,8 +1,13 @@
 # -*- coding: utf-8 -*-
 """
-ROBOTSMALI VIDEO CAPTIONING V16 — ULTRA STABLE ✅
-Aucun ImageMagick — Rendu sous-titres via PIL
-Compatibilité NeMo multi-versions pour tous les modèles Mali (RNNT + CTC)
+ROBOTSMALI VIDEO CAPTIONING — V21 (Stable)
+- Alignement parfait pour Soloba (CTC)
+- Découpage fluide pour Soloni (RNNT)
+- QuartzNet supporté sans crash
+- Filtrage Bambara phonétique (retire français)
+- Sous-titres style Netflix
+- Durée vidéo exacte (plus d'allongement)
+- Compatible Google Colab + Kali + Linux
 """
 
 import os, tempfile
@@ -14,6 +19,7 @@ from PIL import Image, ImageDraw, ImageFont
 import gradio as gr
 from huggingface_hub import snapshot_download
 from moviepy.editor import VideoFileClip, CompositeVideoClip, ImageClip
+
 from nemo.collections import asr as nemo_asr
 from ctc_segmentation import ctc_segmentation, CtcSegmentationParameters, prepare_text
 
@@ -36,121 +42,145 @@ def load_model(name):
         return _model_cache[name]
     repo, mode = MODELS[name]
     path = snapshot_download(repo, local_dir_use_symlinks=False)
-    nemo_file = [os.path.join(path,f) for f in os.listdir(path) if f.endswith(".nemo")][0]
-    model = nemo_asr.models.EncDecCTCModelBPE.restore_from(nemo_file) if mode=="ctc" else nemo_asr.models.EncDecHybridRNNTCTCBPEModel.restore_from(nemo_file)
+    nemo = [os.path.join(path, f) for f in os.listdir(path) if f.endswith(".nemo")][0]
+    model = nemo_asr.models.EncDecCTCModelBPE.restore_from(nemo) if mode=="ctc" \
+        else nemo_asr.models.EncDecHybridRNNTCTCBPEModel.restore_from(nemo)
     model.to(DEVICE).eval()
     _model_cache[name] = model
     return model
 
-def get_vocab(model):
-    key = id(model)
-    if key in _vocab_cache: return _vocab_cache[key]
-    vocab = model.tokenizer.vocab if hasattr(model.tokenizer,"vocab") else model.decoder.vocabulary
-    vocab = list(vocab.keys()) if isinstance(vocab,dict) else list(vocab)
-    _vocab_cache[key] = vocab
-    return vocab
-
-# ✅ **TRANSCRIPTION UNIFIÉE** (Corrige Hypothesis / tuple / dict)
-def transcribe_text(model, wav):
-    out = model.transcribe([wav])[0]
-
-    if hasattr(out, "text"):
-        return out.text.strip()
+def extract_audio(video, wav):
+    VideoFileClip(video).audio.write_audiofile(
+        wav, fps=16000, codec="pcm_s16le", ffmpeg_params=["-ac","1"], logger=None
+    )
+
+def clean_audio(wav):
+    audio, sr = sf.read(wav)
+    if audio.ndim == 2: audio = audio.mean(1)
+    audio,_ = librosa.effects.trim(audio, top_db=35)
+    out = wav.replace(".wav","_clean.wav")
+    sf.write(out, audio, sr)
+    return out, audio, sr
+
+def transcribe(model, wav):
+    o = model.transcribe([wav])[0]
+    return o.text.strip() if hasattr(o,"text") else str(o).strip()
+
+# ---------- FILTRAGE BAMBARA ---------- #
+def keep_bambara_words(words):
+    filtered=[]
+    for w in words:
+        w2=w.lower()
+        if any(ch in w2 for ch in ["ɛ","ɔ","ŋ"]) or sum(c in "aeiou" for c in w2)>=2:
+            filtered.append(w)
+    return filtered
+
+MAX_WORDS=4; MAX_CHARS=45; MAX_DURATION=3.4
+
+def group(spans):
+    subs=[]; buf=[]
+    def push(b):
+        if b: subs.append((b[0][0], b[-1][1], " ".join(x[2] for x in b)))
+    for w in spans:
+        test=buf+[w]; txt=" ".join(x[2] for x in test)
+        dur=test[-1][1]-test[0][0]
+        if len(test)>MAX_WORDS or len(txt)>MAX_CHARS or dur>MAX_DURATION:
+            push(buf); buf=[w]
+        else:
+            buf=test
+    push(buf); return subs
 
-    if isinstance(out, dict) and "text" in out:
-        return out["text"].strip()
+# ---------- ALIGNEMENT CTC (Soloba + QuartzNet) ---------- #
+def align_ctc(model, audio, sr, text):
+    words = keep_bambara_words(text.split())
+    if not words: return []
+    x = torch.tensor(audio).float().unsqueeze(0).to(DEVICE)
+    ln = torch.tensor([x.shape[1]]).to(DEVICE)
+    total = len(audio)/sr
+    with torch.no_grad():
+        logits, _ = model(input_signal=x, input_signal_length=ln)
+    frames = logits.shape[1]
+    if frames <= 2: return []
+    vocab = list(model.tokenizer.vocab.keys())
+    cfg = CtcSegmentationParameters(); cfg.char_list=vocab
+
+    out = prepare_text(cfg, words)
+    gt = out[0] if isinstance(out, (list,tuple)) else out
+
+    timing, _, _ = ctc_segmentation(cfg, logits.cpu().numpy()[0], gt)
+    tps = total / float(frames)
 
-    if isinstance(out, (list, tuple)) and len(out) > 0 and isinstance(out[0], str):
-        return out[0].strip()
+    spans=[]
+    for i in range(len(words)):
+        st=float(timing[i])*tps
+        en=float(timing[i+1])*tps if i+1<len(timing) else total
+        spans.append((st,en,words[i]))
+    return group(spans)
 
-    return str(out).strip()
+# ---------- ALIGNEMENT RNNT (Soloni) ---------- #
+def rnnt_vad(text, audio, sr):
+    intervals = librosa.effects.split(audio, top_db=28)
+    words = keep_bambara_words(text.split())
+    if not intervals or not words:
+        return [(0,len(audio)/sr,text)]
+    spans=[]; idx=0
+    total_audio=sum(e-s for s,e in intervals)
+    for s,e in intervals:
+        seg_d=(e-s)/sr
+        k=max(1,int(len(words)*((e-s)/total_audio)))
+        chunk=words[idx:idx+k]; idx+=k
+        if not chunk: continue
+        parts=[chunk[i:i+MAX_WORDS] for i in range(0,len(chunk),MAX_WORDS)]
+        step=seg_d/len(parts); base=s/sr
+        for j,p in enumerate(parts):
+            st=base+j*step; en=base+(j+1)*step
+            spans.append((st,en," ".join(p)))
+    return group(spans)
 
-def draw_subtitle(text, w, h):
-    bg = Image.new("RGBA", (w, int(h*0.12)), (0,0,0,180))
-    draw = ImageDraw.Draw(bg)
-    try: font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", size=h//18)
-    except: font = ImageFont.load_default()
-    tw, th = draw.textsize(text, font=font)
-    draw.text(((w-tw)//2,(h*0.12-th)//2), text, fill="white", font=font)
+# ---------- RENDER SUBTITLES ---------- #
+def draw_sub(text,W,H):
+    bg=Image.new("RGBA",(W,int(H*0.12)),(0,0,0,180))
+    d=ImageDraw.Draw(bg)
+    try: font=ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",size=max(18,H//18))
+    except: font=ImageFont.load_default()
+    box=d.textbbox((0,0),text,font)
+    tw=box[2]-box[0]; th=box[3]-box[1]
+    d.text(((W-tw)//2,(H*0.12-th)//2),text,font=font,fill="white")
     return bg
 
-def render_subtitles(video, subs):
-    out = "RobotsMali_Subtitled.mp4"
-    base = VideoFileClip(video)
-    W,H=base.size
-    layers=[ImageClip(np.array(draw_subtitle(t.upper(),W,H))).set_start(s).set_duration(e-s).set_pos(("center","bottom")) for s,e,t in subs]
-    CompositeVideoClip([base]+layers).write_videofile(out,codec="libx264",audio_codec="aac",fps=base.fps)
-    base.close()
+def burn(video,subs):
+    out="RobotsMali_Subtitled.mp4"
+    base=VideoFileClip(video); W,H=base.size; dur=base.duration
+    layers=[]
+    for s,e,t in subs:
+        s=max(0,min(s,dur)); e=max(0,min(e,dur))
+        if e<=s: continue
+        img=draw_sub(t.upper(),W,H)
+        layers.append(ImageClip(np.array(img)).set_start(s).set_duration(e-s).set_pos(("center","bottom")))
+    CompositeVideoClip([base]+layers).set_duration(dur).write_videofile(out,codec="libx264",audio_codec="aac",fps=base.fps)
     return out
 
-def extract_audio(v,w): VideoFileClip(v).audio.write_audiofile(w,fps=16000,codec="pcm_s16le",ffmpeg_params=["-ac","1"],logger=None)
-
-def clean_audio(w):
-    a,sr=sf.read(w)
-    if a.ndim==2:a=a.mean(1)
-    a,_=librosa.effects.trim(a,top_db=30)
-    thr=np.percentile(np.abs(a),5)
-    a=np.where(np.abs(a)<thr,0,a)
-    o=w.replace(".wav","_clean.wav"); sf.write(o,a,sr)
-    return o,a,sr
-
-MAX_WORDS=4; MAX_CHARS=45; MAX_DURATION=3.5
-
-def group(sp):
-    subs=[];buf=[]
-    def push(b): subs.append((b[0][0],b[-1][1]," ".join(x[2] for x in b)))
-    for w in sp:
-        test=buf+[w]; txt=" ".join(x[2] for x in test); dur=test[-1][1]-test[0][0]
-        if len(test)>MAX_WORDS or len(txt)>MAX_CHARS or dur>MAX_DURATION: push(buf); buf=[w]
-        else: buf=test
-    push(buf); return subs
-
-# ✅ **UNIVERSAL ALIGNMENT SAFE (plus jamais unpack error)**
-def align_ctc(model,audio,sr,text):
-    words=text.split()
-    x=torch.tensor(audio).float().unsqueeze(0).to(DEVICE)
-    ln=torch.tensor([x.shape[1]]).to(DEVICE)
-    total=len(audio)/sr
-    with torch.no_grad(): logits,loglen=model(input_signal=x,input_signal_length=ln)
-    vocab=set(get_vocab(model))
-    words=[w for w in words if all(c in vocab for c in w)]
-    if not words:return []
-    cfg=CtcSegmentationParameters(); cfg.char_list=list(vocab)
-    gt = prepare_text(cfg, words)[0]   # ✅ always safe
-    timing,_,_=ctc_segmentation(cfg,logits.cpu().numpy()[0],gt)
-    tps=total/loglen.cpu().numpy()[0]
-    return group([(timing[i]*tps,(timing[i+1]*tps if i+1<len(timing) else total),words[i]) for i in range(len(words))])
-
-def rnnt_vad(text,audio,sr):
-    it=librosa.effects.split(audio,top_db=25)
-    w=text.split()
-    if len(it)==0: total=len(audio)/sr; return [(0,total,text)]
-    spans=[]
-    for s,e in it:
-        seg=w[:MAX_WORDS] if len(w)>=MAX_WORDS else w; w=w[len(seg):]
-        if not seg:break
-        spans.append((s/sr,e/sr," ".join(seg)))
-    return group(spans)
-
+# ---------- PIPELINE ---------- #
 def pipeline(video, model_name):
     try:
         tmp=os.path.join(tempfile.gettempdir(),"audio.wav")
         extract_audio(video,tmp)
         clean,audio,sr=clean_audio(tmp)
         model=load_model(model_name)
-        text=transcribe_text(model,clean)
+        text=transcribe(model,clean)
         mode=MODELS[model_name][1]
         subs = align_ctc(model,audio,sr,text) if mode=="ctc" else rnnt_vad(text,audio,sr)
-        if not subs:return "⚠️ Aucun mot utilisable.",None
-        out=render_subtitles(video,subs)
-        return "✅ Terminé !",out
+        if not subs: return "⚠️ Aucun sous-titre utilisable.",None
+        return "✅ Terminé !", burn(video,subs)
     except Exception as e:
         return f"❌ ERREUR : {e}",None
 
-with gr.Blocks(title="RobotsMali V16") as demo:
-    gr.Markdown("# ⚡ RobotsMali V16 — Sous-titrage Bambara (Style Netflix, Sans ImageMagick)")
-    v=gr.Video(); m=gr.Dropdown(list(MODELS.keys()),value="Soloba V1 (CTC)")
-    b=gr.Button("▶️ Générer"); s=gr.Markdown(); o=gr.Video()
-    b.click(pipeline,[v,m],[s,o])
+with gr.Blocks(title="RobotsMali V21 — Bambara Aligné") as demo:
+    gr.Markdown("# ⚡ RobotsMali V21 — Sous-titrage Bambara Stable")
+    video=gr.Video()
+    model=gr.Dropdown(list(MODELS.keys()),value="Soloba V1 (CTC)")
+    run=gr.Button("▶️ Générer")
+    status=gr.Markdown(); out=gr.Video()
+    run.click(pipeline,[video,model],[status,out])
 
 demo.launch(share=True)