Update app.py

app.py

@@ -6,137 +6,114 @@ import time
 from PIL import Image
 import io
 import os
-from dotenv import load_dotenv
 
-# Load environment variables
-load_dotenv()
+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
 
-class SafetyMonitor:
-    def __init__(self, model_name: str = "mixtral-8x7b-vision"):
-        """
-        Initialize the safety monitor using environment variables for API key
-        """
-        api_key = os.getenv("GROQ_API_KEY")
-        if not api_key:
-            raise ValueError("GROQ_API_KEY environment variable is not set")
+        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()
             
-        self.client = Groq(api_key=api_key)
-        self.model_name = model_name
-        
-    def analyze_frame(self, frame: np.ndarray) -> str:
-        """
-        Analyze a single frame using specified vision model
-        """
-        if frame is None:
-            return "No frame received"
+            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."""
             
-        # Convert frame to PIL Image
-        frame_pil = Image.fromarray(frame)
-        
-        # Convert image to bytes
-        img_byte_arr = io.BytesIO()
-        frame_pil.save(img_byte_arr, format='JPEG')
-        img_byte_arr = img_byte_arr.getvalue()
-        
-        # Safety analysis prompt
-        prompt = """Please analyze this image for workplace safety issues. Focus on:
-        1. Required PPE usage (hard hats, safety glasses, reflective vests)
-        2. Unsafe behaviors or positions
-        3. Equipment and machinery safety
-        4. Environmental hazards (spills, obstacles, poor lighting)
-        
-        Provide specific observations and any immediate safety concerns."""
-        
-        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)}"
+            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]:
-        """
-        Process and analyze a single frame
-        """
-        if frame is None:
-            return None, "No frame received"
+        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()
             
-        analysis = self.analyze_frame(frame)
-        
-        # Create a copy of frame for visualization
-        display_frame = frame.copy()
-        
-        # Add semi-transparent overlay for text background
-        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)
-        
-        # Add analysis text
-        cv2.putText(display_frame, "Safety Analysis:", (10, 30),
-                    cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)
-        
-        # Split and display analysis text
-        y_position = 60
-        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
+            # 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
+            return display_frame, analysis
 
-def create_gradio_interface():
-    """
-    Create and launch the Gradio interface with webcam input
-    """
-    try:
-        # Initialize the safety monitor
-        monitor = SafetyMonitor(model_name="mixtral-8x7b-vision")
+    # Create the main interface
+    monitor = SafetyMonitor()
+    
+    with gr.Blocks() as demo:
+        gr.Markdown("# Real-time Safety Monitoring System")
         
-        with gr.Blocks() as demo:
-            gr.Markdown("""
-            # Real-time Safety Monitoring System
-            Click 'Start Webcam' to begin monitoring.
-            """)
+        with gr.Row():
+            webcam = gr.Image(source="webcam", streaming=True, label="Webcam Feed")
+            output_image = gr.Image(label="Analysis Feed")
+        
+        analysis_text = gr.Textbox(label="Safety Analysis", lines=5)
             
-            with gr.Row():
-                # Webcam input
-                webcam = gr.Image(source="webcam", streaming=True, label="Webcam Feed")
-                # Analysis output
-                output_image = gr.Image(label="Analyzed Feed")
+        def analyze_stream(frame):
+            if frame is None:
+                return None, "Webcam not started"
+            processed_frame, analysis = monitor.process_frame(frame)
+            return processed_frame, analysis
             
-            with gr.Row():
-                analysis_text = gr.Textbox(label="Safety Analysis", lines=5)
-                
-            def analyze_stream(frame):
-                if frame is None:
-                    return None, "Webcam not started"
-                processed_frame, analysis = monitor.process_frame(frame)
-                return processed_frame, analysis
-                
-            webcam.stream(
-                fn=analyze_stream,
-                outputs=[output_image, analysis_text],
-                show_progress="hidden"
-            )
+        webcam.stream(
+            fn=analyze_stream,
+            outputs=[output_image, analysis_text],
+            show_progress=False
+        )
 
-        demo.queue()
-        demo.launch()
-        
-    except ValueError as e:
-        print(f"Error: {e}")
-        print("Please make sure to set the GROQ_API_KEY environment variable")
-        
-if __name__ == "__main__":
-    create_gradio_interface()
+    return demo
+
+demo = create_monitor_interface()
+demo.launch()