Multichannel config (#390)

---------

Signed-off-by: mboussarov <mboussarov@gmail.com>

numalogic/config/factory.py

@@ -51,6 +51,7 @@ class PreprocessFactory(_ObjectFactory):
         GaussianNoiseAdder,
         DifferenceTransform,
         FlattenVector,
+        FlattenVectorWithPadding,
         PercentileScaler,
         ExpMovingAverage,
     )
@@ -67,6 +68,7 @@ class PreprocessFactory(_ObjectFactory):
         "GaussianNoiseAdder": GaussianNoiseAdder,
         "DifferenceTransform": DifferenceTransform,
         "FlattenVector": FlattenVector,
+        "FlattenVectorWithPadding": FlattenVectorWithPadding,
         "PercentileScaler": PercentileScaler,
         "ExpMovingAverage": ExpMovingAverage,
     }

numalogic/transforms/__init__.py

@@ -22,6 +22,7 @@
     GaussianNoiseAdder,
     DifferenceTransform,
     FlattenVector,
+    FlattenVectorWithPadding,
 )
 from numalogic.transforms._movavg import ExpMovingAverage, expmov_avg_aggregator
 from numalogic.transforms._postprocess import TanhNorm, tanh_norm, SigmoidNorm
@@ -38,6 +39,7 @@
     "GaussianNoiseAdder",
     "DifferenceTransform",
     "FlattenVector",
+    "FlattenVectorWithPadding",
     "PercentileScaler",
     "SigmoidNorm",
 ]

numalogic/transforms/_stateless.py

@@ -156,3 +156,56 @@ def transform(self, X: npt.NDArray[float], **__) -> npt.NDArray[float]:
 
     def inverse_transform(self, X: npt.NDArray[float]) -> npt.NDArray[float]:
         return X.reshape(-1, self.n_features)
+
+
+class FlattenVectorWithPadding(StatelessTransformer):
+    """A stateless transformer that flattens some of the columns and does padding on the rest.
+
+    Args:
+    ____
+        features: list of all feature names in the order of columns of the payload matrix
+        flatten_features: list of feature names to be flattened
+        padding_with: numerical value to be used for padding, default is 0
+
+    """
+
+    __slots__ = (
+        "features",
+        "flatten_features",
+        "padding_with",
+        "padding_features",
+        "flatten_indexes",
+        "padding_indexes",
+    )
+
+    @staticmethod
+    def _feature_indexes(features_all: list[str], features: list[str]) -> list[int]:
+        return [features_all.index(f) for f in features]
+
+    def __init__(self, features: list[str], flatten_features: list[str], padding_with: float = 0.0):
+        self.features = features
+        self.flatten_features = flatten_features
+        self.padding_with = padding_with
+
+        self.padding_features = list(set(features) - set(flatten_features))
+        if not self.padding_features:
+            raise ValueError("At least one feature should be left for padding.")
+        self.flatten_indexes = self._feature_indexes(features, self.flatten_features)
+        self.padding_indexes = self._feature_indexes(features, self.padding_features)
+
+    def transform(self, X: npt.NDArray[float], **__) -> npt.NDArray[float]:
+        X_flatten = X[:, self.flatten_indexes].flatten().reshape(-1, 1)
+        padding_len = X_flatten.shape[0] - X.shape[0]
+        X_padding = np.pad(
+            X[:, self.padding_indexes],
+            ((0, padding_len), (0, 0)),
+            mode="constant",
+            constant_values=self.padding_with,
+        )
+        return np.concatenate([X_flatten, X_padding], axis=1)
+
+    def inverse_transform(self, X: npt.NDArray[float]) -> npt.NDArray[float]:
+        X_flatten = X[:, 0].reshape(-1, len(self.flatten_features))
+        original_len = X.shape[0] - X_flatten.shape[0]
+        X_padding_removed = X[:original_len, 1:]
+        return np.concatenate([X_flatten, X_padding_removed], axis=1)

tests/transforms/test_transforms.py

@@ -20,6 +20,7 @@
     DifferenceTransform,
     FlattenVector,
     PercentileScaler,
+    FlattenVectorWithPadding,
 )
 
 RNG = np.random.default_rng(42)
@@ -179,6 +180,19 @@ def test_flattenvector():
     assert clf.inverse_transform(data).shape[1] == 2
 
 
+def test_flattenvector_with_padding():
+    x = RNG.random((10, 4))
+    features = ["failed", "degraded", "errorrate", "userimpact"]
+    flatten_features = ["failed", "degraded"]
+
+    clf = FlattenVectorWithPadding(features, flatten_features)
+    X_transformed = clf.transform(x)
+    assert X_transformed.shape[1] == (len(features) - len(flatten_features) + 1)
+
+    X_inverse = clf.inverse_transform(X_transformed)
+    assert X_inverse.shape == x.shape
+
+
 def test_percentile_scaler():
     x = np.abs(RNG.random((10, 3)))
     tx = PercentileScaler(max_percentile=99, eps=1e-6)