Preprocessing Data

[ik362]

Hi There,

I have two questions on how to best implement preprocessing steps within pytorch-lightning framework. For context, I am working with time series data and a shallow 1 layer CNN.

Currently my dataset module is:

class dataset(Dataset):
    def __init__(self, csv_file, root_dir):
        self.dataframe = pd.read_csv(csv_file)
        self.root_dir = root_dir
        self.norm = MinMaxScaler()
 
    def __len__(self):
        return len(self.dataframe)

    def __getitem__(self, idx):
        if torch.is_tensor(idx):
            idx = idx.tolist()

        ts_name = os.path.join(self.root_dir,
                               self.dataframe.iloc[idx, 0])
        ts = np.load(ts_name)
        tsNorm = self.norm.fit_transform(ts.reshape(-1, 1)).reshape(-1)
        tsNorm = torch.from_numpy(tsNorm).float()
        target = self.dataframe.iloc[idx, 1]
        y = torch.tensor(target, dtype=torch.float)
        sample = (tsNorm, y) 
        return sample

And my datamodule is quite generic:

class DataModule(pl.LightningDataModule):

    def __init__(self, batch_size, data_dir, csv_file):
        super().__init__()
        self.data_dir = data_dir
        self.csv_file = csv_file
        self.batch_size = batch_size
        self.num_classes = 4
    
    def setup(self, stage=None, save=False):
        # Assign train/val datasets for use in dataloaders
        dataset = BigBrainDataset(csv_file=self.csv_file,
                                  root_dir=self.data_dir)
        
        train_size, val_size = int(0.7 * len(dataset)), int(0.2 * len(dataset))
        test_size = len(dataset) - train_size - val_size
        
        if stage == 'fit' or stage is None:
            self.train_dataset, self.val_dataset, _ = torch.utils.data.random_split(dataset, [train_size, val_size, test_size])

        # Assign test dataset for use in dataloader(s)
        if stage == 'test' or stage is None:
            _, _, self.test_dataset = torch.utils.data.random_split(dataset, [train_size, val_size, test_size])

    def train_dataloader(self):
        return DataLoader(self.train_dataset, batch_size=self.batch_size, shuffle=True, num_workers=0)
    
    def val_dataloader(self):
        return DataLoader(self.val_dataset, batch_size=self.batch_size, num_workers=0)

    def test_dataloader(self):
        return DataLoader(self.test_dataset, batch_size=self.batch_size, num_workers=0)

What I would like to know is:

1. Is my normalisation/scaling procedure in the correct place or would it be more appropriate to include a prepare_data routine within my data module?
2. I would like to include some preprocessing steps such as torchaudio.transforms.TimeMasking. I would like these to be randomly applied to a certain number of samples in each batch according to a specified probability. How/Where would I incorporate this step?

Thanks!

[rohitgr7]

tsNorm = self.norm.fit_transform(ts.reshape(-1, 1)).reshape(-1)

implementing it here will always to a fit_transform on all the samples one-by-one which is not correct. Ideally fit_transform is done on whole train sample and transform is performed on val/test/inference data using the same instance.

would like to include some preprocessing steps such as torchaudio.transforms.TimeMasking.

if this operation is supposed to happen on a batch individually I'd suggest to checkout on_before_batch_transfer/on_after_batch_transfer hooks.