Self-Learning AI System

# Self-Learning Gauge Reading System
import numpy as np
import pandas as pd
from sklearn.linear_model import SGDClassifier, SGDRegressor
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, mean_squared_error
from sklearn.preprocessing import StandardScaler
import joblib
import json
from datetime import datetime, timedelta
import logging
from collections import deque
import threading
import time

class SelfLearningGaugeSystem:
    def __init__(self, learning_rate=0.01, buffer_size=1000):
        """
        Initialize Self-Learning system
        
        Args:
            learning_rate: Learning rate for online algorithms
            buffer_size: Size of data buffer for learning
        """
        self.learning_rate = learning_rate
        self.buffer_size = buffer_size
        
        # Models for different gauge types
        self.models = {}
        self.scalers = {}
        
        # Data buffers for continuous learning
        self.data_buffers = {}
        self.feedback_buffers = {}
        
        # Performance tracking
        self.performance_history = {}
        self.learning_triggers = {}
        
        # Configuration
        self.config = {
            'min_accuracy_threshold': 0.85,
            'learning_trigger_samples': 50,
            'validation_split': 0.2,
            'max_model_age_days': 30,
            'auto_retrain_interval_hours': 24
        }
        
        # Threading for background learning
        self.learning_thread = None
        self.stop_learning = False
        
        # Logging
        self.setup_logging()
        
    def setup_logging(self):
        """Setup logging for the system"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)
    
    def initialize_model(self, gauge_type, initial_data=None):
        """Initialize models for a new gauge type"""
        self.logger.info(f"Initializing models for gauge type: {gauge_type}")
        
        # Online learning model for real-time adaptation
        self.models[gauge_type] = {
            'classifier': SGDClassifier(
                loss='log', 
                learning_rate='adaptive',
                eta0=self.learning_rate,
                random_state=42
            ),
            'regressor': SGDRegressor(
                loss='squared_error',
                learning_rate='adaptive', 
                eta0=self.learning_rate,
                random_state=42
            ),
            'ensemble': RandomForestClassifier(
                n_estimators=50,
                random_state=42
            )
        }
        
        # Feature scalers
        self.scalers[gauge_type] = {
            'features': StandardScaler(),
            'targets': StandardScaler()
        }
        
        # Data buffers
        self.data_buffers[gauge_type] = deque(maxlen=self.buffer_size)
        self.feedback_buffers[gauge_type] = deque(maxlen=self.buffer_size)
        
        # Performance tracking
        self.performance_history[gauge_type] = {
            'accuracy': [],
            'timestamps': [],
            'sample_counts': []
        }
        
        # Learning triggers
        self.learning_triggers[gauge_type] = {
            'accuracy_drop': False,
            'new_data_available': False,
            'feedback_received': False,
            'scheduled_update': False
        }
        
        # Initial training if data provided
        if initial_data is not None:
            self.initial_training(gauge_type, initial_data)
    
    def initial_training(self, gauge_type, data):
        """Perform initial training with historical data"""
        self.logger.info(f"Initial training for {gauge_type} with {len(data)} samples")
        
        # Extract features and targets
        X, y = self.prepare_training_data(data)
        
        if len(X) == 0:
            self.logger.warning("No valid training data available")
            return
        
        # Scale features
        X_scaled = self.scalers[gauge_type]['features'].fit_transform(X)
        
        # Train ensemble model (for comparison and validation)
        self.models[gauge_type]['ensemble'].fit(X_scaled, y)
        
        # Initialize online models with partial_fit
        unique_classes = np.unique(y)
        self.models[gauge_type]['classifier'].partial_fit(X_scaled, y, classes=unique_classes)
        
        # For regression, convert y to continuous values if needed
        y_continuous = self.extract_continuous_values(y)
        if y_continuous is not None:
            self.models[gauge_type]['regressor'].partial_fit(X_scaled, y_continuous)
        
        # Store initial data in buffer
        for sample in data:
            self.data_buffers[gauge_type].append(sample)
        
        self.logger.info(f"Initial training completed for {gauge_type}")
    
    def prepare_training_data(self, data):
        """Prepare features and targets from raw data"""
        features = []
        targets = []
        
        for sample in data:
            if 'features' in sample and 'target' in sample:
                features.append(sample['features'])
                targets.append(sample['target'])
        
        return np.array(features), np.array(targets)
    
    def extract_continuous_values(self, targets):
        """Extract continuous values from targets for regression"""
        try:
            # Try to convert to float for regression
            return targets.astype(float)
        except:
            # If conversion fails, return None
            return None
    
    def predict(self, gauge_type, features):
        """Make prediction with uncertainty estimation"""
        if gauge_type not in self.models:
            return None
        
        # Scale features
        features_scaled = self.scalers[gauge_type]['features'].transform([features])
        
        # Get predictions from all models
        predictions = {}
        
        try:
            # Classification prediction
            clf_pred = self.models[gauge_type]['classifier'].predict(features_scaled)[0]
            clf_proba = self.models[gauge_type]['classifier'].predict_proba(features_scaled)[0]
            predictions['classification'] = {
                'prediction': clf_pred,
                'confidence': np.max(clf_proba)
            }
        except Exception as e:
            self.logger.warning(f"Classification prediction failed: {e}")
        
        try:
            # Regression prediction
            reg_pred = self.models[gauge_type]['regressor'].predict(features_scaled)[0]
            predictions['regression'] = {
                'prediction': reg_pred,
                'confidence': 0.8  # Simplified confidence
            }
        except Exception as e:
            self.logger.warning(f"Regression prediction failed: {e}")
        
        try:
            # Ensemble prediction
            ens_pred = self.models[gauge_type]['ensemble'].predict(features_scaled)[0]
            ens_proba = self.models[gauge_type]['ensemble'].predict_proba(features_scaled)[0]
            predictions['ensemble'] = {
                'prediction': ens_pred,
                'confidence': np.max(ens_proba)
            }
        except Exception as e:
            self.logger.warning(f"Ensemble prediction failed: {e}")
        
        # Combine predictions (simplified ensemble)
        if predictions:
            # Use the prediction with highest confidence
            best_pred = max(predictions.values(), key=lambda x: x['confidence'])
            return {
                'prediction': best_pred['prediction'],
                'confidence': best_pred['confidence'],
                'all_predictions': predictions,
                'timestamp': datetime.now().isoformat()
            }
        
        return None
    
    def add_feedback(self, gauge_type, features, true_target, predicted_target=None):
        """Add feedback for continuous learning"""
        feedback_sample = {
            'features': features,
            'true_target': true_target,
            'predicted_target': predicted_target,
            'timestamp': datetime.now(),
            'gauge_type': gauge_type
        }
        
        self.feedback_buffers[gauge_type].append(feedback_sample)
        self.learning_triggers[gauge_type]['feedback_received'] = True
        
        self.logger.info(f"Feedback added for {gauge_type}")
        
        # Trigger learning if enough feedback accumulated
        if len(self.feedback_buffers[gauge_type]) >= self.config['learning_trigger_samples']:
            self.trigger_learning(gauge_type)
    
    def trigger_learning(self, gauge_type):
        """Trigger learning process for a gauge type"""
        self.logger.info(f"Triggering learning for {gauge_type}")
        
        # Check if we have enough data
        if len(self.feedback_buffers[gauge_type]) < 10:
            self.logger.warning("Insufficient feedback data for learning")
            return
        
        # Prepare training data from feedback
        feedback_data = list(self.feedback_buffers[gauge_type])
        X, y = self.prepare_training_data(feedback_data)
        
        if len(X) == 0:
            self.logger.warning("No valid feedback data for learning")
            return
        
        # Incremental learning
        self.incremental_update(gauge_type, X, y)
        
        # Evaluate performance
        self.evaluate_performance(gauge_type, X, y)
        
        # Reset learning triggers
        for key in self.learning_triggers[gauge_type]:
            self.learning_triggers[gauge_type][key] = False
    
    def incremental_update(self, gauge_type, X, y):
        """Perform incremental model update"""
        self.logger.info(f"Performing incremental update for {gauge_type}")
        
        try:
            # Scale features (update scaler incrementally if needed)
            X_scaled = self.scalers[gauge_type]['features'].transform(X)
            
            # Update classifier
            self.models[gauge_type]['classifier'].partial_fit(X_scaled, y)
            
            # Update regressor if applicable
            y_continuous = self.extract_continuous_values(y)
            if y_continuous is not None:
                self.models[gauge_type]['regressor'].partial_fit(X_scaled, y_continuous)
            
            # Periodically retrain ensemble model
            if len(self.data_buffers[gauge_type]) > 100:
                # Use recent data for ensemble retraining
                recent_data = list(self.data_buffers[gauge_type])[-100:]
                X_recent, y_recent = self.prepare_training_data(recent_data)
                if len(X_recent) > 0:
                    X_recent_scaled = self.scalers[gauge_type]['features'].transform(X_recent)
                    self.models[gauge_type]['ensemble'].fit(X_recent_scaled, y_recent)
            
            self.logger.info(f"Incremental update completed for {gauge_type}")
            
        except Exception as e:
            self.logger.error(f"Incremental update failed for {gauge_type}: {e}")
    
    def evaluate_performance(self, gauge_type, X_test, y_test):
        """Evaluate model performance"""
        try:
            X_test_scaled = self.scalers[gauge_type]['features'].transform(X_test)
            
            # Evaluate classifier
            y_pred = self.models[gauge_type]['classifier'].predict(X_test_scaled)
            accuracy = accuracy_score(y_test, y_pred)
            
            # Store performance
            self.performance_history[gauge_type]['accuracy'].append(accuracy)
            self.performance_history[gauge_type]['timestamps'].append(datetime.now())
            self.performance_history[gauge_type]['sample_counts'].append(len(X_test))
            
            self.logger.info(f"Performance for {gauge_type}: Accuracy = {accuracy:.3f}")
            
            # Check if performance drop triggers retraining
            if accuracy < self.config['min_accuracy_threshold']:
                self.learning_triggers[gauge_type]['accuracy_drop'] = True
                self.logger.warning(f"Performance drop detected for {gauge_type}: {accuracy:.3f}")
            
            return accuracy
            
        except Exception as e:
            self.logger.error(f"Performance evaluation failed for {gauge_type}: {e}")
            return None
    
    def start_background_learning(self):
        """Start background learning thread"""
        if self.learning_thread is None or not self.learning_thread.is_alive():
            self.stop_learning = False
            self.learning_thread = threading.Thread(target=self._background_learning_loop)
            self.learning_thread.daemon = True
            self.learning_thread.start()
            self.logger.info("Background learning started")
    
    def stop_background_learning(self):
        """Stop background learning thread"""
        self.stop_learning = True
        if self.learning_thread:
            self.learning_thread.join()
        self.logger.info("Background learning stopped")
    
    def _background_learning_loop(self):
        """Background learning loop"""
        while not self.stop_learning:
            try:
                # Check each gauge type for learning triggers
                for gauge_type in self.models.keys():
                    triggers = self.learning_triggers[gauge_type]
                    
                    # Check if any trigger is active
                    if any(triggers.values()):
                        self.trigger_learning(gauge_type)
                    
                    # Check for scheduled updates
                    last_update = self.performance_history[gauge_type]['timestamps']
                    if last_update:
                        time_since_update = datetime.now() - last_update[-1]
                        if time_since_update > timedelta(hours=self.config['auto_retrain_interval_hours']):
                            self.learning_triggers[gauge_type]['scheduled_update'] = True
                
                # Sleep before next check
                time.sleep(300)  # Check every 5 minutes
                
            except Exception as e:
                self.logger.error(f"Background learning error: {e}")
                time.sleep(60)  # Wait before retrying
    
    def save_models(self, filepath):
        """Save models and state to file"""
        state = {
            'models': {},
            'scalers': self.scalers,
            'config': self.config,
            'performance_history': self.performance_history
        }
        
        # Save sklearn models separately
        for gauge_type, models in self.models.items():
            model_dir = f"{filepath}_{gauge_type}"
            joblib.dump(models, f"{model_dir}_models.pkl")
        
        # Save other state
        with open(f"{filepath}_state.json", 'w') as f:
            # Convert non-serializable objects
            serializable_state = self._make_serializable(state)
            json.dump(serializable_state, f, indent=2)
        
        self.logger.info(f"Models saved to {filepath}")
    
    def load_models(self, filepath):
        """Load models and state from file"""
        try:
            # Load state
            with open(f"{filepath}_state.json", 'r') as f:
                state = json.load(f)
            
            self.scalers = state['scalers']
            self.config = state['config']
            self.performance_history = state['performance_history']
            
            # Load sklearn models
            for gauge_type in state['models'].keys():
                model_file = f"{filepath}_{gauge_type}_models.pkl"
                self.models[gauge_type] = joblib.load(model_file)
            
            self.logger.info(f"Models loaded from {filepath}")
            
        except Exception as e:
            self.logger.error(f"Failed to load models: {e}")
    
    def _make_serializable(self, obj):
        """Convert non-serializable objects to serializable format"""
        if isinstance(obj, dict):
            return {k: self._make_serializable(v) for k, v in obj.items()}
        elif isinstance(obj, list):
            return [self._make_serializable(item) for item in obj]
        elif isinstance(obj, datetime):
            return obj.isoformat()
        elif isinstance(obj, deque):
            return list(obj)
        else:
            return obj
    
    def get_system_status(self):
        """Get system status and performance metrics"""
        status = {
            'active_gauge_types': list(self.models.keys()),
            'total_models': len(self.models),
            'learning_active': self.learning_thread is not None and self.learning_thread.is_alive(),
            'performance_summary': {}
        }
        
        for gauge_type in self.models.keys():
            if self.performance_history[gauge_type]['accuracy']:
                latest_accuracy = self.performance_history[gauge_type]['accuracy'][-1]
                total_samples = sum(self.performance_history[gauge_type]['sample_counts'])
                
                status['performance_summary'][gauge_type] = {
                    'latest_accuracy': latest_accuracy,
                    'total_training_samples': total_samples,
                    'buffer_size': len(self.data_buffers[gauge_type]),
                    'feedback_samples': len(self.feedback_buffers[gauge_type])
                }
        
        return status

# Usage example
def main():
    # Initialize self-learning system
    system = SelfLearningGaugeSystem()
    
    # Generate sample initial data
    initial_data = []
    for i in range(100):
        sample = {
            'features': np.random.rand(10),  # 10 features
            'target': np.random.choice(['normal', 'high', 'low'])
        }
        initial_data.append(sample)
    
    # Initialize model for a gauge type
    system.initialize_model('pressure_gauge_01', initial_data)
    
    # Start background learning
    system.start_background_learning()
    
    # Simulate real-time operation
    for i in range(50):
        # Generate test features
        test_features = np.random.rand(10)
        
        # Make prediction
        result = system.predict('pressure_gauge_01', test_features)
        print(f"Prediction {i+1}: {result['prediction']} (confidence: {result['confidence']:.3f})")
        
        # Simulate feedback (ground truth)
        true_target = np.random.choice(['normal', 'high', 'low'])
        system.add_feedback('pressure_gauge_01', test_features, true_target, result['prediction'])
        
        time.sleep(0.1)  # Simulate real-time interval
    
    # Get system status
    status = system.get_system_status()
    print("\nSystem Status:")
    print(json.dumps(status, indent=2))
    
    # Save models
    system.save_models('self_learning_models')
    
    # Stop background learning
    system.stop_background_learning()

if __name__ == "__main__":
    main()