Buckets:
| """ | |
| HuggingFace Inference Endpoint용 InsightFace 핸들러 | |
| 얼굴 임베딩 추출 및 유사도 계산용 | |
| """ | |
| import base64 | |
| import io | |
| from typing import Dict, Any, Optional | |
| from PIL import Image | |
| import numpy as np | |
| import insightface | |
| class EndpointHandler: | |
| def __init__(self, path=""): | |
| """모델 초기화""" | |
| self.face_analyzer = None | |
| self._load_models(path) | |
| def _load_models(self, path: str): | |
| """모델 로드""" | |
| try: | |
| # InsightFace FaceAnalysis 초기화 (얼굴 감지 및 임베딩 추출용) | |
| self.face_analyzer = insightface.app.FaceAnalysis( | |
| name='buffalo_l', | |
| root=path, # 모델 경로 | |
| providers=['CUDAExecutionProvider', 'CPUExecutionProvider'] | |
| ) | |
| self.face_analyzer.prepare(ctx_id=0, det_size=(640, 640)) | |
| print("✅ InsightFace 모델 로드 완료 (얼굴 임베딩 추출용)") | |
| except Exception as e: | |
| print(f"❌ 모델 로드 실패: {e}") | |
| raise | |
| def _base64_to_image(self, base64_str: str) -> Image.Image: | |
| """base64 문자열을 이미지로 변환""" | |
| if "," in base64_str: | |
| base64_str = base64_str.split(",")[1] | |
| img_bytes = base64.b64decode(base64_str) | |
| return Image.open(io.BytesIO(img_bytes)) | |
| def _image_to_base64(self, image: Image.Image) -> str: | |
| """이미지를 base64 문자열로 변환""" | |
| buffer = io.BytesIO() | |
| image.save(buffer, format="PNG") | |
| img_bytes = buffer.getvalue() | |
| return base64.b64encode(img_bytes).decode("utf-8") | |
| def _pil_to_numpy(self, image: Image.Image) -> np.ndarray: | |
| """PIL Image를 BGR numpy 배열로 변환""" | |
| return np.array(image.convert('RGB'))[:, :, ::-1] | |
| def _numpy_to_pil(self, image: np.ndarray) -> Image.Image: | |
| """BGR numpy 배열을 PIL Image로 변환""" | |
| rgb = image[:, :, ::-1] # BGR -> RGB | |
| return Image.fromarray(rgb) | |
| def face_detect(self, data: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| 얼굴 감지 | |
| Args: | |
| data: {"image": "data:image/png;base64,..."} | |
| Returns: | |
| {"faces": [{"bbox": [x1, y1, x2, y2], "embedding": [...], ...}]} | |
| """ | |
| try: | |
| image_b64 = data.get("image", "") | |
| if not image_b64: | |
| return {"error": "이미지가 제공되지 않았습니다"} | |
| # 이미지 변환 | |
| image = self._base64_to_image(image_b64) | |
| image_np = self._pil_to_numpy(image) | |
| # 얼굴 감지 | |
| faces = self.face_analyzer.get(image_np) | |
| # 결과 변환 | |
| faces_data = [] | |
| for face in faces: | |
| face_data = { | |
| "bbox": face.bbox.tolist(), | |
| "kps": face.kps.tolist() if hasattr(face, 'kps') and face.kps is not None else None, | |
| "embedding": face.embedding.tolist() if hasattr(face, 'embedding') and face.embedding is not None else None, | |
| "det_score": float(face.det_score) if hasattr(face, 'det_score') else None, | |
| "gender": int(face.gender) if hasattr(face, 'gender') else None, | |
| "age": int(face.age) if hasattr(face, 'age') else None, | |
| } | |
| faces_data.append(face_data) | |
| return {"faces": faces_data} | |
| except Exception as e: | |
| return {"error": str(e)} | |
| def face_similarity(self, data: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| 두 얼굴 간 유사도 계산 (코사인 유사도) | |
| Args: | |
| data: { | |
| "image1": "data:image/png;base64,...", | |
| "image2": "data:image/png;base64,...", | |
| "face_index1": 0, # image1에서 사용할 얼굴 인덱스 | |
| "face_index2": 0 # image2에서 사용할 얼굴 인덱스 | |
| } | |
| Returns: | |
| { | |
| "similarity": float, # 0.0 ~ 1.0 (1.0이 가장 유사) | |
| "embedding1": [...], # 첫 번째 얼굴 임베딩 | |
| "embedding2": [...], # 두 번째 얼굴 임베딩 | |
| "face1_info": {...}, # 첫 번째 얼굴 정보 | |
| "face2_info": {...} # 두 번째 얼굴 정보 | |
| } | |
| """ | |
| try: | |
| image1_b64 = data.get("image1", "") | |
| image2_b64 = data.get("image2", "") | |
| face_index1 = data.get("face_index1", 0) | |
| face_index2 = data.get("face_index2", 0) | |
| if not image1_b64 or not image2_b64: | |
| return {"error": "이미지1 또는 이미지2가 제공되지 않았습니다"} | |
| # 이미지 변환 | |
| image1 = self._base64_to_image(image1_b64) | |
| image2 = self._base64_to_image(image2_b64) | |
| image1_np = self._pil_to_numpy(image1) | |
| image2_np = self._pil_to_numpy(image2) | |
| # 얼굴 감지 | |
| faces1 = self.face_analyzer.get(image1_np) | |
| faces2 = self.face_analyzer.get(image2_np) | |
| if len(faces1) == 0: | |
| return {"error": "이미지1에서 얼굴을 찾을 수 없습니다"} | |
| if len(faces2) == 0: | |
| return {"error": "이미지2에서 얼굴을 찾을 수 없습니다"} | |
| # 얼굴 선택 | |
| if face_index1 >= len(faces1): | |
| face_index1 = 0 | |
| if face_index2 >= len(faces2): | |
| face_index2 = 0 | |
| face1 = faces1[face_index1] | |
| face2 = faces2[face_index2] | |
| # 임베딩 추출 | |
| if not hasattr(face1, 'embedding') or face1.embedding is None: | |
| return {"error": "이미지1의 얼굴에서 임베딩을 추출할 수 없습니다"} | |
| if not hasattr(face2, 'embedding') or face2.embedding is None: | |
| return {"error": "이미지2의 얼굴에서 임베딩을 추출할 수 없습니다"} | |
| emb1 = face1.embedding | |
| emb2 = face2.embedding | |
| # 코사인 유사도 계산 | |
| # similarity = dot(emb1, emb2) / (norm(emb1) * norm(emb2)) | |
| emb1_norm = np.linalg.norm(emb1) | |
| emb2_norm = np.linalg.norm(emb2) | |
| if emb1_norm == 0 or emb2_norm == 0: | |
| return {"error": "임베딩 벡터의 크기가 0입니다"} | |
| # 코사인 유사도 계산 | |
| # InsightFace 임베딩은 이미 L2 정규화되어 있으므로 내적만으로 유사도 계산 가능 | |
| similarity = np.dot(emb1, emb2) / (emb1_norm * emb2_norm) | |
| # 코사인 유사도는 -1 ~ 1 범위이지만, 얼굴 임베딩은 보통 0 ~ 1 범위 | |
| # 음수 값이 나올 수 있으므로 0 ~ 1로 정규화 (선택사항) | |
| # 실제로는 InsightFace 임베딩이 정규화되어 있어 0 ~ 1 범위가 일반적 | |
| similarity = max(0.0, min(1.0, similarity)) | |
| # 얼굴 정보 추출 | |
| face1_info = { | |
| "bbox": face1.bbox.tolist(), | |
| "det_score": float(face1.det_score) if hasattr(face1, 'det_score') else None, | |
| "gender": int(face1.gender) if hasattr(face1, 'gender') else None, | |
| "age": int(face1.age) if hasattr(face1, 'age') else None, | |
| } | |
| face2_info = { | |
| "bbox": face2.bbox.tolist(), | |
| "det_score": float(face2.det_score) if hasattr(face2, 'det_score') else None, | |
| "gender": int(face2.gender) if hasattr(face2, 'gender') else None, | |
| "age": int(face2.age) if hasattr(face2, 'age') else None, | |
| } | |
| return { | |
| "similarity": float(similarity), | |
| "embedding1": emb1.tolist(), | |
| "embedding2": emb2.tolist(), | |
| "face1_info": face1_info, | |
| "face2_info": face2_info | |
| } | |
| except Exception as e: | |
| return {"error": str(e)} | |
| def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| 메인 엔드포인트 | |
| Args: | |
| data: { | |
| "task": "face-detect" | "face-similarity", | |
| "image": "data:image/png;base64,..." (face-detect용), | |
| "image1": "data:image/png;base64,..." (face-similarity용), | |
| "image2": "data:image/png;base64,..." (face-similarity용), | |
| ... | |
| } | |
| Returns: | |
| 작업 결과 | |
| """ | |
| task = data.get("task", "") | |
| if task == "face-detect": | |
| return self.face_detect(data) | |
| elif task == "face-similarity": | |
| return self.face_similarity(data) | |
| else: | |
| return {"error": f"알 수 없는 작업: {task}. 지원 작업: 'face-detect', 'face-similarity'"} | |
Xet Storage Details
- Size:
- 9.47 kB
- Xet hash:
- 48b654163da68138986406ab4fd1646d92d0c2a41f8b0ca79d973163d60a163f
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.