import gradio as gr
import cv2
import numpy as np
from groq import Groq
import time
from PIL import Image as PILImage
import io
import os
import base64
import random
def create_monitor_interface():
api_key = os.getenv("GROQ_API_KEY")
class SafetyMonitor:
def __init__(self):
self.client = Groq()
self.model_name = "llama-3.2-90b-vision-preview"
self.max_image_size = (800, 800) # Increased size for better visibility
self.colors = [(255, 0, 0), (0, 255, 0), (0, 0, 255), (255, 255, 0), (255, 0, 255)]
def resize_image(self, image):
height, width = image.shape[:2]
aspect = width / height
if width > height:
new_width = min(self.max_image_size[0], width)
new_height = int(new_width / aspect)
else:
new_height = min(self.max_image_size[1], height)
new_width = int(new_height * aspect)
return cv2.resize(image, (new_width, new_height), interpolation=cv2.INTER_AREA)
def analyze_frame(self, frame: np.ndarray) -> str:
if frame is None:
return "No frame received"
# Convert and resize image
if len(frame.shape) == 2:
frame = cv2.cvtColor(frame, cv2.COLOR_GRAY2RGB)
elif len(frame.shape) == 3 and frame.shape[2] == 4:
frame = cv2.cvtColor(frame, cv2.COLOR_RGBA2RGB)
frame = self.resize_image(frame)
frame_pil = PILImage.fromarray(frame)
# High quality image for better analysis
buffered = io.BytesIO()
frame_pil.save(buffered,
format="JPEG",
quality=95,
optimize=True)
img_base64 = base64.b64encode(buffered.getvalue()).decode('utf-8')
image_url = f"data:image/jpeg;base64,{img_base64}"
try:
completion = self.client.chat.completions.create(
model=self.model_name,
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": """As a comprehensive safety expert, analyze this image for ALL potential safety concerns and hazards, including:
- Personal Protective Equipment (PPE)
- Ergonomic issues and posture
- Fire and electrical hazards
- Chemical and environmental hazards
- Machine and equipment safety
- Fall protection and heights
- Material handling and storage
- Emergency access and exits
- Housekeeping and organization
- Lighting and visibility
- Ventilation and air quality
- Tool safety and maintenance
For each safety issue identified, provide:
1. Exact location in the image (be specific: top-left, center-right, bottom, etc.)
2. Type of safety concern
3. Potential risk or hazard
4. Relevant safety violation
Format each observation exactly as:
- position:detailed safety issue description
Example:
- top-right:Exposed electrical wiring creating shock hazard
- bottom-left:Improperly stored chemicals without labeling
Identify ALL safety issues, not just the obvious ones."""
},
{
"type": "image_url",
"image_url": {
"url": image_url
}
}
]
}
],
temperature=0.7,
max_tokens=500,
stream=False
)
return completion.choices[0].message.content
except Exception as e:
print(f"Detailed error: {str(e)}")
return f"Analysis Error: {str(e)}"
def draw_observations(self, image, observations):
"""Draw accurate bounding boxes based on safety issue locations."""
height, width = image.shape[:2]
font = cv2.FONT_HERSHEY_SIMPLEX
font_scale = 0.5
thickness = 2
padding = 10
def get_region_coordinates(position: str) -> tuple:
"""Get coordinates based on position description."""
# Basic regions
regions = {
'top-left': (0, 0, width//3, height//3),
'top': (width//3, 0, 2*width//3, height//3),
'top-right': (2*width//3, 0, width, height//3),
'center-left': (0, height//3, width//3, 2*height//3),
'center': (width//3, height//3, 2*width//3, 2*height//3),
'center-right': (2*width//3, height//3, width, 2*height//3),
'bottom-left': (0, 2*height//3, width//3, height),
'bottom': (width//3, 2*height//3, 2*width//3, height),
'bottom-right': (2*width//3, 2*height//3, width, height),
'left': (0, height//4, width//3, 3*height//4),
'right': (2*width//3, height//4, width, 3*height//4)
}
# Find best matching region
best_match = 'center'
max_words = 0
pos_lower = position.lower()
for region in regions.keys():
words = region.split('-')
matches = sum(1 for word in words if word in pos_lower)
if matches > max_words:
max_words = matches
best_match = region
return regions[best_match]
for idx, obs in enumerate(observations):
color = self.colors[idx % len(self.colors)]
# Parse location and description
parts = obs.split(':')
if len(parts) >= 2:
position = parts[0]
description = ':'.join(parts[1:])
# Get region coordinates
x1, y1, x2, y2 = get_region_coordinates(position)
# Draw rectangle
cv2.rectangle(image, (x1, y1), (x2, y2), color, 2)
# Add label with background
label = description[:50] + "..." if len(description) > 50 else description
label_size, _ = cv2.getTextSize(label, font, font_scale, thickness)
# Position text above the box
text_x = max(0, x1)
text_y = max(label_size[1] + padding, y1 - padding)
# Draw text background
cv2.rectangle(image,
(text_x, text_y - label_size[1] - padding),
(text_x + label_size[0] + padding, text_y),
color, -1)
# Draw text
cv2.putText(image, label,
(text_x + padding//2, text_y - padding//2),
font, font_scale, (255, 255, 255), thickness)
return image
def process_frame(self, frame: np.ndarray) -> tuple[np.ndarray, str]:
if frame is None:
return None, "No image provided"
analysis = self.analyze_frame(frame)
display_frame = self.resize_image(frame.copy())
# Parse observations from the analysis
observations = []
for line in analysis.split('\n'):
line = line.strip()
if line.startswith('-'):
# Extract text between tags if present
if '' in line and '' in line:
start = line.find('') + len('')
end = line.find('')
observation = line[end + len(''):].strip()
else:
observation = line[1:].strip() # Remove the dash
if observation:
observations.append(observation)
# Draw observations on the image
annotated_frame = self.draw_observations(display_frame, observations)
return annotated_frame, analysis
# Create the main interface
monitor = SafetyMonitor()
with gr.Blocks() as demo:
gr.Markdown("# Safety Analysis System powered by Llama 3.2 90b vision")
with gr.Row():
input_image = gr.Image(label="Upload Image")
output_image = gr.Image(label="Annotated Results")
analysis_text = gr.Textbox(label="Detailed Analysis", lines=5)
def analyze_image(image):
if image is None:
return None, "No image provided"
try:
processed_frame, analysis = monitor.process_frame(image)
return processed_frame, analysis
except Exception as e:
print(f"Processing error: {str(e)}")
return None, f"Error processing image: {str(e)}"
input_image.change(
fn=analyze_image,
inputs=input_image,
outputs=[output_image, analysis_text]
)
return demo
demo = create_monitor_interface()
demo.launch()