Upload 9 files

app.py

@@ -1,64 +1,64 @@
-import gradio as gr
-from tts_engine import TTSEngine
-from translation import Translator, CustomTranslator
-from data_manager import save_uploaded_file, convert_to_jsonl
-from training.train_translation import train_from_jsonl
-from stt_engine import STTEngine
-import os
-
-# Init engines
-stt_engine = STTEngine()
-tts_engine = TTSEngine(use_coqui=True)
-translator = CustomTranslator() if os.path.exists("./training/outputs/model") else Translator()
-
-LANGUAGES = ["english", "yoruba", "igbo", "hausa", "pidgin", "esan", "tiv", "calabar", "benin"]
-
-def handle_conversation(audio, src_lang, tgt_lang, clone_voice):
-    if audio is None:
-        return "", None
-
-    # Step 1: Speech to Text
-    text = stt_engine.transcribe(audio, language=src_lang)
-
-    # Step 2: Translate
-    translated = translator.translate(text, src_lang, tgt_lang)
-
-    # Step 3: Text to Speech
-    audio_path = tts_engine.speak(translated, lang=tgt_lang, voice_clone=clone_voice)
-
-    return translated, audio_path
-
-def admin_upload(file):
-    file_path = save_uploaded_file(file, file.name)
-    jsonl_path = convert_to_jsonl(file_path)
-    train_from_jsonl(jsonl_path)
-    return "✅ Training done. Model updated!"
-
-with gr.Blocks(title="🌍 Two-Way Voice Translator") as demo:
-    gr.Markdown("# 🌍 Nigerian Two-Way Voice Translator")
-    with gr.Tab("Translator"):
-        with gr.Row():
-            src_lang = gr.Dropdown(LANGUAGES, value="english", label="Speaker A Language")
-            tgt_lang = gr.Dropdown(LANGUAGES, value="hausa", label="Speaker B Language")
-
-        with gr.Row():
-            audio_in = gr.Audio(sources=["microphone"], type="filepath", label="🎤 Speak")
-            translated = gr.Textbox(label="Translated Text", interactive=False)
-            audio_out = gr.Audio(label="🔊 Translation Audio")
-
-        clone_voice = gr.Checkbox(value=False, label="🎙️ Use my cloned voice (if my_voice.wav exists)")
-
-        audio_in.change(
-            handle_conversation,
-            inputs=[audio_in, src_lang, tgt_lang, clone_voice],
-            outputs=[translated, audio_out]
-        )
-
-    with gr.Tab("Admin (Training)"):
-        gr.Markdown("Upload Hausa ↔ English data (.csv, .xlsx, .tsv, .jsonl)")
-        file_in = gr.File(label="Upload dataset")
-        train_btn = gr.Button("🚀 Train Model")
-        output_box = gr.Textbox(label="Training Status")
-        train_btn.click(admin_upload, inputs=file_in, outputs=output_box)
-
-demo.launch()
+import gradio as gr
+from tts_engine import TTSEngine
+from translation import Translator, CustomTranslator
+from data_manager import save_uploaded_file, convert_to_jsonl
+from training.train_translation import train_from_jsonl
+from stt_engine import STTEngine
+import os
+
+# Init engines
+stt_engine = STTEngine()
+tts_engine = TTSEngine(use_coqui=True)
+translator = CustomTranslator() if os.path.exists("./training/outputs/model") else Translator()
+
+LANGUAGES = ["english", "yoruba", "igbo", "hausa", "pidgin", "esan", "tiv", "calabar", "benin"]
+
+def handle_conversation(audio, src_lang, tgt_lang, clone_voice):
+    if audio is None:
+        return "", None
+
+    # Step 1: Speech to Text
+    text = stt_engine.transcribe(audio, language=src_lang)
+
+    # Step 2: Translate
+    translated = translator.translate(text, src_lang, tgt_lang)
+
+    # Step 3: Text to Speech
+    audio_path = tts_engine.speak(translated, lang=tgt_lang, voice_clone=clone_voice)
+
+    return translated, audio_path
+
+def admin_upload(file):
+    file_path = save_uploaded_file(file, file.name)
+    jsonl_path = convert_to_jsonl(file_path)
+    train_from_jsonl(jsonl_path)
+    return "✅ Training done. Model updated!"
+
+with gr.Blocks(title="🌍 Two-Way Voice Translator") as demo:
+    gr.Markdown("# 🌍 Nigerian Two-Way Voice Translator")
+    with gr.Tab("Translator"):
+        with gr.Row():
+            src_lang = gr.Dropdown(LANGUAGES, value="english", label="Speaker A Language")
+            tgt_lang = gr.Dropdown(LANGUAGES, value="hausa", label="Speaker B Language")
+
+        with gr.Row():
+            audio_in = gr.Audio(sources=["microphone"], type="filepath", label="🎤 Speak")
+            translated = gr.Textbox(label="Translated Text", interactive=False)
+            audio_out = gr.Audio(label="🔊 Translation Audio")
+
+        clone_voice = gr.Checkbox(value=False, label="🎙️ Use my cloned voice (if my_voice.wav exists)")
+
+        audio_in.change(
+            handle_conversation,
+            inputs=[audio_in, src_lang, tgt_lang, clone_voice],
+            outputs=[translated, audio_out]
+        )
+
+    with gr.Tab("Admin (Training)"):
+        gr.Markdown("Upload Hausa ↔ English data (.csv, .xlsx, .tsv, .jsonl)")
+        file_in = gr.File(label="Upload dataset")
+        train_btn = gr.Button("🚀 Train Model")
+        output_box = gr.Textbox(label="Training Status")
+        train_btn.click(admin_upload, inputs=file_in, outputs=output_box)
+
+demo.launch()

data_manager.py

@@ -0,0 +1,42 @@
+import os
+import pandas as pd
+import json
+
+DATA_DIR = "./training/data"
+os.makedirs(DATA_DIR, exist_ok=True)
+
+def save_uploaded_file(file, filename):
+    """Save uploaded file to data folder"""
+    path = os.path.join(DATA_DIR, filename)
+    with open(path, "wb") as f:
+        f.write(file.read())
+    return path
+
+def convert_to_jsonl(file_path, src_col="src", tgt_col="tgt"):
+    """Detect file type (csv, xlsx, tsv, jsonl) and normalize to JSONL"""
+    ext = os.path.splitext(file_path)[-1].lower()
+    data = None
+
+    if ext == ".csv":
+        data = pd.read_csv(file_path)
+    elif ext == ".xlsx":
+        data = pd.read_excel(file_path)
+    elif ext == ".tsv":
+        data = pd.read_csv(file_path, sep="\t")
+    elif ext == ".jsonl":
+        return file_path  # already JSONL
+    else:
+        raise ValueError("Unsupported file format")
+
+    # Ensure we have two columns: src (Hausa) and tgt (English)
+    if len(data.columns) < 2:
+        raise ValueError("Dataset must have at least two columns")
+
+    data = data.rename(columns={data.columns[0]: "src", data.columns[1]: "tgt"})
+    jsonl_path = file_path.rsplit(".", 1)[0] + ".jsonl"
+
+    with open(jsonl_path, "w", encoding="utf-8") as f:
+        for _, row in data.iterrows():
+            f.write(json.dumps({"src": str(row["src"]), "tgt": str(row["tgt"])}, ensure_ascii=False) + "\n")
+
+    return jsonl_path

main.py

@@ -0,0 +1,5 @@
+from assistant import VoiceTranslatorAssistant
+
+if __name__ == "__main__":
+    assistant = VoiceTranslatorAssistant("config.yaml")
+    assistant.run()

stt_engine.py

@@ -0,0 +1,35 @@
+import torch
+from transformers import pipeline
+
+class STTEngine:
+    def __init__(self, model_name="openai/whisper-small"):
+        # Load Whisper pipeline for automatic speech recognition
+        device = 0 if torch.cuda.is_available() else -1
+        self.asr = pipeline(
+            "automatic-speech-recognition",
+            model=model_name,
+            device=device
+        )
+
+    def transcribe(self, audio_path, language="en"):
+        """
+        Transcribe audio file to text using Whisper.
+        Args:
+            audio_path (str): path to .wav file
+            language (str): ISO code ('en', 'ha', 'yo', 'ig')
+        """
+        if audio_path is None:
+            return ""
+
+        # Map Nigerian language names to Whisper codes
+        lang_map = {
+            "english": "en",
+            "hausa": "ha",
+            "yoruba": "yo",
+            "igbo": "ig",
+            "pidgin": "pcm",
+        }
+        whisper_lang = lang_map.get(language.lower(), "en")
+
+        result = self.asr(audio_path, generate_kwargs={"language": whisper_lang})
+        return result["text"].strip()

translation.py

@@ -1,5 +1,5 @@
 import yaml
-from transformers import MarianMTModel, MarianTokenizer
+from transformers import MarianMTModel, MarianTokenizer, pipeline
 
 # ---------------- Load config ----------------
 CONFIG_FILE = "config.yaml"
@@ -79,3 +79,14 @@ class Translator:
                 return text
             else:
                 return f"(⚠️ Dictionary doesn't support {input_lang}→{output_lang})"
+
+
+
+class CustomTranslator:
+    def __init__(self, model_dir="./training/outputs/model"):
+        self.tokenizer = MarianTokenizer.from_pretrained(model_dir)
+        self.model = MarianMTModel.from_pretrained(model_dir)
+        self.pipeline = pipeline("translation", model=self.model, tokenizer=self.tokenizer)
+
+    def translate(self, text):
+        return self.pipeline(text)[0]["translation_text"]

tts_engine.py

@@ -1,61 +1,61 @@
-import os
-from TTS.api import TTS
-
-LANG_MAP = {
-    "english": "en",
-    "yoruba": "en",
-    "hausa": "en",
-    "igbo": "en",
-    "pidgin": "en",
-    "esan": "en",
-    "tiv": "en",
-    "calabar": "en",
-    "benin": "en",
-    "french": "fr-fr",
-    "portuguese": "pt-br"
-}
-
-class TTSEngine:
-    def __init__(self, use_coqui=False):
-        self.use_coqui = use_coqui
-        self.tts = None
-
-        if self.use_coqui:
-            self.tts = TTS(
-                "tts_models/multilingual/multi-dataset/your_tts",
-                progress_bar=False,
-                gpu=False
-            )
-
-    def speak(self, text, lang="english", voice_clone=False):
-        if not text:
-            return None
-
-        out_file = "output.wav"
-
-        if self.use_coqui:
-            lang_code = LANG_MAP.get(lang.lower(), "en")
-
-            if voice_clone and os.path.exists("my_voice.wav"):
-                # ✅ clone your own voice
-                self.tts.tts_to_file(
-                    text=text,
-                    file_path=out_file,
-                    speaker_wav="my_voice.wav",
-                    language=lang_code
-                )
-            else:
-                # ✅ fallback to a demo synthetic voice (pretrained speaker)
-                self.tts.tts_to_file(
-                    text=text,
-                    file_path=out_file,
-                    speaker="female-en-5",   # <-- safe default
-                    language=lang_code
-                )
-        else:
-            import pyttsx3
-            engine = pyttsx3.init()
-            engine.save_to_file(text, out_file)
-            engine.runAndWait()
-
-        return out_file
+import os
+from TTS.api import TTS
+
+LANG_MAP = {
+    "english": "en",
+    "yoruba": "en",
+    "hausa": "en",
+    "igbo": "en",
+    "pidgin": "en",
+    "esan": "en",
+    "tiv": "en",
+    "calabar": "en",
+    "benin": "en",
+    "french": "fr-fr",
+    "portuguese": "pt-br"
+}
+
+class TTSEngine:
+    def __init__(self, use_coqui=False):
+        self.use_coqui = use_coqui
+        self.tts = None
+
+        if self.use_coqui:
+            self.tts = TTS(
+                "tts_models/multilingual/multi-dataset/your_tts",
+                progress_bar=False,
+                gpu=False
+            )
+
+    def speak(self, text, lang="english", voice_clone=False):
+        if not text:
+            return None
+
+        out_file = "output.wav"
+
+        if self.use_coqui:
+            lang_code = LANG_MAP.get(lang.lower(), "en")
+
+            if voice_clone and os.path.exists("my_voice.wav"):
+                # clone your own voice
+                self.tts.tts_to_file(
+                    text=text,
+                    file_path=out_file,
+                    speaker_wav="my_voice.wav",
+                    language=lang_code
+                )
+            else:
+                # fallback to a neutral synthetic voice
+                self.tts.tts_to_file(
+                    text=text,
+                    file_path=out_file,
+                    speaker_wav=None,   # Let model pick default embedding
+                    language=lang_code
+                )
+        else:
+            import pyttsx3
+            engine = pyttsx3.init()
+            engine.save_to_file(text, out_file)
+            engine.runAndWait()
+
+        return out_file