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: "