ml_predictor.py

"""
ML-based predictive cursor movement for improved fluidity.
Uses simple linear regression to predict cursor trajectory and smooth movement.
"""

import numpy as np
import math
import time
from collections import deque
from typing import Tuple, Optional

import config


class CursorPredictor:
    """
    Machine learning-based cursor movement predictor.
    Uses linear regression to predict future cursor positions for smoother movement.
    """
    
    def __init__(self, window_size: int = 5):
        """
        Initialize the cursor predictor.
        
        Args:
            window_size: Number of recent positions to use for prediction
        """
        self.window_size = window_size
        self.position_history: deque = deque(maxlen=window_size)
        self.time_history: deque = deque(maxlen=window_size)
        
        # Prediction parameters
        self.prediction_horizon = config.ML_PREDICTION_HORIZON  # seconds ahead to predict
        self.smoothing_factor = config.ML_SMOOTHING_FACTOR
        
    def add_position(self, x: float, y: float, timestamp: float):
        """Add a new cursor position to the history."""
        self.position_history.append((x, y))
        self.time_history.append(timestamp)
    
    def predict_next_position(self) -> Optional[Tuple[float, float]]:
        """
        Predict the next cursor position using linear regression.
        
        Returns:
            Predicted (x, y) position or None if not enough data
        """
        if len(self.position_history) < 3:
            return None
        
        # Convert to numpy arrays
        positions = np.array(list(self.position_history))
        times = np.array(list(self.time_history))
        
        # Normalize times (relative to most recent)
        times = times - times[-1]
        
        # Fit linear regression for X and Y separately
        try:
            # X prediction
            x_coords = positions[:, 0]
            x_coef = np.polyfit(times, x_coords, 1)  # Linear fit
            predicted_x = np.polyval(x_coef, self.prediction_horizon)
            
            # Y prediction
            y_coords = positions[:, 1]
            y_coef = np.polyfit(times, y_coords, 1)  # Linear fit
            predicted_y = np.polyval(y_coef, self.prediction_horizon)
            
            return (float(predicted_x), float(predicted_y))
        except:
            return None
    
    def smooth_position(
        self,
        current_x: float,
        current_y: float,
        predicted_x: Optional[float] = None,
        predicted_y: Optional[float] = None
    ) -> Tuple[float, float]:
        """
        Smooth cursor position by blending current with predicted.
        Enhanced with adaptive smoothing based on movement speed for stability.
        
        Args:
            current_x: Current X position
            current_y: Current Y position
            predicted_x: Predicted X position (optional)
            predicted_y: Predicted Y position (optional)
            
        Returns:
            Smoothed (x, y) position
        """
        if predicted_x is None or predicted_y is None:
            predicted = self.predict_next_position()
            if predicted is None:
                return (current_x, current_y)
            predicted_x, predicted_y = predicted
        
        # Simple smoothing: blend current with predicted
        smoothed_x = current_x * (1.0 - self.smoothing_factor) + predicted_x * self.smoothing_factor
        smoothed_y = current_y * (1.0 - self.smoothing_factor) + predicted_y * self.smoothing_factor
        
        return (smoothed_x, smoothed_y)
    
    def clear_history(self):
        """Clear position history."""
        self.position_history.clear()
        self.time_history.clear()


class VelocityEstimator:
    """
    ML-enhanced velocity estimator using exponential moving average.
    Provides smoother velocity estimates for better intent prediction.
    """
    
    def __init__(self):
        """Initialize velocity estimator."""
        self.alpha = config.ML_VELOCITY_SMOOTHING  # EMA smoothing factor
        self.last_position: Optional[Tuple[float, float]] = None
        self.last_time: float = 0.0
        self.velocity_x: float = 0.0
        self.velocity_y: float = 0.0
    
    def update(self, x: float, y: float, timestamp: float) -> Tuple[float, float]:
        """
        Update velocity estimate.
        
        Args:
            x: Current X position
            y: Current Y position
            timestamp: Current timestamp
            
        Returns:
            Estimated (velocity_x, velocity_y)
        """
        if self.last_position is None:
            self.last_position = (x, y)
            self.last_time = timestamp
            return (0.0, 0.0)
        
        dt = timestamp - self.last_time
        if dt <= 0:
            return (self.velocity_x, self.velocity_y)
        
        # Calculate instantaneous velocity
        inst_vx = (x - self.last_position[0]) / dt
        inst_vy = (y - self.last_position[1]) / dt
        
        # Apply exponential moving average for smoothness
        self.velocity_x = self.alpha * inst_vx + (1.0 - self.alpha) * self.velocity_x
        self.velocity_y = self.alpha * inst_vy + (1.0 - self.alpha) * self.velocity_y
        
        self.last_position = (x, y)
        self.last_time = timestamp
        
        return (self.velocity_x, self.velocity_y)
    
    def reset(self):
        """Reset velocity estimator."""
        self.last_position = None
        self.last_time = 0.0
        self.velocity_x = 0.0
        self.velocity_y = 0.0