agent.py

import os
import asyncio
import httpx
import xml.etree.ElementTree as ET
import re as _re
from datetime import datetime
import wave
import time as _time
import boto3
from botocore.client import Config
from livekit import rtc
try:
    from zoneinfo import ZoneInfo
except ImportError:
    ZoneInfo = None

from dotenv import load_dotenv
from livekit.agents import AutoSubscribe, JobContext, JobRequest, WorkerOptions, cli, RoomInputOptions
from livekit.agents import AgentSession, Agent
from livekit.plugins import groq, silero, openai
from tts_mms import MmsTTS
from tts_api import ApiTTS
from tts_edge import EdgeTTS

try:
    from livekit.agents import function_tool
except ImportError:
    def function_tool(f):
        return f

load_dotenv()

# ──────────────────────────────────────────────────────────────────────────────
# Voice Registry  (label, tts config, STT language, region, flag)
# ──────────────────────────────────────────────────────────────────────────────
VOICE_INFO = {
    # ─── East Africa ──────────────────────────────────────────────────────────
    "swh": {
        "label": "Kiswahili", "type": "mms", "repo": "facebook/mms-tts-swh",
        "lang_name": "Kiswahili", "whisper_lang": "sw",
        "region": "East Africa", "flag": "🇰🇪",
        "greeting": "Habari yako! Naitwa Fiti. Niko tayari kuanza, ungetaka tuongee kuhusu nini leo?",
        "sample": "Unaweza kuniambia zaidi?",
    },
    "swh_v2": {
        "label": "Kiswahili Daudi", "type": "mms", "repo": "Benjamin-png/swahili-mms-tts-finetuned",
        "lang_name": "Kiswahili", "whisper_lang": "sw",
        "region": "East Africa", "flag": "🇹🇿",
        "greeting": "Habari yako! Naitwa Daudi. Niko tayari kuanza, ungetaka tuongee kuhusu nini leo?",
        "sample": "Ninafurahi kukusaidia.",
    },
    "swh_ke_r": {
        "label": "Kiswahili Rafiki", "type": "edge", "voice_id": "sw-KE-RafikiNeural",
        "lang_name": "Kiswahili", "whisper_lang": "sw",
        "region": "East Africa", "flag": "🇰🇪",
        "greeting": "Hujambo na karibu sana. Naitwa Rafiki. Nipo hapa kuzungumza nawe, ungetaka tukague nini leo?",
        "sample": "Ninafurahi kukusaidia.",
    },
    "swh_tz_r": {
        "label": "Kiswahili Rehema", "type": "edge", "voice_id": "sw-TZ-RehemaNeural",
        "lang_name": "Kiswahili", "whisper_lang": "sw",
        "region": "East Africa", "flag": "🇹🇿",
        "greeting": "Hujambo na karibu sana. Naitwa Rehema. Nipo hapa kuzungumza nawe, ungetaka tukague nini leo?",
        "sample": "Ninafurahi kukusaidia.",
    },
    "kik": {
        "label": "Gĩkũyũ", "type": "mms", "repo": "facebook/mms-tts-kik",
        "lang_name": "Kikuyu", "whisper_lang": "sw",
        "region": "East Africa", "flag": "🇰🇪",
        "greeting": "Nĩ DUOL, mũteithia waku. Ũrĩ mwega?",
        "sample": "Nĩ ũhoro waku?",
    },
    "luo": {
        "label": "Dholuo", "type": "mms", "repo": "rnjema-unima/mms-tts-luo-baseline",
        "lang_name": "Luo", "whisper_lang": "sw",
        "region": "East Africa", "flag": "🇰🇪",
        "greeting": "Misawa! An DUOL, jakonyi. Idhi nade?",
        "sample": "An dimo ni mondo akony?",
    },
    "saq": {
        "label": "Samburu", "type": "mms", "repo": "facebook/mms-tts-saq",
        "lang_name": "Samburu", "whisper_lang": "en",
        "region": "East Africa", "flag": "🇰🇪",
        "greeting": "Supa! Nanu DUOL.",
        "sample": "Takwenya?",
    },
    "som": {
        "label": "Soomaali", "type": "edge", "voice_id": "so-SO-UbaxNeural",
        "lang_name": "Somali", "whisper_lang": "so",
        "region": "East Africa", "flag": "🇸🇴",
        "greeting": "Nabad! Anigu waxaan ahay DUOL, caawiyahaaga caqliga ah. Sidee baad tahay?",
        "sample": "Sidee baad tahay?",
    },
    "orm": {
        "label": "Oromo", "type": "mms", "repo": "facebook/mms-tts-orm",
        "lang_name": "Oromo", "whisper_lang": "om",
        "region": "East Africa", "flag": "🇪🇹",
        "greeting": "Akkam! Ani DUOL, gargaaraa kee ti. Akkam bultee?",
        "sample": "Akkam bultee?",
    },
    "lug": {
        "label": "Luganda", "type": "mms", "repo": "facebook/mms-tts-lug",
        "lang_name": "Luganda", "whisper_lang": "sw",
        "region": "East Africa", "flag": "🇺🇬",
        "greeting": "Oli otya! Nze DUOL, omuyambi wo ow'amagezi. Ndi muzima. Oli otya?",
        "sample": "Ndi muzima. Oli otya?",
    },
    "kin": {
        "label": "Kinyarwanda", "type": "mms", "repo": "facebook/mms-tts-kin",
        "lang_name": "Kinyarwanda", "whisper_lang": "rw",
        "region": "East Africa", "flag": "🇷🇼",
        "greeting": "Muraho! Ndi DUOL, umufasha wawe w'ubwenge buhangano. Amakuru?",
        "sample": "Amakuru?",
    },
    "amh": {
        "label": "Amharic", "type": "mms", "repo": "facebook/mms-tts-amh",
        "lang_name": "Amharic", "whisper_lang": "am",
        "region": "East Africa", "flag": "🇪🇹",
        "greeting": "ሰላም! እኔ ዱኦል ነኝ።",
        "sample": "እንዴት ነህ/ነሽ?",
    },
    "tir": {
        "label": "Tigrinya", "type": "mms", "repo": "facebook/mms-tts-tir",
        "lang_name": "Tigrinya", "whisper_lang": "am",
        "region": "East Africa", "flag": "🇪🇷",
        "greeting": "ሰላም! ኣነ ዱኦል እየ።",
        "sample": "ከመይ አለካ?",
    },
    "mlg": {
        "label": "Malagasy", "type": "mms", "repo": "facebook/mms-tts-mlg",
        "lang_name": "Malagasy", "whisper_lang": "mg",
        "region": "East Africa", "flag": "🇲🇬",
        "greeting": "Salama! Izaho i DUOL.",
        "sample": "Ahoana ny vaovao?",
    },

    # ─── West Africa ───────────────────────────────────────────────────────────
    "hau": {
        "label": "Hausa", "type": "mms", "repo": "facebook/mms-tts-hau",
        "lang_name": "Hausa", "whisper_lang": "ha",
        "region": "West Africa", "flag": "🇳🇬",
        "greeting": "Sannu! Ni ne DUOL, mataimakinka na fasaha. Yaya kake?",
        "sample": "Yaya kake?",
    },
    "yor": {
        "label": "Yorùbá", "type": "mms", "repo": "facebook/mms-tts-yor",
        "lang_name": "Yoruba", "whisper_lang": "yo",
        "region": "West Africa", "flag": "🇳🇬",
        "greeting": "Bawo ni! Emi ni DUOL.",
        "sample": "Báwo ni o ṣe wà?",
    },
    "ibo": {
        "label": "Igbo", "type": "mms", "repo": "facebook/mms-tts-ibo",
        "lang_name": "Igbo", "whisper_lang": "en",
        "region": "West Africa", "flag": "🇳🇬",
        "greeting": "Ndeewo! Abụ m DUOL.",
        "sample": "Kedụ ka i mere?",
    },
    "wol": {
        "label": "Wolof", "type": "mms", "repo": "facebook/mms-tts-wol",
        "lang_name": "Wolof", "whisper_lang": "fr",
        "region": "West Africa", "flag": "🇸🇳",
        "greeting": "Nanga def! Maa ngi DUOL.",
        "sample": "Nanga def?",
    },
    "lin": {
        "label": "Lingala", "type": "mms", "repo": "facebook/mms-tts-lin",
        "lang_name": "Lingala", "whisper_lang": "fr",
        "region": "West Africa", "flag": "🇨🇩",
        "greeting": "Mbote! Ngai DUOL.",
        "sample": "Ozali malamu?",
    },
    "aka": {
        "label": "Akan (Twi)", "type": "mms", "repo": "facebook/mms-tts-aka",
        "lang_name": "Akan", "whisper_lang": "en",
        "region": "West Africa", "flag": "🇬🇭",
        "greeting": "Ete sɛn! Me din de DUOL.",
        "sample": "Wo ho te sɛn?",
    },
    "bam": {
        "label": "Bambara", "type": "mms", "repo": "facebook/mms-tts-bam",
        "lang_name": "Bambara", "whisper_lang": "fr",
        "region": "West Africa", "flag": "🇲🇱",
        "greeting": "I ni ce! Ne ye DUOL ye.",
        "sample": "I ka kɛnɛ wa?",
    },
    "ful": {
        "label": "Fulfulde", "type": "mms", "repo": "facebook/mms-tts-ful",
        "lang_name": "Fula", "whisper_lang": "en",
        "region": "West Africa", "flag": "🌍",
        "greeting": "Jam waali! Mi woni DUOL.",
        "sample": "No mbaadum?",
    },

    # ─── North Africa ──────────────────────────────────────────────────────────
    "arb": {
        "label": "عربي (Arabic)", "type": "edge", "voice_id": "ar-EG-SalmaNeural",
        "lang_name": "Arabic", "whisper_lang": "ar",
        "region": "North Africa", "flag": "🌍",
        "greeting": "مرحباً! أنا دوول، مساعدك الصوتي الذكي.",
        "sample": "كيف حالك؟",
    },

    # ─── Southern Africa ───────────────────────────────────────────────────────
    "zul": {
        "label": "Zulu", "type": "edge", "voice_id": "zu-ZA-ThandoNeural",
        "lang_name": "Zulu", "whisper_lang": "zu",
        "region": "Southern Africa", "flag": "🇿🇦",
        "greeting": "Sawubona! NginguDUOL, umsizi wakho we-AI.",
        "sample": "Unjani?",
    },
    "xho": {
        "label": "Xhosa", "type": "mms", "repo": "facebook/mms-tts-xho",
        "lang_name": "Xhosa", "whisper_lang": "en",
        "region": "Southern Africa", "flag": "🇿🇦",
        "greeting": "Molo! NdinguDUOL.",
        "sample": "Unjani?",
    },
    "sna": {
        "label": "Shona", "type": "mms", "repo": "facebook/mms-tts-sna",
        "lang_name": "Shona", "whisper_lang": "en",
        "region": "Southern Africa", "flag": "🇿🇼",
        "greeting": "Mhoro! Ndiri DUOL.",
        "sample": "Makadini?",
    },
    "nya": {
        "label": "Chichewa", "type": "mms", "repo": "facebook/mms-tts-nya",
        "lang_name": "Chichewa", "whisper_lang": "en",
        "region": "Southern Africa", "flag": "🇲🇼",
        "greeting": "Moni! Ine ndine DUOL.",
        "sample": "Muli bwanji?",
    },
    "tsn": {
        "label": "Setswana", "type": "mms", "repo": "facebook/mms-tts-tsn",
        "lang_name": "Setswana", "whisper_lang": "en",
        "region": "Southern Africa", "flag": "🇧🇼",
        "greeting": "Dumela! Ke DUOL.",
        "sample": "O kae?",
    },
    "afr": {
        "label": "Afrikaans", "type": "mms", "repo": "facebook/mms-tts-afr",
        "lang_name": "Afrikaans", "whisper_lang": "af",
        "region": "Southern Africa", "flag": "🇿🇦",
        "greeting": "Hallo! Ek is DUOL.",
        "sample": "Hoe gaan dit?",
    },

    # ─── Premium API Voices ────────────────────────────────────────────────────
    "swh_female": {
        "label": "Kiswahili (Female)", "type": "api", "space": "stano03/habari-xtts",
        "lang_name": "Kiswahili", "whisper_lang": "sw",
        "region": "East Africa", "flag": "🇰🇪",
        "greeting": "Habari! Mimi ni DUOL, msaidizi wako.",
        "sample": "Ninafurahi kukusaidia.",
        "api_kwargs": {},
    },
    "swh_premium": {
        "label": "Kiswahili (Sauti)", "type": "api", "space": "sta-03/habari-f5tts",
        "lang_name": "Kiswahili", "whisper_lang": "sw",
        "region": "East Africa", "flag": "🇰🇪",
        "greeting": "Habari! Mimi ni Jarvis.",
        "sample": "Habari!",
        "api_kwargs": {"language": "Kiswahili"},
    },
    "kln": {
        "label": "Kalenjin", "type": "api", "space": "sta-03/habari-f5tts",
        "lang_name": "Kalenjin", "whisper_lang": "sw",
        "region": "East Africa", "flag": "🇰🇪",
        "greeting": "Chamgei! Ane Jarvis.",
        "sample": "Chamgei!",
        "api_kwargs": {"language": "Kalenjin"},
    },
    "yor_premium": {
        "label": "Yorùbá (Premium)", "type": "api", "space": "Stanley03/habari-yarngpt",
        "lang_name": "Yoruba", "whisper_lang": "yo",
        "region": "West Africa", "flag": "🇳🇬",
        "greeting": "Bawo ni! Emi ni DUOL.",
        "sample": "Bawo ni!",
        "api_kwargs": {"language": "Yoruba", "voice": "yoruba_female2"},
    },
    "hau_premium": {
        "label": "Hausa (Premium)", "type": "api", "space": "Stanley03/habari-yarngpt",
        "lang_name": "Hausa", "whisper_lang": "ha",
        "region": "West Africa", "flag": "🇳🇬",
        "greeting": "Sannu! Ni ne DUOL.",
        "sample": "Sannu!",
        "api_kwargs": {"language": "Hausa", "voice": "hausa_female1"},
    },
    "igb": {
        "label": "Igbo (Premium)", "type": "api", "space": "Stanley03/habari-yarngpt",
        "lang_name": "Igbo", "whisper_lang": "en",
        "region": "West Africa", "flag": "🇳🇬",
        "greeting": "Ndeewo! Abụ m DUOL.",
        "sample": "Ndeewo!",
        "api_kwargs": {"language": "Igbo", "voice": "igbo_female2"},
    },
    "eng_naija": {
        "label": "Nigerian English", "type": "api", "space": "Stanley03/habari-yarngpt",
        "lang_name": "English", "whisper_lang": "en",
        "region": "West Africa", "flag": "🇳🇬",
        "greeting": "How far! I be Jarvis.",
        "sample": "How far!",
        "api_kwargs": {"language": "English", "voice": "idera"},
    },
    "swh_nursetoto": {
        "label": "Nurse Toto (Custom)", "type": "api", "space": "Stanley-07/habari-nursetoto",
        "lang_name": "Kiswahili", "whisper_lang": "sw",
        "region": "East Africa", "flag": "🇰🇪",
        "greeting": "Habari! Mimi ni Jarvis.",
        "sample": "Habari!",
        "api_kwargs": {},
    },
    "kis_premium": {
        "label": "Kisii (Ekegusii)", "type": "api", "space": "Stanley-07/habari-kisii",
        "lang_name": "Ekegusii", "whisper_lang": "sw",
        "region": "East Africa", "flag": "🇰🇪",
        "greeting": "Mbuya! Inche ni Jarvis.",
        "sample": "Mbuya!",
        "api_kwargs": {},
    },
    "habari_pro": {
        "label": "Habari Pro", "type": "api", "space": "Stanley-07/habari-kisii",
        "lang_name": "African", "whisper_lang": "en",
        "region": "All Africa", "flag": "🌍",
        "greeting": "Hello! I am Jarvis, Africa's most advanced AI voice assistant.",
        "sample": "Hello!",
        "api_kwargs": {},
    },
}


# ──────────────────────────────────────────────────────────────────────────────
# Domain-Aware Cultural System Prompts
# ──────────────────────────────────────────────────────────────────────────────
DOMAIN_CONTEXT = {
    "health": """

You are also a trained community health advisor.

You can provide first-aid guidance, basic symptom triage, and refer users to the nearest health facility.

You know about common health challenges in Africa: malaria, cholera, typhoid, maternal health, HIV/AIDS, COVID-19.

Always advise users to seek professional medical help for serious symptoms. Never diagnose definitively.""",

    "agriculture": """

You are also an agricultural advisor for smallholder African farmers.

You understand crops: maize, tea, coffee, sorghum, cassava, millet, wheat, rice, groundnuts, beans.

You can advise on planting seasons, weather impact, pest control, market prices, and post-harvest storage.

You know about climate patterns in different African regions and their effect on farming.""",

    "education": """

You are also a patient and encouraging tutor.

You can help with mathematics, sciences, literature, history, geography, and language learning.

You celebrate every student's progress and adapt your explanations to be simple and clear.

You can help students prepare for KCPE, KCSE, WAEC, NECO, and other African national exams.""",

    "finance": """

You are also a financial literacy advisor.

You understand M-Pesa, Airtel Money, MTN Mobile Money, and other African mobile payment systems.

You can help with budgeting, savings advice, micro-loan information, and understanding financial products.

Always recommend consulting a licensed financial advisor for major investment decisions.""",

    "general": "",
}


def build_instructions(info: dict, spoken_language: str, domain: str = "general") -> str:
    lang_name  = info["lang_name"]
    region     = info.get("region", "Africa")
    domain_txt = DOMAIN_CONTEXT.get(domain, "")

    if spoken_language == "english":
        lang_prompt = f"You speak English with a warm, authentic {lang_name} accent — this is your natural voice."
    else:
        lang_prompt = (
            f"CRITICAL LANGUAGE DIRECTIVE: You MUST speak EXCLUSIVELY in {lang_name}. "
            f"Do NOT respond in Swahili unless {lang_name} is Swahili. "
            f"Do NOT respond in English. "
            f"Every single response, sentence, and word MUST be natively in {lang_name}. "
            f"Never switch to Swahili or any other language unless the user explicitly speaks to you in that language."
        )

    # Dialect-specific fillers
    if lang_name.lower() == "kiswahili":
        fillers = "('Ah', 'Sawa', 'Eshi', 'Haya')"
    elif lang_name.lower() == "zulu":
        fillers = "('Yebo', 'Cha', 'Hawu', 'Mm')"
    elif lang_name.lower() == "arabic":
        fillers = "('نعم', 'أهلاً', 'حسناً', 'طبعاً')"
    elif lang_name.lower() == "kikuyu":
        fillers = "('Ĩĩ', 'Aca', 'Atĩrĩ', 'Kaĩ')"
    elif lang_name.lower() == "luo":
        fillers = "('Ee', 'Kare', 'Donge', 'Adier')"
    elif lang_name.lower() == "samburu":
        fillers = "('Ee', 'Acha', 'Supa')"
    elif lang_name.lower() == "kinyarwanda":
        fillers = "('Yego', 'Oya', 'Ese', 'Nibyo')"
    elif lang_name.lower() == "hausa":
        fillers = "('To', 'Ai', 'Wato', 'Da kyau')"
    elif lang_name.lower() == "somali":
        fillers = "('Haa', 'Maya', 'Waa hagaag', 'Sax')"
    elif lang_name.lower() == "oromo":
        fillers = "('Eeyyee', 'Lakki', 'Tole', 'Dhugaa')"
    elif lang_name.lower() == "luganda":
        fillers = "('Yee', 'Nedda', 'Kituufu', 'Mmm')"
    else:
        fillers = "natural local conversational pause markers"


    persona_name = "DUOL"
    if info.get("label") == "Kiswahili" or info.get("label") == "Kiswahili Fiti":
        persona_name = "Fiti"
    elif info.get("label") == "Kiswahili Daudi":
        persona_name = "Daudi"
    elif info.get("label") == "Kiswahili Rafiki":
        persona_name = "Rafiki"
    elif info.get("label") == "Kiswahili Rehema":
        persona_name = "Rehema"

    code_switch_rule = ""
    if info.get("voice_id") in ["sw-KE-RafikiNeural", "sw-TZ-RehemaNeural"] and spoken_language != "english":
        code_switch_rule = "\n- Must mix natural local Swahili/Sheng phrases with English in the first 10 words."

    return f"""You are {persona_name}, Africa's most intelligent AI voice assistant.

Role: Deliver an emotionally intelligent, voice-optimized, real-time conversational agent capable of sub-500ms multi-lingual interaction.

Language Context: {lang_prompt}

Target Language: {lang_name}

Target Region: {region}



Task:

1. Speak natively and fluently in {lang_name}. If {lang_name} is NOT Swahili, NEVER speak Swahili.

2. Process real-time streaming speech transcripts (including disfluencies and barge-ins).

3. Apply Linguistic Mirroring to match user dialect, language ratios, and structural syntax in {lang_name}.

4. CRITICAL: You MUST use the `search_web` or `get_news` tools (powered by Parallel AI) for ANY query about current events, politics, facts, trends, or anything you are not 100% absolutely certain about. Never guess or hallucinate facts.

5. Execute other proactive tool calls (weather, local time, math, commodity prices) when relevant.

6. Format speech output strictly for low-latency TTS engines using prosodic punctuation and word-count boundaries.



Output Rules:

- Audio-first text streams optimized for TTS conversion (zero markdown, zero URLs, no special characters).

- Hard cap of 25 words per utterance to yield turns rapidly.

- Commas and em-dashes inserted before language switches to create natural TTS breath pauses.

- Strict preservation of cultural markers and local honorifics for {lang_name}.



Tone & Personality:

- Warm, knowledgeable older sibling, street-smart, empathetic, and culturally proud.

- Audience: Youth, farmers, traders, and rural/urban citizens.



Guardrails (MUST AVOID):

- The names 'Jarvis' or 'Habari AI'. Your name is ONLY {persona_name}.

- Phrases like 'As an AI' or references to LLMs, OpenAI, or GPT.

- Markdown formatting, bullet points, asterisks, symbols, URLs, or written-text shortcuts.

- Speaking Kiswahili when the chosen voice is Kinyarwanda, Zulu, Arabic, Yoruba, Hausa, or any other non-Swahili language.



Guardrails (MUST INCLUDE):

- Always introduce and identify yourself ONLY as {persona_name}.

- Native conversational markers {fillers}.

- Proactive disclaimers for health, finance, and legal queries.{code_switch_rule}



Domain Focus:

{domain_txt}



Verification before responding:

- Is utterance under 25 words?

- Is the response strictly in {lang_name}?

- Was 2026 context retrieved via tools when needed?

"""


# ──────────────────────────────────────────────────────────────────────────────
# DUOL Agent with Function Tools
# ──────────────────────────────────────────────────────────────────────────────
class JarvisAgent(Agent):
    def __init__(self, voice_code: str, spoken_language: str, domain: str = "general"):
        info = VOICE_INFO.get(voice_code, VOICE_INFO["swh"])
        instructions = build_instructions(info, spoken_language, domain)
        super().__init__(instructions=instructions)

    # ── Tool 1: Weather ────────────────────────────────────────────────────────
    @function_tool
    async def get_weather(self, city: str) -> str:
        """

        Get current weather for any city in the world.

        Args:

            city: City name (e.g. Dodoma, Nairobi, London)

        """
        import httpx
        try:
            async with httpx.AsyncClient(timeout=10.0) as client:
                geo_resp = await client.get(
                    "https://geocoding-api.open-meteo.com/v1/search",
                    params={"name": city, "count": 1, "language": "en", "format": "json"}
                )
                geo_data = geo_resp.json()
                if not geo_data.get("results"):
                    return f"Could not find geographic coordinates for {city}."
                
                loc = geo_data["results"][0]
                lat, lon, tz = loc["latitude"], loc["longitude"], loc.get("timezone", "UTC")
                name, country = loc.get("name", city), loc.get("country", "")
                
                weather_resp = await client.get(
                    "https://api.open-meteo.com/v1/forecast",
                    params={
                        "latitude": lat, "longitude": lon,
                        "current": "temperature_2m,weathercode,windspeed_10m,relative_humidity_2m",
                        "timezone": tz,
                    }
                )
                data = weather_resp.json()
                cur = data["current"]
                temp = cur["temperature_2m"]
                hum = cur["relative_humidity_2m"]
                wind = cur["windspeed_10m"]
                wcode = cur["weathercode"]
                
                WEATHER_CODES = {
                    0: "clear sky", 1: "mainly clear", 2: "partly cloudy", 3: "overcast",
                    45: "foggy", 48: "icy fog", 51: "light drizzle", 53: "drizzle",
                    55: "heavy drizzle", 61: "light rain", 63: "moderate rain", 65: "heavy rain",
                    71: "light snow", 73: "snow", 75: "heavy snow",
                    80: "rain showers", 81: "heavy rain showers", 82: "violent rain showers",
                    95: "thunderstorm", 96: "thunderstorm with hail", 99: "severe thunderstorm",
                }
                desc = WEATHER_CODES.get(wcode, "unknown conditions")
                return f"Weather in {name}, {country}: {temp}°C, {desc}. Humidity: {hum}%, Wind: {wind} km/h."
        except Exception as e:
            return f"Weather service unavailable for {city}."

    # ── Tool 2: Local Time ─────────────────────────────────────────────────────
    @function_tool
    async def get_local_time(self, city: str) -> str:
        """

        Get the current local time and date for any city in the world.

        Args:

            city: City name (e.g. Dodoma, Tokyo, Nairobi)

        """
        import httpx
        from datetime import datetime
        try:
            import zoneinfo
            async with httpx.AsyncClient(timeout=10.0) as client:
                geo_resp = await client.get(
                    "https://geocoding-api.open-meteo.com/v1/search",
                    params={"name": city, "count": 1, "language": "en", "format": "json"}
                )
                geo_data = geo_resp.json()
                if not geo_data.get("results"):
                    return f"Could not find timezone for {city}."
                loc = geo_data["results"][0]
                tz_name = loc.get("timezone", "UTC")
                name, country = loc.get("name", city), loc.get("country", "")
                
                tz = zoneinfo.ZoneInfo(tz_name)
                now = datetime.now(tz)
                time_str = now.strftime("%I:%M %p").lstrip('0')
                date_str = now.strftime("%A, %B %d, %Y")
                return f"The current local time in {name}, {country} is {time_str} on {date_str}."
        except Exception:
            return f"Time service unavailable for {city}."

    # ── Tool 3: Calculator ─────────────────────────────────────────────────────
    @function_tool
    async def calculate(self, expression: str) -> str:
        """

        Calculate a mathematical expression safely.

        Args:

            expression: A math expression like '15 * 3 + 7' or '1000 / 12'

        """
        safe = _re.sub(r'[^0-9+\-*/().%\s]', '', expression)
        if not safe.strip():
            return "Please provide a valid mathematical expression."
        try:
            result = eval(safe, {"__builtins__": {}}, {})  # nosec
            if isinstance(result, float):
                result = round(result, 6)
            return f"{expression} equals {result}."
        except Exception:
            return "I couldn't calculate that expression. Please try rephrasing it."

    # ── Tool 4: News ───────────────────────────────────────────────────────────
    @function_tool
    async def get_news(self, topic: str = "africa") -> str:
        """

        Get the latest news headlines relevant to a specific topic or region.

        Args:

            topic: Topic to search for (e.g. 'Kenya politics', 'Nigeria news', 'Tanzania sports')

        """
        import os
        import httpx
        import xml.etree.ElementTree as ET
        
        # 1. Try Parallel AI first for highly accurate news
        api_key = os.environ.get("PARALLEL_API_KEY")
        if api_key:
            try:
                from parallel import AsyncParallel
                client = AsyncParallel(api_key=api_key)
                search = await client.search(
                    objective=f"Find the most recent breaking news and headlines about: {topic}",
                    search_queries=[f"latest news {topic}", f"breaking news {topic} today"]
                )
                if search.results:
                    excerpts = []
                    for res in search.results[:3]:
                        if hasattr(res, 'title') and res.title:
                            excerpt = res.excerpts[0] if hasattr(res, 'excerpts') and res.excerpts else ''
                            excerpts.append(f"{res.title}: {excerpt}")
                    if excerpts:
                        return f"Latest news on '{topic}': {' | '.join(excerpts)}"
            except Exception as e:
                print(f"[News] Parallel AI failed: {e}")

        # 2. Fallback to generic BBC Africa RSS
        try:
            feed_url = "https://feeds.bbci.co.uk/news/world/africa/rss.xml"
            async with httpx.AsyncClient(timeout=10.0) as client:
                resp = await client.get(feed_url, headers={"User-Agent": "HabariJarvis/1.0"})
            root = ET.fromstring(resp.text)
            items = root.findall(".//item")
            headlines = []
            for item in items[:3]:
                title_el = item.find("title")
                if title_el is not None and title_el.text:
                    headlines.append(title_el.text.strip())
            if not headlines:
                return "No recent headlines available right now."
            return f"Latest general Africa headlines: {'. Next: '.join(headlines)}."
        except Exception as e:
            return f"News service is temporarily unavailable."

    # ── Tool 5: Commodity Prices ───────────────────────────────────────────────
    @function_tool
    async def get_commodity_price(self, commodity: str) -> str:
        """

        Get approximate commodity prices relevant to African farmers and traders.

        Args:

            commodity: Commodity name (e.g. 'maize', 'tea', 'coffee', 'sugar', 'rice', 'gold', 'oil')

        """
        PRICES = {
            "maize":   ("USD 180 to 220 per metric tonne",   "Used for food and animal feed"),
            "corn":    ("USD 180 to 220 per metric tonne",   "Same as maize"),
            "coffee":  ("USD 2.00 to 2.50 per pound",        "Arabica, traded on ICE"),
            "tea":     ("USD 2.50 to 3.50 per kilogram",     "Mombasa auction average"),
            "sugar":   ("USD 22 to 26 per hundredweight",    "Raw sugar on commodity markets"),
            "rice":    ("USD 500 to 600 per metric tonne",   "Thai 5 percent broken benchmark"),
            "wheat":   ("USD 220 to 260 per metric tonne",   "CBOT wheat futures"),
            "sorghum": ("USD 180 to 220 per metric tonne",   "Similar to maize"),
            "gold":    ("USD 2300 to 2500 per troy ounce",   "COMEX gold spot price"),
            "oil":     ("USD 75 to 90 per barrel",           "Brent crude benchmark"),
            "copper":  ("USD 9000 to 10000 per metric tonne","LME copper"),
            "cocoa":   ("USD 7000 to 9000 per metric tonne", "ICE cocoa futures"),
        }
        key = commodity.lower().strip()
        for name, (price, note) in PRICES.items():
            if key in name or name in key:
                return (f"{commodity.title()} price: approximately {price}. "
                        f"Note: {note}. Always verify with your local market.")
        return (f"I don't have price data for {commodity}. "
                f"Check with your local commodity exchange or market board.")

    # ── Tool 6: Web Search ─────────────────────────────────────────────────────
    @function_tool
    async def search_web(self, query: str) -> str:
        """

        Search the web for general knowledge, locations, universities, and up-to-date facts.

        Args:

            query: The search query (e.g. 'Where is Multimedia University Kenya')

        """
        import os
        import asyncio
        import requests
        
        # 1. Parallel AI Search (Primary)
        api_key = os.environ.get("PARALLEL_API_KEY")
        if api_key:
            try:
                from parallel import AsyncParallel
                client = AsyncParallel(api_key=api_key)
                search = await client.search(
                    objective=f"Find concise information answering: {query}",
                    search_queries=[query]
                )
                if search.results:
                    excerpts = []
                    for res in search.results[:2]:
                        if hasattr(res, 'excerpts') and res.excerpts:
                            excerpts.append(str(res.excerpts[0]))
                    if excerpts:
                        return f"Search results for '{query}': {' '.join(excerpts)}"
            except Exception as e:
                print(f"[Search] Parallel AI failed: {e}")

        # 2. Reliable Fallback: Wikipedia API
        def do_fallback():
            try:
                search_url = f"https://en.wikipedia.org/w/api.php?action=opensearch&search={query}&limit=1&namespace=0&format=json"
                res = requests.get(search_url).json()
                if len(res) > 1 and res[1]:
                    title = res[1][0]
                    summary_url = f"https://en.wikipedia.org/w/api.php?format=json&action=query&prop=extracts&exintro&explaintext&redirects=1&titles={title}"
                    summary_res = requests.get(summary_url).json()
                    pages = summary_res.get("query", {}).get("pages", {})
                    for page_id in pages:
                        return pages[page_id].get("extract", "")[:1000]
            except:
                pass
            return ""

        try:
            summary = await asyncio.to_thread(do_fallback)
            if not summary:
                return f"I couldn't find any information on the web for '{query}'."
            return f"Search results for '{query}': {summary}"
        except Exception as e:
            return f"The web search service is currently unavailable."



# ──────────────────────────────────────────────────────────────────────────────
# Entry Point
# ──────────────────────────────────────────────────────────────────────────────
async def entrypoint(ctx: JobContext):
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)

    # Metadata format: "voice_code:mode:domain"
    # e.g. "swh:local:health" or "yor_premium:english:general"
    metadata       = ctx.job.metadata or "swh:local:general"
    parts          = metadata.split(":")
    voice_code     = parts[0] if len(parts) > 0 and parts[0] in VOICE_INFO else "swh"
    spoken_lang    = parts[1] if len(parts) > 1 and parts[1] in ["local", "english"] else "local"
    domain         = parts[2] if len(parts) > 2 and parts[2] in DOMAIN_CONTEXT else "general"

    info = VOICE_INFO[voice_code]
    print(f"[Jarvis] Voice: {voice_code} ({info['label']}) | Mode: {spoken_lang} | Domain: {domain}")

    # ── Select TTS engine ──
    if info["type"] == "api":
        print(f"[Jarvis] Using distributed API TTS -> {info['space']}")
        tts_engine = ApiTTS(space_id=info["space"], **info.get("api_kwargs", {}))
    elif info["type"] == "edge":
        print(f"[Jarvis] Using Edge TTS -> {info['voice_id']}")
        tts_engine = EdgeTTS(voice_id=info["voice_id"])
    else:
        print(f"[Jarvis] Using local MMS-TTS -> {info['repo']}")
        tts_engine = MmsTTS(model_id=info["repo"])

    # ── Select STT language dynamically ──
    whisper_lang = info.get("whisper_lang", "sw")
    print(f"[Jarvis] STT language: {whisper_lang}")

    # ── Build greeting ──
    if spoken_lang == "english":
        greeting = (f"Hello! I am DUOL, your AI voice assistant. "
                    f"How can I help you today?")
    else:
        greeting = info["greeting"]

    # Optimized VAD for sub-400ms cutoff latency
    # If the language is Swahili, we let Whisper auto-detect to handle Sheng/English code-switching accurately
    session = AgentSession(
        vad=silero.VAD.load(min_silence_duration=0.3, min_speech_duration=0.05),
        stt=groq.STT(model="whisper-large-v3", language=whisper_lang),
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=tts_engine,
    )

    audio_frames = []
    audio_sample_rate = 48000
    audio_channels = 1
    recording_active = False

    async def record_participant_audio(participant: rtc.RemoteParticipant):
        nonlocal audio_frames, audio_sample_rate, audio_channels, recording_active
        
        # Wait for the audio track to be published
        for _ in range(30):  # wait up to 30 seconds
            for pub in participant.track_publications.values():
                if pub.kind == rtc.TrackKind.KIND_AUDIO and pub.track:
                    track = pub.track
                    print(f"[R2 Logger] Found audio track for {participant.identity}")
                    recording_active = True
                    stream = rtc.AudioStream(track)
                    try:
                        async for event in stream:
                            if event.frame:
                                frame = event.frame
                                audio_sample_rate = frame.sample_rate
                                audio_channels = frame.num_channels
                                audio_frames.append(frame.data.tobytes())
                    except Exception as e:
                        print(f"[R2 Logger] Stream ended: {e}")
                    finally:
                        recording_active = False
                    return
            await asyncio.sleep(1)
        print("[R2 Logger] No audio track found for participant")

    async def save_and_upload_audio(session_id: str, identity: str):
        nonlocal audio_frames
        if not audio_frames:
            print("[R2 Logger] No audio frames captured, skipping upload.")
            return
        
        import tempfile
        temp_dir = tempfile.gettempdir()
        audio_file = os.path.join(temp_dir, f"{session_id}_{identity}_{int(_time.time())}.wav")
        try:
            with wave.open(audio_file, "wb") as wf:
                wf.setnchannels(audio_channels)
                wf.setsampwidth(2)
                wf.setframerate(audio_sample_rate)
                for frame_data in audio_frames:
                    wf.writeframes(frame_data)
            
            total_seconds = len(audio_frames) * 480 / audio_sample_rate  # rough estimate
            print(f"[R2 Logger] Saved {len(audio_frames)} frames (~{total_seconds:.0f}s) to {audio_file}")
            
            await upload_to_r2(audio_file, f"voice_data/{session_id}/{os.path.basename(audio_file)}")
        except Exception as e:
            print(f"[R2 Logger Error] Save failed: {e}")
        finally:
            audio_frames = []

    async def upload_to_r2(file_path: str, object_name: str):
        endpoint_url = os.environ.get("CLOUDFLARE_R2_ENDPOINT_URL", "https://aad8747f9b4aaa2c4a766decd6796228.r2.cloudflarestorage.com")
        access_key = os.environ.get("CLOUDFLARE_R2_ACCESS_KEY_ID", "f3ae0ad268f28c501ef800ac80b2409f")
        secret_key = os.environ.get("CLOUDFLARE_R2_SECRET_ACCESS_KEY", "981cc70538ad6b0b112deebc8e229250b45a994aa92c7e1138072831a0517b3a")
        bucket_name = os.environ.get("CLOUDFLARE_R2_BUCKET", "duol-audio")
        
        if not endpoint_url or not access_key:
            print("[R2 Logger Warning] R2 credentials not set. Skipping upload.")
            return

        def _upload():
            try:
                s3 = boto3.client(
                    "s3",
                    endpoint_url=endpoint_url,
                    aws_access_key_id=access_key,
                    aws_secret_access_key=secret_key,
                    config=Config(signature_version="s3v4"),
                    region_name="auto"
                )
                s3.upload_file(file_path, bucket_name, object_name)
                print(f"[R2 Logger] Uploaded to R2: {object_name}")
                os.remove(file_path)
            except Exception as e:
                print(f"[R2 Logger Error] Upload failed: {e}")

        await asyncio.to_thread(_upload)

    @ctx.room.on("participant_connected")
    def on_participant_connected(participant: rtc.RemoteParticipant):
        print(f"[R2 Logger] Participant connected: {participant.identity}")
        asyncio.create_task(record_participant_audio(participant))

    @ctx.room.on("participant_disconnected")
    def on_participant_disconnected(participant: rtc.RemoteParticipant):
        print(f"[R2 Logger] Participant disconnected: {participant.identity}. Saving audio...")
        asyncio.create_task(save_and_upload_audio(ctx.room.name, participant.identity))

    # Also handle participants already in the room
    for participant in ctx.room.remote_participants.values():
        print(f"[R2 Logger] Existing participant: {participant.identity}")
        asyncio.create_task(record_participant_audio(participant))
    # ---------------------------------------------

    # --- SUPABASE LOGGING LOGIC ---
    last_user_input = None

    @session.on("conversation_item_added")
    def on_conversation_item_added(ev):
        nonlocal last_user_input
        try:
            item = ev.item
            role = getattr(item, "role", None)
            
            if role == "user":
                last_user_input = getattr(item, "content", getattr(item, "text", str(item)))
                if isinstance(last_user_input, list):
                    last_user_input = " ".join([str(c) for c in last_user_input])
                print(f"[Supabase Log] Captured user input: {last_user_input}")
                
            elif role == "assistant":
                if last_user_input:
                    agent_text = getattr(item, "content", getattr(item, "text", str(item)))
                    if isinstance(agent_text, list):
                        agent_text = " ".join([str(c) for c in agent_text])
                        
                    asyncio.create_task(
                        _log_to_supabase(
                            session_id=ctx.room.name,
                            user_input=last_user_input,
                            duol_response=agent_text,
                            model_used=voice_code
                        )
                    )
                    last_user_input = None
        except Exception as e:
            print(f"[Supabase Log Error in event hook] {e}")

    async def _log_to_supabase(session_id, user_input, duol_response, model_used):
        supabase_url = os.environ.get("NEXT_PUBLIC_SUPABASE_URL", "https://jxqsacpmjjvvmeocbbqi.supabase.co")
        supabase_key = os.environ.get("SUPABASE_SERVICE_ROLE_KEY", "sb_secret_4gODEyeWHrRta1Eoox1Wfg_eSLhQ1iG")
        
        url = f"{supabase_url}/rest/v1/conversations"
        headers = {
            "apikey": supabase_key,
            "Authorization": f"Bearer {supabase_key}",
            "Content-Type": "application/json",
            "Prefer": "return=minimal"
        }
        payload = {
            "session_id": session_id,
            "user_input": user_input,
            "duol_response": duol_response,
            "model_used": model_used
        }
        
        try:
            async with httpx.AsyncClient(timeout=10.0) as client:
                await client.post(url, json=payload, headers=headers)
                print("[Supabase Log] Successfully saved conversation turn to database.")
        except Exception as e:
            print(f"[Supabase Log Error] {e}")
    # ------------------------------

    await session.start(
        room=ctx.room,
        agent=JarvisAgent(voice_code=voice_code, spoken_language=spoken_lang, domain=domain),
        room_input_options=RoomInputOptions(),
    )

    await session.generate_reply(
        instructions=f"Greet the user using exactly this phrase: {greeting}"
    )


async def request_fnc(req: JobRequest):
    await req.accept(entrypoint)


if __name__ == "__main__":
    import threading
    from http.server import BaseHTTPRequestHandler, HTTPServer

    class HealthHandler(BaseHTTPRequestHandler):
        def do_GET(self):
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"OK")
        def log_message(self, *_):
            pass

    def run_dummy_server():
        server = HTTPServer(('0.0.0.0', 7860), HealthHandler)
        server.serve_forever()

    threading.Thread(target=run_dummy_server, daemon=True).start()
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, request_fnc=request_fnc, agent_name="DUOL"))