app.py

import gradio as gr
import cv2
import numpy as np
from groq import Groq
import time
from PIL import Image
import io
import os

def create_monitor_interface():
    api_key = os.getenv("GROQ_API_KEY")
    
    if not api_key:
        with gr.Blocks() as demo:
            gr.Markdown("""
            # ⚠️ Groq API Key Required
            
            ## Setup Instructions for Hugging Face Space:
            1. Go to your Space's Settings tab
            2. Scroll down to "Repository Secrets"
            3. Click "New Secret"
            4. Enter:
               - Secret name: `GROQ_API_KEY`
               - Secret value: Your Groq API key
            5. Click "Add secret"
            6. Rebuild the Space
            
            Once configured, the safety monitoring system will be available.
            """)
        return demo

    class SafetyMonitor:
        def __init__(self, model_name: str = "mixtral-8x7b-vision"):
            self.client = Groq(api_key=api_key)
            self.model_name = model_name

        def analyze_frame(self, frame: np.ndarray) -> str:
            if frame is None:
                return "No frame received"
                
            frame_pil = Image.fromarray(frame)
            img_byte_arr = io.BytesIO()
            frame_pil.save(img_byte_arr, format='JPEG')
            img_byte_arr = img_byte_arr.getvalue()
            
            prompt = """Analyze this image for workplace safety issues. Focus on:
            1. PPE usage (helmets, vests, etc.)
            2. Unsafe behaviors
            3. Equipment safety
            4. Environmental hazards
            Provide specific observations."""
            
            try:
                completion = self.client.chat.completions.create(
                    messages=[
                        {
                            "role": "user",
                            "content": [
                                {"type": "text", "text": prompt},
                                {"type": "image", "image": img_byte_arr}
                            ]
                        }
                    ],
                    model=self.model_name,
                    max_tokens=200,
                    temperature=0.2
                )
                return completion.choices[0].message.content
            except Exception as e:
                return f"Analysis Error: {str(e)}"

        def process_frame(self, frame: np.ndarray) -> tuple[np.ndarray, str]:
            if frame is None:
                return None, "No frame received"
                
            analysis = self.analyze_frame(frame)
            display_frame = frame.copy()
            
            # Add text overlay
            overlay = display_frame.copy()
            cv2.rectangle(overlay, (5, 5), (640, 200), (0, 0, 0), -1)
            cv2.addWeighted(overlay, 0.3, display_frame, 0.7, 0, display_frame)
            
            y_position = 30
            for line in analysis.split('\n'):
                cv2.putText(display_frame, line[:80], (10, y_position),
                           cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)
                y_position += 30

            return display_frame, analysis

    # Create the main interface
    monitor = SafetyMonitor()
    
    with gr.Blocks() as demo:
        gr.Markdown("# Real-time Safety Monitoring System")
        
        with gr.Row():
            webcam = gr.Webcam(label="Webcam Feed")
            output_image = gr.Image(label="Analysis Feed")
        
        analysis_text = gr.Textbox(label="Safety Analysis", lines=5)
            
        def analyze_stream(image):
            if image is None:
                return None, "Webcam not started"
            processed_frame, analysis = monitor.process_frame(image)
            return processed_frame, analysis
            
        webcam.change(
            fn=analyze_stream,
            inputs=[webcam],
            outputs=[output_image, analysis_text],
        )

    return demo

demo = create_monitor_interface()
demo.launch()