""" Generates voice response text from sensor data in the farmer's own language. Supports Bambara (bam), Fula (ful), French (fr), and English (en). Bambara/Fula templates use short sentences (≤6 words) for best MMS-TTS quality. Each template has an English equivalent used for the translation display in the UI. """ from __future__ import annotations from typing import TYPE_CHECKING if TYPE_CHECKING: from src.iot.intent_parser import Intent from src.iot.sensor_bridge import SensorData # Alert thresholds SOIL_MOISTURE_LOW = 30.0 # Below this → immediate irrigation recommended SOIL_MOISTURE_HIGH = 70.0 # Above this → drainage warning SOIL_PH_LOW = 5.5 SOIL_PH_HIGH = 7.5 TEMP_HIGH = 38.0 PEST_ALERT_HIGH = 2 # Alert level ≥ 2 → warning # ── Bambara templates (≤6 words per sentence for clear MMS-TTS output) ─────── # Format: (bambara_text, english_translation) BAMBARA_TEMPLATES: dict[str, tuple[str, str]] = { # Greetings / social "greeting": ("I ni ce. N bɛ i dɛmɛ.", "Hello. I am here to help you."), "greeting_morning": ("I ni sogoma. Sɔrɔ ka ɲi.", "Good morning. May the harvest be good."), "greeting_evening": ("I ni wula. Kɛnɛya.", "Good evening. Good health to you."), "thanks": ("Aw ni ce. A ka ɲi.", "Thank you. That is good."), "farewell": ("Kana tɛmɛ. Ne bɛ i kɔnɔ.", "Goodbye. I will be thinking of you."), "not_understood": ("N ma a faamu. A fɔ.", "I did not understand. Please repeat."), # Soil "soil_moisture_low": ("Bunding ji dɔgɔ. I ka foro ji.", "Soil moisture is low. Irrigate your field."), "soil_moisture_high": ("Ji ca kojugu. Foro ma fɛ.", "Too much water. The field is flooded."), "soil_ph_low": ("Bunding kɔnɔ jugu. Kalisi fara a kan.", "Soil is too acidic. Apply lime."), "soil_ph_high": ("Bunding kɔnɔ tɛmɛ. Soufre fara a kan.", "Soil is too alkaline. Apply sulphur."), "soil_ok": ("Bunding ka ɲi. Foro sɔrɔ.", "Soil conditions are good. The field is ready."), # Weather "weather_hot": ("Teliman gbɛlɛ. Tile ma sigi.", "It is very hot. Do not work in the midday sun."), "rain_likely": ("Sanji bɛ na. Sɔrɔ jɔ.", "Rain is coming. Protect the harvest."), "weather_ok": ("Yɔrɔ min ɲi. Kɛ.", "Weather is fine. You can work."), # Pest "pest_high": ("Dɔgɔw bɛ foro kɔnɔ. Bɔ u.", "Pests are in the field. Drive them out."), "pest_ok": ("Dɔgɔw tɛ yen. Foro ka ɲi.", "No pests detected. The field looks healthy."), # Irrigation "irrigation_needed": ("Foro fɛ ji. Ji sira yɔrɔ.", "The field needs water. Open the irrigation."), "irrigation_active": ("Ji bɛ taa. A bɛ kɛ cogo di.", "Irrigation is running. It is working well."), # Default "default": ("Kabako jumanw sɔrɔla.", "Sensor data received. No alerts at this time."), } # ── Fula templates (≤6 words per sentence for clear MMS-TTS output) ────────── # Format: (fula_text, english_translation) FULA_TEMPLATES: dict[str, tuple[str, str]] = { # Greetings / social "greeting": ("Jam waali. Mi woni ɗoo.", "Hello. I am here."), "greeting_morning": ("Jam waali. Yoo barke.", "Good morning. May there be blessings."), "greeting_evening": ("Jam hiiri. Jam waɗaa.", "Good evening. Peace be with you."), "thanks": ("Jaraama. A weli.", "Thank you. That is good."), "farewell": ("Yahdu jam. Mi anndii.", "Go in peace. I will remember you."), "not_understood": ("Mi faamii wanaa. Hol ɗoo.", "I did not understand. Please say again."), # Soil "soil_moisture_low": ("Leydi ndiyam famɗi. Wado ngesa.", "Soil moisture is low. Water the field."), "soil_moisture_high": ("Ndiyam heewi. Leydi famɗaali.", "Too much water. Drainage is needed."), "soil_ph_low": ("Leydi suurii. Waɗ kalisi.", "Soil is acidic. Add lime."), "soil_ph_high": ("Leydi alkalii. Waɗ soufre.", "Soil is alkaline. Add sulphur."), "soil_ok": ("Leydi weli. Ngesa yoodi.", "Soil is good. The field is ready."), # Weather "weather_hot": ("Nguleeki heewi. Muusal.", "It is very hot. Rest during midday."), "rain_likely": ("Ndiyam wadata. Loosu ngesa.", "Rain is coming. Protect the harvest."), "weather_ok": ("Jawdi weli. Waɗ golle.", "Weather is fine. Go work."), # Pest "pest_high": ("Biñ-biñ ngesa nder. Fiil ɗen.", "Pests are in the field. Remove them."), "pest_ok": ("Biñ-biñ alaa. Ngesa weli.", "No pests found. Field looks healthy."), # Irrigation "irrigation_needed": ("Ngesa fɛɗɛli ndiyam. Wado.", "Field needs water. Start irrigation."), "irrigation_active": ("Ndiyam wona jooni.", "Irrigation is running now."), # Default "default": ("Humpito juuti waɗaama.", "Sensor data received. No alerts."), } class VoiceResponder: """Converts sensor readings into actionable voice messages in the farmer's language. generate_response() returns (native_text, english_translation). For French/English, english_translation == native_text. """ def __init__(self, language: str = "fr") -> None: self.language = language def generate_response(self, intent: "Intent", sensor_data: "SensorData") -> tuple[str, str]: """Return (response_in_native_language, english_translation).""" if self.language == "bam": return self._bambara_response(intent, sensor_data) elif self.language == "ful": return self._fula_response(intent, sensor_data) else: text = self._french_response(sensor_data) return text, text # ── Bambara ────────────────────────────────────────────────────────────── def _bambara_response(self, intent: "Intent", sensor_data: "SensorData") -> tuple[str, str]: t = sensor_data.sensor_type v = sensor_data.values T = BAMBARA_TEMPLATES # Greeting intents — checked first regardless of sensor type if intent.action == "greeting": key = "greeting_morning" if v.get("is_morning") else "greeting" return T[key] if intent.action == "thanks": return T["thanks"] if intent.action == "farewell": return T["farewell"] if t == "soil": moisture = v.get("moisture_pct") if moisture is not None: if moisture < SOIL_MOISTURE_LOW: return T["soil_moisture_low"] elif moisture > SOIL_MOISTURE_HIGH: return T["soil_moisture_high"] ph = v.get("ph") if ph is not None: if ph < SOIL_PH_LOW: return T["soil_ph_low"] elif ph > SOIL_PH_HIGH: return T["soil_ph_high"] return T["soil_ok"] elif t == "weather": temp = v.get("temperature_c") rain = v.get("rain_probability_pct") if temp is not None and temp > TEMP_HIGH: return T["weather_hot"] if rain is not None and rain > 70: return T["rain_likely"] return T["weather_ok"] elif t == "irrigation": active = v.get("active") last = v.get("last_irrigation_h_ago") if active: return T["irrigation_active"] if last is not None and last > 24: return T["irrigation_needed"] elif t == "pest": level = int(v.get("alert_level", 0)) if level >= PEST_ALERT_HIGH: return T["pest_high"] return T["pest_ok"] return T["default"] # ── Fula ───────────────────────────────────────────────────────────────── def _fula_response(self, intent: "Intent", sensor_data: "SensorData") -> tuple[str, str]: t = sensor_data.sensor_type v = sensor_data.values T = FULA_TEMPLATES # Greeting intents — checked first regardless of sensor type if intent.action == "greeting": return T["greeting"] if intent.action == "thanks": return T["thanks"] if intent.action == "farewell": return T["farewell"] if t == "soil": moisture = v.get("moisture_pct") if moisture is not None: if moisture < SOIL_MOISTURE_LOW: return T["soil_moisture_low"] elif moisture > SOIL_MOISTURE_HIGH: return T["soil_moisture_high"] ph = v.get("ph") if ph is not None: if ph < SOIL_PH_LOW: return T["soil_ph_low"] elif ph > SOIL_PH_HIGH: return T["soil_ph_high"] return T["soil_ok"] elif t == "weather": temp = v.get("temperature_c") rain = v.get("rain_probability_pct") if temp is not None and temp > TEMP_HIGH: return T["weather_hot"] if rain is not None and rain > 70: return T["rain_likely"] return T["weather_ok"] elif t == "irrigation": active = v.get("active") last = v.get("last_irrigation_h_ago") if active: return T["irrigation_active"] if last is not None and last > 24: return T["irrigation_needed"] elif t == "pest": level = int(v.get("alert_level", 0)) if level >= PEST_ALERT_HIGH: return T["pest_high"] return T["pest_ok"] return T["default"] # ── French (original) ───────────────────────────────────────────────────── def _french_response(self, sensor_data: "SensorData") -> str: t = sensor_data.sensor_type v = sensor_data.values if t == "soil": return self._soil_response(v) elif t == "weather": return self._weather_response(v) elif t == "irrigation": return self._irrigation_response(v) elif t == "pest": return self._pest_response(v) else: return "Données du capteur non disponibles pour le moment." def _soil_response(self, v: dict) -> str: parts = [] moisture = v.get("moisture_pct") ph = v.get("ph") temp = v.get("temperature_c") nitrogen = v.get("nitrogen_ppm") if moisture is not None: parts.append(f"Humidité du sol : {moisture:.0f}%.") if moisture < SOIL_MOISTURE_LOW: parts.append("Irrigation recommandée immédiatement.") elif moisture > SOIL_MOISTURE_HIGH: parts.append("Sol trop humide, risque d'engorgement.") if ph is not None: parts.append(f"pH du sol : {ph:.1f}.") if ph < SOIL_PH_LOW: parts.append("Sol trop acide — envisagez un amendement calcaire.") elif ph > SOIL_PH_HIGH: parts.append("Sol trop alcalin — un apport de soufre peut aider.") if temp is not None: parts.append(f"Température du sol : {temp:.0f}°C.") if nitrogen is not None: parts.append(f"Azote disponible : {nitrogen:.0f} ppm.") if nitrogen < 15: parts.append("Niveau d'azote faible — envisagez un engrais azoté.") return " ".join(parts) if parts else "Données du sol reçues." def _weather_response(self, v: dict) -> str: parts = [] temp = v.get("temperature_c") humidity = v.get("humidity_pct") wind = v.get("wind_speed_kmh") rain = v.get("rain_probability_pct") if temp is not None: parts.append(f"Température : {temp:.0f}°C.") if temp > TEMP_HIGH: parts.append("Chaleur excessive — évitez les travaux aux heures les plus chaudes.") if humidity is not None: parts.append(f"Humidité de l'air : {humidity:.0f}%.") if wind is not None: parts.append(f"Vent : {wind:.0f} km/h.") if rain is not None: parts.append(f"Probabilité de pluie : {rain:.0f}%.") if rain > 70: parts.append("Pluie probable — reportez les traitements pesticides.") return " ".join(parts) if parts else "Données météo reçues." def _irrigation_response(self, v: dict) -> str: parts = [] active = v.get("active") last = v.get("last_irrigation_h_ago") flow = v.get("flow_rate_lph") if active is not None: state = "en marche" if active else "arrêtée" parts.append(f"Irrigation {state}.") if flow is not None and active: parts.append(f"Débit : {flow:.0f} litres par heure.") if last is not None: parts.append(f"Dernière irrigation il y a {last:.0f} heures.") if last > 24: parts.append("Plus de 24 heures sans irrigation — vérifiez les besoins en eau.") return " ".join(parts) if parts else "Statut d'irrigation reçu." def _pest_response(self, v: dict) -> str: level = int(v.get("alert_level", 0)) count = v.get("trap_count_24h") level_labels = {0: "aucune", 1: "faible", 2: "modérée", 3: "élevée"} label = level_labels.get(level, "inconnue") parts = [f"Présence d'insectes nuisibles : niveau {label}."] if count is not None: parts.append(f"{count:.0f} insectes capturés en 24 heures.") if level >= PEST_ALERT_HIGH: parts.append("Traitement recommandé — consultez un agent agricole.") return " ".join(parts)