Cleanup Data
Hapus data tidak terpakai atau sudah lama
🗑 Hapus Data Kosong
Hapus baris yang tidak punya wajah (fc=NULL) dan tidak punya RFID (rfid_uid=NULL). Baris ini tidak berfungsi untuk system.
-
Config ESP32
Atur koneksi WiFi dan Server untuk modul ESP32
⚙ ESP32-S3 CAMERA CONFIG
Pastikan laptop/PC terhubung ke WiFi ESP32 (AP Mode) atau masukkan IP ESP32 jika sudah terhubung ke jaringan.
↻ Reset Server Connection
Kirim perintah ke ESP32 untuk mereset status koneksi server jika ESP32 berhenti merespon (API FAIL).
-
"""
@app.get("/", response_class=HTMLResponse)
def dashboard():
return _html_page("ESP32 Face System", _DASHBOARD_BODY)
@app.get("/debug/recognize_hits")
def debug_recognize_hits():
return {"ok": True, "hits": _recognize_hits}
@app.get("/job/queue")
def job_queue():
return {"ok": True, "pending": _pending_jobs, "count": len(_pending_jobs)}
@app.post("/arm/enroll")
def arm_enroll(
name: str = Query(..., min_length=1, max_length=255),
uid: str | None = Query(None, min_length=1, max_length=64),
):
"""Set identity for the next capture-enroll."""
global _armed_enroll
val = name.strip()
uid_norm = uid.strip().upper() if uid else None
is_uid = _looks_like_rfid_uid(val)
parsed_uid = uid_norm or (val.upper() if is_uid else None)
parsed_name = None if (is_uid and parsed_uid == val.upper() and not uid_norm) else val
row_id = None
conn = get_connection_from_env()
ensure_data_table(conn)
try:
if parsed_name:
row_id = get_or_create_data_row(conn, parsed_name)
if parsed_uid:
upsert_person_rfid(conn, parsed_uid, parsed_name)
except Exception:
pass
finally:
conn.close()
_armed_enroll = {
"uid": parsed_uid,
"name": parsed_name if parsed_name else parsed_uid,
"row_id": row_id,
"armed_at": time.time(),
}
return {"ok": True, "armed": _armed_enroll}
@app.delete("/arm/enroll")
def clear_arm_enroll():
global _armed_enroll
_armed_enroll = None
return {"ok": True}
@app.get("/arm/enroll")
def get_arm_enroll():
if not _armed_enroll:
return {"ok": True, "armed": False, "name": None}
return {"ok": True, "armed": True,
"name": _armed_enroll.get("name"),
"armed_at": _armed_enroll.get("armed_at")}
@app.post("/job/enroll")
def job_enroll(name: str = Query(..., min_length=1, max_length=255)):
job = _enqueue_job("ENROLL", name=name.strip())
return {"ok": True, "job": job}
@app.post("/job/capture_enroll")
def job_capture_enroll():
if not _armed_enroll or not _armed_enroll.get("name"):
raise HTTPException(status_code=409, detail="No armed enroll identity. Call /arm/enroll first")
job = _enqueue_job("ENROLL", name=str(_armed_enroll["name"]))
if _armed_enroll.get("uid"):
job["uid"] = str(_armed_enroll["uid"])
if _armed_enroll.get("row_id"):
job["row_id"] = int(_armed_enroll["row_id"])
job["armed"] = True
return {"ok": True, "job": job, "armed": _armed_enroll}
@app.post("/job/recog")
def job_recog(
threshold: float = Query(0.25, ge=0.0, le=1.0),
uid: str | None = Query(None, min_length=1, max_length=64),
):
job = _enqueue_job("RECOG", name=None)
job["threshold"] = float(threshold)
if uid and uid.strip():
job["uid"] = uid.strip().upper()
return {"ok": True, "job": job}
@app.get("/job/next")
def job_next(device: str | None = Query(None)):
global _pending_jobs
if not _pending_jobs:
return {"ok": True, "job": None}
job = _pending_jobs.pop(0)
job["taken_at"] = time.time()
if device:
job["device"] = device
print(f"[JOB] device={device} take id={job.get('id')} type={job.get('type')}")
return {"ok": True, "job": job}
@app.post("/job/result")
async def job_result(request: Request):
global _last_result
try:
data = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Invalid JSON")
data["received_at"] = time.time()
_last_result = data
print(f"[JOB] result from={data.get('from')} type={data.get('type')} name={data.get('name')}")
return {"ok": True}
@app.get("/job/last")
def job_last():
return {"ok": True, "result": _last_result}
# ============================================================
# FACE HELPERS
# ============================================================
def _parse_embedding(obj) -> np.ndarray:
if obj is None:
raise ValueError("embedding is None")
emb = obj
if isinstance(emb, dict) and "embedding" in emb:
emb = emb["embedding"]
if isinstance(emb, list) and len(emb) > 0 and isinstance(emb[0], dict) and "embedding" in emb[0]:
emb = emb[0]["embedding"]
if isinstance(emb, dict) and "embedding" in emb:
emb = emb["embedding"]
arr = np.array(emb).astype(float).reshape(-1)
if arr.size == 0:
raise ValueError("empty embedding")
return arr
def _cosine_distance(a: np.ndarray, b: np.ndarray) -> float:
a_n = a / (np.linalg.norm(a) + 1e-10)
b_n = b / (np.linalg.norm(b) + 1e-10)
return 1.0 - float(np.dot(a_n, b_n))
def _crop_first_face_bgr(img_bgr: np.ndarray):
gray = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2GRAY)
face_cascade = cv2.CascadeClassifier(CASCADE_PATH)
faces = face_cascade.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=5, minSize=(60, 60))
if len(faces) == 0:
return None
x, y, w, h = faces[0]
crop = img_bgr[y:y+h, x:x+w]
pad = int(0.25 * max(w, h))
padded = cv2.copyMakeBorder(crop, pad, pad, pad, pad, borderType=cv2.BORDER_CONSTANT, value=[0,0,0])
return cv2.resize(padded, (224, 224), interpolation=cv2.INTER_AREA)
def _represent_from_bgr_face(face_bgr_224: np.ndarray) -> np.ndarray:
ts = int(time.time() * 1000)
tmp = os.path.join(tempfile.gettempdir(), f"esp32_face_{ts}.jpg")
cv2.imwrite(tmp, face_bgr_224)
try:
try:
rep = DeepFace.represent(tmp, model_name=DEFAULT_MODEL_NAME, enforce_detection=True)
except Exception:
rep = DeepFace.represent(tmp, model_name=DEFAULT_MODEL_NAME,
enforce_detection=False, detector_backend="opencv")
return _parse_embedding(rep)
finally:
try:
os.remove(tmp)
except Exception:
pass
def _load_gallery_from_db() -> dict:
"""Load all face embeddings from data_fcgntion into memory."""
conn = get_connection_from_env()
ensure_data_table(conn)
rows = load_all_faces(conn)
conn.close()
gallery = {}
for row in rows:
name = row.get("name")
emb_raw = row.get("embedding")
if isinstance(emb_raw, str):
try:
emb_raw = json.loads(emb_raw)
except Exception:
pass
try:
gallery[name] = _parse_embedding(emb_raw)
except Exception:
continue
return gallery
def _build_recognition_response(best_name, best_score, threshold, uid=None, second_score=None) -> dict:
match = is_recognized_match(best_score, threshold, second_score=second_score)
recognized_name = best_name if match else "Unknown"
result = {
"ok": True,
"name": recognized_name,
"score": float(best_score) if best_score is not None else None,
"threshold": float(threshold),
"match": bool(match),
"best": best_name,
}
if uid and uid.strip() and match:
uid_norm = uid.strip().upper()
result["uid"] = uid_norm
conn = get_connection_from_env()
ensure_data_table(conn)
try:
log_attendance(conn, uid=uid_norm, recognized_name=recognized_name, face_ok=True)
finally:
conn.close()
if match and best_name:
print(f"[SUPABASE] Wajah dikenali: '{best_name}' - push ke Supabase...")
_push_to_supabase_async(best_name)
return result
def _recognize_embedding(q: np.ndarray, threshold: float, uid=None) -> dict:
gallery = _load_gallery_from_db()
if not gallery:
uid_norm = uid.strip().upper() if uid and uid.strip() else None
resp = {"ok": True, "name": "Unknown", "score": None,
"reason": "empty_gallery", "threshold": float(threshold)}
if uid_norm:
resp["uid"] = uid_norm
conn = get_connection_from_env()
ensure_data_table(conn)
try:
log_attendance(conn, uid=uid_norm, recognized_name="Unknown", face_ok=False)
finally:
conn.close()
return resp
best_name = None
best_score = float("inf")
second_best_score = float("inf")
for gname, gvec in gallery.items():
d = _cosine_distance(q, gvec)
if d < best_score:
second_best_score = best_score
best_score = d
best_name = gname
elif d < second_best_score:
second_best_score = d
return _build_recognition_response(best_name, best_score, threshold, uid, second_score=second_best_score)
# ============================================================
# RFID ENDPOINTS
# ============================================================
@app.post("/rfid/register")
def rfid_register(
request: Request,
uid: str = Query(..., min_length=1, max_length=64),
name: str | None = Query(None, min_length=1, max_length=128),
):
"""Register/upsert RFID UID in data_fcgntion."""
try:
print(f"[RFID_REGISTER] url={request.url} query={request.url.query}")
except Exception:
pass
uid_norm = uid.strip().upper()
if not uid_norm:
raise HTTPException(status_code=422, detail="uid empty")
raw_name = name.strip() if name else ""
conn = get_connection_from_env()
ensure_data_table(conn)
try:
claimed = False
if raw_name:
claimed = claim_rfid_by_name(conn, raw_name, uid_norm)
if not claimed:
upsert_person_rfid(conn, uid_norm, raw_name)
elif _armed_enroll and _armed_enroll.get("name"):
armed_name = _armed_enroll["name"]
claimed = claim_rfid_by_name(conn, armed_name, uid_norm)
if claimed and _armed_enroll:
_armed_enroll["uid"] = uid_norm
else:
upsert_person_rfid(conn, uid_norm, armed_name)
else:
upsert_person_rfid(conn, uid_norm, None)
row = get_person_by_rfid(conn, uid_norm)
finally:
conn.close()
return {"ok": True, "uid": uid_norm, "name": raw_name or uid_norm, "row": row}
@app.get("/rfid/lookup")
def rfid_lookup(
request: Request,
uid: str = Query(..., min_length=1, max_length=64),
debug: int = Query(0, ge=0, le=1),
):
"""Lookup RFID UID -> person name from data_fcgntion."""
uid_norm = uid.strip().upper()
if not uid_norm:
raise HTTPException(status_code=422, detail="uid empty")
try:
print(f"[RFID_LOOKUP] url={request.url} query={request.url.query}")
except Exception:
pass
conn = get_connection_from_env()
ensure_data_table(conn)
row = get_person_by_rfid(conn, uid_norm)
if not row:
# Cek apakah ada person yang BELUM memiliki RFID
cursor = conn.cursor()
cursor.execute("SELECT id_tabel, person, fc FROM data_fcgntion WHERE rfid_uid IS NULL OR rfid_uid = '' ORDER BY create_at ASC LIMIT 1")
unassigned_row = cursor.fetchone()
if unassigned_row:
person_id = int(unassigned_row[0])
person_name = unassigned_row[1]
has_face = unassigned_row[2] is not None
cursor.execute("UPDATE data_fcgntion SET rfid_uid=%s WHERE id_tabel=%s", (uid_norm, person_id))
conn.commit()
conn.close()
resp = {
"ok": True,
"uid": uid_norm,
"registered": True,
"name": person_name,
"has_face": has_face,
"auto_assigned": True
}
print(f"[RFID_LOOKUP] auto-assigned result={resp}")
return resp
conn.close()
if not row:
resp = {"ok": True, "uid": uid_norm, "registered": False, "name": None}
print(f"[RFID_LOOKUP] result={resp}")
return resp
resp = {
"ok": True,
"uid": row.get("rfid_uid"),
"registered": True,
"name": row.get("person"),
"has_face": row.get("has_face", False),
}
if debug == 1:
resp["create_at"] = str(row.get("create_at"))
print(f"[RFID_LOOKUP] result={resp}")
return resp
# ============================================================
# FACE ENROLLMENT & RECOGNITION ENDPOINTS
# ============================================================
@app.post("/enroll")
async def enroll(
request: Request,
name: str = Query(..., min_length=1),
uid: str | None = Query(None),
):
"""ESP32 sends raw JPEG bytes. Extract face, compute embedding, save to data_fcgntion."""
data = await request.body()
if not data:
raise HTTPException(status_code=422, detail="Missing image body")
img = cv2.imdecode(np.frombuffer(data, dtype=np.uint8), cv2.IMREAD_COLOR)
if img is None:
raise HTTPException(status_code=400, detail="Invalid image")
face = _crop_first_face_bgr(img)
if face is None:
raise HTTPException(status_code=400, detail="No face detected")
vec = _represent_from_bgr_face(face)
uid_norm = uid.strip().upper() if uid else None
conn = get_connection_from_env()
ensure_data_table(conn)
row_id = save_or_update_person(conn, name, json.dumps(vec.tolist()), rfid_uid=uid_norm)
conn.close()
return {"ok": True, "name": name, "dim": int(vec.size), "row_id": row_id}
@app.post("/recognize")
async def recognize(
request: Request,
threshold: float = Query(0.25, ge=0.0, le=1.0),
uid: str | None = Query(None, min_length=1, max_length=64),
):
global _recognize_hits
_recognize_hits += 1
print(f"[RECOGNIZE] uid={uid} threshold={threshold}")
try:
data = await request.body()
except ClientDisconnect:
print("[RECOGNIZE] client disconnected before body received")
raise HTTPException(status_code=400, detail="Client disconnected")
if not data:
raise HTTPException(status_code=422, detail="Missing image body")
img = cv2.imdecode(np.frombuffer(data, dtype=np.uint8), cv2.IMREAD_COLOR)
if img is None:
raise HTTPException(status_code=400, detail="Invalid image")
face = _crop_first_face_bgr(img)
if face is None:
raise HTTPException(status_code=400, detail="No face detected")
q = _represent_from_bgr_face(face)
return _recognize_embedding(q, float(threshold), uid)
@app.post("/rfid/attendance")
async def rfid_attendance(
request: Request,
uid: str = Query(..., min_length=1, max_length=64),
threshold: float = Query(0.25, ge=0.0, le=1.0),
):
print(f"[RFID_ATTENDANCE] uid={uid} threshold={threshold}")
try:
data = await request.body()
except ClientDisconnect:
print(f"[RFID_ATTENDANCE] client disconnected before body received (uid={uid})")
raise HTTPException(status_code=400, detail="Client disconnected")
if not data:
raise HTTPException(status_code=422, detail="Missing image body")
img = cv2.imdecode(np.frombuffer(data, dtype=np.uint8), cv2.IMREAD_COLOR)
if img is None:
raise HTTPException(status_code=400, detail="Invalid image")
face = _crop_first_face_bgr(img)
if face is None:
raise HTTPException(status_code=400, detail="No face detected")
q = _represent_from_bgr_face(face)
uid_norm = uid.strip().upper()
gallery = _load_gallery_from_db()
if not gallery:
return {"ok": True, "match": False, "name": "Unknown", "reason": "empty_gallery"}
conn = get_connection_from_env()
ensure_data_table(conn)
row = get_person_by_rfid(conn, uid_norm)
conn.close()
if not row or not row.get("person"):
return {"ok": True, "match": False, "name": "Unknown", "reason": "uid_not_found"}
target_name = row.get("person")
if target_name not in gallery:
return {"ok": True, "match": False, "name": target_name, "reason": "target_not_loaded"}
target_vec = gallery[target_name]
target_distance = _cosine_distance(q, target_vec)
best_name = None
best_score = float("inf")
second_best_score = float("inf")
for gname, gvec in gallery.items():
d = _cosine_distance(q, gvec)
if d < best_score:
second_best_score = best_score
best_score = d
best_name = gname
elif d < second_best_score:
second_best_score = d
match = (best_name == target_name and
is_recognized_match(target_distance, threshold, second_score=second_best_score, min_confidence=0.35, margin=0.05))
result = {
"ok": True,
"name": target_name if match else "Unknown",
"score": float(target_distance),
"threshold": float(threshold),
"match": bool(match),
"best": best_name if best_name is not None else "Unknown",
"uid": uid_norm
}
# Hanya catat ke database JIKA COCOK. Ini mencegah spam DB saat ESP32 melakukan auto-retry!
if match:
conn = get_connection_from_env()
try:
log_attendance(conn, uid=uid_norm, recognized_name=target_name, face_ok=True)
finally:
conn.close()
print(f"[SUPABASE] Wajah dikenali: '{target_name}' - push ke Supabase...")
_push_to_supabase_async(target_name)
print(f'[RFID_ATTENDANCE] result={result}')
return result
# ============================================================
# PERSON / CLEANUP ENDPOINTS
# ============================================================
@app.delete("/person/delete")
def person_delete(
name: str | None = Query(None),
uid: str | None = Query(None),
):
"""Delete a person from data_fcgntion by name or rfid_uid."""
if not name and not uid:
raise HTTPException(status_code=422, detail="Provide name or uid")
conn = get_connection_from_env()
ensure_data_table(conn)
try:
if uid:
delete_person_by_rfid(conn, uid.strip().upper())
return {"ok": True, "deleted_by": "uid", "uid": uid.strip().upper()}
else:
delete_person_by_name(conn, name.strip())
return {"ok": True, "deleted_by": "name", "name": name.strip()}
finally:
conn.close()
@app.post("/cleanup/unused")
def cleanup_unused():
"""Delete data_fcgntion rows where fc IS NULL AND rfid_uid IS NULL."""
conn = get_connection_from_env()
ensure_data_table(conn)
try:
count = delete_unused_persons(conn)
return {"ok": True, "deleted": count, "msg": f"Dihapus {count} baris data kosong"}
finally:
conn.close()
@app.post("/cleanup/old_logs")
def cleanup_old_logs(days: int = Query(30, ge=1, le=3650)):
"""Delete attendance_log entries older than `days` days."""
conn = get_connection_from_env()
ensure_data_table(conn)
try:
count = delete_old_attendance_logs(conn, days=days)
return {"ok": True, "deleted": count, "msg": f"Dihapus {count} log absen > {days} hari"}
finally:
conn.close()
# ============================================================
# API DATA ENDPOINTS
# ============================================================
@app.get("/api/persons")
def api_list_persons(limit: int = Query(200, ge=1, le=1000)):
conn = get_connection_from_env()
ensure_data_table(conn)
items = list_all_persons(conn, limit=limit)
conn.close()
for item in items:
if item.get("create_at") and not isinstance(item["create_at"], str):
item["create_at"] = str(item["create_at"])
return {"ok": True, "items": items}
@app.get("/api/attendance")
def api_list_attendance(limit: int = Query(200, ge=1, le=1000)):
conn = get_connection_from_env()
ensure_data_table(conn)
items = list_attendance_log(conn, limit=limit)
conn.close()
for item in items:
if item.get("created_at") and not isinstance(item["created_at"], str):
item["created_at"] = str(item["created_at"])
return {"ok": True, "items": items}
# ============================================================
# FRAME / LIVE PREVIEW
# ============================================================
@app.get("/health")
def health():
return {"ok": True}
@app.get("/live", response_class=HTMLResponse)
def live_page():
return """
