Update app.py

app.py

@@ -5,27 +5,33 @@ from groq import Groq
 import time
 from PIL import Image
 import io
-from typing import Optional
+import os
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
 
 class SafetyMonitor:
-    def __init__(self, api_key: str, model_name: str = "mixtral-8x7b-vision"):
+    def __init__(self, model_name: str = "mixtral-8x7b-vision"):
         """
-        Initialize the safety monitor with configurable model
-        
-        Args:
-            api_key (str): Groq API key
-            model_name (str): Name of the vision model to use
+        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")
+            
         self.client = Groq(api_key=api_key)
         self.model_name = model_name
-        self.analysis_interval = 2  # seconds
         
     def analyze_frame(self, frame: np.ndarray) -> str:
         """
         Analyze a single frame using specified vision model
         """
+        if frame is None:
+            return "No frame received"
+            
         # Convert frame to PIL Image
-        frame_pil = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
+        frame_pil = Image.fromarray(frame)
         
         # Convert image to bytes
         img_byte_arr = io.BytesIO()
@@ -38,7 +44,6 @@ class SafetyMonitor:
         2. Unsafe behaviors or positions
         3. Equipment and machinery safety
         4. Environmental hazards (spills, obstacles, poor lighting)
-        5. Emergency exit accessibility
         
         Provide specific observations and any immediate safety concerns."""
         
@@ -55,107 +60,83 @@ class SafetyMonitor:
                 ],
                 model=self.model_name,
                 max_tokens=200,
-                temperature=0.2  # Lower temperature for more focused safety analysis
+                temperature=0.2
             )
             return completion.choices[0].message.content
         except Exception as e:
             return f"Analysis Error: {str(e)}"
 
-    def process_video_stream(self):
+    def process_frame(self, frame: np.ndarray) -> tuple[np.ndarray, str]:
         """
-        Process video stream and yield analyzed frames
+        Process and analyze a single frame
         """
-        cap = cv2.VideoCapture(0)  # Use 0 for webcam
-        last_analysis_time = 0
-        latest_analysis = "Initializing safety analysis..."
-
-        while cap.isOpened():
-            ret, frame = cap.read()
-            if not ret:
-                break
-
-            current_time = time.time()
+        if frame is None:
+            return None, "No frame received"
             
-            # Perform analysis at specified intervals
-            if current_time - last_analysis_time >= self.analysis_interval:
-                latest_analysis = self.analyze_frame(frame)
-                last_analysis_time = current_time
-
-            # 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 latest_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
-
-            yield display_frame
+        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
 
-        cap.release()
+        return display_frame, analysis
 
-def create_gradio_interface(monitor: SafetyMonitor):
+def create_gradio_interface():
     """
-    Create and launch the Gradio interface
+    Create and launch the Gradio interface with webcam input
     """
-    with gr.Blocks() as demo:
-        gr.Markdown(f"""
-        # Real-time Safety Monitoring System
-        Using model: {monitor.model_name}
-        """)
-        
-        with gr.Row():
-            video_output = gr.Image(label="Live Feed with Safety Analysis")
+    try:
+        # Initialize the safety monitor
+        monitor = SafetyMonitor(model_name="mixtral-8x7b-vision")
         
-        with gr.Row():
-            start_button = gr.Button("Start Monitoring", variant="primary")
-            stop_button = gr.Button("Stop")
+        with gr.Blocks() as demo:
+            gr.Markdown("""
+            # Real-time Safety Monitoring System
+            Click 'Start Webcam' to begin monitoring.
+            """)
             
-        with gr.Row():
-            interval_slider = gr.Slider(
-                minimum=1,
-                maximum=10,
-                value=monitor.analysis_interval,
-                step=0.5,
-                label="Analysis Interval (seconds)"
+            with gr.Row():
+                # Webcam input
+                webcam = gr.Image(source="webcam", streaming=True, label="Webcam Feed")
+                # Analysis output
+                output_image = gr.Image(label="Analyzed Feed")
+            
+            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"
             )
 
-        def update_interval(value):
-            monitor.analysis_interval = value
-            return gr.update()
-
-        def start_monitoring():
-            return gr.Image.update(value=monitor.process_video_stream())
-
-        start_button.click(fn=start_monitoring, outputs=[video_output])
-        stop_button.click(fn=lambda: None, outputs=[video_output])
-        interval_slider.change(fn=update_interval, inputs=[interval_slider])
-
-    demo.launch(share=True)
-
-def main():
-    # Replace with your actual API key
-    GROQ_API_KEY = "YOUR_GROQ_API_KEY"
-    
-    # Initialize the safety monitor with desired model
-    monitor = SafetyMonitor(
-        api_key=GROQ_API_KEY,
-        model_name="mixtral-8x7b-vision"  # Replace with your preferred model
-    )
-    
-    # Launch the Gradio interface
-    create_gradio_interface(monitor)
-
+        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__":
-    main()
-
+    create_gradio_interface()