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! Mimi ni Jarvis, msaidizi wako wa akili bandia.", "sample": "Unaweza kuniambia zaidi?", }, "swh_v2": { "label": "Kiswahili v2", "type": "mms", "repo": "Benjamin-png/swahili-mms-tts-finetuned", "lang_name": "Kiswahili", "whisper_lang": "sw", "region": "East Africa", "flag": "🇰🇪", "greeting": "Habari! Mimi ni Jarvis, msaidizi wako wa kisasa.", "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": "Habari! Mimi ni Rafiki, msaidizi wako.", "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": "Habari! Mimi ni Rehema, msaidizi wako.", "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ĩ Jarvis, mũteithia waku.", "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": "Oyaore! An Jarvis.", "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 Jarvis.", "sample": "Takwenya?", }, "som": { "label": "Soomaali", "type": "mms", "repo": "facebook/mms-tts-som", "lang_name": "Somali", "whisper_lang": "so", "region": "East Africa", "flag": "🇸🇴", "greeting": "Nabad! Anigu waxaan ahay Jarvis.", "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 Jarvis.", "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 Jarvis.", "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 Jarvis.", "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 Jarvis.", "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 Jarvis.", "sample": "Yaya kake/kike?", }, "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 Jarvis.", "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 Jarvis.", "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 Jarvis.", "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 Jarvis.", "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 Jarvis.", "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 Jarvis 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 Jarvis.", "sample": "No mbaadum?", }, # ─── North Africa ────────────────────────────────────────────────────────── "arb": { "label": "عربي (Arabic)", "type": "mms", "repo": "facebook/mms-tts-arb", "lang_name": "Arabic", "whisper_lang": "ar", "region": "North Africa", "flag": "🌍", "greeting": "مرحباً! أنا جارفيس.", "sample": "كيف حالك؟", }, # ─── Southern Africa ─────────────────────────────────────────────────────── "zul": { "label": "Zulu", "type": "mms", "repo": "facebook/mms-tts-zul", "lang_name": "Zulu", "whisper_lang": "en", "region": "Southern Africa", "flag": "🇿🇦", "greeting": "Sawubona! NginguJarvis.", "sample": "Unjani?", }, "xho": { "label": "Xhosa", "type": "mms", "repo": "facebook/mms-tts-xho", "lang_name": "Xhosa", "whisper_lang": "en", "region": "Southern Africa", "flag": "🇿🇦", "greeting": "Molo! NdinguJarvis.", "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 Jarvis.", "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 Jarvis.", "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 Jarvis.", "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 Jarvis.", "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 Jarvis, msaidizi wako wa kike.", "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 Jarvis.", "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 Jarvis.", "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 Jarvis.", "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"Respond deeply and fluently in {lang_name}, mirroring local dialects." return f"""You are DUOL, 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 Region: {region} Task: 1. Process real-time streaming speech transcripts (including disfluencies and barge-ins). 2. Apply Linguistic Mirroring to match user dialect, language ratios, and structural syntax across African languages. 3. Execute proactive tool calls for 2026 grounding (web search, weather, local time, math, news, commodity prices). 4. 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 30 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. 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'. - Phrases like 'As an AI' or references to LLMs, OpenAI, or GPT. - Markdown formatting, bullet points, symbols, URLs, or written-text shortcuts. Guardrails (MUST INCLUDE): - Branding as DUOL. - Native disfluency fillers ('Ah', 'Sawa', 'Eshi', 'Haya'). - Proactive disclaimers for health, finance, and legal queries. Domain Focus: {domain_txt} Verification before responding: - Is utterance under 30 words? - Are non-Latin or complex phonemes phonetically anchored for TTS? - Was 2026 context retrieved via tools when needed? - Is the user's exact code-switching ratio mirrored? """ # ────────────────────────────────────────────────────────────────────────────── # Jarvis 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 an African city. Args: city: City name (e.g. Nairobi, Lagos, Accra, Cairo, Johannesburg) """ CITY_COORDS = { "nairobi": (-1.2921, 36.8219, "Africa/Nairobi"), "mombasa": (-4.0435, 39.6682, "Africa/Nairobi"), "kisumu": (-0.0917, 34.7679, "Africa/Nairobi"), "kampala": ( 0.3476, 32.5825, "Africa/Kampala"), "dar es salaam": (-6.7924, 39.2083, "Africa/Dar_es_Salaam"), "kigali": (-1.9441, 30.0619, "Africa/Kigali"), "addis ababa": ( 9.0054, 38.7636, "Africa/Addis_Ababa"), "lagos": ( 6.5244, 3.3792, "Africa/Lagos"), "abuja": ( 9.0765, 7.3986, "Africa/Lagos"), "accra": ( 5.6037, -0.1870, "Africa/Accra"), "dakar": (14.7167, -17.4677, "Africa/Dakar"), "kinshasa": (-4.4419, 15.2663, "Africa/Kinshasa"), "cairo": (30.0444, 31.2357, "Africa/Cairo"), "johannesburg": (-26.2041, 28.0473, "Africa/Johannesburg"), "cape town": (-33.9249, 18.4241, "Africa/Johannesburg"), "durban": (-29.8587, 31.0218, "Africa/Johannesburg"), "harare": (-17.8252, 31.0335, "Africa/Harare"), "lusaka": (-15.3875, 28.3228, "Africa/Lusaka"), "antananarivo": (-18.9137, 47.5361, "Indian/Antananarivo"), "mogadishu": ( 2.0469, 45.3182, "Africa/Mogadishu"), "asmara": (15.3229, 38.9251, "Africa/Asmara"), "bamako": (12.6392, -8.0029, "Africa/Bamako"), "khartoum": (15.5007, 32.5599, "Africa/Khartoum"), } 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", } city_lower = city.lower().strip() coords = CITY_COORDS.get(city_lower) if not coords: for key, val in CITY_COORDS.items(): if key in city_lower or city_lower in key: coords = val city_lower = key break if not coords: return (f"I don't have location data for {city}. " f"Try a major African city like Nairobi, Lagos, or Cairo.") lat, lon, tz = coords try: async with httpx.AsyncClient(timeout=10.0) as client: 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 = resp.json() cur = data["current"] temp = cur["temperature_2m"] code = cur.get("weathercode", 0) wind = cur["windspeed_10m"] hum = cur.get("relative_humidity_2m", "?") desc = WEATHER_CODES.get(code, "variable conditions") return (f"Weather in {city_lower.title()}: {desc}, {temp} degrees Celsius, " f"wind {wind} kilometres per hour, humidity {hum} percent.") except Exception: return "Could not fetch weather right now. Please try again shortly." # ── Tool 2: Local Time ───────────────────────────────────────────────────── @function_tool async def get_local_time(self, city: str) -> str: """ Get the current local time and date in an African city. Args: city: City name (e.g. Nairobi, Lagos, Cairo) """ CITY_TZ = { "nairobi": "Africa/Nairobi", "mombasa": "Africa/Nairobi", "kampala": "Africa/Kampala", "dar es salaam": "Africa/Dar_es_Salaam", "kigali": "Africa/Kigali", "addis ababa": "Africa/Addis_Ababa", "mogadishu": "Africa/Mogadishu", "asmara": "Africa/Asmara", "lagos": "Africa/Lagos", "abuja": "Africa/Lagos", "accra": "Africa/Accra", "dakar": "Africa/Dakar", "kinshasa": "Africa/Kinshasa", "bamako": "Africa/Bamako", "conakry": "Africa/Conakry", "cairo": "Africa/Cairo", "khartoum": "Africa/Khartoum", "johannesburg": "Africa/Johannesburg", "cape town": "Africa/Johannesburg", "durban": "Africa/Johannesburg", "harare": "Africa/Harare", "lusaka": "Africa/Lusaka", "antananarivo": "Indian/Antananarivo", } city_lower = city.lower().strip() tz_str = CITY_TZ.get(city_lower) if not tz_str: for key, val in CITY_TZ.items(): if key in city_lower or city_lower in key: tz_str = val city_lower = key break if not tz_str: return f"I don't have timezone data for {city}. Try a major African city." try: if ZoneInfo: now = datetime.now(ZoneInfo(tz_str)) else: now = datetime.utcnow() formatted = now.strftime("%I:%M %p, %A %d %B %Y") return f"The current time in {city_lower.title()} is {formatted}." except Exception: return f"Could not get the time for {city} right now." # ── 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 Africa. Args: topic: Topic to search for (e.g. 'kenya', 'nigeria', 'africa', 'sports', 'technology') """ 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[:5]: 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." top = headlines[:3] return "Latest Africa headlines from BBC. " + ". Next: ".join(top) + "." except Exception: return "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 asyncio def do_search(): from duckduckgo_search import DDGS with DDGS() as ddgs: return list(ddgs.text(query, max_results=3)) try: results = await asyncio.to_thread(do_search) if not results: return f"I couldn't find any information on the web for '{query}'." summary = " ".join([r.get('body', '') for r in results]) 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 Jarvis. " f"How can I help you today?") else: greeting = info["greeting"] session = AgentSession( vad=silero.VAD.load(), 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"))