# Project export: HazardVision
This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.
## Project metadata
- Hackathon: UC Berkeley AI Hackathon 2025
- Tagline: HazardVision never blinks. While humans overlook risks, our AI watches every tool, every move, and prevents accidents before they occur.
- Devpost: https://devpost.com/software/hazardvision
- GitHub: https://github.com/Pranman1/HazardVision
- Video: https://www.youtube.com/embed/JlA8UPWAgww?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 1 GitHub contributor(s) — Pranav Bhatt (10 commits)
## Devpost submission (written by the team)
### Inspiration
Every day, nearly one million workers are injured on the job—and the majority of these injuries are entirely preventable. Visual hazards like unattended tools, blocked exits, and missing PPE (personal protective equipment) are responsible for most of them. Yet while AI innovation in fields like healthcare and finance accelerates, blue-collar industries—where lives are at stake daily—remain underserved. Workplace safety is a crisis in the U.S.: 4,764 workers died on the job in 2022 (U.S. BLS) That’s one death every 96 minutes Non-fatal injuries cost the U.S. economy $167 billion annually (NSC, 2022) We built HazardVision to bridge this safety gap. Our mission: bring real-time, intelligent oversight to factories, warehouses, and worksites—places where one missed step can cost a life.
### What it does
HazardVision is an AI-powered safety assistant that monitors the workplace through real-time video. It automatically detects and classifies hazards, then alerts workers before accidents occur. It detects: Trip and fall risks (e.g., tools, cables, spills) Improper PPE usage (e.g., missing helmets, gloves) Unsafe equipment handling (e.g., stagnant knives, exposed saws) Blocked emergency exits or cluttered walkways Our system uses: Live webcam input Bounding boxes with severity-coded colors (green/yellow/red) Real-time hazard logs Audio alerts with pitch based on severity It even adjusts classification based on context: a tool in use may be low-risk, but abandoned becomes a critical hazard. And if multiple hazards persist unattended, our AI agent can escalate by notifying a manager or safety officer—ensuring OSHA-level accountability.
### How we built it
We designed a full-stack real-time computer vision system with agentic capabilities: Frontend: HTML/CSS/JavaScript for the dashboard MediaDevices API for webcam access Canvas API for real-time bounding box rendering WebSockets for bi-directional communication AudioContext API for hazard-specific alert tones Backend: FastAPI (Python) + Uvicorn for high-speed async handling Ultralytics YOLOv8 for object detection (PyTorch under the hood) OpenCV for image annotation Custom hazard classification logic (severity scoring, event triggers) Agentic pipeline powered by Vapi and ElevenLabs for escalated alerts The frontend captures frames and sends them via WebSocket to the backend. There, the YOLOv8 model detects tools, objects, and unsafe setups. Based on context, we classify hazard severity and return bounding boxes and labels. If the situation escalates (e.g. abandoned blades or multiple unmitigated risks), our AI voice agent contacts factory management with a violation notice.
### Challenges we ran into
Ensuring inference speed was fast enough for smooth live video Designing a severity framework adaptable to various real-world tool usage Syncing audio, visual annotations, and hazard logs in real-time Learning how to implement agentic workflows using Vapi and ElevenLabs Finding appropriate data to train YOLO for real manufacturing environments Testing the model with real materials (tools, machines, gloves, etc.) One of our teammates had to leave midway—requiring us to re-plan, regroup, and rebuild
### Accomplishments we're proud of
Built a complete real-time detection and alert system from scratch Successfully trained a CNN to identify hazards and contextualize them based on environment Achieved low-latency AI inference using YOLOv8 on webcam input Created a live hazard dashboard with severity mapping, log tracking, and alert sounds Formed a resilient team from across different countries and universities Rebounded and finished the project strong, even after a teammate left mid-hackathon
### What we learned
How to optimize AI models for real-time use in edge environments How to integrate new technologies like Vapi and agentic voice systems How to design for industrial UX—where alert clarity and speed can be life-saving How to collaborate across time zones, adapt when things break, and keep pushing forward
### What's next
Expand detection to fire, gas leaks, electrical sparks, and smoke Partner with manufacturing labs and warehouse operators for field testing Build a plug-and-play camera module for easy installation in existing CCTV setups Continue training on more diverse datasets for better generalization Bring the system to small and mid-sized factories across the U.S.—where safety staff is minimal and the risk is high
## README (from the GitHub repository)
No README available.
## Detected evidence (automated analysis)
Indexed codebase: 9 recognized source files, 54 KB.
- Anthropic (technology) — detected in the code
- CSS (language) — detected in the code
- FastAPI (technology) — detected in the code
- HTML (language) — detected in the code
- JavaScript (language) — detected in the code
- Python (language) — detected in the code
- PyTorch (technology) — claimed on Devpost, not found in the code
## Codebase structure (from repository index)
### Files (14 of 14)
```
.gitignore
backend/claude.py
backend/detect.py
backend/hazard_analysis.py
backend/hazard_log.json
backend/main.py
backend/requirements.txt
backend/utils.py
backend/yolov8m.pt
backend/yolov8n.pt
frontend/public/index.html
frontend/static/app.js
frontend/static/style.css
YOLOV8CNNtuner/labels/data.yaml
```
### Dependencies
- backend/requirements.txt: anthropic, fastapi, opencv-python, requests, ultralytics, uvicorn
### Recent commits (newest first)
- another svae
- finalement savoir
- finalement savoir
- save
- final save
- oui oui
- oui oui
- baguette
- yo
- yo
- savoir moi
## Key source files (fetched from GitHub, selected and truncated for size)
### backend/requirements.txt
```
fastapi
uvicorn
opencv-python
ultralytics
requests
anthropic
```
### backend/main.py
```python
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
import cv2
import base64
import numpy as np
import json
from threading import Thread, Lock
from queue import Queue
import asyncio
from datetime import datetime
import torch
from detect import HAZARD_CATEGORIES, classify_hazard
from ultralytics import YOLO
import logging
from hazard_analysis import process_critical_hazard
import os
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
# Enable CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Mount static files
app.mount("/static", StaticFiles(directory="../frontend/static"), name="static")
app.mount("/snapshots", StaticFiles(directory="../snapshots"), name="snapshots")
# Serve frontend
from fastapi.responses import FileResponse
@app.get("/")
async def read_root():
return FileResponse("../frontend/public/index.html")
# Initialize model with GPU if available
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Using device: {device}")
model = YOLO("yolov8m.pt")
model.to(device)
# Initialize GPU memory management
if torch.cuda.is_available():
torch.cuda.empty_cache()
torch.cuda.set_per_process_memory_fraction(0.8)
torch.backends.cudnn.benchmark = True
# Shared resources
frame_queue = Queue(maxsize=2) # Only keep latest frame
result_lock = Lock()
latest_result = None
# Model configuration
conf_threshold = 0.3 # Confidence threshold for detections
def process_frame(frame):
"""Process a single frame with the model"""
try:
# Convert frame to RGB for YOLO
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
# Run detection
results = model(frame_rgb, conf=conf_threshold)
labels = []
boxes = []
# Process results
for result in results:
for box in result.boxes:
conf = float(box.conf)
label = result.names[int(box.cls)]
if conf >= conf_threshold:
x1, y1, x2, y2 = box.xyxy[0].cpu().numpy()
boxes.append({
"box": [int(x1), int(y1), int(x2), int(y2)],
"label": label,
"confidence": conf
})
labels.append(label)
# Default green
color = (0, 255, 0)
for category, rules in HAZARD_CATEGORIES.items():
if label in rules["objects"]:
if "fall" in category or "sharp" in category or "fire" in category:
color = (0, 0, 255) # Red for serious hazards
elif "trip" in category:
color = (255, 165, 0) # Orange for trip hazards
elif "electrical" in category:
color = (255, 255, 0) # Yellow for electrical
break
cv2.rectangle(frame, (int(x1), int(y1)), (int(x2), int(y2)), color, 2)
label_text = f"{label} {conf:.2f}"
cv2.putText(frame, label_text, (int(x1), int(y1)-10),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2)
# Classify hazard
is_hazardous, hazard_types, severity = classify_hazard(labels, boxes)
# Process critical hazards with LLM and speech synthesis
hazard_analysis = None
if is_hazardous and severity == "critical":
# Save frame for analysis
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
frame_path = f"snapshots/hazard_{timestamp}.jpg"
cv2.imwrite(frame_path, frame)
# Get hazard analysis
hazard_analysis = process_critical_hazard(frame_path, hazard_types, severity)
# Convert frame to base64 for sending
_, buffer = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
frame_b64 = base64.b64encode(buffer).decode('utf-8')
return {
"frame": frame_b64,
"timestamp": datetime.now().strftime("%Y%m%d-%H%M%S"),
"labels": labels,
"boxes": boxes,
"is_hazardous": is_hazardous,
"hazard_types": hazard_types,
"severity": severity,
"hazard_analysis": hazard_analysis
}
except Exception as e:
logger.error(f"Error processing frame: {str(e)}")
return None
def detection_thread():
"""Background thread for processing frames"""
while True:
if not frame_queue.empty():
frame = frame_queue.get()
try:
result = process_frame(frame)
if result:
with result_lock:
global latest_result
latest_result = result
except Exception as e:
logger.error(f"Error in detection thread: {str(e)}")
finally:
if torch.cuda.is_available():
torch.cuda.empty_cache()
# Start detection thread
Thread(target=detection_thread, daemon=True).start()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
logger.info("WebSocket connection accepted")
try:
while True:
try:
# Receive frame data
data = await websocket.receive_text()
# Parse frame data
frame_data = json.loads(data)
if 'frame' not in frame_data:
continue
# Decode frame
frame_bytes = base64.b64decode(
[truncated — 1885 more characters]
```
### frontend/static/app.js
```javascript
const video = document.getElementById('video');
const canvas = document.createElement('canvas');
const ctx = canvas.getContext('2d');
const table = document.getElementById('logTable');
let isRunning = false;
let ws = null;
// FPS tracking
let frameCount = 0;
let lastFpsUpdate = Date.now();
const fpsDisplay = document.getElementById('fpsDisplay');
let skipFrames = 2; // Process every 3rd frame
let frameSkipCount = 0;
// Detection settings
const RECONNECT_TIMEOUT = 3000; // 3 seconds
// Audio feedback for hazards
let audioContext = null;
let lastAudioTime = 0;
const AUDIO_COOLDOWN = 3000; // 3 seconds between alerts
// Audio handling
let currentAudio = null;
// Add message timeout tracking
let messageTimeout = null;
let lastHazardTime = null;
const MESSAGE_DISPLAY_TIME = 45000; // 45 seconds for hazard messages
const FADE_OUT_TIME = 5000; // 5 seconds fade out transition
const NO_HAZARD_COOLDOWN = 30000; // Wait 30 seconds before allowing "No hazards" state
function createAudioContext() {
if (!audioContext) {
audioContext = new (window.AudioContext || window.webkitAudioContext)();
}
return audioContext;
}
function playHazardAlert(severity) {
const now = Date.now();
if (now - lastAudioTime > AUDIO_COOLDOWN) {
const context = createAudioContext();
// Create oscillator for beep
const oscillator = context.createOscillator();
const gainNode = context.createGain();
// Set frequency based on severity
switch(severity) {
case 'high':
oscillator.frequency.setValueAtTime(880, context.currentTime); // A5
break;
case 'medium':
oscillator.frequency.setValueAtTime(440, context.currentTime); // A4
break;
default:
oscillator.frequency.setValueAtTime(220, context.currentTime); // A3
}
// Connect nodes
oscillator.connect(gainNode);
gainNode.connect(context.destination);
// Set volume envelope
gainNode.gain.setValueAtTime(0, context.currentTime);
gainNode.gain.linearRampToValueAtTime(0.5, context.currentTime + 0.1);
gainNode.gain.linearRampToValueAtTime(0, context.currentTime + 0.5);
// Start and stop
oscillator.start(context.currentTime);
oscillator.stop(context.currentTime + 0.5);
lastAudioTime = now;
}
}
function playHazardAudio(audioPath) {
try {
// Stop any currently playing audio
if (currentAudio) {
currentAudio.pause();
currentAudio = null;
}
// Create and play new audio
currentAudio = new Audio(audioPath);
currentAudio.play();
} catch (error) {
console.error('Error playing audio:', error);
}
}
// Helper function to create image elements with proper loading states
function createImage(src, width = null) {
const img = document.createElement("img");
if (width) img.width = width;
img.alt = "Loading...";
img.classList.add("loading");
img.onload = function() {
img.classList.remove("loading");
img.classList.remove("error");
img.alt = "Detection result";
};
img.onerror = function() {
console.error(`Failed to load image: ${src}`);
img.classList.remove("loading");
img.classList.add("error");
img.alt = "Failed to load image";
img.style.backgroundColor = "#f8d7da";
img.style.border = "1px solid #f5c6cb";
};
img.src = src;
return img;
}
// Start camera
async function initCamera() {
try {
// First enumerate devices
const devices = await navigator.mediaDevices.enumerateDevices();
const videoDevices = devices.filter(device => device.kind === 'videoinput');
// Try to find DroidCam
const droidcam = videoDevices.find(device => device.label.toLowerCase().includes('droidcam'));
// Get video stream
const stream = await navigator.mediaDevices.getUserMedia({
video: {
width: { ideal: 640 },
height: { ideal: 480 },
frameRate: { ideal: 30 },
deviceId: droidcam ? droidcam.deviceId : undefined
}
});
video.srcObject = stream;
video.addEventListener('loadedmetadata', () => {
canvas.width = video.videoWidth;
canvas.height = video.videoHeight;
});
console.log('Camera initialized successfully');
console.log('Available video devices:', videoDevices.map(d => d.label));
} catch (err) {
console.error("Camera error:", err);
}
}
// Initialize camera
initCamera();
function connectWebSocket() {
if (ws) {
ws.close();
}
ws = new WebSocket('ws://localhost:8000/ws');
ws.onopen = () => {
console.log('WebSocket connected');
if (isRunning) {
startDetection();
}
};
ws.onclose = () => {
console.log('WebSocket disconnected');
setTimeout(connectWebSocket, RECONNECT_TIMEOUT);
};
ws.onerror = (error) => {
console.error('WebSocket error:', error);
};
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
// Update hazard info display
updateHazardInfo(data);
// Update detection image
const annotatedImg = document.getElementById('annotatedImg');
if (!annotatedImg) return;
// Create new image from base64 data
annotatedImg.src = 'data:image/jpeg;base64,' + data.frame;
// Play alert if hazardous
if (data.is_hazardous) {
if (data.severity === 'critical') {
// Play beep alert
playHazardAlert(data.severity);
[truncated — 6403 more characters]
```
### backend/claude.py
```python
import json
from anthropic import Anthropic
def summarize_logs():
with open("backend/hazard_log.json") as f:
logs = json.load(f)
text = json.dumps(logs[-10:], indent=2)
prompt = f"Summarize these workplace hazard logs:\n{text}"
client = Anthropic(api_key="sk-your-key-here")
response = client.messages.create(
model="claude-3-opus-20240229",
messages=[{"role": "user", "content": prompt}],
max_tokens=300
)
return {"summary": response.content[0].text}
```
### backend/utils.py
```python
import json
import os
from datetime import datetime
import cv2
def log_hazard(labels, timestamp, image_path):
"""Log hazard detection to file - minimal logging"""
try:
log_entry = {
"timestamp": timestamp,
"labels": labels,
"image": os.path.basename(image_path)
}
log_file = "hazard_log.json"
# Keep log file small by limiting entries
entries = []
if os.path.exists(log_file):
try:
with open(log_file, 'r') as f:
entries = json.load(f)
# Keep only last 100 entries
entries = entries[-100:]
except:
entries = []
entries.append(log_entry)
with open(log_file, 'w') as f:
json.dump(entries[-100:], f) # Only keep last 100 entries
except Exception as e:
print(f"Error logging hazard: {str(e)}")
def save_snapshot(frame, filename):
"""Save a frame to disk with compression"""
try:
# Save with reduced quality
cv2.imwrite(filename, frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
return True
except Exception as e:
print(f"Error saving snapshot: {str(e)}")
return False
```
### backend/hazard_analysis.py
```python
import google.generativeai as genai
import requests
import base64
import json
import os
from datetime import datetime
from PIL import Image
# API Configuration
GEMINI_API_KEY = "AIzaSyAl6X4cZYC2GE4KRTyOc__3k-ujX4ayEno"
ELEVEN_LABS_API_KEY = "sk_0cd0f86badc873f3524f5f349598f0bdca2291223d13754c"
# Configure Gemini
genai.configure(api_key=GEMINI_API_KEY)
model = genai.GenerativeModel('gemini-pro-vision')
def encode_image_base64(image_path):
"""Convert image to base64 string"""
with open(image_path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode('utf-8')
def analyze_hazard_with_llm(image_path, hazard_types, severity):
"""Use Gemini to analyze the hazard situation"""
try:
# Prepare the system prompt with more specific context
system_prompt = """You are a workplace safety expert AI. Your task is to analyze workplace safety situations and provide specific, actionable feedback. Focus on:
1. The exact hazard type(s) detected
2. The specific dangers they present
3. Clear, immediate actions needed to resolve the situation
Keep responses concise and actionable, under 100 words.
Format: "Safety Alert: [specific hazard]. This is dangerous because [specific reason]. Recommended Action: [specific action]."
Important: Never use underscores in your response. Use spaces instead.
"""
# Load image using PIL
image = Image.open(image_path)
# Create a more detailed prompt with specific hazard context
hazard_descriptions = {
"fall_hazards": "potential falls from height",
"floor obstacles": "items blocking walkways or creating trip hazards",
"sharp_objects": "dangerous sharp tools or objects",
"fire_hazards": "potential fire sources",
"electrical_hazards": "unsafe electrical conditions",
"chemical_hazards": "hazardous chemicals",
"unattended sharp object": "sharp objects left unattended",
"unsafe sharp object handling": "improper handling of sharp tools",
"electrical hazard": "unsafe electrical situation",
"bag": "bag left on floor creating obstacle",
"backpack": "backpack blocking walkway",
"box": "box creating trip hazard",
"cord": "cord creating trip hazard",
"cable": "cable creating trip hazard",
"bottle": "bottle left on floor"
}
# Build detailed hazard description
hazard_details = []
for hazard in hazard_types:
# Convert any underscores to spaces first
hazard = hazard.replace("_", " ")
if hazard in hazard_descriptions:
hazard_details.append(hazard_descriptions[hazard])
else:
hazard_details.append(hazard)
prompt = f"""Analyze this workplace safety situation.
Detected hazards: {', '.join(hazard_details)}.
Severity level: {severity.upper()}.
Provide specific details about the visible hazards and clear actions needed.
Remember: Use spaces instead of underscores in your response."""
# Get LLM response
response = model.generate_content([system_prompt, prompt, image])
if response.text:
# Ensure no underscores in response
return response.text.replace("_", " ")
else:
return f"Safety Alert: {', '.join(hazard_details)} detected. Severity: {severity}. Immediate inspection required."
except Exception as e:
print(f"Error in LLM analysis: {str(e)}")
# Ensure no underscores in error message
hazard_list = [h.replace("_", " ") for h in hazard_types]
return f"Safety Alert: {', '.join(hazard_list)} detected. Severity: {severity}. Please check the area immediately."
def generate_speech(text):
"""Generate speech using ElevenLabs API"""
try:
url = "https://api.elevenlabs.io/v1/text-to-speech/21m00Tcm4TlvDq8ikWAM" # Using Rachel voice
headers = {
"Accept": "audio/mpeg",
"Content-Type": "application/json",
"xi-api-key": ELEVEN_LABS_API_KEY
}
data = {
"text": text,
"model_id": "eleven_monolingual_v1",
"voice_settings": {
"stability": 0.75,
"similarity_boost": 0.75
}
}
response = requests.post(url, json=data, headers=headers)
if response.status_code == 200:
# Save the audio file with timestamp
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
# Update path to use the correct frontend directory
audio_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "frontend", "static", "alerts", f"alert_{timestamp}.mp3")
# Ensure directory exists
os.makedirs(os.path.dirname(audio_path), exist_ok=True)
with open(audio_path, "wb") as f:
f.write(response.content)
# Return the correct URL path for the frontend
return f"/static/alerts/alert_{timestamp}.mp3"
else:
print(f"Error generating speech: {response.status_code}")
return None
except Exception as e:
print(f"Error in speech generation: {str(e)}")
return None
def process_critical_hazard(image_path, hazard_types, severity):
"""Process a critical hazard with LLM analysis and speech synthesis"""
try:
# Get LLM analysis
analysis = analyze_hazard_with_llm(image_path, hazard_types, severity)
# Generate speech
audio_path = generate_speech(analysis)
return {
"analysis": analysis,
"audio_path": audio_path
}
except Exception as e:
print(f"Error processing critical hazard: {str(e)}")
[truncated — 13 more characters]
```
### backend/detect.py
```python
import cv2
import os
import torch
import numpy as np
from datetime import datetime, timedelta
from ultralytics import YOLO
from utils import save_snapshot, log_hazard
import glob
from pathlib import Path
import requests
import logging
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Check for CUDA availability and set device
device = 'cuda' if torch.cuda.is_available() else 'cpu'
print(f"Using device: {device}")
# Load the YOLOv8 model with CUDA support and better confidence threshold
model = YOLO("yolov8m.pt")
model.to(device) # Move model to GPU
conf_threshold = 0.3 # Lower confidence threshold for better knife detection
# Enhanced OSHA-aligned hazard categories
HAZARD_CATEGORIES = {
"fall_hazards": {
"objects": {"ladder", "chair", "platform", "scaffold", "stairs"},
"min_height": 4, # feet - OSHA requires fall protection above 4 feet
"severity": "high"
},
"floor_obstacles": { # New category specifically for floor hazards
"objects": {"bag", "backpack", "suitcase", "box", "cord", "cable", "bottle"},
"severity": "critical", # Increased severity for floor obstacles
"ground_level": True
},
"sharp_objects": {
"objects": {"knife", "scissors", "tool", "saw", "drill", "blade", "cutter"},
"confidence_threshold": 0.3,
"safe_handling": {
"required_context": ["hand", "person"],
"safe_distance": 50,
"severity": {
"unattended": "critical",
"improper_handling": "high",
"proper_handling": "medium"
}
}
},
"fire_hazards": {
"objects": {"fire", "smoke", "cigarette", "matches", "lighter"},
"always_hazard": True,
"severity": "critical"
},
"electrical_hazards": {
"objects": {"cord", "wire", "cable", "outlet", "power strip", "electrical panel"},
"context_sensitive": True,
"severity": "high"
},
"chemical_hazards": {
"objects": {"bottle", "container", "spray", "tank"},
"context_sensitive": True,
"severity": "high"
}
}
# Enhanced hazardous object combinations
HAZARD_COMBINATIONS = [
# Fall hazards - person on elevated surface
{"person", "ladder"},
{"person", "chair"},
{"person", "platform"},
{"person", "scaffold"},
{"person", "stairs"},
# Floor obstacles - objects that shouldn't be on the floor
{"bag"},
{"backpack"},
{"suitcase"},
{"box"},
{"cord"},
{"cable"},
{"bottle"},
# Sharp object hazards
{"person", "knife"},
{"person", "scissors"},
{"person", "saw"},
{"person", "drill"},
# Fire hazards
{"fire"},
{"smoke"},
{"person", "cigarette"},
# Electrical hazards
{"person", "cord"},
{"person", "wire"},
{"person", "cable"},
{"person", "outlet"},
{"person", "electrical panel"},
# Multiple people in hazardous situation
{"person", "person", "ladder"},
{"person", "person", "scaffold"},
]
# Add cooldown tracking
HAZARD_COOLDOWN = 9 # seconds (3x the base cooldown of 3 seconds)
last_hazard_detection = {} # Store timestamp of last detection for each hazard type
critical_alert_count = 0 # Track number of critical alerts
def is_hazard_in_cooldown(hazard_type):
"""Check if a hazard type is still in cooldown period"""
global last_hazard_detection
current_time = datetime.now()
last_time = last_hazard_detection.get(hazard_type)
if last_time is None:
return False
time_diff = current_time - last_time
return time_diff.total_seconds() < HAZARD_COOLDOWN
def update_hazard_timestamp(hazard_type):
"""Update the last detection timestamp for a hazard type"""
global last_hazard_detection, critical_alert_count
last_hazard_detection[hazard_type] = datetime.now()
# Increment critical alert count if severity is critical
if hazard_type in HAZARD_CATEGORIES and HAZARD_CATEGORIES[hazard_type].get("severity") == "critical":
critical_alert_count += 1
# Check if we've reached the threshold
if critical_alert_count >= 5:
trigger_webhook()
critical_alert_count = 0 # Reset counter after triggering
def calculate_box_overlap(box1, box2):
"""
Calculate how much box1 overlaps with box2.
Returns the percentage of box1 that is inside box2.
"""
# Get coordinates
x1_min, y1_min = box1["box"][0], box1["box"][1]
x1_max, y1_max = box1["box"][2], box1["box"][3]
x2_min, y2_min = box2["box"][0], box2["box"][1]
x2_max, y2_max = box2["box"][2], box2["box"][3]
# Calculate intersection
x_left = max(x1_min, x2_min)
y_top = max(y1_min, y2_min)
x_right = min(x1_max, x2_max)
y_bottom = min(y1_max, y2_max)
if x_right < x_left or y_bottom < y_top:
return 0.0
# Calculate areas
intersection_area = (x_right - x_left) * (y_bottom - y_top)
box1_area = (x1_max - x1_min) * (y1_max - y1_min)
# Return percentage of box1 that overlaps with box2
return intersection_area / box1_area if box1_area > 0 else 0.0
def is_object_held_safely(person_or_hand_box, object_box, overlap_threshold=0.3):
"""
Determine if an object is being held safely by checking if it overlaps significantly
with a person or hand bounding box
"""
# If it's a hand detection, use the original distance check
if person_or_hand_box["label"] == "hand":
x1_hand = (person_or_hand_box["box"][0] + person_or_hand_box["box"][2]) / 2
y1_hand = (person_or_hand_box["box"][1] + person_or_hand_box["box"][3]) / 2
x2_obj = (object_box["box"][0] + object_box["box"][2]) / 2
y2_obj = (object_box["box"][1] + object_box["box"][3]) / 2
# Calculate distance between centers
distance = np.sqrt((x1_hand - x2_obj)**2 + (y1
[truncated — 13980 more characters]
```
### frontend/public/index.html
```html
HazardVision - Workplace Safety Monitoring