RSK World - TensorFlow Deep Learning - Project Files | RSK World - Free Programming Resources & Source Code

.gitignore model_evaluation.py model_deployment.py

.gitignore

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg

# Virtual Environment
venv/
env/
ENV/
.venv

# Jupyter Notebook
.ipynb_checkpoints
*.ipynb_checkpoints

# TensorFlow
*.pb
*.h5
*.ckpt
*.index
*.meta
saved_models/
checkpoints/
logs/
tuning/

# Data
# Keep data directory structure but ignore generated files
data/*.npy
data/*.csv
data/*.json
data/visualizations/
!data/.gitkeep
!data/README.md
*.pkl
*.pickle
*.npz

# IDE
.vscode/
.idea/
*.swp
*.swo
*~

# OS
.DS_Store
Thumbs.db

# Project specific
models/*.h5
models/*.pb
*.tflite
tfjs_model/

72 lines•744 B

text

src/model_evaluation.py

Raw Download

"""
Model Evaluation and Metrics with TensorFlow
Author: RSK World
Website: https://rskworld.in
Email: help@rskworld.in
Phone: +91 93305 39277

This module provides comprehensive model evaluation and metrics.
"""

import tensorflow as tf
from tensorflow import keras
import numpy as np
from sklearn.metrics import (
    accuracy_score, precision_score, recall_score, f1_score,
    confusion_matrix, classification_report, roc_auc_score,
    roc_curve, precision_recall_curve, average_precision_score
)
import matplotlib.pyplot as plt
import seaborn as sns

class ModelEvaluator:
    """
    Model evaluation utilities.
    Author: RSK World - https://rskworld.in
    """
    
    def __init__(self, model):
        """
        Initialize model evaluator.
        
        Args:
            model: Trained Keras model
        """
        self.model = model
    
    def evaluate_classification(self, X_test, y_test, class_names=None):
        """
        Evaluate classification model.
        
        Args:
            X_test: Test features
            y_test: Test labels
            class_names: List of class names
        
        Returns:
            Dictionary with evaluation metrics
        """
        # Get predictions
        y_pred_proba = self.model.predict(X_test, verbose=0)
        y_pred = np.argmax(y_pred_proba, axis=1)
        
        # Handle one-hot encoded labels
        if len(y_test.shape) > 1 and y_test.shape[1] > 1:
            y_test = np.argmax(y_test, axis=1)
        
        # Calculate metrics
        accuracy = accuracy_score(y_test, y_pred)
        precision = precision_score(y_test, y_pred, average='weighted', zero_division=0)
        recall = recall_score(y_test, y_pred, average='weighted', zero_division=0)
        f1 = f1_score(y_test, y_pred, average='weighted', zero_division=0)
        
        # Confusion matrix
        cm = confusion_matrix(y_test, y_pred)
        
        # Classification report
        report = classification_report(y_test, y_pred, target_names=class_names, output_dict=True)
        
        metrics = {
            'accuracy': accuracy,
            'precision': precision,
            'recall': recall,
            'f1_score': f1,
            'confusion_matrix': cm,
            'classification_report': report,
            'predictions': y_pred,
            'prediction_probabilities': y_pred_proba
        }
        
        return metrics
    
    def evaluate_regression(self, X_test, y_test):
        """
        Evaluate regression model.
        
        Args:
            X_test: Test features
            y_test: Test labels
        
        Returns:
            Dictionary with evaluation metrics
        """
        # Get predictions
        y_pred = self.model.predict(X_test, verbose=0).flatten()
        
        # Calculate metrics
        mse = np.mean((y_test - y_pred) ** 2)
        rmse = np.sqrt(mse)
        mae = np.mean(np.abs(y_test - y_pred))
        r2 = 1 - (np.sum((y_test - y_pred) ** 2) / np.sum((y_test - np.mean(y_test)) ** 2))
        
        metrics = {
            'mse': mse,
            'rmse': rmse,
            'mae': mae,
            'r2_score': r2,
            'predictions': y_pred
        }
        
        return metrics
    
    def plot_confusion_matrix(self, X_test, y_test, class_names=None, figsize=(10, 8)):
        """
        Plot confusion matrix.
        
        Args:
            X_test: Test features
            y_test: Test labels
            class_names: List of class names
            figsize: Figure size
        """
        metrics = self.evaluate_classification(X_test, y_test, class_names)
        cm = metrics['confusion_matrix']
        
        plt.figure(figsize=figsize)
        sns.heatmap(cm, annot=True, fmt='d', cmap='Blues',
                    xticklabels=class_names, yticklabels=class_names)
        plt.title('Confusion Matrix')
        plt.ylabel('True Label')
        plt.xlabel('Predicted Label')
        plt.tight_layout()
        plt.show()
    
    def plot_roc_curve(self, X_test, y_test, class_index=0):
        """
        Plot ROC curve for binary classification.
        
        Args:
            X_test: Test features
            y_test: Test labels
            class_index: Class index for ROC curve
        """
        y_pred_proba = self.model.predict(X_test, verbose=0)
        
        if len(y_pred_proba.shape) > 1:
            y_scores = y_pred_proba[:, class_index]
        else:
            y_scores = y_pred_proba
        
        # Handle one-hot encoded labels
        if len(y_test.shape) > 1:
            y_true = y_test[:, class_index]
        else:
            y_true = (y_test == class_index).astype(int)
        
        fpr, tpr, thresholds = roc_curve(y_true, y_scores)
        auc = roc_auc_score(y_true, y_scores)
        
        plt.figure(figsize=(8, 6))
        plt.plot(fpr, tpr, label=f'ROC curve (AUC = {auc:.2f})')
        plt.plot([0, 1], [0, 1], 'k--', label='Random')
        plt.xlabel('False Positive Rate')
        plt.ylabel('True Positive Rate')
        plt.title('ROC Curve')
        plt.legend()
        plt.grid(True)
        plt.tight_layout()
        plt.show()
    
    def plot_precision_recall_curve(self, X_test, y_test, class_index=0):
        """
        Plot precision-recall curve.
        
        Args:
            X_test: Test features
            y_test: Test labels
            class_index: Class index
        """
        y_pred_proba = self.model.predict(X_test, verbose=0)
        
        if len(y_pred_proba.shape) > 1:
            y_scores = y_pred_proba[:, class_index]
        else:
            y_scores = y_pred_proba
        
        # Handle one-hot encoded labels
        if len(y_test.shape) > 1:
            y_true = y_test[:, class_index]
        else:
            y_true = (y_test == class_index).astype(int)
        
        precision, recall, thresholds = precision_recall_curve(y_true, y_scores)
        ap = average_precision_score(y_true, y_scores)
        
        plt.figure(figsize=(8, 6))
        plt.plot(recall, precision, label=f'PR curve (AP = {ap:.2f})')
        plt.xlabel('Recall')
        plt.ylabel('Precision')
        plt.title('Precision-Recall Curve')
        plt.legend()
        plt.grid(True)
        plt.tight_layout()
        plt.show()
    
    def get_model_summary(self):
        """
        Get model summary information.
        
        Returns:
            Dictionary with model information
        """
        total_params = self.model.count_params()
        trainable_params = sum([tf.keras.backend.count_params(w) for w in self.model.trainable_weights])
        non_trainable_params = sum([tf.keras.backend.count_params(w) for w in self.model.non_trainable_weights])
        
        summary = {
            'total_params': total_params,
            'trainable_params': trainable_params,
            'non_trainable_params': non_trainable_params,
            'num_layers': len(self.model.layers),
            'input_shape': self.model.input_shape,
            'output_shape': self.model.output_shape
        }
        
        return summary

def calculate_model_complexity(model):
    """
    Calculate model complexity metrics.
    
    Args:
        model: Keras model
    
    Returns:
        Dictionary with complexity metrics
    """
    total_params = model.count_params()
    trainable_params = sum([tf.keras.backend.count_params(w) for w in model.trainable_weights])
    
    # Estimate model size (assuming float32)
    model_size_mb = total_params * 4 / (1024 * 1024)
    
    complexity = {
        'total_parameters': total_params,
        'trainable_parameters': trainable_params,
        'non_trainable_parameters': total_params - trainable_params,
        'estimated_size_mb': model_size_mb,
        'number_of_layers': len(model.layers)
    }
    
    return complexity

def compare_models(models, X_test, y_test, model_names=None):
    """
    Compare multiple models.
    
    Args:
        models: List of models
        X_test: Test features
        y_test: Test labels
        model_names: List of model names
    
    Returns:
        DataFrame with comparison results
    """
    import pandas as pd
    
    results = []
    
    for i, model in enumerate(models):
        evaluator = ModelEvaluator(model)
        metrics = evaluator.evaluate_classification(X_test, y_test)
        
        model_name = model_names[i] if model_names else f'Model {i+1}'
        
        results.append({
            'Model': model_name,
            'Accuracy': metrics['accuracy'],
            'Precision': metrics['precision'],
            'Recall': metrics['recall'],
            'F1 Score': metrics['f1_score']
        })
    
    return pd.DataFrame(results)

def example_usage():
    """
    Example usage of model evaluation functions.
    """
    # Create a simple model for demonstration
    from tensorflow import keras
    from tensorflow.keras import layers
    
    model = keras.Sequential([
        layers.Dense(128, activation='relu', input_shape=(784,)),
        layers.Dense(64, activation='relu'),
        layers.Dense(10, activation='softmax')
    ])
    
    model.compile(
        optimizer='adam',
        loss='sparse_categorical_crossentropy',
        metrics=['accuracy']
    )
    
    # Generate dummy data
    X_test = np.random.randn(100, 784).astype('float32')
    y_test = np.random.randint(0, 10, 100)
    
    # Train briefly
    X_train = np.random.randn(1000, 784).astype('float32')
    y_train = np.random.randint(0, 10, 1000)
    model.fit(X_train, y_train, epochs=1, verbose=0)
    
    # Evaluate model
    evaluator = ModelEvaluator(model)
    metrics = evaluator.evaluate_classification(X_test, y_test)
    
    print("Model Evaluation Metrics:")
    print(f"Accuracy: {metrics['accuracy']:.4f}")
    print(f"Precision: {metrics['precision']:.4f}")
    print(f"Recall: {metrics['recall']:.4f}")
    print(f"F1 Score: {metrics['f1_score']:.4f}")
    
    # Get model summary
    summary = evaluator.get_model_summary()
    print(f"\nModel Summary:")
    print(f"Total Parameters: {summary['total_params']:,}")
    print(f"Trainable Parameters: {summary['trainable_params']:,}")
    
    return evaluator, metrics

if __name__ == '__main__':
    print("Model Evaluation and Metrics with TensorFlow")
    print("Author: RSK World - https://rskworld.in")
    evaluator, metrics = example_usage()

337 lines•10.5 KB

python

src/model_deployment.py

Raw Download

"""
Model Deployment and Serving with TensorFlow
Author: RSK World
Website: https://rskworld.in
Email: help@rskworld.in
Phone: +91 93305 39277

This module demonstrates model saving, loading, and deployment strategies.
"""

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import numpy as np
import os
import json

def save_model_complete(model, model_dir='./saved_models'):
    """
    Save model in multiple formats for different deployment scenarios.
    
    Args:
        model: Keras model to save
        model_dir: Directory to save models
    """
    os.makedirs(model_dir, exist_ok=True)
    
    # 1. Save as SavedModel format (recommended)
    savedmodel_path = os.path.join(model_dir, 'savedmodel')
    model.save(savedmodel_path, save_format='tf')
    print(f"Model saved as SavedModel: {savedmodel_path}")
    
    # 2. Save as H5 format
    h5_path = os.path.join(model_dir, 'model.h5')
    model.save(h5_path, save_format='h5')
    print(f"Model saved as H5: {h5_path}")
    
    # 3. Save only weights
    weights_path = os.path.join(model_dir, 'weights.h5')
    model.save_weights(weights_path)
    print(f"Weights saved: {weights_path}")
    
    # 4. Save model architecture as JSON
    json_path = os.path.join(model_dir, 'model_architecture.json')
    model_json = model.to_json()
    with open(json_path, 'w') as f:
        json.dump(json.loads(model_json), f, indent=2)
    print(f"Model architecture saved: {json_path}")
    
    return savedmodel_path, h5_path, weights_path, json_path

def load_model_from_savedmodel(model_path):
    """
    Load model from SavedModel format.
    
    Args:
        model_path: Path to SavedModel directory
    
    Returns:
        Loaded Keras model
    """
    model = keras.models.load_model(model_path)
    print(f"Model loaded from: {model_path}")
    return model

def load_model_from_h5(h5_path):
    """
    Load model from H5 format.
    
    Args:
        h5_path: Path to H5 file
    
    Returns:
        Loaded Keras model
    """
    model = keras.models.load_model(h5_path)
    print(f"Model loaded from: {h5_path}")
    return model

def convert_to_tflite(model, tflite_path='./model.tflite', quantize=False):
    """
    Convert model to TensorFlow Lite format for mobile deployment.
    
    Args:
        model: Keras model to convert
        tflite_path: Path to save TFLite model
        quantize: Whether to apply quantization
    
    Returns:
        Path to TFLite model
    """
    converter = tf.lite.TFLiteConverter.from_keras_model(model)
    
    if quantize:
        # Apply quantization
        converter.optimizations = [tf.lite.Optimize.DEFAULT]
    
    tflite_model = converter.convert()
    
    with open(tflite_path, 'wb') as f:
        f.write(tflite_model)
    
    print(f"TFLite model saved: {tflite_path}")
    return tflite_path

def convert_to_tensorflow_js(model, js_dir='./tfjs_model'):
    """
    Convert model to TensorFlow.js format for web deployment.
    
    Args:
        model: Keras model to convert
        js_dir: Directory to save TensorFlow.js model
    """
    try:
        import tensorflowjs as tfjs
        os.makedirs(js_dir, exist_ok=True)
        tfjs.converters.save_keras_model(model, js_dir)
        print(f"TensorFlow.js model saved: {js_dir}")
    except ImportError:
        print("Warning: tensorflowjs not installed. Install it with: pip install tensorflowjs")
        raise

def create_tf_serving_model(model, serving_dir='./serving_model'):
    """
    Prepare model for TensorFlow Serving.
    
    Args:
        model: Keras model to prepare
        serving_dir: Directory to save serving model
    """
    os.makedirs(serving_dir, exist_ok=True)
    
    # Save model with version number (required by TF Serving)
    version_dir = os.path.join(serving_dir, '1')
    os.makedirs(version_dir, exist_ok=True)
    
    model.save(version_dir, save_format='tf')
    print(f"Model prepared for TF Serving: {serving_dir}")

def create_prediction_function(model):
    """
    Create a prediction function wrapper for easier deployment.
    
    Args:
        model: Trained Keras model
    
    Returns:
        Prediction function
    """
    def predict(input_data):
        """
        Make predictions on input data.
        
        Args:
            input_data: Input data (numpy array or list)
        
        Returns:
            Predictions
        """
        # Preprocess input if needed
        if isinstance(input_data, list):
            input_data = np.array(input_data)
        
        # Make prediction
        predictions = model.predict(input_data, verbose=0)
        
        return predictions
    
    return predict

def create_rest_api_wrapper(model, model_name='tensorflow_model'):
    """
    Create a REST API wrapper template for model serving.
    
    Args:
        model: Trained Keras model
        model_name: Name of the model
    
    Returns:
        Flask app code template (as string)
    """
    flask_code = f"""
# Flask REST API for {model_name}
# Author: RSK World - https://rskworld.in

from flask import Flask, request, jsonify
import numpy as np
import tensorflow as tf
from tensorflow import keras

app = Flask(__name__)

# Load model
model = keras.models.load_model('./saved_models/savedmodel')

@app.route('/predict', methods=['POST'])
def predict():
    try:
        # Get input data
        data = request.json
        input_data = np.array(data['input'])
        
        # Make prediction
        predictions = model.predict(input_data, verbose=0)
        
        # Return results
        return jsonify({{
            'success': True,
            'predictions': predictions.tolist()
        }})
    except Exception as e:
        return jsonify({{
            'success': False,
            'error': str(e)
        }}), 400

@app.route('/health', methods=['GET'])
def health():
    return jsonify({{'status': 'healthy'}})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)
"""
    
    return flask_code

def benchmark_model(model, test_data, batch_sizes=[1, 8, 16, 32, 64]):
    """
    Benchmark model inference performance.
    
    Args:
        model: Keras model to benchmark
        test_data: Test data for benchmarking
        batch_sizes: List of batch sizes to test
    
    Returns:
        Dictionary with benchmark results
    """
    results = {}
    
    for batch_size in batch_sizes:
        # Warm up
        _ = model.predict(test_data[:batch_size], verbose=0)
        
        # Benchmark
        import time
        start_time = time.time()
        _ = model.predict(test_data[:batch_size*10], batch_size=batch_size, verbose=0)
        elapsed_time = time.time() - start_time
        
        results[batch_size] = {
            'time': elapsed_time,
            'samples_per_second': (batch_size * 10) / elapsed_time
        }
    
    return results

def example_usage():
    """
    Example usage of deployment functions.
    """
    # Create a simple model
    model = keras.Sequential([
        layers.Dense(128, activation='relu', input_shape=(784,)),
        layers.Dense(64, activation='relu'),
        layers.Dense(10, activation='softmax')
    ])
    
    model.compile(
        optimizer='adam',
        loss='sparse_categorical_crossentropy',
        metrics=['accuracy']
    )
    
    # Train model (using dummy data)
    X_train = np.random.randn(1000, 784).astype('float32')
    y_train = np.random.randint(0, 10, 1000)
    
    model.fit(X_train, y_train, epochs=5, verbose=0)
    
    # Save model in multiple formats
    savedmodel_path, h5_path, weights_path, json_path = save_model_complete(model)
    
    # Convert to TFLite
    tflite_path = convert_to_tflite(model, quantize=False)
    
    # Create prediction function
    predict_fn = create_prediction_function(model)
    
    # Test prediction
    test_input = np.random.randn(1, 784).astype('float32')
    predictions = predict_fn(test_input)
    print(f"\nPredictions shape: {predictions.shape}")
    
    # Benchmark model
    test_data = np.random.randn(100, 784).astype('float32')
    benchmark_results = benchmark_model(model, test_data)
    print("\nBenchmark Results:")
    for batch_size, result in benchmark_results.items():
        print(f"Batch size {batch_size}: {result['samples_per_second']:.2f} samples/sec")
    
    return model

if __name__ == '__main__':
    print("Model Deployment and Serving with TensorFlow")
    print("Author: RSK World - https://rskworld.in")
    model = example_usage()

308 lines•8.7 KB

python

Theme Settings

Color Scheme

Display Options

Font Size

Theme Settings

Color Scheme

Display Options

Font Size

🚀 Support RSK World