Upload 4 files

S3_101_102_103_validation_epoch_10.pth

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:15b55be6c0e57c14ddfece3466f7ab50415f68105ef9105ff0c1ea69616baa50
+size 1100794

data_loader_interface.py

@@ -0,0 +1,115 @@
+import math
+import random
+import matplotlib.pyplot as plt
+import scipy
+import statistics as st
+import os
+import numpy as np
+
+from sklearn.base import TransformerMixin
+from sklearn.preprocessing import StandardScaler, RobustScaler, MinMaxScaler, LabelEncoder
+
+
+class NDStandardScaler(TransformerMixin):
+    def __init__(self, **kwargs):
+        self._scaler = StandardScaler(copy=True, **kwargs)
+        self._orig_shape = None
+
+    def fit(self, X, **kwargs):
+        X = np.array(X)
+        # Save the original shape to reshape the flattened X later
+        # back to its original shape
+        if len(X.shape) > 1:
+            self._orig_shape = X.shape[1:]
+        
+        X = self._flatten(X)    # reshape data into two-dimensioal format suitable for standardScaler
+        self._scaler.fit(X, **kwargs)
+        return self
+
+    def transform(self, X, **kwargs):
+        X = np.array(X)         # convert X to numpy array
+        X = self._flatten(X)    # shape data into the same format used during fit    
+        X = self._scaler.transform(X, **kwargs)
+        X = self._reshape(X)    # reshape in original shape    
+        return X
+
+    def _flatten(self, X):
+        # Reshape X to <= 2 dimensions -> [10, 28, 28, 3] to [10, 28*28*3]
+        if len(X.shape) > 2:
+            n_dims = np.prod(self._orig_shape)  # calculates the number of elements in original shape
+            X = X.reshape(-1, n_dims)           # reshapes the data into two dimensions
+        return X
+
+    def _reshape(self, X):
+        # Reshape X back to it's original shape
+        if len(X.shape) >= 2:
+            X = X.reshape(-1, *self._orig_shape)
+        return X
+
+
+# substraction of mean of each column from the data: reduce bias, normalization
+def mean_sub(data):
+    EPSILON = 1e-12
+    for i in range(data.shape[1]):
+        data[:, i] = data[:, i] - np.mean(data[:, i])  # / np.std(data[subject][action]['imu_signals'][:,i,c])
+    return data
+
+
+def degree_to_radians(signals):
+    return np.array(list(map(lambda signal: np.array(list(map(math.radians, signal))), signals)))
+
+
+def get_start_indices(n_timesteps, window_len, step_size):
+    n_timesteps_valid = n_timesteps - window_len + 1
+    if step_size <= 0:
+        step_size = 1
+
+    start_indices = np.arange(0, n_timesteps_valid, step_size,
+        dtype=int)
+    return start_indices
+
+
+def get_data(data):
+    window_len = 243
+    step_size = 1
+    X = []
+
+    print("data shape: ",data.shape)
+
+    start_indices = get_start_indices(data.shape[0], window_len=window_len, step_size=step_size)
+    for k in start_indices:
+        this_window_data = data[k:k + window_len, :]
+        this_window_data = mean_sub(this_window_data)
+
+        X.append(this_window_data)
+
+    print("data shape after: ",np.array(X).shape)
+
+    return X
+
+
+def select_random_samples(X, sample_percent):
+    num_samples = len(X)
+    num_samples_to_select = int(num_samples * sample_percent)
+    selected_indices = random.sample(range(num_samples), num_samples_to_select)
+    selected_indices.sort()
+    selected_samplesX = [X[i] for i in selected_indices]
+    return selected_samplesX
+
+
+def load_imu(data_array):
+    test_data = []
+
+    X = get_data(data_array)
+    print("X shape: ",np.array(X).shape)
+
+    samplesX = select_random_samples(X, 1)
+    print("samplesX shape: ",np.array(samplesX).shape)
+
+    test_data.extend(samplesX)
+    test_data = np.array(test_data)
+
+    print('')
+    print(len(test_data))
+
+    return test_data

testing_interface.py

@@ -0,0 +1,83 @@
+import datetime
+import numpy as np
+import torch
+import torch.nn as nn
+import torch.optim as optim
+
+from matplotlib import pyplot as plt
+from data_loader_interface import load_imu
+from sklearn.metrics import accuracy_score, f1_score, confusion_matrix, classification_report, average_precision_score
+from torch.utils.data import Dataset, DataLoader, TensorDataset, ConcatDataset
+from einops import rearrange
+from torch.optim.lr_scheduler import StepLR
+
+
+def test(model, test_loader, output_file):
+    running_loss = 0.0
+    predlist = torch.zeros(0, dtype=torch.long, device='cpu')
+    lbllist = torch.zeros(0, dtype=torch.long, device='cpu')
+    number_of_nan_loss = 0
+    
+    with torch.no_grad(), open(output_file, 'w') as f:
+        for batch_idx, (data) in enumerate(test_loader):
+            data = data.float().to('cpu')
+            outputs = model(data)
+            _, predicted = torch.max(outputs.data, 1)
+            predlist = torch.cat([predlist, predicted.view(-1).cpu()])
+            # lbllist = torch.cat([lbllist, labels.view(-1).cpu()])
+
+            for i, prediction in enumerate(predicted):
+                f.write(f"{prediction.item()}\n")
+
+    return running_loss / len(test_loader), predlist
+
+
+class MyDataset_labeled(Dataset):
+    def __init__(self, x):
+        self.x_data = torch.from_numpy(np.array(x)).to(torch.float)
+        self.len = x.shape[0]
+
+    def __getitem__(self, idx):
+        return self.x_data[idx]
+
+    def __len__(self):
+        return self.len
+
+
+def model_defining(test_data_array, name_model, output_file):
+    
+    test_losses = []
+    accuracies = []
+    f1_scores = []
+    cm_list = []
+    cr_list = []
+
+    model = torch.load(str(name_model) + ".pth", map_location=torch.device('cpu'))
+    model.eval()
+
+    test_data = load_imu(test_data_array)
+    test_data = rearrange(test_data, 'n t c -> n c t')
+
+    test_dataset = MyDataset_labeled(test_data)
+    test_dataloaders = DataLoader(test_dataset, batch_size=256, shuffle=False, drop_last=False)
+
+    print("Data loaded ...")
+
+    test_loss, predlist = test(model, test_dataloaders, output_file)
+    test_losses.append(test_loss)
+
+
+def main():
+
+    name_model = "./models_chadwick/training_3S_model_validation/S3_101_102_103_validation_epoch_10"
+    train_subjects = []
+    test_subjects = ['104'] 
+    output_file = "S3_101_102_103_validation_epoch_10_tested_104_predicted.txt"
+    label_file = "S3_101_102_103_validation_epoch_10_tested_104_label.txt"
+    classification_file = "S3_101_102_103_validation_epoch_10_tested_104_classification.txt"
+
+    model_defining(train_subjects, test_subjects, name_model, output_file, label_file, classification_file)
+
+
+if __name__ == "__main__":
+    main()

tryout_v3_model.py

@@ -0,0 +1,156 @@
+import torch.nn as nn
+import torch.nn.functional as F
+
+
+class TemporalModelBase(nn.Module):
+    """
+    Do not instantiate this class.
+    """
+
+    def __init__(self, num_joints_in, in_features, num_joints_out,
+                 filter_widths, causal, dropout, channels, sagittal=0, freezing=0, fusion=0):
+        super().__init__()
+
+        # Validate input
+        for fw in filter_widths:
+            assert fw % 2 != 0, 'Only odd filter widths are supported'
+
+        self.num_joints_in = num_joints_in
+        self.in_features = in_features
+        self.num_joints_out = 2
+        self.filter_widths = filter_widths
+
+        # Initialize layers
+        self.drop = nn.Dropout(dropout)         #for regularization
+        self.relu = nn.ReLU(inplace=True)       #introducing non-linearity
+
+        self.pad = [ filter_widths[0] // 2 ]
+        self.expand_bn = nn.BatchNorm1d(channels, momentum=0.1)         #normalize channels
+        self.expand_bn2 = nn.BatchNorm2d(in_features, momentum=0.1)     #normalize features
+
+        self.shrink = nn.Conv1d(channels, num_joints_out, 1)
+        self.flatten = nn.Flatten()
+
+    def set_bn_momentum(self, momentum):
+        """
+        Batch normalization is a technique used to normalize the inputs of each layer, which helps stabilize and speed up the training process. The momentum parameter determines how much of the statistics from the current batch should contribute to the running mean and variance of the batch normalization layer.
+        In some cases, especially during fine-tuning or transfer learning, it might be beneficial to adjust the momentum dynamically. For example, when fine-tuning on a new dataset, you may want to decrease the momentum to adapt faster to the new data distribution. This method provides a way to adjust the momentum value dynamically during training.
+        """
+        self.expand_bn.momentum = momentum
+        for bn in self.layers_bn:
+            bn.momentum = momentum
+
+    def receptive_field(self):
+        """
+        Return the total receptive field of this model as # of frames.
+        The receptive field represents the region in the input data that influences a particular output of the neural network. It is determined by the filter sizes and strides of the convolutional layers in the network.
+        """
+        frames = 0
+        for f in self.pad:
+            frames += f
+        return 1 + 2*frames
+
+    def total_causal_shift(self):
+        """
+        Return the asymmetric offset for sequence padding.
+        The returned value is typically 0 if causal convolutions are disabled,
+        otherwise it is half the receptive field.
+        """
+        frames = self.causal_shift[0]
+        next_dilation = self.filter_widths[0]
+        for i in range(1, len(self.filter_widths)):
+            frames += self.causal_shift[i] * next_dilation
+            next_dilation *= self.filter_widths[i]
+        return frames
+
+    def forward(self, x):
+
+        x = self._forward_blocks(x)
+        return x
+
+class TemporalModel(TemporalModelBase):
+    """
+    Reference 3D pose estimation model with temporal convolutions.
+    This implementation can be used for all use-cases.
+    """
+
+    def __init__(self, num_joints_in, in_features, num_joints_out,
+                 filter_widths, causal=False, dropout=0.25, channels=1024, dense=False, sagittal=0, freezing=0, fusion=0):
+        """
+        Initialize this model.
+
+        Arguments:
+        num_joints_in -- number of input joints (e.g. 17 for Human3.6M)
+        in_features -- number of input features for each joint (typically 2 for 2D input)
+        num_joints_out -- number of output joints (can be different than input)
+        filter_widths -- list of convolution widths, which also determines the # of blocks and receptive field
+        causal -- use causal convolutions instead of symmetric convolutions (for real-time applications)
+        dropout -- dropout probability
+        channels -- number of convolution channels
+        dense -- use regular dense convolutions instead of dilated convolutions (ablation experiment)
+        """
+        super().__init__(num_joints_in, in_features, num_joints_out, filter_widths, causal, dropout, channels, sagittal=sagittal, freezing=freezing, fusion=fusion)
+
+        self.expand_conv = nn.Conv1d(num_joints_in*in_features, channels, filter_widths[0], bias=False)
+
+        layers_conv = []
+        layers_bn = []
+
+        self.causal_shift = [ (filter_widths[0]) // 2 if causal else 0 ]
+        next_dilation = filter_widths[0]
+        for i in range(1, len(filter_widths)):
+            self.pad.append((filter_widths[i] - 1)*next_dilation // 2)
+            print(self.pad)
+            print(next_dilation)
+            self.causal_shift.append((filter_widths[i]//2 * next_dilation) if causal else 0)
+            print("shift", self.causal_shift)
+            layers_conv.append(nn.Conv1d(channels, channels,
+                                         filter_widths[i] if not dense else (2*self.pad[-1] + 1),
+                                         dilation=next_dilation if not dense else 1,
+                                         bias=False))
+            layers_bn.append(nn.BatchNorm1d(channels, momentum=0.1))
+            layers_conv.append(nn.Conv1d(channels, channels, 1, dilation=1, bias=False))
+            layers_bn.append(nn.BatchNorm1d(channels, momentum=0.1))
+
+            next_dilation *= filter_widths[i]
+        
+        self.layers_conv = nn.ModuleList(layers_conv)
+        self.layers_bn = nn.ModuleList(layers_bn)
+
+    def set_freezing(self, freezing, expand_bn, expand_bn2, shrink):
+        for param in self.shrink.parameters():
+            param.requires_grad = False
+        for param in self.expand_bn.parameters():
+            param.requires_grad = False
+        if freezing > 0:
+        # Freeze initial layers (expand_conv) and potentially the first n layers_conv
+            for param in self.expand_conv.parameters():
+                param.requires_grad = False
+            for i in range(freezing):
+                print(i)
+                print("i%2 = ", i%2)
+                if (i%2 == 0):
+                    for param in self.layers_conv[i].parameters():
+                        param.requires_grad = False  # Freeze Conv1d layer
+                for param in self.layers_bn[i].parameters():
+                    param.requires_grad = False  # Freeze BatchNorm layer (optional)
+        
+    def _forward_blocks(self, x):
+
+        #First block of Conv1D + BatchNorm1D, Relu, Dropout
+        x = self.drop(self.relu(self.expand_bn(self.expand_conv(x))))
+
+        #Repeaterd block of Conv1D + BatchNorm1D, Relu, Dropout
+        for i in range(len(self.pad) - 1): #same length as filter_widths
+            pad = self.pad[i+1]
+            # print(pad)
+            shift = self.causal_shift[i+1]
+            res = x[:, :, pad + shift : x.shape[2] - pad + shift]
+            # print("i voor eerste x = ", i)
+            x = self.drop(self.relu(self.layers_bn[2*i](self.layers_conv[2*i](x))))
+            # print("i voor tweede x = ", i)
+            x = res + self.drop(self.relu(self.layers_bn[2*i + 1](self.layers_conv[2*i + 1](x))))
+
+        x = self.shrink(x)  #1D convolution
+        x = self.flatten(x)
+        return x