Spaces:
Sleeping
Sleeping
File size: 3,222 Bytes
207fcc9 1833ece 207fcc9 1833ece 207fcc9 1724c8d 207fcc9 1833ece 207fcc9 37876bc 1724c8d 37876bc 1724c8d 37876bc 1724c8d 37876bc 1833ece 1724c8d 1833ece 37876bc 1833ece 1724c8d 207fcc9 1724c8d 207fcc9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | import json
import base64
import cv2
import numpy as np
import asyncio
from fastapi import WebSocket, WebSocketDisconnect
from services.vision import process_frame_synchronous
from services.attendance import mark_attendance
from core.state import active_connections
async def websocket_endpoint(websocket: WebSocket):
"""
Hard-Locked WebSocket Connection with Visual Debugging.
Returns crops and bounding boxes alongside the Truth Report.
"""
await websocket.accept()
active_connections.append(websocket)
try:
while True:
data = await websocket.receive_text()
payload = json.loads(data)
if payload.get("type") == "heartbeat":
await websocket.send_json({"type": "heartbeat_ack"})
continue
if payload.get("type") == "frame":
encoded_data = payload["image"].split(',')[1]
nparr = np.frombuffer(base64.b64decode(encoded_data), np.uint8)
frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
report, stats = await asyncio.to_thread(process_frame_synchronous, frame)
results_summary = []
# Prepare visual data for frontend
client_faces = []
for face in report:
name = face["name"]
score = face["score"]
status = face["status"]
# Store data for drawing boxes and showing crops
client_faces.append({
"name": name,
"score": score,
"status": status,
"box": face["box"],
"crop": face["crop_b64"]
})
if status == "match":
# Business Logic: Mark Attendance
status_db, time_str = mark_attendance(name)
results_summary.append(f"✅ {name} ({score}%)")
# Push to session attendance list
await websocket.send_json({
"type": "attendance",
"name": name,
"time": time_str or "Just Now",
"status": "success"
})
else:
results_summary.append(f"❌ {name} ({score}%)")
# Detailed Terminal Output
debug_msg = f"Faces: {stats['detected']} | "
debug_msg += " | ".join(results_summary) if results_summary else "No faces found."
# Send frame processing result back to browser
await websocket.send_json({
"type": "ready",
"debug": debug_msg,
"faces": client_faces # ALL visual debug data included here
})
except WebSocketDisconnect:
if websocket in active_connections:
active_connections.remove(websocket)
|