Game Analytics Platform Computer Vision

Game Analytics Platform - Computer Vision Fitness Tracker

Skill by ara.so — Data Skills collection.

What This Project Does

Game Analytics Platform is a local-first, real-time computer vision system that tracks user movements across 16 fitness exercises using webcam input. It combines:

• YOLO v8 for object detection and tracking (balls, cones, people)
• MediaPipe for skeletal pose estimation and form validation
• Spring Boot (Java 17) backend for process orchestration
• React + Vite frontend dashboard for game control
• Python AI scripts that export workout metrics to CSV
• pyttsx3 for real-time audio coaching

The architecture runs entirely locally with a 3-tier design: React UI → Spring Boot API → Python AI processes.

Installation

Prerequisites

Install these first:

• Python 3.10+ (ensure "Add to PATH" is checked)
• Java 17 (from Adoptium)
• Node.js LTS

Auto-Install

# Windows
python install.py

# Mac/Linux
python3 install.py

This creates a Python virtual environment, installs dependencies, downloads YOLO models, and builds the frontend.

Manual Setup (if auto-install fails)

# 1. Create Python virtual environment
python -m venv venv

# 2. Activate it
# Windows:
venv\Scripts\activate
# Mac/Linux:
source venv/bin/activate

# 3. Install Python dependencies
pip install ultralytics mediapipe opencv-python pandas pyttsx3

# 4. Build frontend
cd frontend
npm install
npm run build
cd ..

# 5. Build backend
cd backend
mvn clean package
cd ..

Starting the Platform

# Windows
start.bat

# Mac/Linux
./start.sh

Access dashboard at http://localhost:8080

Architecture Components

1. Spring Boot Backend (Java)

The backend orchestrates Python AI processes via REST API.

Key Files:

• backend/src/main/java/com/gameanalytics/controller/GameController.java
• backend/src/main/java/com/gameanalytics/service/ProcessService.java

REST API Endpoints:

// Start a game
POST /api/games/{id}/start
// Response: 200 OK or 400 if game already running

// Stop a game
POST /api/games/{id}/stop
// Response: 200 OK

// Get available CSV data files
GET /api/games/data
// Response: ["workout_20260601_143022.csv", ...]

// Get list of all games
GET /api/games
// Response: [{"id": 1, "name": "YOLO Ball Counter", ...}, ...]

Process Management Pattern:

// ProcessService.java
public class ProcessService {
    private Process currentProcess;
    private final Object lock = new Object();

    public boolean startGame(int gameId) {
        synchronized (lock) {
            if (currentProcess != null && currentProcess.isAlive()) {
                return false; // Game already running
            }
            
            String pythonPath = System.getProperty("os.name").toLowerCase().contains("win")
                ? "venv\\Scripts\\python.exe"
                : "venv/bin/python";
            
            String scriptPath = "games/exe_" + gameId + ".py";
            
            ProcessBuilder pb = new ProcessBuilder(pythonPath, scriptPath);
            pb.directory(new File(System.getProperty("user.dir")));
            pb.redirectErrorStream(true);
            
            try {
                currentProcess = pb.start();
                
                // Stream logs asynchronously
                new Thread(() -> {
                    try (BufferedReader reader = new BufferedReader(
                            new InputStreamReader(currentProcess.getInputStream()))) {
                        String line;
                        while ((line = reader.readLine()) != null) {
                            System.out.println("[Python] " + line);
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }).start();
                
                return true;
            } catch (IOException e) {
                e.printStackTrace();
                return false;
            }
        }
    }

    public boolean stopGame() {
        synchronized (lock) {
            if (currentProcess != null && currentProcess.isAlive()) {
                currentProcess.destroy();
                try {
                    currentProcess.waitFor(5, TimeUnit.SECONDS);
                } catch (InterruptedException e) {
                    currentProcess.destroyForcibly();
                }
                currentProcess = null;
                return true;
            }
            return false;
        }
    }
}

2. Python AI Vision Scripts

Each game is a standalone Python script in games/exe_*.py.

Template for New Game:

import cv2
import pandas as pd
import numpy as np
from ultralytics import YOLO
import mediapipe as mp
import pyttsx3
import signal
import sys
from datetime import datetime
import threading

# Global state
running = True
event_buffer = []
tts_engine = None

def signal_handler(sig, frame):
    """Handle SIGTERM from Java backend"""
    global running
    print("Received stop signal, cleaning up...")
    running = False

def tts_worker(queue):
    """Async text-to-speech thread"""
    global tts_engine
    tts_engine = pyttsx3.init()
    while running:
        if not queue.empty():
            message = queue.get()
            tts_engine.say(message)
            tts_engine.runAndWait()

def main():
    global running, event_buffer
    
    # Register signal handler
    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)
    
    # Initialize models
    yolo_model = YOLO('models/yolov8n.pt')  # Nano model for speed
    mp_pose = mp.solutions.pose
    pose = mp_pose.Pose(
        min_detection_confidence=0.5,
        min_tracking_confidence=0.5
    )
    
    # Start TTS thread
    from queue import Queue
    tts_queue = Queue()
    tts_thread = threading.Thread(target=tts_worker, args=(tts_queue,))
    tts_thread.daemon = True
    tts_thread.start()
    
    # Open webcam
    cap = cv2.VideoCapture(0)
    if not cap.isOpened():
        print("ERROR: Cannot open webcam")
        return
    
    # Game state
    rep_count = 0
    last_state = None
    
    print("Starting game loop...")
    
    while running:
        ret, frame = cap.read()
        if not ret:
            break
        
        # Resize for performance
        frame = cv2.resize(frame, (640, 480))
        rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        
        # YOLO object detection
        yolo_results = yolo_model.track(frame, persist=True, verbose=False)
        
        # MediaPipe pose detection
        pose_results = pose.process(rgb_frame)
        
        # Game logic example: squat counter
        if pose_results.pose_landmarks:
            landmarks = pose_results.pose_landmarks.landmark
            
            # Get hip and knee angles
            left_hip = landmarks[mp_pose.PoseLandmark.LEFT_HIP.value]
            left_knee = landmarks[mp_pose.PoseLandmark.LEFT_KNEE.value]
            left_ankle = landmarks[mp_pose.PoseLandmark.LEFT_ANKLE.value]
            
            # Calculate knee angle (simplified)
            hip_y = left_hip.y
            knee_y = left_knee.y
            angle = abs(hip_y - knee_y) * 100  # Normalize to 0-100
            
            # State machine
            if angle < 30 and last_state != 'down':
                last_state = 'down'
            elif angle > 70 and last_state == 'down':
                rep_count += 1
                last_state = 'up'
                tts_queue.put(f"Rep {rep_count}")
                event_buffer.append({
                    'timestamp': datetime.now().isoformat(),
                    'event': 'rep_completed',
                    'count': rep_count,
                    'angle': angle
                })
            
            # Draw skeleton
            mp.solutions.drawing_utils.draw_landmarks(
                frame, pose_results.pose_landmarks, mp_pose.POSE_CONNECTIONS
            )
        
        # Draw UI overlay
        cv2.putText(frame, f"Reps: {rep_count}", (10, 50),
                    cv2.FONT_HERSHEY_SIMPLEX, 1.5, (0, 255, 0), 3)
        
        cv2.imshow('Game', frame)
        
        if cv2.waitKey(1) & 0xFF == ord('q'):
            running = False
    
    # Cleanup
    cap.release()
    cv2.destroyAllWindows()
    
    # Export data
    if event_buffer:
        df = pd.DataFrame(event_buffer)
        output_file = f"data/workout_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        df.to_csv(output_file, index=False)
        print(f"Saved workout data to {output_file}")
    
    print("Game stopped cleanly")

if __name__ == "__main__":
    main()

3. React Frontend

API Integration Pattern:

// frontend/src/services/gameService.js
const API_BASE = 'http://localhost:8080/api/games';

export const startGame = async (gameId) => {
  const response = await fetch(`${API_BASE}/${gameId}/start`, {
    method: 'POST',
  });
  
  if (!response.ok) {
    const error = await response.text();
    throw new Error(error || 'Failed to start game');
  }
  
  return response.json();
};

export const stopGame = async (gameId) => {
  const response = await fetch(`${API_BASE}/${gameId}/stop`, {
    method: 'POST',
  });
  
  return response.json();
};

export const getWorkoutData = async () => {
  const response = await fetch(`${API_BASE}/data`);
  return response.json();
};

// Polling pattern for CSV updates
export const pollForNewData = (callback, interval = 2000) => {
  const poller = setInterval(async () => {
    const files = await getWorkoutData();
    callback(files);
  }, interval);
  
  return () => clearInterval(poller);
};

Configuration

Each game has a JSON config in configs/game_{id}.json:

{
  "game_id": 1,
  "name": "YOLO Ball Counter",
  "yolo_model": "models/yolov8n.pt",
  "confidence_threshold": 0.5,
  "tracking_persistence": true,
  "audio_coaching": true,
  "target_fps": 30,
  "resolution": [640, 480],
  "coaching_triggers": {
    "milestone_reps": [5, 10, 20],
    "form_warning_angle": 45
  }
}

Loading config in Python:

import json

def load_game_config(game_id):
    with open(f'configs/game_{game_id}.json', 'r') as f:
        return json.load(f)

config = load_game_config(1)
yolo_model = YOLO(config['yolo_model'])
confidence = config['confidence_threshold']

Common Patterns

1. Adding a New Exercise Game

# 1. Create Python script
touch games/exe_17.py

# 2. Create config
cat > configs/game_17.json << EOF
{
  "game_id": 17,
  "name": "Jumping Jacks Counter",
  "yolo_model": "models/yolov8n-pose.pt",
  "confidence_threshold": 0.6
}
EOF

# 3. Update backend game list
# Edit: backend/src/main/resources/games.json
# Add: {"id": 17, "name": "Jumping Jacks Counter", "description": "..."}

2. Combining YOLO + MediaPipe

# Detect objects with YOLO, track pose with MediaPipe
yolo_results = yolo_model(frame)
pose_results = pose.process(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))

# Example: Check if person's hand crosses detected ball
if pose_results.pose_landmarks and len(yolo_results) > 0:
    hand = pose_results.pose_landmarks.landmark[mp_pose.PoseLandmark.LEFT_WRIST.value]
    
    for detection in yolo_results[0].boxes:
        if detection.cls == 32:  # Sports ball class
            ball_x, ball_y = detection.xywh[0][:2]
            hand_x = hand.x * frame.shape[1]
            hand_y = hand.y * frame.shape[0]
            
            distance = np.sqrt((hand_x - ball_x)**2 + (hand_y - ball_y)**2)
            if distance < 50:  # Pixels
                print("Hand touched ball!")

3. CSV Data Export Pattern

# Track events during game
event_buffer = []

# During game loop
event_buffer.append({
    'timestamp': datetime.now().isoformat(),
    'event_type': 'crossing',
    'player_position_x': x,
    'player_position_y': y,
    'speed_estimate': speed,
    'rep_count': reps
})

# On game stop
df = pd.DataFrame(event_buffer)
df['session_id'] = datetime.now().strftime('%Y%m%d_%H%M%S')
df.to_csv(f"data/workout_{df['session_id'].iloc[0]}.csv", index=False)

4. Thread-Safe Audio Coaching

from queue import Queue
import threading
import pyttsx3

def tts_worker(queue):
    engine = pyttsx3.init()
    while True:
        message = queue.get()
        if message is None:
            break
        engine.say(message)
        engine.runAndWait()
        queue.task_done()

tts_queue = Queue()
tts_thread = threading.Thread(target=tts_worker, args=(tts_queue,))
tts_thread.daemon = True
tts_thread.start()

# In game loop
if rep_count % 5 == 0:
    tts_queue.put(f"Great job! {rep_count} reps completed")

Troubleshooting

Python Process Won't Stop

// In ProcessService.java, add forceful termination
public boolean stopGame() {
    synchronized (lock) {
        if (currentProcess != null && currentProcess.isAlive()) {
            currentProcess.destroy();
            try {
                if (!currentProcess.waitFor(3, TimeUnit.SECONDS)) {
                    currentProcess.destroyForcibly();
                    currentProcess.waitFor(2, TimeUnit.SECONDS);
                }
            } catch (InterruptedException e) {
                currentProcess.destroyForcibly();
            }
            currentProcess = null;
            return true;
        }
        return false;
    }
}

Webcam Not Found

# Test all camera indices
for i in range(5):
    cap = cv2.VideoCapture(i)
    if cap.isOpened():
        print(f"Camera found at index {i}")
        cap.release()
        break

YOLO Model Loading Fails

import os
from ultralytics import YOLO

model_path = 'models/yolov8n.pt'

if not os.path.exists(model_path):
    print("Downloading YOLO model...")
    model = YOLO('yolov8n.pt')  # Auto-downloads
    os.makedirs('models', exist_ok=True)
    # Model cached in ultralytics directory
else:
    model = YOLO(model_path)

CORS Issues (if running frontend separately)

// backend/src/main/java/com/gameanalytics/config/WebConfig.java
@Configuration
public class WebConfig implements WebMvcConfigurer {
    @Override
    public void addCorsMappings(CorsRegistry registry) {
        registry.addMapping("/api/**")
                .allowedOrigins("http://localhost:5173")  // Vite dev server
                .allowedMethods("GET", "POST", "PUT", "DELETE");
    }
}

CSV Not Appearing in Frontend

// Ensure polling starts after game stops
const handleStopGame = async (gameId) => {
  await stopGame(gameId);
  
  // Wait for Python to write CSV
  setTimeout(async () => {
    const files = await getWorkoutData();
    setWorkoutFiles(files);
  }, 2000);
};

Performance Optimization

# Use YOLO tracking instead of detection for speed
results = model.track(frame, persist=True, tracker="bytetrack.yaml")

# Reduce frame processing
frame_skip = 2
frame_count = 0

while running:
    ret, frame = cap.read()
    frame_count += 1
    
    if frame_count % frame_skip != 0:
        continue  # Process every 2nd frame
    
    # Your inference code...

Environment Variables

# .env (create in project root)
YOLO_MODEL_PATH=models/yolov8n.pt
MEDIAPIPE_MODEL_COMPLEXITY=1
TTS_RATE=150
WEBCAM_INDEX=0
OUTPUT_DIR=data

# Load in Python
import os
from dotenv import load_dotenv

load_dotenv()

model_path = os.getenv('YOLO_MODEL_PATH', 'models/yolov8n.pt')
webcam_index = int(os.getenv('WEBCAM_INDEX', '0'))