Spaces:
Sleeping
Sleeping
# app_gradio.py | |
import gradio as gr | |
import numpy as np | |
import torch | |
import os, yaml, soundfile as sf | |
from dotenv import load_dotenv | |
from threading import Thread | |
import logging | |
# --- TTS & AI Imports --- | |
# from parler_tts import ParlerTTSForConditionalGeneration | |
# from transformers import AutoTokenizer, AutoFeatureExtractor | |
# from streamer import ParlerTTSStreamer # local file | |
from src.detection.factory import get_detector | |
from src.alerting.alert_system import FileAlertSystem | |
# ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
# CONFIG & BACKEND SET-UP | |
# ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
load_dotenv() | |
logging.basicConfig( | |
level=logging.INFO, | |
format="%(asctime)s %(levelname)s β %(message)s", | |
datefmt="%H:%M:%S", | |
) | |
with open("config.yaml", "r") as f: | |
config = yaml.safe_load(f) | |
secrets = {"gemini_api_key": os.getenv("GEMINI_API_KEY")} | |
print("Initializing detector and alerter β¦") | |
detector = get_detector(config) | |
alerter = FileAlertSystem(CONFIG) | |
if alerter.audio_bytes is None: | |
logging.warning("No alert sound loaded; driver will not hear any audio!") | |
print("Backend ready.") | |
# ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
# FRAME PROCESSOR | |
# ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
def process_live_frame(frame): | |
if frame is None: | |
return np.zeros((480, 640, 3), np.uint8), "Status: Inactive", None | |
t0 = time.time() | |
processed, indicators, _ = detector.process_frame(frame) | |
level = indicators.get("drowsiness_level", "Awake") | |
lighting = indicators.get("lighting", "Good") | |
score = indicators.get("details", {}).get("Score", 0) | |
dt_ms = (time.time() - t0) * 1000.0 | |
logging.info(f"{dt_ms:6.1f} ms β {lighting:<4} β {level:<14} β score={score:.2f}") | |
status_txt = f"Lighting: {lighting}\n" | |
status_txt += ("Detection paused due to low light." | |
if lighting == "Low" | |
else f"Status: {level}\nScore: {score:.2f}") | |
audio_out = None | |
audio_out = None | |
if level != "Awake" and lighting != "Low": | |
audio_bytes = alerter.trigger_alert(level=level) | |
logging.info(f"Printing {audio_bytes}") | |
if audio_bytes: | |
audio_out = audio_bytes | |
return processed, status_txt, audio_out | |
# ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
# GRADIO UI | |
# ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
with gr.Blocks(theme=gr.themes.Default(primary_hue="blue")) as app: | |
gr.Markdown("# π Drive Paddy β Drowsiness Detection") | |
gr.Markdown("Live detection with real-time voice alerts.") | |
with gr.Row(): | |
with gr.Column(scale=2): | |
webcam = gr.Image(sources=["webcam"], streaming=True, | |
label="Live Camera Feed") | |
with gr.Column(scale=1): | |
processed_img = gr.Image(label="Processed Feed") | |
status_box = gr.Textbox(label="Live Status", lines=3, interactive=False) | |
alert_audio = gr.Audio(label="Alert", | |
autoplay=True, | |
streaming=True) | |
webcam.stream( | |
fn=process_live_frame, | |
inputs=webcam, | |
outputs=[processed_img, status_box, alert_audio], | |
) | |
if __name__ == "__main__": | |
logging.info("Launching Gradio app β¦") | |
app.launch(debug=True) | |