Merge pull request #26 from mathysgrapotte/composition_data_types

Author: mathysgrapotte

bin/src/data/csv_parser.py

@@ -97,7 +97,10 @@ def get_and_encode(self, dictionary: dict, idx: Any) -> dict:
 
             # encode the data at given index
             # For that, it first retrieves the data object and then calls the encode_all method to encode the data
-            output[name] = self.experiment.__getattribute__(data_type.lower()).encode_all(data)
+
+            
+            output[name] = self.experiment.get_encoding_all(data_type)(dictionary[key][idx])
+
 
         return output
 

bin/src/data/data_types/data_types.py

@@ -89,3 +89,33 @@ def decode(self, data: np.array) -> str:
         """
         return self.encoder.inverse_transform(data)
 
+class FloatEncoder(AbstractEncoder):
+    """
+    Encoder for float data.
+    """
+    def encode(self, data: float) -> float:
+        """
+        Encodes the data. 
+        This method takes as input a single data point, should be mappable to a single output. 
+        """
+        return float(data)
+    
+    def encode_all(self, data: list) -> list:
+        """
+        Encodes the data. 
+        This method takes as input a list of data points, should be mappable to a single output. 
+        """
+
+        # check if data is a string, in that case it should use the encode sequence method
+        if isinstance(data, str):
+            return [self.encode(data)]
+        else:
+            return [float(d) for d in data]
+    
+    def decode(self, data: float) -> float:
+        """
+        Decodes the data.
+        """
+        return data
+    
+    

bin/src/data/data_types/encoding/encoders.py renamed to bin/src/data/encoding/encoders.py

@@ -9,9 +9,11 @@
 
 from abc import ABC, abstractmethod
 from typing import Any
-from .data_types import data_types as data_types
 from .spliters import spliters as spliters
+from .encoding import encoders as encoders
+from .noise import noise_generators as noise_generators
 from copy import deepcopy
+
 import numpy as np
 
 class AbstractExperiment(ABC):
@@ -24,8 +26,6 @@ def __init__(self, seed: float = None) -> None:
         # allow ability to add a seed for reproducibility
         self.seed = seed
 
-        #self.random_splitter = spliters.RandomSplitter(seed=seed) 
-
 
     def get_split_indexes(self, data: list, split: tuple) -> list | list | list:
         """
@@ -51,48 +51,26 @@ def get_keys_based_on_name_data_type_or_input(self, data: dict, column_name: str
 
         if column_name is not None:
             return [key for key in data if column_name in key.split(':')[0]]
+        
         if data_type is not None:
             if category is not None:
                 return [key for key in data if data_type in key.split(':')[1] and category in key.split(':')[2]]
             else:
                 return [key for key in data if data_type in key.split(':')[1]]
-
- 
-    def noise(self, data: Any, noise_method: str, **noise_params: dict) -> Any:
+        
+    def get_encoding_all(self, data_type: str) -> Any:
         """
-        Adds noise to the data, using function defined in self.noise
+        This method gets the encoding function for a specific data type.
         """
-        # check if noise_method exist in the class, if it does, call it with the associated **noise_params, if not raise an error
-
-        if hasattr(self, noise_method):
-            return getattr(self, noise_method)(data, **noise_params)
-        else:
-            raise NotImplementedError(f"No noise method {noise_method} in the class {self.__class__.__name__}")
-
+        return getattr(self, data_type)['encoder'].encode_all
 
 class DnaToFloatExperiment(AbstractExperiment):
     """
     Class for dealing with DNA to float predictions (for instance regression from DNA sequence to CAGE value)
     """
-
-    def __init__(self, seed: float = None, **parameters) -> None:
-        super().__init__(seed)
-        self.dna = data_types.Dna(**parameters)
-        self.float = data_types.Float(**parameters)
-
-    def noise_dna_uniform_masker(self, data: dict, **noise_params) -> dict:
-        """
-        Adds noise to the data of a single input.
-        Applied on all input keys that have the dna data type.
-        """
-
-
-        dna_type_keys = self.get_keys_based_on_name_data_type_or_input(data, data_type='dna')
-
-        for key in dna_type_keys:
-            data[key] = self.dna.add_noise_uniform_text_masker_all_inputs(data[key], seed=self.seed, **noise_params)
-        
-        return data
-
-
+    def __init__(self):
+        super().__init__()
+        self.dna = {'encoder': encoders.TextOneHotEncoder(alphabet='acgt'), 'noise_generators': {'uniform_text_masker': noise_generators.UniformTextMasker()}}
+        self.float = {'encoder': encoders.FloatEncoder(), 'noise_generators': {'uniform_float_masker': noise_generators.GaussianNoise()}}
+        #self.protein = {'encoder': encoders.TextOneHotEncoder(), 'noise_generators': {'uniform_text_masker': noise_generators.UniformTextMasker()}}
 

bin/src/data/experiments.py

@@ -21,10 +21,16 @@ def convert_list_of_numpy_arrays_to_tensor(self, data: list) -> Tuple[torch.Tens
         """
         Converts a list of numpy arrays to a tensor. 
         If the list includes numpy arrays of different shapes, padd the numpy arrays and return a mask tensor, otherwise mask tensor is set to None
+
+        # TODO: This method utilizes ifs to check the shape of the data. this is not ideal. Performance improvement could be done here.
         """
         if len(data) > 1:
+            # check if data is a flat list (of float or integers):
+            if isinstance(data[0], (float, int)):
+                return torch.tensor(data), None
+
             # check if the data is of different shapes
-            if len(set([d.shape for d in data])) == 1:
+            elif len(set([d.shape for d in data])) == 1:
                 return torch.tensor(np.array(data)), None
 
             # otherwise, pad the data and build a mask tensor that points to where the data has been padded.

bin/src/data/handlertorch.py

@@ -25,14 +25,13 @@ def add_noise(self, data: Any, seed: float = None) -> Any:
         #  np.random.seed(seed)
         raise NotImplementedError
 
-    def add_noise_multiprocess(self, data: list, seed: float = None) -> list:
+    @abstractmethod
+    def add_noise_all(self, data: list, seed: float = None) -> list:
         """
-        Adds noise to the data using multiprocessing.
+        Adds noise to the data.
         """
-        with mp.Pool(mp.cpu_count()) as pool:
-            # reshaping the inputs of this function to meet starmap requirements, basically adding into a tuple the list[elem] + seed
-            function_specific_input = [(item, seed) for item in data]
-            return pool.starmap(self.add_noise, function_specific_input)
+        #  np.random.seed(seed)
+        raise NotImplementedError
 
 
 class UniformTextMasker(AbstractNoiseGenerator):
@@ -47,13 +46,16 @@ def add_noise(self, data: str, probability: float = 0.1, mask='N', seed: float =
         np.random.seed(seed)
         return ''.join([c if np.random.rand() > probability else mask for c in data])
 
-    def add_noise_multiprocess(self, data: list, probability: float = 0.1, mask='N', seed: float = None) -> list:
+
+    def add_noise_all(self, data: list, probability: float = 0.1, mask='N', seed: float = None) -> list:
+
         """
         Adds noise to the data using multiprocessing.
         """
         with mp.Pool(mp.cpu_count()) as pool:
             function_specific_input = [(item, probability, mask, seed) for item in data]
             return pool.starmap(self.add_noise, function_specific_input)
+        
 
 class GaussianNoise(AbstractNoiseGenerator):
     """
@@ -67,10 +69,11 @@ def add_noise(self, data: float, mean: float = 0, std: float= 0, seed: float = N
         np.random.seed(seed)
         return data + np.random.normal(mean, std)
 
-    def add_noise_multiprocess(self, data: list, mean: float = 0, std: float = 0, seed: float = None) -> list:
+    def add_noise_all(self, data: list, mean: float = 0, std: float = 0, seed: float = None) -> list:
         """
         Adds noise to the data using np arrays
         # TODO return a np array to gain performance.
         """
         np.random.seed(seed)
-        return list(np.array(data) + np.random.normal(mean, std, len(data)))
+        return list(np.array(data) + np.random.normal(mean, std, len(data)))
+