Merge pull request #24 from mathysgrapotte/json_to_noise_func_call

Json to noise func call

Author: alessiovignoli

.gitignore

@@ -11,4 +11,6 @@ works
 /singularity_cache
 /results/
 .coverage
-.vscode
+.vscode/
+bin/.vscode/
+

bin/src/data/__init__.py

@@ -29,12 +29,13 @@ def encode_all(self, data: list) -> np.array:
 
 class Dna(AbstractType):
     """
-    class for dealing with DNA data    
+    class for dealing with DNA data
+    # TODO make a Text base class for this class and other text based classes (rna, protein etc...)    
     """
 
     def __init__(self, **parameters) -> None:
         self.one_hot_encoder = encoders.TextOneHotEncoder(alphabet=parameters.get("one_hot_encoder_alphabet", "acgt"))
-        self.uniform_text_masker = noise_generators.UniformTextMasker(probability=parameters.get("text_masker_probability", 0.1))
+        self.uniform_text_masker = noise_generators.UniformTextMasker()
 
     def one_hot_encode(self, data: str) -> np.array:
         """
@@ -62,26 +63,30 @@ def encode_all(self, data: list, encoder: Literal['one_hot'] = 'one_hot') -> lis
             raise ValueError(f"Unknown encoder {encoder}")
 
 
-    def add_noise_uniform_text_masker(self, data: str, seed: float = None) -> str:
+    def add_noise_uniform_text_masker(self, data: str, seed: float = None, **noise_params) -> str:
         """
         Adds noise to the data of a single input.
         """
-        return self.uniform_text_masker.add_noise(data, seed=seed)
+        # get the probability param from noise_params, default value is set to 0.1
+        probability = noise_params.get("probability", 0.1)
+        return self.uniform_text_masker.add_noise(data, probability=probability, seed=seed)
 
-    def add_noise_uniform_text_masker_all_inputs(self, data: list, seed: float = None) -> list:
+    def add_noise_uniform_text_masker_all_inputs(self, data: list, seed: float = None, **noise_params) -> list:
         """
         Adds noise to the data of multiple inputs.
         """
-        return self.uniform_text_masker.add_noise_multiprocess(data, seed=seed)
+        # get the probability param from noise_params, default value is set to 0.1 
+        probability = noise_params.get("probability", 0.1)
+        return self.uniform_text_masker.add_noise_multiprocess(data, probability=probability, seed=seed)
 
 
 class Float():
     """
     class for dealing with float data
     """
 
-    def __init__(self, **parameters) -> None:
-        self.gaussian_noise = noise_generators.GaussianNoise(mean=parameters.get("gaussian_noise_mean", 0), std=parameters.get("gaussian_noise_std", 1))
+    def __init__(self) -> None:
+        self.gaussian_noise = noise_generators.GaussianNoise()
 
     def add_noise_gaussian_noise(self, data: float, seed: float = None) -> float:
         """

bin/src/data/data_types/data_types.py

@@ -40,39 +40,43 @@ class UniformTextMasker(AbstractNoiseGenerator):
     This noise generators replace characters with 'N' with a given probability.
     """
 
-    def __init__(self, probability: float = 0.1) -> None:
-        self.probability = probability
 
-
-    def add_noise(self, data: str, seed: float = None) -> str:
+    def add_noise(self, data: str, probability: float = 0.1, seed: float = None) -> str:
         """
         Adds noise to the data.
         """
 
         np.random.seed(seed)
-        return ''.join([c if np.random.rand() > self.probability else 'N' for c in data])
-    
+        return ''.join([c if np.random.rand() > probability else 'N' for c in data])
+
+    def add_noise_multiprocess(self, data: list, probability: float = 0.1, seed: float = None) -> list:
+        """
+        Adds noise to the data using multiprocessing.
+        """
+
+        with mp.Pool(mp.cpu_count()) as pool:
+            function_specific_input = [(item, probability, seed) for item in data]
+            return pool.starmap(self.add_noise, function_specific_input)
+
 class GaussianNoise(AbstractNoiseGenerator):
     """
     This noise generator adds gaussian noise to float values
     """
 
-    def __init__(self, mean: float = 0, std: float = 1) -> None:
-        self.mean = mean
-        self.std = std
 
-    def add_noise(self, data: float, seed: float = None) -> float:
+    def add_noise(self, data: float, mean: float = 0, std: float= 0, seed: float = None) -> float:
         """
         Adds noise to a single point of data.
         """
 
         np.random.seed(seed)
-        return data + np.random.normal(self.mean, self.std)
+        return data + np.random.normal(mean, std)
 
-    def add_noise_multiprocess(self, data: list, seed: float = None) -> list:
+    def add_noise_multiprocess(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(self.mean, self.std, len(data)))
+        return list(np.array(data) + np.random.normal(mean, std, len(data)))

bin/src/data/data_types/noise/noise_generators.py

@@ -11,6 +11,7 @@
 from typing import Any
 from .data_types import data_types as data_types
 from .spliters import spliters as spliters
+from copy import deepcopy
 import numpy as np
 
 class AbstractExperiment(ABC):
@@ -32,12 +33,42 @@ def get_split_indexes(self, data: list, split: tuple) -> list | list | list:
         """
         raise NotImplementedError
 
+    def get_keys_based_on_name_data_type_or_input(self, data: dict, column_name: str = None, data_type: str = None, category = None) -> list:
+        """
+        Returns the keys of the data that are of a specific type, name or category.
+        If the column_name is specified, it will return all the keys that contain the column_name in their name. 
+        If the data_type is specified, it will return all the keys that contain the data_type in their name.
+        If the data_type and the category are specified, it will return all the keys that contain the data_type and the category in their name.
+        """
+
+        # Check that one of column_name, data_type or category is not None
+        if column_name is None and data_type is None and category is None:
+            raise ValueError("At least one of column_name, data_type or category should be specified.")
+        
+        # Check that category is not the only one specified
+        if category is not None and column_name is None and data_type is None:
+            raise ValueError("category cannot be the only one specified.")
+        
+        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) -> Any:
+    def noise(self, data: Any, noise_method: str, **noise_params: dict) -> Any:
         """
-        Adds noise to the data.
+        Adds noise to the data, using function defined in self.noise
         """
-        raise NotImplementedError
+        # 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__}")
+
 
 class DnaToFloatExperiment(AbstractExperiment):
     """
@@ -49,18 +80,19 @@ def __init__(self, seed: float = None, **parameters) -> None:
         self.dna = data_types.Dna(**parameters)
         self.float = data_types.Float(**parameters)
 
-    def add_noise(self, data: list) -> list:
+    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.
         """
-        return self.dna.add_noise_uniform_text_masker_all_inputs(data, seed=self.seed)
-    
-    def noise_scheme(self, data: list, params: dict) -> dict:
-        output = {}
-        for key in params: 
-            output[key] = self.add_noise(data, params[key])
 
-        return output
+
+        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
 
 
 

bin/src/data/experiments.py

@@ -2,9 +2,32 @@
 import numpy.testing as npt
 import unittest
 from bin.src.data.experiments import DnaToFloatExperiment
+from copy import deepcopy
 
 class TestDnaToFloatExperiment(unittest.TestCase):
 
         def setUp(self):
             self.dna_to_float_experiment = DnaToFloatExperiment()
-    
+
+        def test_noise(self):
+            # Test calling the noise method using a kwargs dictionary
+            noise_method_noise_dna_uniform_masker = 'noise_dna_uniform_masker'
+            kwarg_dict = {'probability': 0.5}
+            original_data = {
+                "sequences1:dna:input": ["ACGTACGT", "ACGTACGT", "ACGTACGT", "ACGTACGT"],
+                "sequences2:dna:input": ["ACGTACGT", "ACGTACGT", "ACGTACGT", "ACGTACGT"],
+                "float1:float:label": [1.0, 2.0, 3.0, 4.0],
+                "float2:float:label": [1.0, 2.0, 3.0, 4.0]
+            }
+
+            data = deepcopy(original_data)
+            noisy_data = self.dna_to_float_experiment.noise(data, noise_method_noise_dna_uniform_masker, **kwarg_dict)
+            self.assertIsInstance(noisy_data, dict)
+            self.assertEqual(len(noisy_data), 4)
+            # checking if the noise was applied to the correct keys, meaning that the sequences have changed
+            self.assertNotEqual(noisy_data["sequences1:dna:input"], original_data["sequences1:dna:input"])
+            self.assertNotEqual(noisy_data["sequences2:dna:input"], original_data["sequences2:dna:input"])
+            # checking if the noise was not applied to the float keys
+            self.assertEqual(noisy_data["float1:float:label"], original_data["float1:float:label"])
+            self.assertEqual(noisy_data["float2:float:label"], original_data["float2:float:label"])
+

bin/tests/test_experiments.py

@@ -26,15 +26,15 @@ def test_add_noise(self):
 class TestUniformTextMasker(unittest.TestCase):
     def test_add_noise_single(self):
         # Test adding noise to a single string
-        masker = UniformTextMasker(probability=0.1)
+        masker = UniformTextMasker()
         noisy_data = masker.add_noise("ACGTACGT", seed=42)
         self.assertIsInstance(noisy_data, str)      # making sure output is of correct type
         self.assertEqual(noisy_data, "ACGTACNT")    # checking if given a seed the noise happens in the same way
 
     def test_add_noise_multiprocess(self):
         # Test adding noise to a list of strings using multiprocessing
-        masker = UniformTextMasker(probability=0.1)
-        noisy_data_list = masker.add_noise_multiprocess(["ATCGATCGATCG", "ATCG"], seed=42)
+        masker = UniformTextMasker()
+        noisy_data_list = masker.add_noise_multiprocess(["ATCGATCGATCG", "ATCG"], seed=42, probability=0.1 )
         self.assertIsInstance(noisy_data_list, list)     # making sure output is of correct type
         self.assertIsInstance(noisy_data_list[0], str)
         self.assertIsInstance(noisy_data_list[1], str)
@@ -44,14 +44,14 @@ def test_add_noise_multiprocess(self):
 class TestGaussianNoise(unittest.TestCase):
     def test_add_noise_single(self):
         # Test adding noise to a single float value
-        noise_generator = GaussianNoise(mean=0, std=1)
-        noisy_data = noise_generator.add_noise(5.0, seed=42)
+        noise_generator = GaussianNoise()
+        noisy_data = noise_generator.add_noise(5.0, seed=42, mean=0, std=1)
         self.assertIsInstance(noisy_data, float)
         self.assertAlmostEqual(noisy_data, 5.4967141530)     # there might be float point variation across systems so not all decimals have to be identical
 
     def test_add_noise_multiprocess(self):
         # Test adding noise to a list of float values using multiprocessing
-        noise_generator = GaussianNoise(mean=0, std=1)
+        noise_generator = GaussianNoise()
         noisy_data = noise_generator.add_noise_multiprocess([1.0, 2.0, 3.0])
         self.assertIsInstance(noisy_data, list)
         self.assertIsInstance(noisy_data[0], float)