rep_tracker / server.py
bhardwaj08sarthak's picture
Upload 12 files
59c4adb verified
"""
FastAPI + WebSocket server for real-time squat rep tracking.
Protocol (client ↔ server over WebSocket):
→ Client sends JSON:
{
"frame": "<base64-encoded JPEG or PNG>",
"target_reps": 15 // optional, only read on first frame or reset
}
← Server responds JSON:
{
"status": "Squatting" | "Standing" | "Error",
"view": "Frontal" | "Profile" | "Oblique" | "Unknown",
"depth_ratio": 1.23,
"knee_angle": 110.5,
"facing_ratio": 0.45,
"reps": 3,
"message": "Good form, keep going.",
"annotated_frame": "<base64-encoded JPEG>" // optional, if ?annotate=true
}
→ Client sends JSON to reset:
{ "command": "reset", "target_reps": 15 }
"""
import base64
import json
import logging
import os
from contextlib import asynccontextmanager
import cv2
import numpy as np
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from inference import PoseEstimator
from detectors import SquatDetector
from state_machine import RepCounter
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("rep_tracker")
CONFIG_PATH = os.getenv("CONFIG_PATH", "config.json")
# Shared singleton for the static-image health-check endpoint
_estimator: PoseEstimator | None = None
_detector: SquatDetector | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global _estimator, _detector
_detector = SquatDetector(CONFIG_PATH)
_estimator = PoseEstimator(static_image_mode=True)
log.info("Rep-tracker server started — MediaPipe BlazePose loaded")
yield
if _estimator:
_estimator.close()
log.info("Rep-tracker server shut down")
app = FastAPI(
title="TuebiFit Rep Tracker",
version="0.1.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/health")
async def health():
return {"status": "ok", "model": "BlazePose", "complexity": 2}
def _decode_frame(b64: str) -> np.ndarray:
"""Decode a base64-encoded image string into a BGR numpy frame."""
raw = base64.b64decode(b64)
arr = np.frombuffer(raw, dtype=np.uint8)
frame = cv2.imdecode(arr, cv2.IMREAD_COLOR)
if frame is None:
raise ValueError("Failed to decode base64 frame into an image")
return frame
def _encode_frame(frame: np.ndarray) -> str:
"""Encode a BGR numpy frame as a base64 JPEG string."""
_, buf = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 80])
return base64.b64encode(buf).decode("ascii")
@app.websocket("/ws")
async def ws_rep_tracker(websocket: WebSocket, annotate: bool = Query(False)):
await websocket.accept()
# Per-connection pipeline: dedicated estimator in video-tracking mode
estimator = PoseEstimator(static_image_mode=False)
detector = SquatDetector(CONFIG_PATH)
counter = RepCounter(target_reps=15)
log.info("WebSocket connection opened")
try:
while True:
raw_msg = await websocket.receive_text()
data = json.loads(raw_msg)
# Handle reset command
if data.get("command") == "reset":
target = data.get("target_reps", 15)
counter = RepCounter(target_reps=target)
await websocket.send_json({"reps": 0, "message": "Let's go!", "status": "reset"})
continue
# On first meaningful message, allow overriding target_reps
if counter.reps_completed == 0 and "target_reps" in data:
counter.target_reps = int(data["target_reps"])
b64_frame = data.get("frame")
if not b64_frame:
await websocket.send_json({"error": "missing 'frame' field"})
continue
try:
frame = _decode_frame(b64_frame)
_, keypoints, sx, sy = estimator.extract_keypoints_from_frame(frame)
except ValueError as e:
await websocket.send_json({"error": str(e)})
continue
metrics = detector.evaluate_form(keypoints)
rep_state = counter.update(metrics["status"])
response = {
"status": rep_state["confirmed_state"],
"raw_status": metrics["status"],
"view": metrics["view"],
"depth_ratio": round(metrics["depth_ratio"], 3),
"knee_angle": round(metrics["knee_angle"], 1),
"facing_ratio": round(metrics["facing_ratio"], 3),
"reps": rep_state["reps"],
"message": rep_state["message"],
}
if annotate:
display_text = (
f"[{metrics['view']}] {rep_state['confirmed_state']} | "
f"Reps: {rep_state['reps']} | {rep_state['message']}"
)
annotated = detector.draw_skeleton(frame.copy(), keypoints, sx, sy, display_text)
response["annotated_frame"] = _encode_frame(annotated)
await websocket.send_json(response)
except WebSocketDisconnect:
log.info("WebSocket connection closed by client")
except Exception as exc:
log.exception("Unexpected error in WebSocket handler")
try:
await websocket.send_json({"error": str(exc)})
except Exception:
pass
finally:
estimator.close()
@app.post("/analyze")
async def analyze_frame(payload: dict):
"""
One-shot REST endpoint for testing.
Accepts { "frame": "<base64>" } and returns metrics (no rep counting).
"""
b64 = payload.get("frame")
if not b64:
return JSONResponse({"error": "missing 'frame' field"}, status_code=400)
try:
frame = _decode_frame(b64)
_, keypoints, _, _ = _estimator.extract_keypoints_from_frame(frame)
except ValueError as e:
return JSONResponse({"error": str(e)}, status_code=422)
metrics = _detector.evaluate_form(keypoints)
return {
"status": metrics["status"],
"view": metrics["view"],
"depth_ratio": round(metrics["depth_ratio"], 3),
"knee_angle": round(metrics["knee_angle"], 1),
"facing_ratio": round(metrics["facing_ratio"], 3),
}
if __name__ == "__main__":
import uvicorn
uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=True)