FCTX commited on
Commit
c54cd04
·
verified ·
1 Parent(s): 5207a73

Upload 7 files

Browse files
Files changed (8) hide show
  1. .gitattributes +1 -0
  2. .gitignore +26 -0
  3. README.md +164 -10
  4. app.py +607 -0
  5. requirements.txt +6 -0
  6. sample_images/README.md +12 -0
  7. sample_images/test.png +3 -0
  8. yolov8n.pt +3 -0
.gitattributes CHANGED
@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
33
  *.zip filter=lfs diff=lfs merge=lfs -text
34
  *.zst filter=lfs diff=lfs merge=lfs -text
35
  *tfevents* filter=lfs diff=lfs merge=lfs -text
 
 
33
  *.zip filter=lfs diff=lfs merge=lfs -text
34
  *.zst filter=lfs diff=lfs merge=lfs -text
35
  *tfevents* filter=lfs diff=lfs merge=lfs -text
36
+ sample_images/test.png filter=lfs diff=lfs merge=lfs -text
.gitignore ADDED
@@ -0,0 +1,26 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ __pycache__/
2
+ *.py[cod]
3
+ *$py.class
4
+
5
+ .Python
6
+ .venv/
7
+ venv/
8
+ env/
9
+
10
+ .env
11
+ .env.*
12
+
13
+ .gradio/
14
+ runs/
15
+ weights/
16
+ models/
17
+
18
+ *.pt
19
+ *.onnx
20
+ *.engine
21
+
22
+ .DS_Store
23
+ Thumbs.db
24
+
25
+ .idea/
26
+ .vscode/
README.md CHANGED
@@ -1,15 +1,169 @@
1
  ---
2
- title: AI PPE Detection System
3
- emoji: 🌖
4
- colorFrom: gray
5
- colorTo: red
6
  sdk: gradio
7
- sdk_version: 6.17.3
8
- python_version: '3.13'
9
  app_file: app.py
10
- pinned: false
11
- license: mit
12
- short_description: AI PPE detection system for industrial safety monitoring
13
  ---
14
 
15
- Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
  ---
2
+ title: AI-PPE-Detection-System
 
 
 
3
  sdk: gradio
 
 
4
  app_file: app.py
 
 
 
5
  ---
6
 
7
+ # AI-PPE-Detection-System
8
+
9
+ AI-powered PPE detection system for industrial safety monitoring.
10
+
11
+ ## Overview
12
+
13
+ This project detects workers and personal protective equipment such as helmets and safety vests from workplace images.
14
+ It is designed as a prototype for factory, warehouse, and construction site safety monitoring.
15
+
16
+ Unlike a simple YOLO object detection demo, this system associates PPE detections with each detected worker and produces an operational safety dashboard. The output is designed to be useful for supervisors, operations teams, AI portfolio reviewers, and prototype discussions with industrial clients.
17
+
18
+ ## Features
19
+
20
+ - Worker detection
21
+ - Helmet detection
22
+ - Safety vest detection
23
+ - PPE compliance check
24
+ - Worker count
25
+ - Helmet compliance summary
26
+ - Vest compliance summary
27
+ - Per-worker PPE assessment table
28
+ - Missing PPE item summary
29
+ - Detection details table with assigned / unassigned PPE status
30
+ - Adjustable confidence threshold
31
+ - Adjustable IoU threshold
32
+ - Annotated detection image
33
+ - Gradio web UI
34
+ - Hugging Face Spaces compatible
35
+
36
+ ## Industrial AI Use Cases
37
+
38
+ This system can be used as a prototype for industrial safety monitoring applications.
39
+
40
+ - Factory PPE Monitoring
41
+ - Safety Compliance Check
42
+ - Worker Safety Analytics
43
+ - Smart Factory AI
44
+ - Construction Site Safety Monitoring
45
+ - Warehouse Safety Monitoring
46
+
47
+ ## Demo Behavior
48
+
49
+ Upload a factory, warehouse, construction site, or worker image. The app returns:
50
+
51
+ - An annotated image with detected workers and PPE
52
+ - Site Safety Status
53
+ - PPE Check Result table
54
+ - Worker Safety Summary table
55
+ - Per-worker PPE assessment table
56
+ - Detection details table
57
+
58
+ PPE judgment rules:
59
+
60
+ - Helmet is `OK` when helmet or hardhat evidence is detected and associated with a worker.
61
+ - Safety Vest is `OK` when safety vest evidence is detected and associated with a worker.
62
+ - Overall Result is `OK` only when all detected workers are compliant.
63
+ - If any detected worker is missing a helmet or safety vest, the site safety status is `NG`.
64
+
65
+ Worker safety summary rules:
66
+
67
+ - `Workers` is the number of detected `person` objects.
68
+ - `Helmet Compliance` is the number of workers with a helmet inside their person box divided by total workers.
69
+ - `Vest Compliance` is the number of workers with a safety vest inside their person box divided by total workers.
70
+ - `Fully Compliant Workers` is the number of workers with both required PPE items divided by total workers.
71
+ - If no workers are detected, the summary shows `No workers detected`.
72
+
73
+ ## System Logic
74
+
75
+ The application runs two object detection passes:
76
+
77
+ 1. A COCO YOLO model detects workers using the `person` class.
78
+ 2. A PPE YOLO model detects helmet, hardhat, safety vest, and missing-PPE labels.
79
+
80
+ PPE boxes are assigned to the most likely worker using box overlap and center-point checks. This makes the result more useful than a raw object list because the app can answer operational questions such as:
81
+
82
+ - Which worker is missing required PPE?
83
+ - How many workers are fully compliant?
84
+ - What is the current helmet compliance ratio?
85
+ - What is the current safety vest compliance ratio?
86
+ - Which PPE detections could not be assigned to a worker?
87
+
88
+ ## Tech Stack
89
+
90
+ - Python
91
+ - Gradio
92
+ - YOLO
93
+ - OpenCV
94
+ - Pillow
95
+ - NumPy
96
+ - Hugging Face Hub
97
+ - Hugging Face Spaces
98
+
99
+ ## Models
100
+
101
+ Person detection:
102
+
103
+ - Model: `yolov8n.pt`
104
+ - Source: Ultralytics COCO pretrained model
105
+ - Class used: `person`
106
+
107
+ PPE detection:
108
+
109
+ - Model repository: [`Hexmon/vyra-yolo-ppe-detection`](https://huggingface.co/Hexmon/vyra-yolo-ppe-detection)
110
+ - Weight file: `best.pt`
111
+ - Key classes include helmet / hardhat and safety vest labels
112
+
113
+ The PPE model is downloaded at runtime with `huggingface_hub`. The COCO person model is loaded by Ultralytics at runtime, so large model files do not need to be committed to this repository.
114
+
115
+ ## How to Run
116
+
117
+ ```bash
118
+ pip install -r requirements.txt
119
+ python app.py
120
+ ```
121
+
122
+ Open the local Gradio URL shown in the terminal, upload an image, and click **Run Site Safety Check**.
123
+
124
+ ## Hugging Face Spaces
125
+
126
+ This repository is ready for Hugging Face Spaces.
127
+
128
+ 1. Create a new Space.
129
+ 2. Select **Gradio** as the SDK.
130
+ 3. Upload this repository.
131
+ 4. The Space will run `app.py` automatically.
132
+
133
+ ## Accuracy Notes
134
+
135
+ Detection quality depends on image quality, lighting, occlusion, worker distance, camera angle, and the model training data. For better results:
136
+
137
+ - Use clear images where workers, helmets, and vests are visible.
138
+ - Avoid images where workers are extremely small.
139
+ - Lower confidence thresholds if valid detections are missed.
140
+ - Raise confidence thresholds if false detections are frequent.
141
+ - Treat this project as a prototype, not as the only safety-control mechanism in a real workplace.
142
+
143
+ ## File Structure
144
+
145
+ ```text
146
+ AI-PPE-Detection-System/
147
+ +-- app.py
148
+ +-- requirements.txt
149
+ +-- README.md
150
+ +-- .gitignore
151
+ +-- sample_images/
152
+ +-- README.md
153
+ ```
154
+
155
+ ## Portfolio Point
156
+
157
+ This project demonstrates practical AI implementation for industrial safety monitoring, combining object detection, rule-based safety judgment, and user-friendly visualization.
158
+
159
+ It shows how YOLO-based detection can be turned into a workplace-oriented AI safety monitoring prototype for factories, warehouses, and construction sites.
160
+
161
+ ## Contact
162
+
163
+ Consultation Form: https://forms.gle/SaWGZFu8J7DgbytL7
164
+
165
+ ## Follow My Work
166
+
167
+ - GitHub: https://github.com/futurecortexlabs
168
+ - note: https://note.com/future_cortex
169
+ - Hugging Face: https://huggingface.co/FCTX
app.py ADDED
@@ -0,0 +1,607 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from dataclasses import dataclass
2
+ from functools import lru_cache
3
+ from inspect import signature
4
+ from pathlib import Path
5
+
6
+ import gradio as gr
7
+ import numpy as np
8
+ from huggingface_hub import hf_hub_download
9
+ from PIL import Image, ImageDraw
10
+ from ultralytics import YOLO
11
+
12
+
13
+ APP_NAME = "AI-PPE-Detection-System"
14
+ PPE_MODEL_REPO = "Hexmon/vyra-yolo-ppe-detection"
15
+ PPE_MODEL_FILE = "best.pt"
16
+ PERSON_MODEL_NAME = "yolov8n.pt"
17
+
18
+ PERSON_CONFIDENCE_THRESHOLD = 0.40
19
+ PPE_CONFIDENCE_THRESHOLD = 0.50
20
+ IOU_THRESHOLD = 0.45
21
+ IMAGE_SIZE = 960
22
+ PPE_PERSON_OVERLAP_THRESHOLD = 0.20
23
+
24
+ HELMET_LABELS = {"helmet", "hardhat", "hard hat", "safety helmet", "with helmet"}
25
+ VEST_LABELS = {"vest", "safety vest", "with vest", "reflective vest", "safety jacket"}
26
+ NO_HELMET_LABELS = {"no helmet", "no hardhat", "no hard hat", "without helmet"}
27
+ NO_VEST_LABELS = {"no vest", "no safety vest", "without vest"}
28
+ SUPPORTED_PPE_LABELS = HELMET_LABELS | VEST_LABELS | NO_HELMET_LABELS | NO_VEST_LABELS
29
+
30
+ STATUS_OK = "OK"
31
+ STATUS_NG = "NG"
32
+ STATUS_UNKNOWN = "Unknown"
33
+
34
+
35
+ @dataclass(frozen=True)
36
+ class DetectionBox:
37
+ x1: float
38
+ y1: float
39
+ x2: float
40
+ y2: float
41
+ confidence: float
42
+ label: str
43
+ category: str
44
+
45
+ @property
46
+ def area(self) -> float:
47
+ return max(0.0, self.x2 - self.x1) * max(0.0, self.y2 - self.y1)
48
+
49
+ @property
50
+ def center(self) -> tuple[float, float]:
51
+ return (self.x1 + self.x2) / 2, (self.y1 + self.y2) / 2
52
+
53
+
54
+ @dataclass(frozen=True)
55
+ class WorkerAssessment:
56
+ worker_id: int
57
+ worker_box: DetectionBox
58
+ helmet_status: str
59
+ vest_status: str
60
+ helmet_confidence: float | None
61
+ vest_confidence: float | None
62
+
63
+ @property
64
+ def overall_status(self) -> str:
65
+ return STATUS_OK if self.helmet_status == STATUS_OK and self.vest_status == STATUS_OK else STATUS_NG
66
+
67
+ @property
68
+ def missing_items(self) -> str:
69
+ missing = []
70
+ if self.helmet_status != STATUS_OK:
71
+ missing.append("Helmet")
72
+ if self.vest_status != STATUS_OK:
73
+ missing.append("Safety Vest")
74
+ return ", ".join(missing) if missing else "None"
75
+
76
+
77
+ @lru_cache(maxsize=1)
78
+ def load_ppe_model() -> YOLO:
79
+ model_path = hf_hub_download(repo_id=PPE_MODEL_REPO, filename=PPE_MODEL_FILE)
80
+ return YOLO(model_path)
81
+
82
+
83
+ @lru_cache(maxsize=1)
84
+ def load_person_model() -> YOLO:
85
+ return YOLO(PERSON_MODEL_NAME)
86
+
87
+
88
+ def normalize_label(label: str) -> str:
89
+ return label.strip().lower().replace("_", " ").replace("-", " ")
90
+
91
+
92
+ def classify_ppe_label(label: str) -> str:
93
+ normalized = normalize_label(label)
94
+ if normalized in HELMET_LABELS:
95
+ return "helmet"
96
+ if normalized in VEST_LABELS:
97
+ return "vest"
98
+ if normalized in NO_HELMET_LABELS:
99
+ return "no_helmet"
100
+ if normalized in NO_VEST_LABELS:
101
+ return "no_vest"
102
+ return "other"
103
+
104
+
105
+ def intersection_area(first: DetectionBox, second: DetectionBox) -> float:
106
+ x1 = max(first.x1, second.x1)
107
+ y1 = max(first.y1, second.y1)
108
+ x2 = min(first.x2, second.x2)
109
+ y2 = min(first.y2, second.y2)
110
+ return max(0.0, x2 - x1) * max(0.0, y2 - y1)
111
+
112
+
113
+ def point_in_box(point: tuple[float, float], box: DetectionBox, margin_ratio: float = 0.08) -> bool:
114
+ center_x, center_y = point
115
+ width = box.x2 - box.x1
116
+ height = box.y2 - box.y1
117
+ margin_x = width * margin_ratio
118
+ margin_y = height * margin_ratio
119
+
120
+ return (
121
+ box.x1 - margin_x <= center_x <= box.x2 + margin_x
122
+ and box.y1 - margin_y <= center_y <= box.y2 + margin_y
123
+ )
124
+
125
+
126
+ def person_association_score(ppe_box: DetectionBox, worker_box: DetectionBox) -> float:
127
+ overlap_ratio = intersection_area(ppe_box, worker_box) / max(ppe_box.area, 1.0)
128
+ center_bonus = 1.0 if point_in_box(ppe_box.center, worker_box) else 0.0
129
+ return overlap_ratio + center_bonus
130
+
131
+
132
+ def assign_ppe_to_workers(
133
+ workers: list[DetectionBox],
134
+ ppe_boxes: list[DetectionBox],
135
+ ) -> dict[int, list[DetectionBox]]:
136
+ assignments = {index: [] for index in range(len(workers))}
137
+
138
+ for ppe_box in ppe_boxes:
139
+ scored_workers = [
140
+ (index, person_association_score(ppe_box, worker_box))
141
+ for index, worker_box in enumerate(workers)
142
+ ]
143
+ if not scored_workers:
144
+ continue
145
+
146
+ best_worker_index, best_score = max(scored_workers, key=lambda item: item[1])
147
+ overlap_ratio = intersection_area(ppe_box, workers[best_worker_index]) / max(ppe_box.area, 1.0)
148
+ if best_score >= 1.0 or overlap_ratio >= PPE_PERSON_OVERLAP_THRESHOLD:
149
+ assignments[best_worker_index].append(ppe_box)
150
+
151
+ return assignments
152
+
153
+
154
+ def build_ppe_assignment_lookup(worker_ppe: dict[int, list[DetectionBox]]) -> dict[DetectionBox, int]:
155
+ return {
156
+ ppe_box: worker_index
157
+ for worker_index, assigned_boxes in worker_ppe.items()
158
+ for ppe_box in assigned_boxes
159
+ }
160
+
161
+
162
+ def pick_best_confidence(boxes: list[DetectionBox], categories: set[str]) -> float | None:
163
+ candidates = [box.confidence for box in boxes if box.category in categories]
164
+ return max(candidates) if candidates else None
165
+
166
+
167
+ def assess_workers(
168
+ workers: list[DetectionBox],
169
+ worker_ppe: dict[int, list[DetectionBox]],
170
+ ) -> list[WorkerAssessment]:
171
+ assessments = []
172
+ for index, worker_box in enumerate(workers):
173
+ assigned_boxes = worker_ppe.get(index, [])
174
+
175
+ helmet_confidence = pick_best_confidence(assigned_boxes, {"helmet"})
176
+ vest_confidence = pick_best_confidence(assigned_boxes, {"vest"})
177
+ no_helmet_confidence = pick_best_confidence(assigned_boxes, {"no_helmet"})
178
+ no_vest_confidence = pick_best_confidence(assigned_boxes, {"no_vest"})
179
+
180
+ helmet_status = STATUS_OK if helmet_confidence is not None else STATUS_NG
181
+ vest_status = STATUS_OK if vest_confidence is not None else STATUS_NG
182
+
183
+ if no_helmet_confidence is not None and (helmet_confidence is None or no_helmet_confidence > helmet_confidence):
184
+ helmet_status = STATUS_NG
185
+ if no_vest_confidence is not None and (vest_confidence is None or no_vest_confidence > vest_confidence):
186
+ vest_status = STATUS_NG
187
+
188
+ assessments.append(
189
+ WorkerAssessment(
190
+ worker_id=index + 1,
191
+ worker_box=worker_box,
192
+ helmet_status=helmet_status,
193
+ vest_status=vest_status,
194
+ helmet_confidence=helmet_confidence,
195
+ vest_confidence=vest_confidence,
196
+ )
197
+ )
198
+
199
+ return assessments
200
+
201
+
202
+ def extract_person_boxes(person_result) -> list[DetectionBox]:
203
+ boxes = []
204
+ for box in person_result.boxes:
205
+ x1, y1, x2, y2 = box.xyxy[0].tolist()
206
+ boxes.append(DetectionBox(x1, y1, x2, y2, float(box.conf[0]), "person", "worker"))
207
+ return boxes
208
+
209
+
210
+ def extract_ppe_boxes(ppe_result) -> list[DetectionBox]:
211
+ boxes = []
212
+ for box in ppe_result.boxes:
213
+ class_id = int(box.cls[0])
214
+ label = normalize_label(ppe_result.names[class_id])
215
+ if label not in SUPPORTED_PPE_LABELS:
216
+ continue
217
+ x1, y1, x2, y2 = box.xyxy[0].tolist()
218
+ boxes.append(DetectionBox(x1, y1, x2, y2, float(box.conf[0]), label, classify_ppe_label(label)))
219
+ return boxes
220
+
221
+
222
+ def format_confidence(confidence: float | None) -> str:
223
+ return f"{confidence:.2f}" if confidence is not None else "-"
224
+
225
+
226
+ def build_ppe_markdown(assessments: list[WorkerAssessment], ppe_boxes: list[DetectionBox]) -> str:
227
+ detected_helmet = any(box.category == "helmet" for box in ppe_boxes)
228
+ detected_vest = any(box.category == "vest" for box in ppe_boxes)
229
+ all_workers_ok = bool(assessments) and all(worker.overall_status == STATUS_OK for worker in assessments)
230
+
231
+ return f"""
232
+ ### PPE Check Result
233
+
234
+ | Item | Status |
235
+ |---|---|
236
+ | Helmet | {STATUS_OK if detected_helmet else STATUS_NG} |
237
+ | Safety Vest | {STATUS_OK if detected_vest else STATUS_NG} |
238
+ | Overall Result | {STATUS_OK if all_workers_ok else STATUS_NG} |
239
+ """
240
+
241
+
242
+ def build_worker_summary_markdown(assessments: list[WorkerAssessment]) -> str:
243
+ worker_count = len(assessments)
244
+ if worker_count == 0:
245
+ return """
246
+ ### Worker Safety Summary
247
+
248
+ | Metric | Value |
249
+ |---|---|
250
+ | Workers | 0 |
251
+ | Fully Compliant Workers | No workers detected |
252
+ | Helmet Compliance | No workers detected |
253
+ | Vest Compliance | No workers detected |
254
+ """
255
+
256
+ helmet_ok_count = sum(worker.helmet_status == STATUS_OK for worker in assessments)
257
+ vest_ok_count = sum(worker.vest_status == STATUS_OK for worker in assessments)
258
+ fully_compliant_count = sum(worker.overall_status == STATUS_OK for worker in assessments)
259
+
260
+ return f"""
261
+ ### Worker Safety Summary
262
+
263
+ | Metric | Value |
264
+ |---|---|
265
+ | Workers | {worker_count} |
266
+ | Fully Compliant Workers | {fully_compliant_count}/{worker_count} |
267
+ | Helmet Compliance | {helmet_ok_count}/{worker_count} |
268
+ | Vest Compliance | {vest_ok_count}/{worker_count} |
269
+ """
270
+
271
+
272
+ def build_status_markdown(assessments: list[WorkerAssessment]) -> str:
273
+ if not assessments:
274
+ return """
275
+ ### Site Safety Status
276
+
277
+ **NG - No workers detected**
278
+
279
+ Upload a clearer image or lower the worker confidence threshold.
280
+ """
281
+
282
+ non_compliant = [worker for worker in assessments if worker.overall_status == STATUS_NG]
283
+ if not non_compliant:
284
+ return """
285
+ ### Site Safety Status
286
+
287
+ **OK - All detected workers are PPE compliant**
288
+
289
+ Every detected worker has both helmet and safety vest evidence.
290
+ """
291
+
292
+ missing_summary = "; ".join(
293
+ f"Worker {worker.worker_id}: {worker.missing_items}" for worker in non_compliant
294
+ )
295
+ return f"""
296
+ ### Site Safety Status
297
+
298
+ **NG - {len(non_compliant)} worker(s) need attention**
299
+
300
+ {missing_summary}
301
+ """
302
+
303
+
304
+ def build_worker_table(assessments: list[WorkerAssessment]) -> list[list[str]]:
305
+ return [
306
+ [
307
+ f"Worker {worker.worker_id}",
308
+ worker.helmet_status,
309
+ worker.vest_status,
310
+ worker.overall_status,
311
+ worker.missing_items,
312
+ format_confidence(worker.worker_box.confidence),
313
+ format_confidence(worker.helmet_confidence),
314
+ format_confidence(worker.vest_confidence),
315
+ ]
316
+ for worker in assessments
317
+ ]
318
+
319
+
320
+ def build_detection_table(
321
+ workers: list[DetectionBox],
322
+ ppe_boxes: list[DetectionBox],
323
+ ppe_assignment: dict[DetectionBox, int],
324
+ ) -> list[list[str]]:
325
+ rows = [
326
+ [
327
+ "worker",
328
+ box.label,
329
+ format_confidence(box.confidence),
330
+ "-",
331
+ f"{int(box.x1)}, {int(box.y1)}, {int(box.x2)}, {int(box.y2)}",
332
+ ]
333
+ for box in workers
334
+ ]
335
+ rows.extend(
336
+ [
337
+ box.category,
338
+ box.label,
339
+ format_confidence(box.confidence),
340
+ f"Worker {ppe_assignment[box] + 1}" if box in ppe_assignment else "Unassigned",
341
+ f"{int(box.x1)}, {int(box.y1)}, {int(box.x2)}, {int(box.y2)}",
342
+ ]
343
+ for box in ppe_boxes
344
+ )
345
+ return rows
346
+
347
+
348
+ def draw_label(draw: ImageDraw.ImageDraw, x: int, y: int, label: str, fill: tuple[int, int, int]) -> None:
349
+ text_bbox = draw.textbbox((x, y), label)
350
+ text_width = text_bbox[2] - text_bbox[0]
351
+ text_height = text_bbox[3] - text_bbox[1]
352
+ label_y = max(0, y - text_height - 8)
353
+ draw.rectangle((x, label_y, x + text_width + 8, y), fill=fill)
354
+ draw.text((x + 4, label_y + 2), label, fill=(255, 255, 255))
355
+
356
+
357
+ def draw_detection_boxes(
358
+ image: Image.Image,
359
+ workers: list[DetectionBox],
360
+ ppe_boxes: list[DetectionBox],
361
+ assessments: list[WorkerAssessment],
362
+ ppe_assignment: dict[DetectionBox, int],
363
+ ) -> Image.Image:
364
+ annotated_image = image.copy()
365
+ draw = ImageDraw.Draw(annotated_image)
366
+ assessment_by_id = {assessment.worker_id: assessment for assessment in assessments}
367
+
368
+ for index, worker_box in enumerate(workers, start=1):
369
+ assessment = assessment_by_id.get(index)
370
+ is_ok = assessment is not None and assessment.overall_status == STATUS_OK
371
+ color = (30, 150, 85) if is_ok else (220, 80, 60)
372
+ x1, y1, x2, y2 = map(int, (worker_box.x1, worker_box.y1, worker_box.x2, worker_box.y2))
373
+ draw.rectangle((x1, y1, x2, y2), outline=color, width=4)
374
+ status = assessment.overall_status if assessment else STATUS_UNKNOWN
375
+ draw_label(draw, x1, y1, f"Worker {index}: {status}", color)
376
+
377
+ for box in ppe_boxes:
378
+ if box not in ppe_assignment:
379
+ color = (100, 116, 139)
380
+ elif box.category in {"helmet", "vest"}:
381
+ color = (35, 160, 80)
382
+ elif box.category in {"no_helmet", "no_vest"}:
383
+ color = (220, 80, 60)
384
+ else:
385
+ color = (90, 100, 120)
386
+
387
+ x1, y1, x2, y2 = map(int, (box.x1, box.y1, box.x2, box.y2))
388
+ draw.rectangle((x1, y1, x2, y2), outline=color, width=3)
389
+ draw_label(draw, x1, y1, f"{box.label} {box.confidence:.2f}", color)
390
+
391
+ return annotated_image
392
+
393
+
394
+ def empty_outputs(message: str) -> tuple[Image.Image | None, str, str, list[list[str]], list[list[str]], str]:
395
+ return None, message, "", [], [], ""
396
+
397
+
398
+ def detect_ppe(
399
+ image: Image.Image,
400
+ person_confidence: float,
401
+ ppe_confidence: float,
402
+ iou_threshold: float,
403
+ ) -> tuple[Image.Image | None, str, str, list[list[str]], list[list[str]], str]:
404
+ if image is None:
405
+ return empty_outputs("### PPE Check Result\n\nPlease upload an image.")
406
+
407
+ try:
408
+ ppe_model = load_ppe_model()
409
+ person_model = load_person_model()
410
+ except Exception as exc:
411
+ return (
412
+ image,
413
+ "### Model Loading Error\n\n"
414
+ "The detection model could not be loaded. Check the Hugging Face Space logs, "
415
+ "network access, or model repository settings.\n\n"
416
+ f"`{type(exc).__name__}: {exc}`",
417
+ "",
418
+ [],
419
+ [],
420
+ "",
421
+ )
422
+
423
+ try:
424
+ rgb_image = image.convert("RGB")
425
+ image_array = np.array(rgb_image)
426
+
427
+ person_results = person_model.predict(
428
+ source=image_array,
429
+ conf=person_confidence,
430
+ iou=iou_threshold,
431
+ imgsz=IMAGE_SIZE,
432
+ classes=[0],
433
+ verbose=False,
434
+ )
435
+ ppe_results = ppe_model.predict(
436
+ source=image_array,
437
+ conf=ppe_confidence,
438
+ iou=iou_threshold,
439
+ imgsz=IMAGE_SIZE,
440
+ verbose=False,
441
+ )
442
+
443
+ workers = extract_person_boxes(person_results[0])
444
+ ppe_boxes = extract_ppe_boxes(ppe_results[0])
445
+ worker_ppe = assign_ppe_to_workers(workers, ppe_boxes)
446
+ ppe_assignment = build_ppe_assignment_lookup(worker_ppe)
447
+ associated_ppe_boxes = sorted(
448
+ {box for assigned_boxes in worker_ppe.values() for box in assigned_boxes},
449
+ key=lambda box: (box.y1, box.x1),
450
+ )
451
+ assessments = assess_workers(workers, worker_ppe)
452
+
453
+ annotated_image = draw_detection_boxes(rgb_image, workers, ppe_boxes, assessments, ppe_assignment)
454
+ dashboard = "\n".join(
455
+ [
456
+ build_status_markdown(assessments),
457
+ build_ppe_markdown(assessments, associated_ppe_boxes),
458
+ build_worker_summary_markdown(assessments),
459
+ ]
460
+ )
461
+
462
+ return (
463
+ annotated_image,
464
+ dashboard,
465
+ "### Per-Worker PPE Assessment",
466
+ build_worker_table(assessments),
467
+ build_detection_table(workers, ppe_boxes, ppe_assignment),
468
+ f"Processed {len(workers)} worker(s), {len(associated_ppe_boxes)} associated PPE detection(s), and {len(ppe_boxes) - len(associated_ppe_boxes)} unassigned PPE detection(s).",
469
+ )
470
+ except Exception as exc:
471
+ return (
472
+ image,
473
+ "### Detection Error\n\n"
474
+ "The image could not be processed. Try another image or adjust the thresholds.\n\n"
475
+ f"`{type(exc).__name__}: {exc}`",
476
+ "",
477
+ [],
478
+ [],
479
+ "",
480
+ )
481
+
482
+
483
+ def get_example_images() -> list[str]:
484
+ sample_dir = Path("sample_images")
485
+ if not sample_dir.exists():
486
+ return []
487
+ return [
488
+ str(path)
489
+ for path in sample_dir.iterdir()
490
+ if path.suffix.lower() in {".jpg", ".jpeg", ".png", ".webp"}
491
+ ]
492
+
493
+
494
+ CUSTOM_CSS = """
495
+ .gradio-container {
496
+ max-width: 1240px !important;
497
+ }
498
+ .app-subtitle {
499
+ color: #475569;
500
+ font-size: 1.03rem;
501
+ line-height: 1.6;
502
+ }
503
+ .status-note textarea {
504
+ font-weight: 650;
505
+ }
506
+ """
507
+
508
+ THEME = gr.themes.Soft(primary_hue="blue", neutral_hue="slate")
509
+ BLOCKS_KWARGS = {"title": APP_NAME}
510
+ LAUNCH_KWARGS = {}
511
+
512
+ if "theme" in signature(gr.Blocks).parameters:
513
+ BLOCKS_KWARGS.update({"theme": THEME, "css": CUSTOM_CSS})
514
+ else:
515
+ LAUNCH_KWARGS.update({"theme": THEME, "css": CUSTOM_CSS})
516
+
517
+
518
+ with gr.Blocks(**BLOCKS_KWARGS) as demo:
519
+ gr.Markdown(
520
+ f"""
521
+ # {APP_NAME}
522
+
523
+ <div class="app-subtitle">
524
+ Industrial safety monitoring prototype for factories, warehouses, and construction sites.
525
+ Upload a workplace image to detect workers, associate PPE with each worker, and generate
526
+ an operational OK / NG safety summary.
527
+ </div>
528
+ """
529
+ )
530
+
531
+ with gr.Row():
532
+ with gr.Column(scale=5):
533
+ input_image = gr.Image(
534
+ type="pil",
535
+ label="Upload workplace safety image",
536
+ sources=["upload"],
537
+ height=430,
538
+ )
539
+ with gr.Row():
540
+ person_confidence_slider = gr.Slider(
541
+ minimum=0.1,
542
+ maximum=0.9,
543
+ value=PERSON_CONFIDENCE_THRESHOLD,
544
+ step=0.05,
545
+ label="Worker confidence threshold",
546
+ )
547
+ ppe_confidence_slider = gr.Slider(
548
+ minimum=0.1,
549
+ maximum=0.9,
550
+ value=PPE_CONFIDENCE_THRESHOLD,
551
+ step=0.05,
552
+ label="PPE confidence threshold",
553
+ )
554
+ iou_slider = gr.Slider(
555
+ minimum=0.1,
556
+ maximum=0.9,
557
+ value=IOU_THRESHOLD,
558
+ step=0.05,
559
+ label="IoU threshold",
560
+ )
561
+ detect_button = gr.Button("Run Site Safety Check", variant="primary", size="lg")
562
+
563
+ with gr.Column(scale=7):
564
+ output_image = gr.Image(label="Annotated detection result", height=430)
565
+ status_note = gr.Textbox(label="Processing note", lines=1, elem_classes=["status-note"])
566
+
567
+ with gr.Tabs():
568
+ with gr.Tab("Safety Dashboard"):
569
+ dashboard_markdown = gr.Markdown()
570
+ with gr.Tab("Worker Assessment"):
571
+ worker_title = gr.Markdown("### Per-Worker PPE Assessment")
572
+ worker_table = gr.Dataframe(
573
+ headers=[
574
+ "Worker",
575
+ "Helmet",
576
+ "Safety Vest",
577
+ "Overall",
578
+ "Missing Items",
579
+ "Worker Conf.",
580
+ "Helmet Conf.",
581
+ "Vest Conf.",
582
+ ],
583
+ datatype=["str", "str", "str", "str", "str", "str", "str", "str"],
584
+ interactive=False,
585
+ wrap=True,
586
+ )
587
+ with gr.Tab("Detection Details"):
588
+ detection_table = gr.Dataframe(
589
+ headers=["Type", "Label", "Confidence", "Assigned Worker", "Box"],
590
+ datatype=["str", "str", "str", "str", "str"],
591
+ interactive=False,
592
+ wrap=True,
593
+ )
594
+
595
+ examples = get_example_images()
596
+ if examples:
597
+ gr.Examples(examples=examples, inputs=input_image, label="Sample Images")
598
+
599
+ detect_button.click(
600
+ fn=detect_ppe,
601
+ inputs=[input_image, person_confidence_slider, ppe_confidence_slider, iou_slider],
602
+ outputs=[output_image, dashboard_markdown, worker_title, worker_table, detection_table, status_note],
603
+ )
604
+
605
+
606
+ if __name__ == "__main__":
607
+ demo.launch(**LAUNCH_KWARGS)
requirements.txt ADDED
@@ -0,0 +1,6 @@
 
 
 
 
 
 
 
1
+ gradio>=4.44.0
2
+ ultralytics>=8.3.0
3
+ opencv-python-headless>=4.10.0
4
+ huggingface_hub>=0.24.0
5
+ pillow>=10.0.0
6
+ numpy>=1.26.0
sample_images/README.md ADDED
@@ -0,0 +1,12 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ # Sample Images
2
+
3
+ Place optional demo images in this directory.
4
+
5
+ Recommended examples:
6
+
7
+ - A worker wearing both a helmet and safety vest
8
+ - A worker missing a helmet
9
+ - A worker missing a safety vest
10
+ - Multiple workers with mixed PPE compliance
11
+
12
+ Do not commit private workplace photos or images without permission.
sample_images/test.png ADDED

Git LFS Details

  • SHA256: 5e39a7b3100e4bd36d29bc8fdc9826cfa243c1629624d06c223cb875ddf287be
  • Pointer size: 132 Bytes
  • Size of remote file: 2.86 MB
yolov8n.pt ADDED
@@ -0,0 +1,3 @@
 
 
 
 
1
+ version https://git-lfs.github.com/spec/v1
2
+ oid sha256:f59b3d833e2ff32e194b5bb8e08d211dc7c5bdf144b90d2c8412c47ccfc83b36
3
+ size 6549796