Added modularity

app.py

@@ -14,215 +14,11 @@ import torch.fft as fft
 import xgboost as xgb
 from torch.utils.data import DataLoader, TensorDataset
 import time
-
-# Define the TCN model
-class TCN(nn.Module):
-    def __init__(self, input_size, hidden_size, output_size, num_layers=3, dropout=0.1):
-        super(TCN, self).__init__()
-        
-        # List to hold convolutional layers
-        self.convs = nn.ModuleList()
-        dropout = dropout if num_layers > 1 else 0  # No dropout if only one layer
-        self.dropout = nn.Dropout(dropout)
-        
-        # Create the convolutional layers
-        for i in range(num_layers):
-            in_channels = input_size if i == 0 else hidden_size  # First layer uses input_size, others use hidden_size
-            out_channels = hidden_size  # All layers have the same hidden size
-            self.convs.append(nn.Conv1d(in_channels, out_channels, kernel_size=2, padding=1))
-        
-        # Fully connected output layer
-        self.fc = nn.Linear(hidden_size, output_size)
-
-    def forward(self, x):
-        x = x.permute(0, 2, 1)  # Change to (batch_size, features, timesteps)
-        
-        # Apply each convolutional layer followed by dropout
-        for conv in self.convs:
-            x = torch.relu(conv(x))
-            x = self.dropout(x)  # Apply dropout after each convolution
-        
-        x = torch.mean(x, dim=2)  # Global average pooling
-        x = self.fc(x)  # Output layer
-        return x
-
-# Define the Temporal Fusion Transformer (Temporal Fusion Transformer) model
-class TemporalFusionTransformer(nn.Module):
-    def __init__(self, input_size, hidden_size, output_size, num_layers=3, dropout=0.1):
-        super(TemporalFusionTransformer, self).__init__()
-        # Encoder and Decoder LSTMs with multiple layers
-        self.encoder = nn.LSTM(input_size, hidden_size, num_layers=num_layers, batch_first=True, dropout=dropout)
-        self.decoder = nn.LSTM(hidden_size, hidden_size, num_layers=num_layers, batch_first=True, dropout=dropout)     
-        
-        self.attention = nn.MultiheadAttention(hidden_size, num_heads=4, batch_first=True) # Attention mechanism
-        self.fc = nn.Linear(hidden_size, output_size) # Fully connected output layer
-        self.dropout = nn.Dropout(dropout) # Dropout layer
-
-    def forward(self, x):
-        encoder_output, _ = self.encoder(x) # Encoder output
-        decoder_output, _ = self.decoder(encoder_output) # Decoder output
-        attention_output, _ = self.attention(decoder_output, encoder_output, encoder_output) # Attention output
-        attention_output = self.dropout(attention_output) # Apply dropout
-        output = self.fc(attention_output[:, -1, :]) # Take the last time step from the attention output
-        return output
-    
-
-# Build the ETSformer Class: Encoder, Trend, Seasonality, Exponential Smoothing, and Output Layer
-class ETSformer(nn.Module):
-    def __init__(self, input_size, hidden_size, output_size, num_layers=3, dropout=0.1):
-        super(ETSformer, self).__init__()
-
-        # Encoder: LSTM with multiple layers and dropout
-        self.encoder = nn.LSTM(
-            input_size, 
-            hidden_size, 
-            num_layers=num_layers, 
-            batch_first=True, 
-            dropout=dropout if num_layers > 1 else 0.0  # Dropout only applies if num_layers > 1
-        )
-
-        # Trend, Seasonality, Exponential Modules
-        self.trend_module = nn.Sequential(
-            nn.Linear(hidden_size, hidden_size),
-            nn.Dropout(dropout)  # Dropout in the trend module
-        )
-        self.seasonality_module = nn.Sequential(
-            nn.Linear(hidden_size, hidden_size),
-            nn.Dropout(dropout)  # Dropout in the seasonality module
-        )
-        self.exponential_module = nn.Sequential(
-            nn.Linear(hidden_size, hidden_size),
-            nn.Dropout(dropout)  # Dropout in the exponential module
-        )
-
-        self.fc = nn.Linear(hidden_size, output_size) # Fully connected layer for output
-
-    def forward(self, x):
-        encoder_output, _ = self.encoder(x) # Encode the input sequence
-        trend = self.trend_module(encoder_output )# Trend Component
-        # Seasonality Component
-        freq = fft.fft(encoder_output, dim=1)  # Frequency domain transformation
-        seasonality = fft.ifft(self.seasonality_module(torch.abs(freq)), dim=1).real
-        exponential = torch.sigmoid(self.exponential_module(encoder_output)) # Exponential Smoothing Component
-        combined = trend + seasonality + exponential # Combine the components
-        # Output layer: Use the last time step for predictions
-        output = self.fc(combined[:, -1, :])
-        return output
-
-# Updated BiLSTM to handle variable layers
-class BiLSTM(nn.Module):
-    def __init__(self, input_size, hidden_size, output_size, num_layers=2, dropout=0.1):
-        super(BiLSTM, self).__init__()
-        self.bilstm = nn.LSTM(
-            input_size, 
-            hidden_size, 
-            num_layers=num_layers, 
-            batch_first=True, 
-            bidirectional=True, 
-            dropout=dropout if num_layers > 1 else 0  # Dropout only applies for num_layers > 1
-        )
-        self.fc = nn.Linear(hidden_size * 2, output_size)  # Multiply hidden_size by 2 for bidirectional
-        
-    def forward(self, x):
-        bilstm_output, _ = self.bilstm(x)
-        output = self.fc(bilstm_output[:, -1, :])  # Use the last time step
-        return output
-
-class RespFusion(nn.Module):
-    def __init__(self, tft_model, tcn_model, ets_model, bilstm_model, meta_learner_path=None, weights=None, strategy='stacking',):
-        super(RespFusion, self).__init__()
-        self.tft = tft_model
-        self.tcn = tcn_model
-        self.ets = ets_model
-        self.bilstm = bilstm_model
-        self.strategy = strategy  # 'stacking' or other strategies
-
-        # Initialize XGBoost meta-learner
-        self.meta_learner = xgb.XGBRegressor()  # Or XGBClassifier for classification
-
-        # Load the meta-learner if a path is provided
-        if meta_learner_path is not None:
-            self.meta_learner.load_model(meta_learner_path)
-            print(f"Meta-learner loaded from {meta_learner_path}")
-
-        # Storage for stacking training data
-        self.stacking_features = []
-        self.stacking_targets = []
-        
-        # Set model weights for ensembling, default to equal weights for weighted_average strategy
-        if strategy == 'weighted_average':
-            if weights is None:
-                self.weights = [1.0, 1.0, 1.0, 1.0]
-            else:
-                assert len(weights) == 4, "Weights must match the number of models."
-                self.weights = weights
-                
-        
-    def forward(self, x):
-        # Get predictions from each base model
-        tft_output = self.tft(x).detach().cpu().numpy()
-        tcn_output = self.tcn(x).detach().cpu().numpy()
-        ets_output = self.ets(x).detach().cpu().numpy()
-        bilstm_output = self.bilstm(x).detach().cpu().numpy()
-
-        if self.strategy == 'stacking':
-            # Combine outputs into features for the meta-learner
-            features = np.column_stack((tft_output, tcn_output, ets_output, bilstm_output))
-            # During inference, use the meta-learner to make predictions
-            ensemble_output = self.meta_learner.predict(features)
-            return torch.tensor(ensemble_output).to(x.device).float()
-        
-        elif self.strategy == 'voting':
-            # For soft voting, calculate the average
-            ensemble_output = torch.mean(torch.stack([tft_output, tcn_output, ets_output, bilstm_output], dim=0), dim=0)
-            return ensemble_output
-
-        elif self.strategy == 'weighted_average':
-            # Weighted average of outputs
-            ensemble_output = (
-                self.weights[0] * tft_output +
-                self.weights[1] * tcn_output +
-                self.weights[2] * ets_output +
-                self.weights[3] * bilstm_output
-            ) / sum(self.weights)
-            return ensemble_output
-        
-        elif self.strategy == 'simple_average':
-            # Simple average of outputs
-            ensemble_output = (tft_output + tcn_output + ets_output + bilstm_output) / 4
-            return ensemble_output
-        
-        
-        else:
-            raise ValueError(f"Invalid strategy: {self.strategy}. Currently supports only 'stacking', 'voting', 'weighted_average', and 'simple_average'.")
-
-    def collect_stacking_data(self, x, y):
-        """Collect base model outputs and corresponding targets for meta-learner training."""
-        tft_output = self.tft(x).detach().cpu().numpy()
-        tcn_output = self.tcn(x).detach().cpu().numpy()
-        ets_output = self.ets(x).detach().cpu().numpy()
-        bilstm_output = self.bilstm(x).detach().cpu().numpy()
-
-        # Stack features and store
-        features = np.column_stack((tft_output, tcn_output, ets_output, bilstm_output))
-        self.stacking_features.append(features)
-        self.stacking_targets.append(y.detach().cpu().numpy())
-
-    def train_meta_learner(self, save_path=None):
-        """Train the XGBoost meta-learner on collected data and save the model."""
-        # Concatenate all collected features and targets
-        X = np.vstack(self.stacking_features)
-        y = np.concatenate(self.stacking_targets)
-
-        # Train the XGBoost model
-        self.meta_learner.fit(X, y)
-        print("Meta-learner trained successfully!")
-
-        # Save the trained meta-learner
-        if save_path:
-            self.meta_learner.save_model(save_path)
-            print(f"Meta-learner saved to {save_path}")
-
+from models.temporal_fusion_transformer import TemporalFusionTransformer
+from models.tcn import TCN
+from models.etsformer import ETSformer
+from models.bilstm import BiLSTM
+from models.respfusion import RespFusion 
 
 def process_video(video_path):
     # Parameters

models/bilstm.py

@@ -1,19 +1,5 @@
-import gradio as gr
-import torch
-import cv2
-import numpy as np
-import pandas as pd
-from scipy.signal import find_peaks, savgol_filter
-from collections import Counter
-from tqdm import tqdm
-import time
-import os
-import torch
 import torch.nn as nn
-import torch.fft as fft
-import xgboost as xgb
-from torch.utils.data import DataLoader, TensorDataset
-import time
+
 
 # Updated BiLSTM to handle variable layers
 class BiLSTM(nn.Module):

models/etsformer.py

@@ -1,19 +1,6 @@
-import gradio as gr
-import torch
-import cv2
-import numpy as np
-import pandas as pd
-from scipy.signal import find_peaks, savgol_filter
-from collections import Counter
-from tqdm import tqdm
-import time
-import os
 import torch
 import torch.nn as nn
 import torch.fft as fft
-import xgboost as xgb
-from torch.utils.data import DataLoader, TensorDataset
-import time
 
 # Build the ETSformer Class: Encoder, Trend, Seasonality, Exponential Smoothing, and Output Layer
 class ETSformer(nn.Module):

models/respfusion.py

@@ -0,0 +1,100 @@
+import torch
+import torch.nn as nn
+import xgboost as xgb
+import numpy as np
+
+class RespFusion(nn.Module):
+    def __init__(self, tft_model, tcn_model, ets_model, bilstm_model, meta_learner_path=None, weights=None, strategy='stacking',):
+        super(RespFusion, self).__init__()
+        self.tft = tft_model
+        self.tcn = tcn_model
+        self.ets = ets_model
+        self.bilstm = bilstm_model
+        self.strategy = strategy  # 'stacking' or other strategies
+
+        # Initialize XGBoost meta-learner
+        self.meta_learner = xgb.XGBRegressor()  # Or XGBClassifier for classification
+
+        # Load the meta-learner if a path is provided
+        if meta_learner_path is not None:
+            self.meta_learner.load_model(meta_learner_path)
+            print(f"Meta-learner loaded from {meta_learner_path}")
+
+        # Storage for stacking training data
+        self.stacking_features = []
+        self.stacking_targets = []
+        
+        # Set model weights for ensembling, default to equal weights for weighted_average strategy
+        if strategy == 'weighted_average':
+            if weights is None:
+                self.weights = [1.0, 1.0, 1.0, 1.0]
+            else:
+                assert len(weights) == 4, "Weights must match the number of models."
+                self.weights = weights
+                
+        
+    def forward(self, x):
+        # Get predictions from each base model
+        tft_output = self.tft(x).detach().cpu().numpy()
+        tcn_output = self.tcn(x).detach().cpu().numpy()
+        ets_output = self.ets(x).detach().cpu().numpy()
+        bilstm_output = self.bilstm(x).detach().cpu().numpy()
+
+        if self.strategy == 'stacking':
+            # Combine outputs into features for the meta-learner
+            features = np.column_stack((tft_output, tcn_output, ets_output, bilstm_output))
+            # During inference, use the meta-learner to make predictions
+            ensemble_output = self.meta_learner.predict(features)
+            return torch.tensor(ensemble_output).to(x.device).float()
+        
+        elif self.strategy == 'voting':
+            # For soft voting, calculate the average
+            ensemble_output = torch.mean(torch.stack([tft_output, tcn_output, ets_output, bilstm_output], dim=0), dim=0)
+            return ensemble_output
+
+        elif self.strategy == 'weighted_average':
+            # Weighted average of outputs
+            ensemble_output = (
+                self.weights[0] * tft_output +
+                self.weights[1] * tcn_output +
+                self.weights[2] * ets_output +
+                self.weights[3] * bilstm_output
+            ) / sum(self.weights)
+            return ensemble_output
+        
+        elif self.strategy == 'simple_average':
+            # Simple average of outputs
+            ensemble_output = (tft_output + tcn_output + ets_output + bilstm_output) / 4
+            return ensemble_output
+        
+        
+        else:
+            raise ValueError(f"Invalid strategy: {self.strategy}. Currently supports only 'stacking', 'voting', 'weighted_average', and 'simple_average'.")
+
+    def collect_stacking_data(self, x, y):
+        """Collect base model outputs and corresponding targets for meta-learner training."""
+        tft_output = self.tft(x).detach().cpu().numpy()
+        tcn_output = self.tcn(x).detach().cpu().numpy()
+        ets_output = self.ets(x).detach().cpu().numpy()
+        bilstm_output = self.bilstm(x).detach().cpu().numpy()
+
+        # Stack features and store
+        features = np.column_stack((tft_output, tcn_output, ets_output, bilstm_output))
+        self.stacking_features.append(features)
+        self.stacking_targets.append(y.detach().cpu().numpy())
+
+    def train_meta_learner(self, save_path=None):
+        """Train the XGBoost meta-learner on collected data and save the model."""
+        # Concatenate all collected features and targets
+        X = np.vstack(self.stacking_features)
+        y = np.concatenate(self.stacking_targets)
+
+        # Train the XGBoost model
+        self.meta_learner.fit(X, y)
+        print("Meta-learner trained successfully!")
+
+        # Save the trained meta-learner
+        if save_path:
+            self.meta_learner.save_model(save_path)
+            print(f"Meta-learner saved to {save_path}")
+

models/tcn.py

@@ -1,19 +1,5 @@
-import gradio as gr
-import torch
-import cv2
-import numpy as np
-import pandas as pd
-from scipy.signal import find_peaks, savgol_filter
-from collections import Counter
-from tqdm import tqdm
-import time
-import os
 import torch
 import torch.nn as nn
-import torch.fft as fft
-import xgboost as xgb
-from torch.utils.data import DataLoader, TensorDataset
-import time
 
 # Define the TCN model
 class TCN(nn.Module):

models/{temporalfusiontransformer.py → temporal_fusion_transformer.py}

@@ -1,19 +1,5 @@
-import gradio as gr
-import torch
-import cv2
-import numpy as np
-import pandas as pd
-from scipy.signal import find_peaks, savgol_filter
-from collections import Counter
-from tqdm import tqdm
-import time
-import os
-import torch
 import torch.nn as nn
-import torch.fft as fft
-import xgboost as xgb
-from torch.utils.data import DataLoader, TensorDataset
-import time
+
 
 # Define the Temporal Fusion Transformer (Temporal Fusion Transformer) model
 class TemporalFusionTransformer(nn.Module):