Makatia
/

adaptive-serdes-lstm-controller

@@ -1,319 +0,0 @@
-#!/usr/bin/env python3
-"""
-Hugging Face Model Interface for Adaptive SerDes LSTM Controller
-================================================================
-This module provides a Hugging Face compatible interface for the
-Adaptive SerDes LSTM Controller model.
-"""
-import torch
-import torch.nn as nn
-import numpy as np
-import json
-from typing import Dict, List, Union, Optional
-from pathlib import Path
-class AdaptiveSerDesLSTM(nn.Module):
-    """
-    LSTM-based Adaptive SerDes Controller
-    Dynamically optimizes 31 SerDes parameters based on 12 channel characteristics
-    for real-time signal integrity optimization in high-speed digital communications.
-    """
-    def __init__(self, input_size=12, hidden_size=256, num_layers=3, output_size=31, dropout=0.2):
-        super(AdaptiveSerDesLSTM, self).__init__()
-        self.input_size = input_size
-        self.hidden_size = hidden_size
-        self.num_layers = num_layers
-        self.output_size = output_size
-        # Input normalization
-        self.input_norm = nn.BatchNorm1d(input_size)
-        # LSTM layers
-        self.lstm1 = nn.LSTM(input_size, hidden_size, batch_first=True, dropout=dropout)
-        self.lstm2 = nn.LSTM(hidden_size, hidden_size, batch_first=True, dropout=dropout)
-        self.lstm3 = nn.LSTM(hidden_size, hidden_size, batch_first=True, dropout=dropout)
-        # Dropout for regularization
-        self.dropout = nn.Dropout(dropout)
-        # Fully connected layers
-        self.fc_layers = nn.Sequential(
-            nn.Linear(hidden_size, 128),
-            nn.ReLU(),
-            nn.Dropout(dropout),
-            nn.Linear(128, 64),
-            nn.ReLU(),
-            nn.Dropout(dropout),
-            nn.Linear(64, output_size),
-            nn.Tanh()  # Output in [-1, 1] range
-        )
-        # Output normalization
-        self.output_norm = nn.BatchNorm1d(output_size)
-    def forward(self, x):
-        """
-        Forward pass through the LSTM controller
-        Args:
-            x: Input tensor of shape (batch_size, sequence_length, input_size)
-               Channel characteristics: [insertion_loss_db, return_loss_db, group_delay_ps, ...]
-        Returns:
-            tensor: Optimized SerDes parameters of shape (batch_size, output_size)
-        """
-        batch_size = x.size(0)
-        # Handle single-step input (most common case)
-        if x.dim() == 2:
-            x = x.unsqueeze(1)  # Add sequence dimension
-        # Normalize input features
-        if x.size(0) > 1:  # Only if batch size > 1
-            x = x.view(-1, x.size(-1))
-            x = self.input_norm(x)
-            x = x.view(batch_size, -1, self.input_size)
-        # LSTM forward pass
-        lstm_out, _ = self.lstm1(x)
-        lstm_out, _ = self.lstm2(lstm_out)
-        lstm_out, _ = self.lstm3(lstm_out)
-        # Take the last time step output
-        lstm_out = lstm_out[:, -1, :]
-        # Apply dropout
-        lstm_out = self.dropout(lstm_out)
-        # Fully connected layers
-        output = self.fc_layers(lstm_out)
-        # Output normalization
-        if output.size(0) > 1:  # Only if batch size > 1
-            output = self.output_norm(output)
-        return output
-class SerDesController:
-    """
-    High-level interface for the Adaptive SerDes LSTM Controller
-    """
-    def __init__(self, model_path: str = "adaptive_serdes_lstm_controller.pth"):
-        """
-        Initialize the SerDes controller
-        Args:
-            model_path: Path to the trained model file
-        """
-        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
-        self.model = None
-        self.config = None
-        self.load_model(model_path)
-    def load_model(self, model_path: str):
-        """Load the trained model"""
-        try:
-            # Load the model checkpoint
-            checkpoint = torch.load(model_path, map_location=self.device, weights_only=False)
-            # Initialize model architecture
-            self.model = AdaptiveSerDesLSTM()
-            # Load the state dict
-            if 'model_state_dict' in checkpoint:
-                self.model.load_state_dict(checkpoint['model_state_dict'])
-            else:
-                self.model.load_state_dict(checkpoint)
-            self.model.to(self.device)
-            self.model.eval()
-            print(f"✅ Model loaded successfully from {model_path}")
-            print(f"📱 Using device: {self.device}")
-        except Exception as e:
-            raise RuntimeError(f"Failed to load model: {e}")
-    def load_config(self, config_path: str = "config.json"):
-        """Load model configuration"""
-        try:
-            with open(config_path, 'r') as f:
-                self.config = json.load(f)
-        except FileNotFoundError:
-            print("⚠️ Config file not found, using defaults")
-            self.config = self._default_config()
-    def _default_config(self) -> Dict:
-        """Default configuration if config.json is not found"""
-        return {
-            "input_features": [
-                "insertion_loss_db", "return_loss_db", "group_delay_ps", "data_rate_gbps",
-                "nyquist_freq_ghz", "eye_height_v", "eye_width_ui", "snr_db",
-                "ber_estimate", "jitter_rms_ui", "amplitude_v", "quality_factor"
-            ],
-            "output_parameters": [
-                f"ffe_tap_{i}" for i in range(7)
-            ] + [
-                f"dfe_tap_{i}" for i in range(8)
-            ] + [
-                "tx_swing_v", "tx_pre_emphasis", "tx_post_emphasis", "tx_slew_rate",
-                "tx_drive_strength", "tx_offset", "tx_skew", "tx_jitter_control",
-                "rx_ctle_gain", "rx_ctle_bandwidth", "rx_vga_gain", "rx_offset_compensation",
-                "rx_dfe_enable", "rx_lms_adaptation", "rx_threshold", "rx_hysteresis"
-            ]
-        }
-    def predict(self, channel_data: Union[Dict, List, np.ndarray, torch.Tensor]) -> Dict:
-        """
-        Predict optimal SerDes parameters for given channel characteristics
-        Args:
-            channel_data: Channel characteristics as dict, list, numpy array, or torch tensor
-        Returns:
-            dict: Optimized SerDes parameters with parameter names and values
-        """
-        # Convert input to tensor
-        if isinstance(channel_data, dict):
-            # Convert dictionary to tensor using feature order
-            features = self.config.get("input_features", [])
-            input_tensor = torch.tensor([channel_data[feat] for feat in features], dtype=torch.float32)
-        elif isinstance(channel_data, (list, np.ndarray)):
-            input_tensor = torch.tensor(channel_data, dtype=torch.float32)
-        elif isinstance(channel_data, torch.Tensor):
-            input_tensor = channel_data.float()
-        else:
-            raise ValueError("Unsupported input type")
-        # Ensure proper shape
-        if input_tensor.dim() == 1:
-            input_tensor = input_tensor.unsqueeze(0)  # Add batch dimension
-        # Move to device
-        input_tensor = input_tensor.to(self.device)
-        # Predict
-        with torch.no_grad():
-            predictions = self.model(input_tensor)
-        # Convert to numpy and create result dictionary
-        predictions_np = predictions.cpu().numpy().flatten()
-        # Get parameter names
-        param_names = self.config.get("output_parameters", [f"param_{i}" for i in range(31)])
-        result = {
-            "parameters": dict(zip(param_names, predictions_np.tolist())),
-            "raw_output": predictions_np.tolist(),
-            "input_shape": list(input_tensor.shape),
-            "output_shape": list(predictions.shape)
-        }
-        return result
-    def analyze_channel(self, s4p_file: Optional[str] = None, **channel_params) -> Dict:
-        """
-        Analyze channel and predict optimal SerDes parameters
-        Args:
-            s4p_file: Optional S4P file path for channel characterization
-            **channel_params: Direct channel parameters
-        Returns:
-            dict: Analysis results with channel characteristics and optimal parameters
-        """
-        if s4p_file:
-            # TODO: Implement S4P file parsing
-            pass
-        # Use provided parameters
-        if not channel_params:
-            # Default example channel
-            channel_params = {
-                "insertion_loss_db": -18.22,
-                "return_loss_db": -16.38,
-                "group_delay_ps": 45.2,
-                "data_rate_gbps": 25.78125,
-                "nyquist_freq_ghz": 12.89,
-                "eye_height_v": 0.85,
-                "eye_width_ui": 0.65,
-                "snr_db": 12.5,
-                "ber_estimate": 1e-12,
-                "jitter_rms_ui": 0.15,
-                "amplitude_v": 2.1,
-                "quality_factor": 0.92
-            }
-        # Predict optimal parameters
-        result = self.predict(channel_params)
-        return {
-            "channel_characteristics": channel_params,
-            "optimal_parameters": result["parameters"],
-            "analysis": {
-                "model_confidence": "high",  # Could be computed from model uncertainty
-                "adaptation_needed": True,
-                "estimated_improvement": {
-                    "eye_height": "+30-50%",
-                    "snr": "+20-35%",
-                    "ber": "1-2 orders of magnitude"
-                }
-            }
-        }
-# Hugging Face compatible interface
-def from_pretrained(model_name_or_path: str) -> SerDesController:
-    """
-    Load model in Hugging Face style
-    Args:
-        model_name_or_path: Local path or Hugging Face model identifier
-    Returns:
-        SerDesController: Loaded model controller
-    """
-    return SerDesController(model_name_or_path)
-# Example usage and testing
-if __name__ == "__main__":
-    # Initialize controller
-    controller = SerDesController()
-    # Example channel data
-    example_channel = {
-        "insertion_loss_db": -18.22,
-        "return_loss_db": -16.38,
-        "group_delay_ps": 45.2,
-        "data_rate_gbps": 25.78125,
-        "nyquist_freq_ghz": 12.89,
-        "eye_height_v": 0.85,
-        "eye_width_ui": 0.65,
-        "snr_db": 12.5,
-        "ber_estimate": 1e-12,
-        "jitter_rms_ui": 0.15,
-        "amplitude_v": 2.1,
-        "quality_factor": 0.92
-    }
-    # Predict optimal parameters
-    result = controller.predict(example_channel)
-    print("🎯 Predicted SerDes Parameters:")
-    for param, value in result["parameters"].items():
-        print(f"  {param}: {value:.4f}")
-    # Full channel analysis
-    analysis = controller.analyze_channel(**example_channel)
-    print(f"\n📊 Analysis Summary:")
-    print(f"  Adaptation needed: {analysis['analysis']['adaptation_needed']}")
-    print(f"  Expected improvements: {analysis['analysis']['estimated_improvement']}")