Roboflow를 활용한 실시간 PPE 탐지 및 작업자 안전 모니터링 시스템 구축

def run(self, predictions): if predictions is None or not hasattr(predictions, 'data') or predictions.data is None: return {'safe_count': 0, 'unsafe_count': 0} class_names = predictions.data.get('class_name', []) if hasattr(predictions, 'xyxy'): bboxes = predictions.xyxy else: bboxes = [] if len(bboxes) == 0 or len(class_names) == 0: return {'safe_count': 0, 'unsafe_count': 0} persons = [] helmets = [] vests = [] for i in range(min(len(class_names), len(bboxes))): cls = str(class_names[i]).strip().lower() bbox = bboxes[i] if cls == 'person': persons.append(bbox) elif cls == 'helmet': helmets.append(bbox) elif cls == 'vest': vests.append(bbox) def center_in(boxA, boxB): cx = (boxA[0] + boxA[2]) / 2 cy = (boxA[1] + boxA[3]) / 2 return (boxB[0] <= cx <= boxB[2]) and (boxB[1] <= cy <= boxB[3]) safe = 0 unsafe = 0 for person in persons: has_helmet = any(center_in(h, person) for h in helmets) has_vest = any(center_in(v, person) for v in vests) if has_helmet and has_vest: safe += 1 else: unsafe += 1 return {'safe_count': safe, 'unsafe_count': unsafe}