Spaces:
Sleeping
Sleeping
| import cv2 as cv | |
| import numpy as np | |
| from PIL import Image | |
| import torch | |
| import pandas as pd | |
| from .setConfig import efficientnet_model, face_detector, transform, pca_xgb, faiss, load_db | |
| import os | |
| import requests | |
| def ImgPreprocessing(img): | |
| if len(img.shape) == 2 or img.shape[2] == 1: | |
| img = cv.cvtColor(img, cv.COLOR_GRAY2BGR) | |
| img_yuv = cv.cvtColor(img, cv.COLOR_BGR2YUV) | |
| clahe = cv.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8)) | |
| img_yuv[:, :, 0] = clahe.apply(img_yuv[:, :, 0]) | |
| img = cv.cvtColor(img_yuv, cv.COLOR_YUV2BGR) | |
| return img | |
| def YoloFaceDetection(img): | |
| results = face_detector.predict(img, conf=0.7) | |
| keyReturn = { | |
| 'message': "", | |
| 'status': False, | |
| 'coordinate': [] | |
| } | |
| if len(results) == 0 or results[0].boxes is None or len(results[0].boxes) == 0: | |
| keyReturn['message'] = "Tidak ada wajah terdeteksi" | |
| return keyReturn | |
| boxes = results[0].boxes.xyxy.cpu().numpy() | |
| confs = results[0].boxes.conf.cpu().numpy() | |
| max_idx = confs.argmax() | |
| x1, y1, x2, y2 = map(int, boxes[max_idx]) | |
| keyReturn['message'] = "Face detected" | |
| keyReturn['status'] = True | |
| keyReturn['coordinate'] = [x1, y1, x2, y2] | |
| return keyReturn | |
| def FaceValidationPredict(**face_crop): | |
| x1, y1, x2, y2 = face_crop['coordinate'] | |
| face_crop = face_crop['img'][y1:y2, x1:x2] | |
| face_crop = cv.cvtColor(face_crop, cv.COLOR_BGR2GRAY) | |
| face_crop = cv.cvtColor(face_crop, cv.COLOR_GRAY2RGB) | |
| face_pil = Image.fromarray(face_crop) | |
| face_tensor = transform(face_pil).unsqueeze(0) | |
| with torch.no_grad(): | |
| features = efficientnet_model(face_tensor).cpu().numpy() | |
| pred = pca_xgb.predict(features)[0] | |
| return pred, features, face_crop | |
| # feature | |
| def FaceValidation(frame: np.ndarray): | |
| if frame is None: | |
| return "No frame captured from webcam" | |
| if isinstance(frame, dict): | |
| frame = frame['image'] | |
| pred = '' | |
| img = ImgPreprocessing(frame) | |
| keyReturn = YoloFaceDetection(img) | |
| if keyReturn['status'] is False: | |
| return keyReturn['message'] | |
| results = keyReturn['coordinate'] | |
| x1, y1, x2, y2 = results | |
| validation, features, face_crop = FaceValidationPredict(coordinate=[x1, y1, x2, y2], img=img) | |
| pred = 'Wajah Valid' if validation == 0 else 'Wajah Tidak Valid' | |
| return f"Predicted class: {pred}" | |
| def FaceRecord(img, name, photo_idx): | |
| if img is None or name is None: | |
| return f"Foto {photo_idx}] Gagal: Tidak ada gambar atau nama" | |
| os.makedirs('users', exist_ok=True) | |
| user_dir = os.path.join("users", name) | |
| os.makedirs(user_dir, exist_ok=True) | |
| img = np.array(img) | |
| img = ImgPreprocessing(img) | |
| keyReturn = YoloFaceDetection(img) | |
| if keyReturn['status'] is False: | |
| return keyReturn['message'] | |
| x1, y1, x2, y2 = keyReturn['coordinate'] | |
| pred, features, face_crop = FaceValidationPredict(coordinate=[x1, y1, x2, y2], img=img) | |
| if pred == 1: | |
| return f"[Foto {photo_idx}] Gagal: wajah tidak valid" | |
| save_path = os.path.join(user_dir, f"photo_{photo_idx}.jpg") | |
| cv.imwrite(save_path, face_crop) | |
| csv_path = "users/face_features.csv" | |
| row = pd.DataFrame({ | |
| "label": [name], | |
| "features": [features.flatten().tolist()] | |
| }) | |
| if not os.path.exists(csv_path): | |
| row.to_csv(csv_path, index=False) | |
| else: | |
| row.to_csv(csv_path, mode="a", index=False, header=False) | |
| total_fotos = len([f for f in os.listdir(user_dir) if f.endswith(".jpg")]) | |
| if total_fotos < 4: | |
| return f"[Foto {photo_idx}] Berhasil disimpan ke {save_path}. ({total_fotos}/4)" | |
| else: | |
| return f"[Foto {photo_idx}] Berhasil disimpan ke {save_path}. β Semua 4 foto sudah lengkap!" | |
| def Recognize(frame: np.ndarray): | |
| if frame is None: | |
| return "No frame captured from webcam" | |
| if isinstance(frame, dict): | |
| frame = frame['image'] | |
| img = ImgPreprocessing(frame) | |
| keyReturn = YoloFaceDetection(img) | |
| if not keyReturn['status']: | |
| return keyReturn['message'] | |
| results = keyReturn['coordinate'] | |
| x1, y1, x2, y2 = results | |
| pred, features, face_crop = FaceValidationPredict(coordinate=[x1, y1, x2, y2], img=img) | |
| faiss.normalize_L2(features) | |
| if pred == 1: | |
| return 'Wajah tidak terdeteksi' | |
| faiss_index, labels, db = load_db("users/face_features.csv") | |
| if faiss_index is None: | |
| return "Database kosong" | |
| D, I = faiss_index.search(features, k=1) | |
| score = float(D[0][0]) | |
| idx = int(I[0][0]) | |
| if score < 0.7: | |
| return f"Tidak dikenali" | |
| else: | |
| return f"Terkenali sebagai: {labels[idx]} - (score={score:.2f})" | |
| def UploadVoice(audio_file): | |
| if audio_file is None: | |
| return "β Tidak ada file audio. Coba rekam lagi.", "", "" | |
| try: | |
| with open(audio_file, "rb") as f: | |
| response = requests.post( | |
| "https://n8n.smartid.co.id/webhook/voice-upload", | |
| data=f, | |
| headers={"Content-Type": "audio/wav"}, | |
| ) | |
| if response.status_code != 200: | |
| return f"β Gagal upload. Status: {response.status_code}", "", "" | |
| try: | |
| data = response.json() | |
| except Exception: | |
| return f"β Server tidak mengirim JSON. Balasan:\n{response}", "", "" | |
| transcription = data.get("transcribe", "(Tidak ada transkripsi)") | |
| summary = data.get("summary", "(Tidak ada ringkasan)") | |
| return "β Rekaman berhasil diproses", f"Transcribe: {transcription}", f"Summary: {summary}" | |
| except Exception as e: | |
| return f"β Gagal upload: {e}.", f"Transcribe: ", f"Summary: " |