128 lines
4.6 KiB
Python
128 lines
4.6 KiB
Python
import re
|
|
|
|
with open('esp32_http_server.py', 'r', encoding='utf8') as f:
|
|
content = f.read()
|
|
|
|
# 1. Update /rfid/attendance to only check against the specific person linked to the RFID
|
|
old_rfid_attendance = '''@app.post("/rfid/attendance")
|
|
async def rfid_attendance(
|
|
request: Request,
|
|
uid: str = Query(..., min_length=1, max_length=64),
|
|
threshold: float = Query(0.38, 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)
|
|
return _recognize_embedding(q, float(threshold), uid)'''
|
|
|
|
new_rfid_attendance = '''@app.post("/rfid/attendance")
|
|
async def rfid_attendance(
|
|
request: Request,
|
|
uid: str = Query(..., min_length=1, max_length=64),
|
|
threshold: float = Query(0.38, 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()
|
|
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")
|
|
emb_raw = row.get("embedding")
|
|
if not emb_raw:
|
|
return {"ok": True, "match": False, "name": target_name, "reason": "no_face_enrolled"}
|
|
|
|
try:
|
|
if isinstance(emb_raw, str):
|
|
import json
|
|
emb_raw = json.loads(emb_raw)
|
|
gvec = _parse_embedding(emb_raw)
|
|
except Exception:
|
|
return {"ok": True, "match": False, "name": target_name, "reason": "invalid_embedding"}
|
|
|
|
d = _cosine_distance(q, gvec)
|
|
match = d <= threshold
|
|
|
|
result = {
|
|
"ok": True,
|
|
"name": target_name if match else "Unknown",
|
|
"score": float(d),
|
|
"threshold": float(threshold),
|
|
"match": bool(match),
|
|
"best": target_name,
|
|
"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)
|
|
|
|
return result'''
|
|
|
|
content = content.replace(old_rfid_attendance, new_rfid_attendance)
|
|
|
|
# 2. Update _build_recognition_response to not log failed attendances (just in case it's used elsewhere with UID)
|
|
old_build_resp = ''' if uid and uid.strip():
|
|
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=bool(match))
|
|
finally:
|
|
conn.close()'''
|
|
|
|
new_build_resp = ''' 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()'''
|
|
|
|
content = content.replace(old_build_resp, new_build_resp)
|
|
|
|
with open('esp32_http_server.py', 'w', encoding='utf8') as f:
|
|
f.write(content)
|
|
|
|
print("Update complete")
|