Simplify data preprocessor schema arg

Author: Francisco Santos

src/ydata_synthetic/tests/custom_layers/test_activation_interface.py

@@ -1,41 +1,53 @@
 "Activation Interface layer test suite."
-from numpy import cumsum, isin, split
+from itertools import cycle, islice
+from re import search
+
+from numpy import array, cumsum, isin, split
 from numpy import sum as npsum
 from numpy.random import normal
+from pandas import DataFrame, concat
 from pytest import fixture
 from tensorflow.keras import Model
 from tensorflow.keras.layers import Dense, Input
 
+from ydata_synthetic.preprocessing.regular.processor import \
+    RegularDataProcessor
 from ydata_synthetic.utils.gumbel_softmax import ActivationInterface
 
+BATCH_SIZE = 10
 
 @fixture(name='noise_batch')
 def fixture_noise_batch():
     "Sample noise for mock output generation."
-    return normal(size=(10, 16))
-
-@fixture(name='mock_col_map')
-def fixture_mock_col_map():
-    "Mock data processing column map (var blocks i/o names)."
-    return {'numerical': [
-        [f'nfeat{n}' for n in range(6)],
-        [f'nfeat{n}' for n in range(6)]],
-        'categorical': [
-        [f'cfeat{n}' for n in range(2)],
-        sum([[f'cfeat0_{i}' for i in range(4)], [f'cfeat1_{i}' for i in range(2)]],[])]}
+    return normal(size=(BATCH_SIZE, 16))
+
+@fixture(name='mock_data')
+def fixture_mock_data():
+    "Creates mock data for the tests."
+    num_block = DataFrame(normal(size=(BATCH_SIZE, 6)), columns = [f'num_{i}' for i in range(6)])
+    cat_block_1 = DataFrame(array(list(islice(cycle(range(2)), BATCH_SIZE))), columns = ['cat_0'])
+    cat_block_2 = DataFrame(array(list(islice(cycle(range(4)), BATCH_SIZE))), columns = ['cat_1'])
+    return concat([num_block, cat_block_1, cat_block_2], axis = 1)
+
+@fixture(name='mock_processor')
+def fixture_mock_processor(mock_data):
+    "Creates a mock data processor for the mock data."
+    num_cols = [col for col in mock_data.columns if col.startswith('num')]
+    cat_cols = [col for col in mock_data.columns if col.startswith('cat')]
+    return RegularDataProcessor(num_cols, cat_cols).fit(mock_data)
 
 # pylint: disable=C0103
 @fixture(name='mock_generator')
-def fixture_mock_generator(noise_batch, mock_col_map):
+def fixture_mock_generator(noise_batch, mock_processor):
     "A mock generator with the Activation Interface as final layer."
-    input_ = Input(shape=noise_batch.shape[1], batch_size = noise_batch.shape[0])
+    input_ = Input(shape=noise_batch.shape[1], batch_size = BATCH_SIZE)
     dim = 15
     data_dim = 12
     x = Dense(dim, activation='relu')(input_)
     x = Dense(dim * 2, activation='relu')(x)
     x = Dense(dim * 4, activation='relu')(x)
     x = Dense(data_dim)(x)
-    x = ActivationInterface(mock_col_map, name='act_itf')(x)
+    x = ActivationInterface(processor_info=mock_processor.col_transform_info, name='act_itf')(x)
     return Model(inputs=input_, outputs=x)
 
 @fixture(name='mock_output')
@@ -44,16 +56,17 @@ def fixture_mock_output(noise_batch, mock_generator):
     return mock_generator(noise_batch).numpy()
 
 # pylint: disable=W0632
-def test_io(noise_batch, mock_col_map, mock_output):
+def test_io(mock_processor, mock_output):
     "Tests the output format of the activation interface for a known input."
-    num_lens = len(mock_col_map.get('numerical')[1])
-    cat_lens = len(mock_col_map.get('categorical')[1])
-    assert mock_output.shape == (len(noise_batch), num_lens + cat_lens), "The output has wrong shape."
+    num_lens = len(mock_processor.col_transform_info.numerical.feat_names_out)
+    cat_lens = len(mock_processor.col_transform_info.categorical.feat_names_out)
+    assert mock_output.shape == (BATCH_SIZE, num_lens + cat_lens), "The output has wrong shape."
     num_part, cat_part = split(mock_output, [num_lens], 1)
     assert not isin(num_part, [0, 1]).all(), "The numerical block is not expected to contain 0 or 1."
     assert isin(cat_part, [0, 1]).all(), "The categorical block is expected to contain only 0 or 1."
-    cat_i, cat_o = mock_col_map.get('categorical')
-    cat_blocks = cumsum([len([col for col in cat_o if ''.join(col.split('_')[:-1]) == feat]) for feat in cat_i])
+    cat_i, cat_o = mock_processor.col_transform_info.categorical
+    cat_blocks = cumsum([len([col for col in cat_o if col.startswith(feat) and search('_[0-9]*$', col)]) \
+        for feat in cat_i])
     cat_blocks = split(cat_part, cat_blocks[:-1], 1)
-    assert all(npsum(abs(block)) == noise_batch.shape[0] for block in cat_blocks), "There are non one-hot encoded \
+    assert all(npsum(abs(block)) == BATCH_SIZE for block in cat_blocks), "There are non one-hot encoded \
         categorical blocks."

src/ydata_synthetic/utils/gumbel_softmax.py

@@ -1,6 +1,7 @@
 """Gumbel-Softmax layer implementation.
 Reference: https://arxiv.org/pdf/1611.04051.pdf"""
-from typing import Dict, List, Optional
+from re import search
+from typing import NamedTuple, Optional
 
 # pylint: disable=E0401
 from tensorflow import (Tensor, TensorShape, concat, one_hot, split, squeeze,
@@ -42,34 +43,28 @@ class ActivationInterface(Layer):
     Processed features of different kind are sent to a TanH activation.
     Finally all output parts are concatenated and returned in the same order.
 
-    The parts of an incoming tensor are qualified by leveraging a data processor's in/out feature map.
+    The parts of an incoming tensor are qualified by leveraging a namedtuple pointing to each of the used data \
+        processor's pipelines in/out feature maps. For simplicity this object can be taken directly from the data \
+        processor col_transform_info."""
 
-    Example of how to get a col_map from a Data Processor ProcessorInfo attribute:
-    >>> col_map = {k: [v.feat_names_in, v.feat_names_out] for k, v in ProcessorInfo._asdict().items() if v}"""
-
-    def __init__(self, col_map: Dict[str, List[List[str]]], name: Optional[str] = None):
+    def __init__(self, processor_info: NamedTuple, name: Optional[str] = None):
         """Arguments:
-            col_map (Dict[str, List[List[str]]]): A map defining the processor pipelines input/output features.
+            col_map (NamedTuple): Defines each of the processor pipelines input/output features.
             name (Optional[str]): Name of the layer"""
         super().__init__(name)
 
-        self.cat_names_i, cat_names_o = col_map.get("categorical", [[],[]])
-        num_names_i, num_names_o = col_map.get("numerical", [[],[]])
-
-        self._cat_lens = None
-        self._num_lens = None
+        self.cat_feats = processor_info.categorical
+        self.num_feats = processor_info.numerical
 
-        if self.cat_names_i:  # Get the length of each processed categorical feature's output block
-            self._cat_lens = [len([col for col in cat_names_o \
-            if ''.join(col.split('_')[:-1]) == cat_feat]) for cat_feat in self.cat_names_i]
-        if num_names_i:  # Get the length of the numerical features output block
-            self._num_lens = len(num_names_o)
+        self._cat_lens = [len([col for col in self.cat_feats.feat_names_out if \
+            col.startswith(cat_feat) and search('_[0-9]*$', col)]) for cat_feat in self.cat_feats.feat_names_in]
+        self._num_lens = len(self.num_feats.feat_names_out)
 
     def call(self, _input):  # pylint: disable=W0221
-        num_cols, cat_cols = split(_input, [self._num_lens if self._num_lens else 0, -1], 1, name='split_num_cats')
-        cat_cols = split(cat_cols, self._cat_lens if self._cat_lens else 1, 1, name='split_cats')
+        num_cols, cat_cols = split(_input, [self._num_lens, -1], 1, name='split_num_cats')
+        cat_cols = split(cat_cols, self._cat_lens, 1, name='split_cats')
 
-        num_cols = [Activation('tanh', name='num_cols_activation')(num_cols)] if self._num_lens else []
-        cat_cols = [GumbelSoftmaxLayer(name=name).call(col)[0] for name, col in zip(self.cat_names_i, cat_cols)] \
-            if self._cat_lens else []
+        num_cols = [Activation('tanh', name='num_cols_activation')(num_cols)]
+        cat_cols = [GumbelSoftmaxLayer(name=name)(col)[0] for name, col in \
+            zip(self.cat_feats.feat_names_in, cat_cols)]
         return concat(num_cols+cat_cols, 1)