Spaces:
Runtime error
Runtime error
| # drive_paddy/alerting/alert_system.py | |
| import time, os, io, google.generativeai as genai | |
| from gtts import gTTS | |
| class BaseAlerter: | |
| def __init__(self, config): | |
| self.config = config['alerting'] | |
| self.cooldown = self.config['alert_cooldown_seconds'] | |
| self.last_alert_time = 0 | |
| self.alert_on = False | |
| def trigger_alert(self, level="Very Drowsy"): raise NotImplementedError | |
| def reset_alert(self): | |
| if self.alert_on: print("Resetting Alert."); self.alert_on = False | |
| class FileAlertSystem(BaseAlerter): | |
| def __init__(self, config): | |
| super().__init__(config) | |
| self.audio_bytes = None | |
| try: | |
| if os.path.exists(config['alerting']['alert_sound_path']): | |
| with open(config['alerting']['alert_sound_path'], "rb") as f: self.audio_bytes = f.read() | |
| except Exception as e: print(f"Warning: Could not load audio file. Error: {e}.") | |
| def trigger_alert(self, level="Very Drowsy"): | |
| current_time = time.time() | |
| if (current_time - self.last_alert_time) > self.cooldown and not self.alert_on and self.audio_bytes: | |
| self.last_alert_time = current_time; self.alert_on = True | |
| print("Triggering Static Alert!") | |
| return self.audio_bytes | |
| return None | |
| class GeminiAlertSystem(BaseAlerter): | |
| def __init__(self, config, api_key): | |
| super().__init__(config) | |
| try: genai.configure(api_key=api_key); self.model = genai.GenerativeModel('gemini-pro') | |
| except Exception as e: print(f"Error initializing Gemini: {e}."); self.model = None | |
| def _generate_audio_data(self, level): | |
| if not self.model: return None | |
| if level == "Slightly Drowsy": | |
| prompt = "You are an AI driving assistant. Generate a short, gentle reminder (under 10 words) for a driver showing minor signs of fatigue." | |
| else: # Very Drowsy | |
| prompt = "You are an AI driving assistant. Generate a short, firm, and urgent alert (under 10 words) for a driver who is very drowsy." | |
| try: | |
| response = self.model.generate_content(prompt) | |
| alert_text = response.text.strip().replace('*', '') | |
| print(f"Generated Alert Text ({level}): '{alert_text}'") | |
| mp3_fp = io.BytesIO(); tts = gTTS(text=alert_text, lang='en'); tts.write_to_fp(mp3_fp) | |
| mp3_fp.seek(0); return mp3_fp.getvalue() | |
| except Exception as e: print(f"Error generating TTS audio: {e}"); return None | |
| def trigger_alert(self, level="Very Drowsy"): | |
| current_time = time.time() | |
| if (current_time - self.last_alert_time) > self.cooldown and not self.alert_on and self.model: | |
| self.last_alert_time = current_time; self.alert_on = True | |
| return self._generate_audio_data(level) | |
| return None | |
| def get_alerter(config, api_key=None): | |
| if config.get('gemini_api', {}).get('enabled', False) and api_key: return GeminiAlertSystem(config, api_key) | |
| return FileAlertSystem(config) | |