# safeguard_pro.py - AI-POWERED WORKPLACE SAFETY MONITORING SYSTEM import streamlit as st import numpy as np from PIL import Image import time from datetime import datetime, timedelta import pandas as pd import plotly.graph_objects as go import plotly.express as px import matplotlib.pyplot as plt import warnings import sqlite3 import json import hashlib from typing import Dict, List, Optional, Any import uuid import random warnings.filterwarnings('ignore') # ========== PASSWORD UTILITIES ========== def hash_password(password: str) -> str: """Hash a password using SHA-256""" return hashlib.sha256(password.encode()).hexdigest() def verify_password(password: str, hashed: str) -> bool: """Verify a password against its hash""" return hash_password(password) == hashed # ========== DATABASE SETUP ========== class SafetyDatabase: def __init__(self, db_name='safeguard.db'): self.db_name = db_name self.init_database() def get_connection(self): return sqlite3.connect(self.db_name, check_same_thread=False) def init_database(self): conn = self.get_connection() cursor = conn.cursor() # Users table (safety managers, admins) cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( user_id TEXT PRIMARY KEY, username TEXT UNIQUE NOT NULL, password_hash TEXT NOT NULL, full_name TEXT NOT NULL, email TEXT, role TEXT NOT NULL, department TEXT, created_at TEXT, last_login TEXT, is_active INTEGER DEFAULT 1 ) ''') # Employees table cursor.execute(''' CREATE TABLE IF NOT EXISTS employees ( employee_id TEXT PRIMARY KEY, name TEXT NOT NULL, age INTEGER, gender TEXT, department TEXT, shift TEXT, designation TEXT, contact TEXT, emergency_contact TEXT, hire_date TEXT, medical_conditions TEXT, certification_level TEXT, created_at TEXT, updated_at TEXT, created_by TEXT, FOREIGN KEY (created_by) REFERENCES users (user_id) ) ''') # PPE Compliance records cursor.execute(''' CREATE TABLE IF NOT EXISTS ppe_records ( record_id TEXT PRIMARY KEY, employee_id TEXT, timestamp TEXT, hard_hat INTEGER DEFAULT 0, safety_vest INTEGER DEFAULT 0, safety_glasses INTEGER DEFAULT 0, gloves INTEGER DEFAULT 0, steel_toe_boots INTEGER DEFAULT 0, ear_protection INTEGER DEFAULT 0, respirator INTEGER DEFAULT 0, harness INTEGER DEFAULT 0, compliance_score INTEGER, location TEXT, image_path TEXT, alert_triggered INTEGER DEFAULT 0, notes TEXT, recorded_by TEXT, FOREIGN KEY (employee_id) REFERENCES employees (employee_id), FOREIGN KEY (recorded_by) REFERENCES users (user_id) ) ''') # Hazard Detection records cursor.execute(''' CREATE TABLE IF NOT EXISTS hazard_records ( hazard_id TEXT PRIMARY KEY, detected_at TEXT, hazard_type TEXT, severity TEXT, location TEXT, description TEXT, status TEXT DEFAULT 'active', resolved_at TEXT, resolved_by TEXT, image_path TEXT, notes TEXT, FOREIGN KEY (resolved_by) REFERENCES users (user_id) ) ''') # Incident Reports cursor.execute(''' CREATE TABLE IF NOT EXISTS incidents ( incident_id TEXT PRIMARY KEY, timestamp TEXT, incident_type TEXT, severity TEXT, location TEXT, description TEXT, employees_involved TEXT, injuries TEXT, root_cause TEXT, actions_taken TEXT, reported_by TEXT, status TEXT DEFAULT 'investigating', closed_at TEXT, FOREIGN KEY (reported_by) REFERENCES users (user_id) ) ''') # Environmental Monitoring cursor.execute(''' CREATE TABLE IF NOT EXISTS environmental_data ( reading_id TEXT PRIMARY KEY, timestamp TEXT, location TEXT, temperature REAL, humidity REAL, noise_level REAL, air_quality REAL, co2_level REAL, voc_level REAL, illuminance REAL, notes TEXT ) ''') # Safety Inspections cursor.execute(''' CREATE TABLE IF NOT EXISTS inspections ( inspection_id TEXT PRIMARY KEY, date TEXT, inspector_id TEXT, area TEXT, findings TEXT, violations TEXT, corrective_actions TEXT, due_date TEXT, status TEXT DEFAULT 'pending', completed_at TEXT, FOREIGN KEY (inspector_id) REFERENCES users (user_id) ) ''') # Equipment Safety Checks cursor.execute(''' CREATE TABLE IF NOT EXISTS equipment_checks ( check_id TEXT PRIMARY KEY, equipment_id TEXT, equipment_name TEXT, check_date TEXT, next_check_date TEXT, status TEXT, issues_found TEXT, maintenance_required INTEGER DEFAULT 0, checked_by TEXT, notes TEXT, FOREIGN KEY (checked_by) REFERENCES users (user_id) ) ''') # Activity log table cursor.execute(''' CREATE TABLE IF NOT EXISTS activity_log ( log_id TEXT PRIMARY KEY, user_id TEXT, action TEXT, details TEXT, timestamp TEXT, FOREIGN KEY (user_id) REFERENCES users (user_id) ) ''') conn.commit() # Create default admin and safety manager users self.create_default_users() conn.close() def create_default_users(self): """Create default users if none exist""" conn = self.get_connection() cursor = conn.cursor() cursor.execute("SELECT COUNT(*) FROM users") count = cursor.fetchone()[0] if count == 0: # Create default admin admin_id = f"USR-{uuid.uuid4().hex[:6].upper()}" cursor.execute(''' INSERT INTO users (user_id, username, password_hash, full_name, email, role, department, created_at, is_active) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( admin_id, "admin", hash_password("admin123"), "System Administrator", "admin@safeguard.com", "admin", "IT", datetime.now().isoformat(), 1 )) # Create safety manager manager_id = f"USR-{uuid.uuid4().hex[:6].upper()}" cursor.execute(''' INSERT INTO users (user_id, username, password_hash, full_name, email, role, department, created_at, is_active) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( manager_id, "safety", hash_password("safety123"), "Safety Manager", "safety@safeguard.com", "safety_manager", "Safety", datetime.now().isoformat(), 1 )) # Create supervisor supervisor_id = f"USR-{uuid.uuid4().hex[:6].upper()}" cursor.execute(''' INSERT INTO users (user_id, username, password_hash, full_name, email, role, department, created_at, is_active) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( supervisor_id, "supervisor", hash_password("super123"), "Floor Supervisor", "supervisor@safeguard.com", "supervisor", "Production", datetime.now().isoformat(), 1 )) conn.commit() conn.close() def authenticate_user(self, username: str, password: str) -> Optional[Dict]: """Authenticate user with username and password""" conn = self.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT user_id, username, password_hash, full_name, email, role, department, is_active FROM users WHERE username = ? AND is_active = 1 ''', (username,)) row = cursor.fetchone() conn.close() if row and verify_password(password, row[2]): self.update_last_login(row[0]) self.log_activity(row[0], "login", f"User {username} logged in") return { 'user_id': row[0], 'username': row[1], 'full_name': row[3], 'email': row[4], 'role': row[5], 'department': row[6] } return None def update_last_login(self, user_id: str): conn = self.get_connection() cursor = conn.cursor() cursor.execute('UPDATE users SET last_login = ? WHERE user_id = ?', (datetime.now().isoformat(), user_id)) conn.commit() conn.close() def log_activity(self, user_id: str, action: str, details: str = ""): conn = self.get_connection() cursor = conn.cursor() log_id = f"LOG-{uuid.uuid4().hex[:8].upper()}" cursor.execute(''' INSERT INTO activity_log (log_id, user_id, action, details, timestamp) VALUES (?, ?, ?, ?, ?) ''', (log_id, user_id, action, details, datetime.now().isoformat())) conn.commit() conn.close() def get_user_permissions(self, role: str) -> Dict: permissions = { 'admin': { 'can_manage_users': True, 'can_add_employees': True, 'can_edit_employees': True, 'can_delete_employees': True, 'can_view_all_data': True, 'can_run_ppe_detection': True, 'can_manage_hazards': True, 'can_report_incidents': True, 'can_view_analytics': True, 'can_export_data': True, 'can_configure_system': True, 'can_acknowledge_alerts': True, 'can_schedule_inspections': True, 'can_approve_corrective_actions': True }, 'safety_manager': { 'can_manage_users': False, 'can_add_employees': True, 'can_edit_employees': True, 'can_delete_employees': False, 'can_view_all_data': True, 'can_run_ppe_detection': True, 'can_manage_hazards': True, 'can_report_incidents': True, 'can_view_analytics': True, 'can_export_data': True, 'can_configure_system': False, 'can_acknowledge_alerts': True, 'can_schedule_inspections': True, 'can_approve_corrective_actions': True }, 'supervisor': { 'can_manage_users': False, 'can_add_employees': False, 'can_edit_employees': True, 'can_delete_employees': False, 'can_view_all_data': True, 'can_run_ppe_detection': True, 'can_manage_hazards': True, 'can_report_incidents': True, 'can_view_analytics': True, 'can_export_data': False, 'can_configure_system': False, 'can_acknowledge_alerts': True, 'can_schedule_inspections': False, 'can_approve_corrective_actions': False }, 'inspector': { 'can_manage_users': False, 'can_add_employees': False, 'can_edit_employees': False, 'can_delete_employees': False, 'can_view_all_data': True, 'can_run_ppe_detection': True, 'can_manage_hazards': True, 'can_report_incidents': True, 'can_view_analytics': True, 'can_export_data': False, 'can_configure_system': False, 'can_acknowledge_alerts': True, 'can_schedule_inspections': True, 'can_approve_corrective_actions': False }, 'viewer': { 'can_manage_users': False, 'can_add_employees': False, 'can_edit_employees': False, 'can_delete_employees': False, 'can_view_all_data': True, 'can_run_ppe_detection': False, 'can_manage_hazards': False, 'can_report_incidents': False, 'can_view_analytics': True, 'can_export_data': False, 'can_configure_system': False, 'can_acknowledge_alerts': False, 'can_schedule_inspections': False, 'can_approve_corrective_actions': False } } return permissions.get(role, permissions['viewer']) def get_all_users(self) -> List[Dict]: conn = self.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT user_id, username, full_name, email, role, department, created_at, last_login, is_active FROM users ORDER BY full_name ''') users = [] columns = [column[0] for column in cursor.description] for row in cursor.fetchall(): users.append(dict(zip(columns, row))) conn.close() return users def add_user(self, user_data: Dict) -> str: conn = self.get_connection() cursor = conn.cursor() user_id = f"USR-{uuid.uuid4().hex[:6].upper()}" cursor.execute(''' INSERT INTO users (user_id, username, password_hash, full_name, email, role, department, created_at, is_active) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( user_id, user_data['username'], hash_password(user_data['password']), user_data['full_name'], user_data.get('email', ''), user_data['role'], user_data.get('department', ''), datetime.now().isoformat(), 1 )) conn.commit() conn.close() return user_id def update_user(self, user_id: str, user_data: Dict): conn = self.get_connection() cursor = conn.cursor() if 'password' in user_data and user_data['password']: cursor.execute(''' UPDATE users SET full_name=?, email=?, role=?, department=?, password_hash=? WHERE user_id=? ''', ( user_data['full_name'], user_data.get('email', ''), user_data['role'], user_data.get('department', ''), hash_password(user_data['password']), user_id )) else: cursor.execute(''' UPDATE users SET full_name=?, email=?, role=?, department=? WHERE user_id=? ''', ( user_data['full_name'], user_data.get('email', ''), user_data['role'], user_data.get('department', ''), user_id )) conn.commit() conn.close() def delete_user(self, user_id: str): conn = self.get_connection() cursor = conn.cursor() cursor.execute('UPDATE users SET is_active = 0 WHERE user_id = ?', (user_id,)) conn.commit() conn.close() def get_activity_log(self, limit: int = 100) -> List[Dict]: conn = self.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT a.log_id, a.user_id, u.username, u.full_name, a.action, a.details, a.timestamp FROM activity_log a LEFT JOIN users u ON a.user_id = u.user_id ORDER BY a.timestamp DESC LIMIT ? ''', (limit,)) logs = [] columns = [column[0] for column in cursor.description] for row in cursor.fetchall(): logs.append(dict(zip(columns, row))) conn.close() return logs # ========== EMPLOYEE MANAGEMENT METHODS ========== def add_employee(self, employee_data: Dict) -> str: conn = self.get_connection() cursor = conn.cursor() employee_id = f"EMP-{uuid.uuid4().hex[:6].upper()}" current_time = datetime.now().isoformat() cursor.execute(''' INSERT INTO employees (employee_id, name, age, gender, department, shift, designation, contact, emergency_contact, hire_date, medical_conditions, certification_level, created_at, updated_at, created_by) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( employee_id, employee_data.get('name'), employee_data.get('age'), employee_data.get('gender'), employee_data.get('department'), employee_data.get('shift'), employee_data.get('designation'), employee_data.get('contact'), employee_data.get('emergency_contact'), employee_data.get('hire_date'), employee_data.get('medical_conditions', 'None'), employee_data.get('certification_level', 'Level 1'), current_time, current_time, employee_data.get('created_by') )) conn.commit() conn.close() return employee_id def update_employee(self, employee_id: str, employee_data: Dict): conn = self.get_connection() cursor = conn.cursor() cursor.execute(''' UPDATE employees SET name=?, age=?, gender=?, department=?, shift=?, designation=?, contact=?, emergency_contact=?, medical_conditions=?, certification_level=?, updated_at=? WHERE employee_id=? ''', ( employee_data.get('name'), employee_data.get('age'), employee_data.get('gender'), employee_data.get('department'), employee_data.get('shift'), employee_data.get('designation'), employee_data.get('contact'), employee_data.get('emergency_contact'), employee_data.get('medical_conditions'), employee_data.get('certification_level'), datetime.now().isoformat(), employee_id )) conn.commit() conn.close() def delete_employee(self, employee_id: str): conn = self.get_connection() cursor = conn.cursor() cursor.execute('DELETE FROM ppe_records WHERE employee_id=?', (employee_id,)) cursor.execute('DELETE FROM employees WHERE employee_id=?', (employee_id,)) conn.commit() conn.close() def get_all_employees(self) -> List[Dict]: conn = self.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT employee_id, name, age, gender, department, shift, designation, contact, emergency_contact, hire_date, medical_conditions, certification_level, created_at, updated_at FROM employees ORDER BY name ''') employees = [] columns = [column[0] for column in cursor.description] for row in cursor.fetchall(): employees.append(dict(zip(columns, row))) conn.close() return employees def get_employee(self, employee_id: str) -> Optional[Dict]: conn = self.get_connection() cursor = conn.cursor() cursor.execute('SELECT * FROM employees WHERE employee_id=?', (employee_id,)) row = cursor.fetchone() if not row: conn.close() return None columns = [column[0] for column in cursor.description] employee = dict(zip(columns, row)) conn.close() return employee # ========== PPE RECORDS METHODS ========== def add_ppe_record(self, employee_id: str, ppe_data: Dict) -> str: conn = self.get_connection() cursor = conn.cursor() record_id = f"PPE-{uuid.uuid4().hex[:8].upper()}" # Calculate compliance score ppe_items = [ ppe_data.get('hard_hat', 0), ppe_data.get('safety_vest', 0), ppe_data.get('safety_glasses', 0), ppe_data.get('gloves', 0), ppe_data.get('steel_toe_boots', 0), ppe_data.get('ear_protection', 0), ppe_data.get('respirator', 0), ppe_data.get('harness', 0) ] compliance_score = int((sum(ppe_items) / len(ppe_items)) * 100) cursor.execute(''' INSERT INTO ppe_records (record_id, employee_id, timestamp, hard_hat, safety_vest, safety_glasses, gloves, steel_toe_boots, ear_protection, respirator, harness, compliance_score, location, alert_triggered, notes, recorded_by) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( record_id, employee_id, datetime.now().isoformat(), ppe_data.get('hard_hat', 0), ppe_data.get('safety_vest', 0), ppe_data.get('safety_glasses', 0), ppe_data.get('gloves', 0), ppe_data.get('steel_toe_boots', 0), ppe_data.get('ear_protection', 0), ppe_data.get('respirator', 0), ppe_data.get('harness', 0), compliance_score, ppe_data.get('location', 'Unknown'), 1 if compliance_score < 80 else 0, ppe_data.get('notes', ''), ppe_data.get('recorded_by') )) conn.commit() conn.close() return record_id def get_employee_ppe_history(self, employee_id: str, limit: int = 20) -> List[Dict]: conn = self.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT * FROM ppe_records WHERE employee_id=? ORDER BY timestamp DESC LIMIT ? ''', (employee_id, limit)) records = [] columns = [column[0] for column in cursor.description] for row in cursor.fetchall(): records.append(dict(zip(columns, row))) conn.close() return records def get_recent_ppe_violations(self, hours: int = 24) -> List[Dict]: conn = self.get_connection() cursor = conn.cursor() cutoff = (datetime.now() - timedelta(hours=hours)).isoformat() cursor.execute(''' SELECT p.*, e.name as employee_name, e.department FROM ppe_records p JOIN employees e ON p.employee_id = e.employee_id WHERE p.timestamp > ? AND p.alert_triggered = 1 ORDER BY p.timestamp DESC ''', (cutoff,)) violations = [] columns = [column[0] for column in cursor.description] for row in cursor.fetchall(): violations.append(dict(zip(columns, row))) conn.close() return violations # ========== HAZARD MANAGEMENT METHODS ========== def add_hazard(self, hazard_data: Dict) -> str: conn = self.get_connection() cursor = conn.cursor() hazard_id = f"HAZ-{uuid.uuid4().hex[:8].upper()}" cursor.execute(''' INSERT INTO hazard_records (hazard_id, detected_at, hazard_type, severity, location, description, status, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ''', ( hazard_id, datetime.now().isoformat(), hazard_data.get('hazard_type'), hazard_data.get('severity'), hazard_data.get('location'), hazard_data.get('description'), 'active', hazard_data.get('notes', '') )) conn.commit() conn.close() return hazard_id def resolve_hazard(self, hazard_id: str, resolved_by: str, notes: str = ""): conn = self.get_connection() cursor = conn.cursor() cursor.execute(''' UPDATE hazard_records SET status='resolved', resolved_at=?, resolved_by=?, notes=? WHERE hazard_id=? ''', ( datetime.now().isoformat(), resolved_by, notes, hazard_id )) conn.commit() conn.close() def get_active_hazards(self) -> List[Dict]: conn = self.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT * FROM hazard_records WHERE status='active' ORDER BY CASE severity WHEN 'Critical' THEN 1 WHEN 'High' THEN 2 WHEN 'Medium' THEN 3 WHEN 'Low' THEN 4 END, detected_at DESC ''') hazards = [] columns = [column[0] for column in cursor.description] for row in cursor.fetchall(): hazards.append(dict(zip(columns, row))) conn.close() return hazards # ========== INCIDENT MANAGEMENT METHODS ========== def add_incident(self, incident_data: Dict) -> str: conn = self.get_connection() cursor = conn.cursor() incident_id = f"INC-{uuid.uuid4().hex[:8].upper()}" cursor.execute(''' INSERT INTO incidents (incident_id, timestamp, incident_type, severity, location, description, employees_involved, injuries, root_cause, actions_taken, reported_by, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( incident_id, datetime.now().isoformat(), incident_data.get('incident_type'), incident_data.get('severity'), incident_data.get('location'), incident_data.get('description'), incident_data.get('employees_involved', ''), incident_data.get('injuries', ''), incident_data.get('root_cause', ''), incident_data.get('actions_taken', ''), incident_data.get('reported_by'), 'investigating' )) conn.commit() conn.close() return incident_id def get_recent_incidents(self, days: int = 30) -> List[Dict]: conn = self.get_connection() cursor = conn.cursor() cutoff = (datetime.now() - timedelta(days=days)).isoformat() cursor.execute(''' SELECT * FROM incidents WHERE timestamp > ? ORDER BY timestamp DESC ''', (cutoff,)) incidents = [] columns = [column[0] for column in cursor.description] for row in cursor.fetchall(): incidents.append(dict(zip(columns, row))) conn.close() return incidents # ========== ENVIRONMENTAL DATA METHODS ========== def add_environmental_reading(self, env_data: Dict) -> str: conn = self.get_connection() cursor = conn.cursor() reading_id = f"ENV-{uuid.uuid4().hex[:8].upper()}" cursor.execute(''' INSERT INTO environmental_data (reading_id, timestamp, location, temperature, humidity, noise_level, air_quality, co2_level, voc_level, illuminance, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( reading_id, datetime.now().isoformat(), env_data.get('location'), env_data.get('temperature'), env_data.get('humidity'), env_data.get('noise_level'), env_data.get('air_quality'), env_data.get('co2_level'), env_data.get('voc_level'), env_data.get('illuminance'), env_data.get('notes', '') )) conn.commit() conn.close() return reading_id def get_environmental_history(self, location: str = None, hours: int = 24) -> List[Dict]: conn = self.get_connection() cursor = conn.cursor() cutoff = (datetime.now() - timedelta(hours=hours)).isoformat() if location: cursor.execute(''' SELECT * FROM environmental_data WHERE timestamp > ? AND location = ? ORDER BY timestamp DESC ''', (cutoff, location)) else: cursor.execute(''' SELECT * FROM environmental_data WHERE timestamp > ? ORDER BY timestamp DESC ''', (cutoff,)) records = [] columns = [column[0] for column in cursor.description] for row in cursor.fetchall(): records.append(dict(zip(columns, row))) conn.close() return records # ========== INSPECTION METHODS ========== def add_inspection(self, inspection_data: Dict) -> str: conn = self.get_connection() cursor = conn.cursor() inspection_id = f"INS-{uuid.uuid4().hex[:8].upper()}" cursor.execute(''' INSERT INTO inspections (inspection_id, date, inspector_id, area, findings, violations, corrective_actions, due_date, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( inspection_id, inspection_data.get('date'), inspection_data.get('inspector_id'), inspection_data.get('area'), inspection_data.get('findings', ''), inspection_data.get('violations', ''), inspection_data.get('corrective_actions', ''), inspection_data.get('due_date'), 'pending' )) conn.commit() conn.close() return inspection_id def complete_inspection(self, inspection_id: str): conn = self.get_connection() cursor = conn.cursor() cursor.execute(''' UPDATE inspections SET status='completed', completed_at=? WHERE inspection_id=? ''', ( datetime.now().isoformat(), inspection_id )) conn.commit() conn.close() def get_pending_inspections(self) -> List[Dict]: conn = self.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT i.*, u.full_name as inspector_name FROM inspections i LEFT JOIN users u ON i.inspector_id = u.user_id WHERE i.status='pending' ORDER BY i.due_date ASC ''') inspections = [] columns = [column[0] for column in cursor.description] for row in cursor.fetchall(): inspections.append(dict(zip(columns, row))) conn.close() return inspections # ========== EQUIPMENT CHECKS METHODS ========== def add_equipment_check(self, check_data: Dict) -> str: conn = self.get_connection() cursor = conn.cursor() check_id = f"CHK-{uuid.uuid4().hex[:8].upper()}" next_check = (datetime.now() + timedelta(days=check_data.get('check_interval_days', 30))).strftime('%Y-%m-%d') cursor.execute(''' INSERT INTO equipment_checks (check_id, equipment_id, equipment_name, check_date, next_check_date, status, issues_found, maintenance_required, checked_by, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( check_id, check_data.get('equipment_id'), check_data.get('equipment_name'), datetime.now().strftime('%Y-%m-%d'), next_check, check_data.get('status'), check_data.get('issues_found', ''), check_data.get('maintenance_required', 0), check_data.get('checked_by'), check_data.get('notes', '') )) conn.commit() conn.close() return check_id def get_equipment_due_for_check(self, days: int = 7) -> List[Dict]: conn = self.get_connection() cursor = conn.cursor() cutoff = (datetime.now() + timedelta(days=days)).strftime('%Y-%m-%d') cursor.execute(''' SELECT * FROM equipment_checks WHERE next_check_date <= ? ORDER BY next_check_date ASC ''', (cutoff,)) equipment = [] columns = [column[0] for column in cursor.description] for row in cursor.fetchall(): equipment.append(dict(zip(columns, row))) conn.close() return equipment # Initialize database db = SafetyDatabase() # ========== AI DETECTION CLASSES ========== class PPEDetector: """Simulate PPE detection from images""" def detect_ppe(self, image=None): """Detect PPE items in image""" np.random.seed(int(time.time())) # Simulate detection results return { 'hard_hat': np.random.choice([0, 1], p=[0.1, 0.9]), 'safety_vest': np.random.choice([0, 1], p=[0.15, 0.85]), 'safety_glasses': np.random.choice([0, 1], p=[0.2, 0.8]), 'gloves': np.random.choice([0, 1], p=[0.25, 0.75]), 'steel_toe_boots': np.random.choice([0, 1], p=[0.1, 0.9]), 'ear_protection': np.random.choice([0, 1], p=[0.3, 0.7]), 'respirator': np.random.choice([0, 1], p=[0.4, 0.6]), 'harness': np.random.choice([0, 1], p=[0.5, 0.5]), 'detection_confidence': np.random.uniform(0.75, 0.98), 'persons_detected': np.random.randint(1, 4) } def calculate_safety_score(self, detections): """Calculate overall safety score""" ppe_items = [ detections.get('hard_hat', 0), detections.get('safety_vest', 0), detections.get('safety_glasses', 0), detections.get('gloves', 0), detections.get('steel_toe_boots', 0), detections.get('ear_protection', 0), detections.get('respirator', 0), detections.get('harness', 0) ] # Weight different PPE items weights = [0.15, 0.15, 0.1, 0.1, 0.1, 0.1, 0.1, 0.2] weighted_score = sum(p * w for p, w in zip(ppe_items, weights)) * 100 return { 'compliance_score': int(weighted_score), 'missing_items': [item for item, present in zip( ['Hard Hat', 'Safety Vest', 'Safety Glasses', 'Gloves', 'Steel Toe Boots', 'Ear Protection', 'Respirator', 'Harness'], ppe_items) if present == 0], 'alert_level': 'HIGH' if weighted_score < 70 else 'MEDIUM' if weighted_score < 85 else 'LOW' } class HazardDetector: """Simulate hazard detection""" def detect_hazards(self, image=None): """Detect workplace hazards""" np.random.seed(int(time.time() * 1000)) hazard_types = [ 'Spill/Chemical Leak', 'Blocked Emergency Exit', 'Unsafe Stacking', 'Electrical Hazard', 'Trip Hazard', 'Fire Hazard', 'Machine Guard Missing', 'Poor Housekeeping', 'Confined Space Risk', 'Working at Height Unsafe' ] severities = ['Low', 'Medium', 'High', 'Critical'] probabilities = [0.4, 0.3, 0.2, 0.1] # Randomly decide if hazard is detected (30% chance) if np.random.random() < 0.3: hazard_index = np.random.randint(0, len(hazard_types)) severity = np.random.choice(severities, p=probabilities) return { 'hazard_detected': True, 'hazard_type': hazard_types[hazard_index], 'severity': severity, 'confidence': np.random.uniform(0.6, 0.95), 'location': f"Zone {np.random.choice(['A', 'B', 'C', 'D'])}-{np.random.randint(1, 10)}", 'recommendation': f"Immediate action required: {hazard_types[hazard_index]} detected" } else: return { 'hazard_detected': False, 'message': 'No hazards detected', 'confidence': np.random.uniform(0.85, 0.99) } class EnvironmentalMonitor: """Simulate environmental monitoring""" def get_readings(self, location="Zone A"): """Get simulated environmental readings""" np.random.seed(int(time.time())) return { 'location': location, 'temperature': 22 + np.random.uniform(-3, 5), 'humidity': 45 + np.random.uniform(-10, 15), 'noise_level': 70 + np.random.uniform(-10, 20), 'air_quality': 85 + np.random.uniform(-15, 10), # AQI scale (0-100, higher is better) 'co2_level': 400 + np.random.uniform(-50, 150), 'voc_level': np.random.uniform(0, 5), 'illuminance': 300 + np.random.uniform(-100, 200), 'timestamp': datetime.now().isoformat() } def check_alerts(self, readings): """Check if any readings are outside safe limits""" alerts = [] # Define safe ranges safe_ranges = { 'temperature': (18, 28), 'humidity': (30, 70), 'noise_level': (0, 85), 'air_quality': (50, 100), 'co2_level': (0, 1000), 'voc_level': (0, 3), 'illuminance': (200, 1000) } for param, (min_val, max_val) in safe_ranges.items(): value = readings.get(param) if value: if value < min_val: alerts.append({ 'parameter': param, 'value': value, 'threshold': f'below {min_val}', 'severity': 'Low' if (min_val - value) < 5 else 'Medium' }) elif value > max_val: alerts.append({ 'parameter': param, 'value': value, 'threshold': f'above {max_val}', 'severity': 'Low' if (value - max_val) < 5 else 'Medium' }) return alerts # Initialize detectors ppe_detector = PPEDetector() hazard_detector = HazardDetector() env_monitor = EnvironmentalMonitor() # ========== HELPER FUNCTIONS ========== def display_ppe_status(detections, safety_score): """Display PPE detection results""" # Create PPE status grid col1, col2, col3, col4 = st.columns(4) ppe_items = [ ('Hard Hat', detections.get('hard_hat', 0), '๐Ÿช–'), ('Safety Vest', detections.get('safety_vest', 0), '๐Ÿฆบ'), ('Safety Glasses', detections.get('safety_glasses', 0), '๐Ÿ‘“'), ('Gloves', detections.get('gloves', 0), '๐Ÿงค'), ('Steel Toe Boots', detections.get('steel_toe_boots', 0), '๐Ÿ‘ข'), ('Ear Protection', detections.get('ear_protection', 0), '๐ŸŽง'), ('Respirator', detections.get('respirator', 0), '๐Ÿ˜ท'), ('Harness', detections.get('harness', 0), 'โ›“๏ธ') ] for i, (item, present, icon) in enumerate(ppe_items): col = [col1, col2, col3, col4][i % 4] with col: status = "โœ…" if present else "โŒ" color = "green" if present else "red" st.markdown(f"""
{icon} {status} {item}
""", unsafe_allow_html=True) # Safety score gauge st.subheader("Safety Compliance Score") fig = go.Figure(go.Indicator( mode="gauge+number", value=safety_score['compliance_score'], title={'text': "PPE Compliance"}, domain={'x': [0, 1], 'y': [0, 1]}, gauge={ 'axis': {'range': [0, 100]}, 'bar': {'color': "#28a745"}, 'steps': [ {'range': [0, 70], 'color': "#dc3545"}, {'range': [70, 85], 'color': "#ffc107"}, {'range': [85, 100], 'color': "#28a745"} ], 'threshold': { 'line': {'color': "black", 'width': 4}, 'thickness': 0.75, 'value': safety_score['compliance_score'] } } )) fig.update_layout(height=250) st.plotly_chart(fig, use_container_width=True) # Missing items alert if safety_score['missing_items']: st.warning(f"โš ๏ธ Missing PPE: {', '.join(safety_score['missing_items'])}") if safety_score['alert_level'] == 'HIGH': st.error("๐Ÿšจ HIGH ALERT: Critical PPE missing!") else: st.success("โœ… All required PPE detected!") def check_permission(permission: str): return st.session_state.user_permissions.get(permission, False) def show_login_page(): st.markdown(""" """, unsafe_allow_html=True) col1, col2, col3 = st.columns([1, 2, 1]) with col2: st.markdown("""

๐Ÿ›ก๏ธ SafeGuard Pro

AI-Powered Workplace Safety Monitoring System

""", unsafe_allow_html=True) with st.form("login_form"): st.subheader("Safety Officer Login") username = st.text_input("Username", placeholder="Enter your username") password = st.text_input("Password", type="password", placeholder="Enter your password") submitted = st.form_submit_button("Login", type="primary", use_container_width=True) if submitted: if username and password: user = db.authenticate_user(username, password) if user: st.session_state.authenticated = True st.session_state.current_user = user st.session_state.user_permissions = db.get_user_permissions(user['role']) st.success(f"Welcome, Safety Officer {user['full_name']}!") st.rerun() else: st.error("Invalid username or password") else: st.warning("Please enter both username and password") st.markdown("""

Default credentials:

Admin: admin / admin123

Safety Manager: safety / safety123

Supervisor: supervisor / super123

""", unsafe_allow_html=True) def logout(): if st.session_state.current_user: db.log_activity(st.session_state.current_user['user_id'], "logout", f"User {st.session_state.current_user['username']} logged out") st.session_state.authenticated = False st.session_state.current_user = None st.session_state.user_permissions = {} st.session_state.selected_employee_id = None st.rerun() def show_add_employee_form(): with st.form("add_employee_form", clear_on_submit=True): st.subheader("โž• Add New Employee") col1, col2 = st.columns(2) with col1: name = st.text_input("Full Name*", placeholder="John Doe") age = st.number_input("Age", min_value=18, max_value=75, value=30) gender = st.selectbox("Gender", ["Male", "Female", "Other"]) department = st.selectbox( "Department", ["Production", "Assembly", "Warehouse", "Maintenance", "Quality Control", "Logistics", "Administration"] ) designation = st.text_input("Designation", placeholder="Machine Operator") with col2: shift = st.selectbox( "Shift", ["Morning (6AM-2PM)", "Afternoon (2PM-10PM)", "Night (10PM-6AM)", "Rotating"] ) contact = st.text_input("Contact Number", placeholder="+1234567890") emergency_contact = st.text_input("Emergency Contact", placeholder="+1234567890") hire_date = st.date_input("Hire Date", value=datetime.now()) certification_level = st.selectbox( "Safety Certification", ["Level 1 (Basic)", "Level 2 (Intermediate)", "Level 3 (Advanced)", "Level 4 (Expert)"] ) medical_conditions = st.text_area("Medical Conditions / Allergies", placeholder="Enter any relevant medical information...") submitted = st.form_submit_button("Add Employee", type="primary") if submitted: if not name: st.error("Name is required!") return employee_data = { 'name': name, 'age': age, 'gender': gender, 'department': department, 'shift': shift, 'designation': designation, 'contact': contact, 'emergency_contact': emergency_contact, 'hire_date': hire_date.strftime('%Y-%m-%d'), 'medical_conditions': medical_conditions, 'certification_level': certification_level, 'created_by': st.session_state.current_user['user_id'] if st.session_state.current_user else None } employee_id = db.add_employee(employee_data) if st.session_state.current_user: db.log_activity(st.session_state.current_user['user_id'], "add_employee", f"Added employee: {name} (ID: {employee_id})") st.success(f"โœ… Employee added successfully! Employee ID: {employee_id}") st.session_state.employees = db.get_all_employees() st.rerun() # ========== SESSION STATE INITIALIZATION ========== if 'authenticated' not in st.session_state: st.session_state.authenticated = False if 'current_user' not in st.session_state: st.session_state.current_user = None if 'user_permissions' not in st.session_state: st.session_state.user_permissions = {} if 'selected_employee_id' not in st.session_state: st.session_state.selected_employee_id = None if 'employees' not in st.session_state: try: st.session_state.employees = db.get_all_employees() except Exception as e: st.session_state.employees = [] if 'active_alerts' not in st.session_state: st.session_state.active_alerts = [] if 'emergency' not in st.session_state: st.session_state.emergency = False # ========== PAGE CONFIG & CSS ========== st.set_page_config( page_title="SafeGuard Pro", page_icon="๐Ÿ›ก๏ธ", layout="wide", initial_sidebar_state="expanded" ) st.markdown(""" """, unsafe_allow_html=True) # ========== CHECK AUTHENTICATION ========== if not st.session_state.authenticated: show_login_page() st.stop() # ========== SIDEBAR ========== with st.sidebar: st.markdown(f"""

Welcome, {st.session_state.current_user['full_name']}!

{st.session_state.current_user['role'].replace('_', ' ').upper()}

{st.session_state.current_user.get('department', '')}

""", unsafe_allow_html=True) st.divider() # Active Alerts Counter recent_violations = db.get_recent_ppe_violations(hours=24) active_hazards = db.get_active_hazards() total_alerts = len(recent_violations) + len(active_hazards) if total_alerts > 0: st.markdown(f"""
๐Ÿšจ {total_alerts} ACTIVE ALERTS
""", unsafe_allow_html=True) st.divider() # Employee Selection employees = st.session_state.employees if employees: default_option = "Select employee..." employee_options = [default_option] + [f"{e['name']} ({e['employee_id']})" for e in employees] selected_employee_display = st.selectbox( "๐Ÿ‘ค Select Employee", employee_options, index=0 ) if selected_employee_display != default_option: selected_name = selected_employee_display.split(" (")[0] selected_employee = next((e for e in employees if e['name'] == selected_name), None) if selected_employee: st.session_state.selected_employee_id = selected_employee['employee_id'] with st.expander("๐Ÿ‘ค Employee Info", expanded=True): st.write(f"**Name:** {selected_employee['name']}") st.write(f"**ID:** {selected_employee['employee_id']}") st.write(f"**Dept:** {selected_employee['department']}") st.write(f"**Designation:** {selected_employee['designation']}") st.write(f"**Shift:** {selected_employee['shift']}") st.write(f"**Certification:** {selected_employee['certification_level']}") # Recent compliance ppe_history = db.get_employee_ppe_history(selected_employee['employee_id'], limit=1) if ppe_history: latest = ppe_history[0] st.write(f"**Last Check:** {datetime.fromisoformat(latest['timestamp']).strftime('%H:%M')}") st.write(f"**Compliance:** {latest['compliance_score']}%") else: if check_permission('can_add_employees'): st.info("No employees. Add employees first.") else: st.info("No employees available.") st.session_state.selected_employee_id = None st.divider() # Quick Actions st.subheader("Quick Actions") if check_permission('can_run_ppe_detection'): if st.button("๐Ÿช– Quick PPE Scan", use_container_width=True): st.session_state.run_ppe_scan = True if check_permission('can_manage_hazards'): if st.button("โš ๏ธ Report Hazard", use_container_width=True): st.session_state.report_hazard = True if check_permission('can_report_incidents'): if st.button("๐Ÿ“‹ Report Incident", use_container_width=True): st.session_state.report_incident = True # User Management (admin only) if check_permission('can_manage_users'): st.divider() st.subheader("๐Ÿ‘ฅ User Management") if st.button("Manage Users", use_container_width=True): st.session_state.show_user_management = True # Emergency Button if check_permission('can_acknowledge_alerts'): st.divider() if st.button("๐Ÿšจ EMERGENCY ALERT", type="primary", use_container_width=True): st.session_state.emergency = True if st.session_state.current_user: db.log_activity(st.session_state.current_user['user_id'], "emergency", "Emergency alert triggered") st.error("๐Ÿšจ EMERGENCY PROTOCOL ACTIVATED!") st.divider() if st.button("๐Ÿšช Logout", use_container_width=True): logout() # ========== MAIN APP ========== st.markdown("""

๐Ÿ›ก๏ธ SafeGuard Pro

AI-Powered Workplace Safety Monitoring System

Welcome, {} | Role: {}

""".format(st.session_state.current_user['full_name'], st.session_state.current_user['role'].replace('_', ' ').upper()), unsafe_allow_html=True) # ========== MAIN TABS ========== tab_list = ["๐Ÿญ Dashboard"] if check_permission('can_add_employees') or check_permission('can_edit_employees'): tab_list.append("๐Ÿ‘ฅ Employees") if check_permission('can_run_ppe_detection'): tab_list.append("๐Ÿช– PPE Detection") tab_list.append("โš ๏ธ Hazard Detection") tab_list.append("๐ŸŒก๏ธ Environment") if check_permission('can_view_analytics'): tab_list.append("๐Ÿ“Š Safety Analytics") if check_permission('can_schedule_inspections'): tab_list.append("๐Ÿ” Inspections") if check_permission('can_manage_users'): tab_list.append("๐Ÿ‘ค Users") if check_permission('can_view_audit_logs'): tab_list.append("๐Ÿ“‹ Audit Log") tabs = st.tabs(tab_list) tab_index = 0 # ========== TAB 1: DASHBOARD ========== with tabs[0]: st.header("Safety Dashboard") # Top metrics col1, col2, col3, col4 = st.columns(4) with col1: total_employees = len(st.session_state.employees) st.metric("Total Employees", total_employees) with col2: st.metric("Active Hazards", len(active_hazards), delta=-len(active_hazards) if active_hazards else 0) with col3: st.metric("PPE Violations (24h)", len(recent_violations)) with col4: # Calculate safety score ppe_records = [] for emp in st.session_state.employees[:10]: # Sample last 10 employees records = db.get_employee_ppe_history(emp['employee_id'], limit=1) if records: ppe_records.append(records[0]['compliance_score']) avg_compliance = np.mean(ppe_records) if ppe_records else 85 st.metric("Safety Score", f"{avg_compliance:.0f}%", delta=f"{avg_compliance-85:.0f}%" if ppe_records else None) # Alerts section if total_alerts > 0: st.subheader("๐Ÿšจ Active Alerts") alert_col1, alert_col2 = st.columns(2) with alert_col1: if recent_violations: with st.expander(f"PPE Violations ({len(recent_violations)})", expanded=True): for v in recent_violations[:5]: st.warning(f"**{v.get('employee_name', 'Unknown')}** - {v['compliance_score']}% compliance") st.caption(f"Location: {v.get('location', 'Unknown')} | {datetime.fromisoformat(v['timestamp']).strftime('%H:%M')}") with alert_col2: if active_hazards: with st.expander(f"Active Hazards ({len(active_hazards)})", expanded=True): for h in active_hazards[:5]: color = "red" if h['severity'] in ['Critical', 'High'] else "orange" st.markdown(f"โš ๏ธ **{h['hazard_type']}** ({h['severity']})", unsafe_allow_html=True) st.caption(f"Location: {h['location']}") # Recent incidents st.subheader("Recent Incidents") recent_incidents = db.get_recent_incidents(days=7) if recent_incidents: for inc in recent_incidents[:5]: with st.container(): col1, col2, col3 = st.columns([2, 2, 1]) with col1: st.write(f"**{inc['incident_type']}**") with col2: st.write(f"Location: {inc['location']}") with col3: status_color = "orange" if inc['status'] == 'investigating' else "green" st.markdown(f"{inc['status'].upper()}", unsafe_allow_html=True) st.caption(f"{datetime.fromisoformat(inc['timestamp']).strftime('%Y-%m-%d %H:%M')}") st.divider() else: st.info("No recent incidents reported") # Environmental summary st.subheader("Environmental Summary") env_readings = env_monitor.get_readings() env_col1, env_col2, env_col3, env_col4 = st.columns(4) with env_col1: temp_color = "green" if 18 <= env_readings['temperature'] <= 28 else "orange" st.markdown(f"๐ŸŒก๏ธ **Temperature:** {env_readings['temperature']:.1f}ยฐC", unsafe_allow_html=True) with env_col2: noise_color = "green" if env_readings['noise_level'] <= 85 else "orange" st.markdown(f"๐Ÿ”Š **Noise:** {env_readings['noise_level']:.0f} dB", unsafe_allow_html=True) with env_col3: air_color = "green" if env_readings['air_quality'] >= 70 else "orange" st.markdown(f"๐Ÿ’จ **Air Quality:** {env_readings['air_quality']:.0f}", unsafe_allow_html=True) with env_col4: light_color = "green" if 200 <= env_readings['illuminance'] <= 1000 else "orange" st.markdown(f"๐Ÿ’ก **Light:** {env_readings['illuminance']:.0f} lux", unsafe_allow_html=True) # ========== TAB 2: EMPLOYEE MANAGEMENT ========== if "๐Ÿ‘ฅ Employees" in tab_list: tab_index += 1 with tabs[tab_index]: st.header("๐Ÿ‘ฅ Employee Management") emp_tab1, emp_tab2 = st.tabs(["๐Ÿ“‹ Employee List", "โž• Add New Employee"]) with emp_tab1: employees = st.session_state.employees if not employees: st.info("No employees found. Add your first employee.") else: search_col, filter_col = st.columns([2, 1]) with search_col: search_query = st.text_input("๐Ÿ” Search employees by name or ID:", placeholder="Enter name or employee ID...") with filter_col: departments = list(set(e.get('department', 'Unknown') for e in employees)) filter_dept = st.selectbox( "Filter by Department:", ["All Departments"] + departments ) filtered_employees = employees if search_query: filtered_employees = [ e for e in filtered_employees if search_query.lower() in e['name'].lower() or search_query.lower() in e['employee_id'].lower() ] if filter_dept != "All Departments": filtered_employees = [e for e in filtered_employees if e.get('department') == filter_dept] for emp in filtered_employees: with st.container(): col1, col2, col3 = st.columns([3, 2, 1]) with col1: st.markdown(f"**{emp['name']}** ({emp.get('age', 'N/A')} years)") st.caption(f"ID: {emp['employee_id']} | {emp.get('department', 'N/A')} | {emp.get('designation', 'N/A')}") with col2: ppe_history = db.get_employee_ppe_history(emp['employee_id'], limit=1) if ppe_history: latest = ppe_history[0] score = latest.get('compliance_score', 0) if score >= 85: status = "โœ… Good" color = "green" elif score >= 70: status = "โš ๏ธ Fair" color = "orange" else: status = "โŒ Poor" color = "red" st.markdown(f"Compliance: {status}", unsafe_allow_html=True) st.caption(f"Score: {score}%") else: st.caption("No PPE data") with col3: if st.button("Select", key=f"select_{emp['employee_id']}"): st.session_state.selected_employee_id = emp['employee_id'] st.rerun() if check_permission('can_delete_employees'): if st.button("๐Ÿ—‘๏ธ", key=f"delete_{emp['employee_id']}"): if st.session_state.selected_employee_id == emp['employee_id']: st.session_state.selected_employee_id = None db.delete_employee(emp['employee_id']) db.log_activity(st.session_state.current_user['user_id'], "delete_employee", f"Deleted employee: {emp['name']}") st.success(f"Employee {emp['name']} deleted.") st.session_state.employees = db.get_all_employees() st.rerun() st.divider() with emp_tab2: if check_permission('can_add_employees'): show_add_employee_form() else: st.warning("You don't have permission to add new employees.") # ========== TAB 3: PPE DETECTION ========== if "๐Ÿช– PPE Detection" in tab_list: tab_index += 1 with tabs[tab_index]: st.header("๐Ÿช– AI PPE Detection") if not st.session_state.selected_employee_id: st.info("๐Ÿ‘ˆ Please select an employee from the sidebar for PPE verification.") else: employee = db.get_employee(st.session_state.selected_employee_id) st.subheader(f"PPE Check for {employee['name']}") # Detection options detection_type = st.radio( "Select Input Method:", ["Camera Live Feed", "Upload Image", "Manual Check"], horizontal=True ) if detection_type == "Camera Live Feed": st.info("๐Ÿ“น Using camera for real-time PPE detection") camera_input = st.camera_input("Take photo for PPE analysis") if camera_input: image = Image.open(camera_input) col1, col2 = st.columns(2) with col1: st.image(image, caption="Capture", use_column_width=True) if st.button("๐Ÿ” Analyze PPE", type="primary", use_container_width=True): with st.spinner("Analyzing PPE compliance..."): # Detect PPE detections = ppe_detector.detect_ppe() safety_score = ppe_detector.calculate_safety_score(detections) # Save record record_id = db.add_ppe_record( employee['employee_id'], { **detections, 'location': 'Production Floor', 'notes': 'Camera detection', 'recorded_by': st.session_state.current_user['user_id'] } ) db.log_activity(st.session_state.current_user['user_id'], "ppe_scan", f"PPE scan for {employee['name']}, Score: {safety_score['compliance_score']}%") with col2: st.success(f"โœ… Analysis Complete! Record ID: {record_id}") display_ppe_status(detections, safety_score) # Recommendations if safety_score['missing_items']: st.subheader("๐Ÿ“‹ Required Actions") for item in safety_score['missing_items']: st.write(f"โ€ข Don {item}") if 'analysis_done' not in st.session_state: with col2: st.info("๐Ÿ‘ˆ Click 'Analyze PPE' to start detection") elif detection_type == "Upload Image": st.info("๐Ÿ“ค Upload an image for PPE detection") uploaded_file = st.file_uploader( "Choose an image", type=['jpg', 'jpeg', 'png'], help="Upload a clear photo of the employee" ) if uploaded_file: image = Image.open(uploaded_file) col1, col2 = st.columns(2) with col1: st.image(image, caption="Uploaded Image", use_column_width=True) with col2: if st.button("๐Ÿ” Analyze Uploaded Image", type="primary", use_container_width=True): with st.spinner("Analyzing PPE compliance..."): detections = ppe_detector.detect_ppe() safety_score = ppe_detector.calculate_safety_score(detections) record_id = db.add_ppe_record( employee['employee_id'], { **detections, 'location': 'Upload Check', 'notes': 'Image upload analysis', 'recorded_by': st.session_state.current_user['user_id'] } ) db.log_activity(st.session_state.current_user['user_id'], "ppe_upload", f"PPE upload for {employee['name']}, Score: {safety_score['compliance_score']}%") st.success(f"โœ… Analysis Complete! Record ID: {record_id}") display_ppe_status(detections, safety_score) if safety_score['missing_items']: st.warning(f"Missing: {', '.join(safety_score['missing_items'])}") else: # Manual Check st.info("๐Ÿ“‹ Manual PPE verification checklist") with st.form("manual_ppe_form"): st.subheader("PPE Checklist") col1, col2 = st.columns(2) with col1: hard_hat = st.checkbox("Hard Hat", value=True) safety_vest = st.checkbox("Safety Vest", value=True) safety_glasses = st.checkbox("Safety Glasses", value=True) gloves = st.checkbox("Gloves", value=True) with col2: steel_toe_boots = st.checkbox("Steel Toe Boots", value=True) ear_protection = st.checkbox("Ear Protection", value=False) respirator = st.checkbox("Respirator", value=False) harness = st.checkbox("Safety Harness", value=False) location = st.text_input("Location", value="Production Floor") notes = st.text_area("Additional Notes") submitted = st.form_submit_button("Submit PPE Check", type="primary") if submitted: ppe_data = { 'hard_hat': 1 if hard_hat else 0, 'safety_vest': 1 if safety_vest else 0, 'safety_glasses': 1 if safety_glasses else 0, 'gloves': 1 if gloves else 0, 'steel_toe_boots': 1 if steel_toe_boots else 0, 'ear_protection': 1 if ear_protection else 0, 'respirator': 1 if respirator else 0, 'harness': 1 if harness else 0, 'location': location, 'notes': notes, 'recorded_by': st.session_state.current_user['user_id'] } record_id = db.add_ppe_record(employee['employee_id'], ppe_data) # Calculate compliance for display ppe_items = [hard_hat, safety_vest, safety_glasses, gloves, steel_toe_boots, ear_protection, respirator, harness] compliance = int((sum(ppe_items) / len(ppe_items)) * 100) st.success(f"โœ… Manual PPE check recorded! Compliance: {compliance}%") if compliance < 80: missing = [item for item, present in zip( ['Hard Hat', 'Safety Vest', 'Safety Glasses', 'Gloves', 'Steel Toe Boots', 'Ear Protection', 'Respirator', 'Harness'], ppe_items) if not present] st.warning(f"Missing PPE: {', '.join(missing)}") # ========== TAB 4: HAZARD DETECTION ========== if "โš ๏ธ Hazard Detection" in tab_list: tab_index += 1 with tabs[tab_index]: st.header("โš ๏ธ AI Hazard Detection") hazard_tab1, hazard_tab2, hazard_tab3 = st.tabs(["๐Ÿ” Detect Hazards", "๐Ÿ“‹ Active Hazards", "โž• Report Hazard"]) with hazard_tab1: st.subheader("AI Hazard Detection") detection_method = st.radio( "Select detection method:", ["Camera Feed", "Upload Image", "Simulate Detection"], horizontal=True ) if detection_method == "Camera Feed": camera_input = st.camera_input("Scan area for hazards") if camera_input: if st.button("๐Ÿ” Analyze for Hazards", type="primary"): with st.spinner("Analyzing for safety hazards..."): time.sleep(2) detection_result = hazard_detector.detect_hazards() if detection_result['hazard_detected']: st.error(f"๐Ÿšจ HAZARD DETECTED!") st.warning(f"**Type:** {detection_result['hazard_type']}") st.warning(f"**Severity:** {detection_result['severity']}") st.warning(f"**Location:** {detection_result['location']}") st.warning(f"**Confidence:** {detection_result['confidence']*100:.1f}%") # Add to database hazard_id = db.add_hazard({ 'hazard_type': detection_result['hazard_type'], 'severity': detection_result['severity'], 'location': detection_result['location'], 'description': detection_result['recommendation'], 'notes': f"Detected with {detection_result['confidence']*100:.1f}% confidence" }) st.info(f"Hazard logged with ID: {hazard_id}") if st.button("Resolve Hazard"): db.resolve_hazard(hazard_id, st.session_state.current_user['user_id'], "Resolved") st.success("Hazard marked as resolved!") st.rerun() else: st.success(f"โœ… No hazards detected (Confidence: {detection_result['confidence']*100:.1f}%)") elif detection_method == "Upload Image": uploaded_file = st.file_uploader("Upload area image", type=['jpg', 'jpeg', 'png']) if uploaded_file and st.button("Analyze Image"): with st.spinner("Analyzing for hazards..."): time.sleep(2) detection_result = hazard_detector.detect_hazards() if detection_result['hazard_detected']: st.error("๐Ÿšจ HAZARD DETECTED!") st.json(detection_result) else: st.success("โœ… Area appears safe") else: # Simulate st.info("Running simulated hazard detection across facility") zones = ['Zone A', 'Zone B', 'Zone C', 'Zone D'] results = [] for zone in zones: result = hazard_detector.detect_hazards() results.append({ 'zone': zone, 'hazard': result['hazard_type'] if result['hazard_detected'] else 'None', 'severity': result['severity'] if result['hazard_detected'] else 'None', 'status': 'โš ๏ธ' if result['hazard_detected'] else 'โœ…' }) st.subheader("Facility Scan Results") results_df = pd.DataFrame(results) st.dataframe(results_df, use_container_width=True, hide_index=True) with hazard_tab2: st.subheader("Active Hazards") active_hazards = db.get_active_hazards() if active_hazards: for hazard in active_hazards: with st.container(): severity_color = { 'Critical': '๐Ÿ”ด', 'High': '๐ŸŸ ', 'Medium': '๐ŸŸก', 'Low': '๐ŸŸข' }.get(hazard['severity'], 'โšช') col1, col2, col3 = st.columns([3, 2, 1]) with col1: st.markdown(f"{severity_color} **{hazard['hazard_type']}**") st.caption(f"Location: {hazard['location']}") with col2: st.write(f"Severity: {hazard['severity']}") st.caption(f"Detected: {datetime.fromisoformat(hazard['detected_at']).strftime('%H:%M %d/%m')}") with col3: if st.button("Resolve", key=f"resolve_{hazard['hazard_id']}"): db.resolve_hazard(hazard['hazard_id'], st.session_state.current_user['user_id'], "Resolved") st.success("Hazard resolved!") st.rerun() st.divider() else: st.success("โœ… No active hazards detected!") with hazard_tab3: st.subheader("Report New Hazard") with st.form("report_hazard_form"): col1, col2 = st.columns(2) with col1: hazard_type = st.selectbox( "Hazard Type*", ["Spill/Chemical Leak", "Blocked Emergency Exit", "Unsafe Stacking", "Electrical Hazard", "Trip Hazard", "Fire Hazard", "Machine Guard Missing", "Poor Housekeeping", "Confined Space Risk", "Working at Height Unsafe", "Other"] ) severity = st.selectbox( "Severity*", ["Low", "Medium", "High", "Critical"] ) with col2: location = st.text_input("Location*", placeholder="Building/Floor/Zone") custom_type = st.text_input("If Other, specify:") description = st.text_area("Description*", placeholder="Describe the hazard in detail...") notes = st.text_area("Additional Notes", placeholder="Any additional information...") submitted = st.form_submit_button("Report Hazard", type="primary") if submitted: if not location or not description: st.error("Location and Description are required!") else: final_type = custom_type if hazard_type == "Other" and custom_type else hazard_type hazard_id = db.add_hazard({ 'hazard_type': final_type, 'severity': severity, 'location': location, 'description': description, 'notes': notes }) db.log_activity(st.session_state.current_user['user_id'], "report_hazard", f"Reported hazard: {final_type} at {location}") st.success(f"โœ… Hazard reported successfully! Hazard ID: {hazard_id}") st.balloons() # ========== TAB 5: ENVIRONMENTAL MONITORING ========== if "๐ŸŒก๏ธ Environment" in tab_list: tab_index += 1 with tabs[tab_index]: st.header("๐ŸŒก๏ธ Environmental Monitoring") col1, col2 = st.columns([2, 1]) with col2: st.subheader("Current Readings") location = st.selectbox( "Select Zone", ["Zone A", "Zone B", "Zone C", "Zone D", "Warehouse", "Assembly Line"] ) if st.button("Refresh Readings", type="primary"): st.rerun() # Get current readings readings = env_monitor.get_readings(location) alerts = env_monitor.check_alerts(readings) with col1: st.subheader(f"Environmental Dashboard - {location}") # Display gauges gauge_col1, gauge_col2, gauge_col3 = st.columns(3) with gauge_col1: # Temperature gauge fig = go.Figure(go.Indicator( mode="gauge+number", value=readings['temperature'], title={'text': "Temperature (ยฐC)"}, domain={'x': [0, 1], 'y': [0, 1]}, gauge={ 'axis': {'range': [0, 50]}, 'bar': {'color': "orange"}, 'steps': [ {'range': [0, 18], 'color': "lightblue"}, {'range': [18, 28], 'color': "lightgreen"}, {'range': [28, 50], 'color': "lightcoral"} ], 'threshold': { 'line': {'color': "red", 'width': 4}, 'thickness': 0.75, 'value': 28 } } )) fig.update_layout(height=200, margin=dict(l=10, r=10, t=50, b=10)) st.plotly_chart(fig, use_container_width=True) with gauge_col2: # Humidity gauge fig = go.Figure(go.Indicator( mode="gauge+number", value=readings['humidity'], title={'text': "Humidity (%)"}, domain={'x': [0, 1], 'y': [0, 1]}, gauge={ 'axis': {'range': [0, 100]}, 'bar': {'color': "blue"}, 'steps': [ {'range': [0, 30], 'color': "lightyellow"}, {'range': [30, 70], 'color': "lightgreen"}, {'range': [70, 100], 'color': "lightblue"} ] } )) fig.update_layout(height=200, margin=dict(l=10, r=10, t=50, b=10)) st.plotly_chart(fig, use_container_width=True) with gauge_col3: # Noise gauge fig = go.Figure(go.Indicator( mode="gauge+number", value=readings['noise_level'], title={'text': "Noise Level (dB)"}, domain={'x': [0, 1], 'y': [0, 1]}, gauge={ 'axis': {'range': [0, 120]}, 'bar': {'color': "purple"}, 'steps': [ {'range': [0, 70], 'color': "lightgreen"}, {'range': [70, 85], 'color': "lightyellow"}, {'range': [85, 120], 'color': "lightcoral"} ], 'threshold': { 'line': {'color': "red", 'width': 4}, 'thickness': 0.75, 'value': 85 } } )) fig.update_layout(height=200, margin=dict(l=10, r=10, t=50, b=10)) st.plotly_chart(fig, use_container_width=True) # Alert section if alerts: st.subheader("โš ๏ธ Environmental Alerts") for alert in alerts: severity_color = "orange" if alert['severity'] == 'Medium' else "yellow" st.warning(f"**{alert['parameter'].replace('_', ' ').title()}**: {alert['value']:.1f} is {alert['threshold']}") # Historical data st.subheader("Historical Data") history = db.get_environmental_history(location, hours=24) if history: hist_df = pd.DataFrame(history) hist_df['timestamp'] = pd.to_datetime(hist_df['timestamp']) hist_df = hist_df.sort_values('timestamp') fig = px.line(hist_df, x='timestamp', y=['temperature', 'humidity', 'noise_level'], title=f"Environmental Trends - {location} (Last 24h)", labels={'value': 'Reading', 'variable': 'Parameter'}) st.plotly_chart(fig, use_container_width=True) else: # Generate sample data times = pd.date_range(end=datetime.now(), periods=24, freq='H') sample_data = pd.DataFrame({ 'timestamp': times, 'temperature': [22 + np.random.uniform(-3, 3) for _ in range(24)], 'humidity': [45 + np.random.uniform(-10, 10) for _ in range(24)], 'noise_level': [70 + np.random.uniform(-5, 15) for _ in range(24)] }) fig = px.line(sample_data, x='timestamp', y=['temperature', 'humidity', 'noise_level'], title=f"Sample Environmental Trends - {location}", labels={'value': 'Reading', 'variable': 'Parameter'}) st.plotly_chart(fig, use_container_width=True) st.info("Sample data shown. Real monitoring data will appear here.") # Save current reading if st.button("๐Ÿ’พ Save Current Reading"): reading_id = db.add_environmental_reading(readings) st.success(f"Reading saved! ID: {reading_id}") # ========== TAB 6: SAFETY ANALYTICS ========== if "๐Ÿ“Š Safety Analytics" in tab_list: tab_index += 1 with tabs[tab_index]: st.header("๐Ÿ“Š Safety Analytics") date_range = st.selectbox( "Select Time Range", ["Last 7 Days", "Last 30 Days", "Last 90 Days", "Year to Date"] ) days = { "Last 7 Days": 7, "Last 30 Days": 30, "Last 90 Days": 90, "Year to Date": 365 }[date_range] col1, col2 = st.columns(2) with col1: # PPE Compliance Trend st.subheader("PPE Compliance Trend") # Generate sample compliance data dates = pd.date_range(end=datetime.now(), periods=days, freq='D') compliance_data = pd.DataFrame({ 'date': dates, 'compliance': [85 + np.random.uniform(-10, 5) for _ in range(days)] }) fig = px.line(compliance_data, x='date', y='compliance', title="Overall PPE Compliance", labels={'compliance': 'Compliance %'}) fig.add_hline(y=85, line_dash="dash", line_color="green", annotation_text="Target (85%)") st.plotly_chart(fig, use_container_width=True) with col2: # Hazard Distribution st.subheader("Hazards by Type") hazard_types = ['Spill', 'Electrical', 'Fire', 'Trip', 'Machine Guard', 'Housekeeping'] hazard_counts = [np.random.randint(0, 10) for _ in hazard_types] fig = px.pie(values=hazard_counts, names=hazard_types, title="Hazard Distribution") st.plotly_chart(fig, use_container_width=True) col3, col4 = st.columns(2) with col3: # Incidents by Department st.subheader("Incidents by Department") depts = ['Production', 'Assembly', 'Warehouse', 'Maintenance', 'Logistics'] incident_counts = [np.random.randint(0, 5) for _ in depts] fig = px.bar(x=depts, y=incident_counts, title="Incident Count by Department", labels={'x': 'Department', 'y': 'Incidents'}) st.plotly_chart(fig, use_container_width=True) with col4: # Safety Score by Shift st.subheader("Safety Score by Shift") shifts = ['Morning', 'Afternoon', 'Night'] safety_scores = [88, 82, 75] fig = px.bar(x=shifts, y=safety_scores, title="Average Safety Score by Shift", labels={'x': 'Shift', 'y': 'Safety Score %'}, color=safety_scores, color_continuous_scale=['red', 'yellow', 'green']) st.plotly_chart(fig, use_container_width=True) # Key Metrics st.subheader("Key Safety Metrics") m1, m2, m3, m4 = st.columns(4) with m1: st.metric("Days Without Incident", np.random.randint(5, 30), delta=np.random.randint(-2, 5)) with m2: st.metric("Open Hazards", len(db.get_active_hazards()), delta=-np.random.randint(0, 3)) with m3: st.metric("Avg Compliance", f"{np.random.randint(82, 92)}%", delta=f"{np.random.randint(-2, 3)}%") with m4: st.metric("Inspections Due", np.random.randint(3, 12), delta=np.random.randint(-2, 2)) if check_permission('can_export_data'): if st.button("๐Ÿ“ฅ Export Analytics Report"): csv_data = "Sample report data" st.download_button( label="Download CSV", data=csv_data, file_name=f"safety_report_{datetime.now().strftime('%Y%m%d')}.csv", mime="text/csv" ) # ========== TAB 7: INSPECTIONS ========== if "๐Ÿ” Inspections" in tab_list: tab_index += 1 with tabs[tab_index]: st.header("๐Ÿ” Safety Inspections") insp_tab1, insp_tab2 = st.tabs(["๐Ÿ“‹ Pending Inspections", "โž• Schedule Inspection"]) with insp_tab1: pending = db.get_pending_inspections() if pending: for insp in pending: with st.container(): col1, col2, col3 = st.columns([3, 2, 1]) with col1: st.write(f"**Area:** {insp['area']}") st.caption(f"Findings: {insp['findings'][:100]}...") with col2: due_date = datetime.fromisoformat(insp['due_date']).strftime('%Y-%m-%d') st.write(f"Due: {due_date}") st.write(f"Inspector: {insp.get('inspector_name', 'Unknown')}") with col3: if st.button("Complete", key=f"comp_{insp['inspection_id']}"): db.complete_inspection(insp['inspection_id']) st.success("Inspection completed!") st.rerun() st.divider() else: st.success("โœ… No pending inspections") with insp_tab2: with st.form("schedule_inspection"): col1, col2 = st.columns(2) with col1: area = st.text_input("Area to Inspect*") inspector = st.selectbox( "Inspector*", [u['full_name'] for u in db.get_all_users() if u['is_active']] ) with col2: due_date = st.date_input("Due Date*", min_value=datetime.now()) findings = st.text_area("Initial Findings (if any)") corrective_actions = st.text_area("Recommended Corrective Actions") submitted = st.form_submit_button("Schedule Inspection", type="primary") if submitted: if not area or not inspector: st.error("Area and Inspector are required!") else: inspector_id = next(u['user_id'] for u in db.get_all_users() if u['full_name'] == inspector) insp_data = { 'date': datetime.now().strftime('%Y-%m-%d'), 'inspector_id': inspector_id, 'area': area, 'findings': findings, 'violations': '', 'corrective_actions': corrective_actions, 'due_date': due_date.strftime('%Y-%m-%d') } insp_id = db.add_inspection(insp_data) st.success(f"โœ… Inspection scheduled! ID: {insp_id}") st.balloons() # ========== TAB 8: USER MANAGEMENT ========== if "๐Ÿ‘ค Users" in tab_list: tab_index += 1 with tabs[tab_index]: st.header("๐Ÿ‘ค User Management") user_tab1, user_tab2, user_tab3 = st.tabs(["๐Ÿ“‹ Users", "โž• Add User", "๐Ÿ”‘ Change Password"]) with user_tab1: users = db.get_all_users() if users: for user in users: with st.container(): col1, col2, col3 = st.columns([3, 2, 1]) with col1: st.markdown(f"**{user['full_name']}** ({user['username']})") st.caption(f"ID: {user['user_id']} | Role: {user['role'].upper()} | Dept: {user.get('department', 'N/A')}") with col2: status = "๐ŸŸข Active" if user['is_active'] else "๐Ÿ”ด Inactive" st.write(f"Status: {status}") if user.get('last_login'): last_login = datetime.fromisoformat(user['last_login']).strftime('%Y-%m-%d %H:%M') st.caption(f"Last login: {last_login}") with col3: if user['user_id'] != st.session_state.current_user['user_id']: if st.button("๐Ÿ—‘๏ธ", key=f"del_user_{user['user_id']}"): db.delete_user(user['user_id']) db.log_activity(st.session_state.current_user['user_id'], "delete_user", f"Deleted user: {user['username']}") st.success(f"User {user['username']} deactivated.") st.rerun() st.divider() else: st.info("No users found.") with user_tab2: st.subheader("Add New User") with st.form("add_user_form"): col1, col2 = st.columns(2) with col1: username = st.text_input("Username*") full_name = st.text_input("Full Name*") password = st.text_input("Password*", type="password") with col2: email = st.text_input("Email") role = st.selectbox("Role*", ["admin", "safety_manager", "supervisor", "inspector", "viewer"]) department = st.text_input("Department") submitted = st.form_submit_button("Add User", type="primary") if submitted: if not username or not full_name or not password: st.error("Username, Full Name, and Password are required!") else: user_data = { 'username': username, 'password': password, 'full_name': full_name, 'email': email, 'role': role, 'department': department } try: user_id = db.add_user(user_data) db.log_activity(st.session_state.current_user['user_id'], "add_user", f"Added user: {username}") st.success(f"โœ… User added successfully! User ID: {user_id}") st.rerun() except sqlite3.IntegrityError: st.error("Username already exists.") with user_tab3: st.subheader("Change Password") with st.form("change_password_form"): current_user = st.session_state.current_user st.write(f"Changing password for: **{current_user['full_name']}**") new_password = st.text_input("New Password*", type="password") confirm_password = st.text_input("Confirm New Password*", type="password") submitted = st.form_submit_button("Change Password", type="primary") if submitted: if not new_password: st.error("Password cannot be empty!") elif new_password != confirm_password: st.error("Passwords do not match!") elif len(new_password) < 6: st.error("Password must be at least 6 characters long!") else: db.update_user(current_user['user_id'], {'password': new_password}) db.log_activity(current_user['user_id'], "change_password", "User changed their password") st.success("โœ… Password changed successfully!") # ========== TAB 9: AUDIT LOG ========== if "๐Ÿ“‹ Audit Log" in tab_list: tab_index += 1 with tabs[tab_index]: st.header("๐Ÿ“‹ Audit Log") logs = db.get_activity_log(limit=200) if logs: log_df = pd.DataFrame(logs) log_df['timestamp'] = pd.to_datetime(log_df['timestamp']).dt.strftime('%Y-%m-%d %H:%M:%S') st.dataframe( log_df[['timestamp', 'username', 'full_name', 'action', 'details']], use_container_width=True, hide_index=True ) if st.button("๐Ÿ“ฅ Export Logs"): csv = log_df.to_csv(index=False) st.download_button( label="Download CSV", data=csv, file_name=f"audit_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv", mime="text/csv" ) else: st.info("No activity logs available.") # ========== EMERGENCY OVERLAY ========== if st.session_state.get('emergency', False): st.markdown(f"""

๐Ÿšจ EMERGENCY ALERT ๐Ÿšจ

EMERGENCY PROTOCOL ACTIVATED

Emergency Services: 911

Safety Officer: Ext. 1111

Medical Station: Building A, Ground Floor

Assembly Point: Parking Lot A

""", unsafe_allow_html=True) # ========== FOOTER ========== st.divider() employees_count = len(st.session_state.employees) active_hazards_count = len(db.get_active_hazards()) recent_incidents_count = len(db.get_recent_incidents(days=7)) st.markdown(f"""

๐Ÿ›ก๏ธ SafeGuard Pro v1.0 | AI-Powered Workplace Safety Monitoring

๐Ÿ“Š Employees: {employees_count} | Active Hazards: {active_hazards_count} | Incidents (7d): {recent_incidents_count}

๐Ÿ‘ค Logged in as: {st.session_state.current_user['full_name']} ({st.session_state.current_user['role'].replace('_', ' ').upper()})

โš ๏ธ This system aids safety monitoring but doesn't replace physical safety inspections.

๐Ÿ“ž Emergency: 911 | Safety Hotline: 1-800-SAFETY | copyright ยฉ Sabiha Noor/BCA

""", unsafe_allow_html=True)