Repository: eyecan-ai/pipelime (language: python)

Issue:
Feature/groupby



Discussion / hints:

Fix patch:
diff --git a/.coveragerc b/.coveragerc
new file mode 100644
index 0000000..48a4e18
--- /dev/null
+++ b/.coveragerc
@@ -0,0 +1,7 @@
+[report]
+exclude_lines =
+    # Have to re-enable the standard pragma
+    pragma: no cover
+
+    # Don't complain if tests don't hit defensive assertion code:
+    raise NotImplementedError
\ No newline at end of file
diff --git a/.flake8 b/.flake8
new file mode 100644
index 0000000..98b7c02
--- /dev/null
+++ b/.flake8
@@ -0,0 +1,3 @@
+ 
+[flake8]
+max-line-length = 140
\ No newline at end of file
diff --git a/.github/python-package.yml b/.github/python-package.yml
new file mode 100644
index 0000000..bd67ed6
--- /dev/null
+++ b/.github/python-package.yml
@@ -0,0 +1,39 @@
+# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
+# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions
+
+name: Python package
+
+on:
+  push:
+    branches: [ main, develop ]
+  pull_request:
+    branches: [ main, develop ]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [3.7, 3.8, 3.9]
+
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v2
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install flake8 pytest
+        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+    - name: Lint with flake8
+      run: |
+        # stop the build if there are Python syntax errors or undefined names
+        flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+        # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
+        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+    - name: Test with pytest
+      run: |
+        pytest
diff --git a/.travis.yml b/.travis.yml_old
similarity index 97%
rename from .travis.yml
rename to .travis.yml_old
index a51bcd5..5efbbcc 100644
--- a/.travis.yml
+++ b/.travis.yml_old
@@ -2,6 +2,7 @@
 
 language: python
 python:
+  - 3.9
   - 3.8
   - 3.7
   - 3.6
diff --git a/README.rst b/README.rst
index bda054c..ff0eac4 100644
--- a/README.rst
+++ b/README.rst
@@ -1,6 +1,9 @@
-========
-pipelime
-========
+===========
+🍋 Pipelime
+===========
+
+data pipeline 101
+
 
 
 .. image:: https://img.shields.io/pypi/v/pipelime.svg
@@ -13,29 +16,7 @@ pipelime
         :target: https://pipelime.readthedocs.io/en/latest/?badge=latest
         :alt: Documentation Status
 
-
 .. image:: https://pyup.io/repos/github/eyecan-ai/pipelime/shield.svg
      :target: https://pyup.io/repos/github/eyecan-ai/pipelime/
      :alt: Updates
 
-
-
-data pipeline 101
-
-
-* Free software: GNU General Public License v3
-* Documentation: https://pipelime.readthedocs.io.
-
-
-Features
---------
-
-* TODO
-
-Credits
--------
-
-This package was created with Cookiecutter_ and the `audreyr/cookiecutter-pypackage`_ project template.
-
-.. _Cookiecutter: https://github.com/audreyr/cookiecutter
-.. _`audreyr/cookiecutter-pypackage`: https://github.com/audreyr/cookiecutter-pypackage
diff --git a/examples/albumentations_tst.py b/examples/albumentations_tst.py
new file mode 100644
index 0000000..a63baa6
--- /dev/null
+++ b/examples/albumentations_tst.py
@@ -0,0 +1,31 @@
+from pipelime.augmentations.factory import AugmentationsFactory
+from choixe.configurations import XConfig
+from albumentations.core.serialization import SerializableMeta
+from albumentations.core.six import add_metaclass
+import rich
+import albumentations as A
+import cv2
+import pipelime as PL
+# A.core.serialization.SerializableMeta.__new__(PadIfNeededV2, 'PadIfNeededV2', )
+
+
+transform = A.Compose([
+    A.RandomCrop(width=450, height=450),
+    A.HorizontalFlip(p=0.5),
+    A.RandomBrightnessContrast(p=0.2),
+],
+    bbox_params=A.BboxParams(format='coco'),
+    keypoint_params=A.KeypointParams(format='xy', label_fields=['category_id'])
+)
+
+
+#
+#transform = A.RGBShift()
+
+
+# d = A.to_dict(transform)
+
+
+# cfg = XConfig.from_dict(d)
+# rich.print(cfg.to_dict(discard_private_qualifiers=False))
+# cfg.save_to('/tmp/out.yml')
diff --git a/examples/example_underfolder.py b/examples/example_underfolder.py
new file mode 100644
index 0000000..56a7a2f
--- /dev/null
+++ b/examples/example_underfolder.py
@@ -0,0 +1,102 @@
+from pipelime.augmentations.transforms import PadIfNeededV2
+import rich
+import albumentations as A
+from albumentations.augmentations.functional import scale
+from pipelime.sequences.samples import Sample
+from pipelime.sequences.stages import StageAugmentations, StageIdentity, StageRemap
+from pipelime.sequences.readers.filesystem import UnderfolderReader
+import cv2
+from pipelime.sequences.operations import OperationFilterByQuery, OperationGroupBy, OperationIdentity
+import numpy as np
+
+
+def draw_keypoints(keypoints: np.ndarray, image: np.ndarray):
+
+    h, w = image.shape[:2]
+    for kp in keypoints:
+        print("KP", kp)
+        pos = np.array(kp[0:2])  # * np.array([w, h])
+        a = kp[2]
+        s = kp[3] * 0.7
+        direction = np.array([np.cos(-a), np.sin(-a)])
+
+        cv2.circle(image, tuple(pos.astype(int)), radius=5, color=(255, 255, 255))
+
+        pos2 = pos + direction * s
+        cv2.arrowedLine(image, tuple(pos.astype(int)), tuple(pos2.astype(int)), (255, 255, 255), 2)
+
+
+# folder = '/home/daniele/Desktop/experiments/2021-01-28.PlaygroundDatasets/flow_00'
+folder = '/tmp/ds'
+dataset = UnderfolderReader(folder=folder)
+print(len(dataset))
+
+op = OperationGroupBy(field='`metadata.counter`')
+op = OperationIdentity()  # OperationFilterByQuery(query='`metadata.')
+dataset = op(dataset)
+
+transform = A.Compose([
+    # A.RandomCrop(width=450, height=450),
+    # A.HorizontalFlip(p=0.5)
+    A.ShiftScaleRotate(shift_limit=0.2, scale_limit=0.6, rotate_limit=180, border_mode=cv2.BORDER_CONSTANT, p=1.0),
+    PadIfNeededV2(min_height=1800, min_width=1000),
+    # A.RandomBrightnessContrast(p=0.2),
+],
+    keypoint_params=A.KeypointParams(format='xyas', remove_invisible=True, angle_in_degrees=False),
+    # bbox_params=A.BboxParams(format='yolo')
+)
+
+stage = StageAugmentations(
+    transform_cfg=A.to_dict(transform),
+    targets={
+        'image': 'image',
+        'keypoints': 'keypoints',
+        'mask': 'mask',
+        'inst': 'mask'
+    }
+)
+
+
+# stage = StageRemap({
+#     'image': 'a'
+# }, remove_missing=False)
+# s2 = StageIdentity()
+[cv2.namedWindow(x, cv2.WINDOW_NORMAL) for x in ["image", "mask", "inst"]]
+
+for sample in dataset:
+
+    sample = stage(sample)
+
+    image = sample['image']
+    h, w = image.shape[:2]
+    keypoints = sample['keypoints']
+
+    mask = sample['mask']
+    inst = sample['inst']
+
+    # idx = keypoints[:, 0]
+
+    # # keypoints = keypoints[:, 1:5]
+    # keypoints[:, 0] *= w
+    # keypoints[:, 1] *= h
+
+    # print("pre", keypoints)
+    # transform.add_targets({'inst': 'mask', 'keypoints': 'keypoints'})
+    # _t = transform(image=image, keypoints=keypoints, mask=mask, inst=inst)  # , class_labels=idx)
+    # image = _t['image']
+    # keypoints = _t['keypoints']
+    # # idx = _t['class_labels']
+    # mask = _t['mask']
+    # inst = _t['inst']
+    # print("post", keypoints)
+    # print(idx)
+
+    draw_keypoints(keypoints, image)
+
+    mask = (mask - mask.min())/(mask.max() - mask.min())
+    inst = (inst - inst.min())/(inst.max() - inst.min())
+    cv2.imshow("image", image)
+    cv2.imshow("mask", mask)
+    cv2.imshow("inst", inst)
+    if ord('q') == cv2.waitKey(0):
+        break
diff --git a/examples/gkit_example.py b/examples/gkit_example.py
new file mode 100644
index 0000000..c4f2efb
--- /dev/null
+++ b/examples/gkit_example.py
@@ -0,0 +1,30 @@
+from operator import mul, sub
+import graphkit
+from graphkit import compose, operation
+# Computes |a|^p.
+from pipelime.sequences.operations import OperationShuffle, OperationSum
+
+
+def abspow(a, p):
+    c = abs(a) ** p
+    return c
+
+
+# Compose the mul, sub, and abspow operations into a computation graph.
+graph = compose(name="graph")(
+    operation(name="l1", needs=["a", "b"], provides=["ab"])(OperationSum()),
+    operation(name="l2", needs=["a", "ab"], provides=["aab"])(OperationSum()),
+    operation(name="l3", needs=["aab"], provides=["out"])(OperationShuffle())
+)
+
+# Run the graph and request all of the outputs.
+out = graph({'a': 2, 'b': 5})
+
+# Prints "{'a': 2, 'a_minus_ab': -8, 'b': 5, 'ab': 10, 'abs_a_minus_ab_cubed': 512}".
+print(out)
+
+# Run the graph and request a subset of the outputs.
+out = graph({'a': 2, 'b': 5}, outputs=["a_minus_ab"])
+
+# Prints "{'a_minus_ab': -8}".
+print(out)
diff --git a/examples/groupby_example.py b/examples/groupby_example.py
new file mode 100644
index 0000000..b3609ef
--- /dev/null
+++ b/examples/groupby_example.py
@@ -0,0 +1,63 @@
+import rich
+from pipelime.sequences.samples import PlainSample, SamplesSequence
+import numpy as np
+
+from pipelime.sequences.operations import OperationGroupBy
+
+
+def generate_plain_samples(
+    namespace: str,
+    N: int,
+    group_each: int = 5,
+    heavy_data: bool = False
+):
+
+    samples = []
+    for i in range(N):
+        data = {
+            'idx': f'{namespace}{i}',
+            'number': i,
+            'fraction': i / 1000.,
+            'odd': i % 2 == 1,
+            'data0': np.random.uniform(0.0, 1.0, (64, 64, 3)) if heavy_data else 1.64643,
+            'data1': np.random.uniform(0, 255, (8, 8)).astype(np.uint8) if heavy_data else 25588,
+            'metadata': {
+                'even': i % 2 == 0,
+                'name': f'{namespace}{i}',
+                'N': i,
+                'deep': {
+                    'super_deep': 0,
+                    'groupby_field': int(i / group_each),
+                    'groupby_field2': int(i % 3),
+                    'groupby_field3': int(i % 10),
+                }
+            }
+        }
+        samples.append(PlainSample(data=data))
+
+    return samples
+
+
+d = SamplesSequence(samples=generate_plain_samples('d0', 128))
+
+op = OperationGroupBy(field='metadata.deep.groupby_field')
+
+out = op(d)
+for sample in out:
+    for k, v in sample.items():
+        rich.print(k, v)
+
+op2 = OperationGroupBy(field='metadata.deep.groupby_field')
+
+out2 = op2(out)
+# for sample in out2:
+#     for k, v in sample.items():
+#         rich.print(k, v)
+
+op3 = OperationGroupBy(field='metadata.even.0.0')
+
+out3 = op3(out2)
+# for sample in out3:
+#     for k, v in sample.items():
+#         rich.print(k, v)
+print(len(out), len(out2), len(out3))
diff --git a/examples/pipe_test.py b/examples/pipe_tst.py
similarity index 100%
rename from examples/pipe_test.py
rename to examples/pipe_tst.py
diff --git a/examples/pipe_tst_with_graph.py b/examples/pipe_tst_with_graph.py
new file mode 100644
index 0000000..3c101cb
--- /dev/null
+++ b/examples/pipe_tst_with_graph.py
@@ -0,0 +1,308 @@
+
+
+import hashlib
+import random
+import os
+import tempfile
+from networkx.drawing.nx_agraph import graphviz_layout, to_agraph
+from typing import Union
+import rich
+from pipelime.lib import AddOp, PlainSample, SamplesSequence, SequenceOpFactory
+from choixe.configurations import XConfig
+import networkx as nx
+import matplotlib.pyplot as plt
+
+
+def sample_dataset(namespace: str, N: int):
+
+    samples = []
+    for i in range(N):
+        data = {
+            'idx': f'{namespace}{i}',
+            'number': i,
+            'fraction': i / 1000.,
+            'metadata': {
+                'name': f'{namespace}{i}',
+                'N': i,
+                'deep': {
+                    'super_deep': 0
+                }
+            }
+        }
+        samples.append(PlainSample(data=data))
+
+    return SamplesSequence(samples=samples)
+
+
+class PipeNode(object):
+    ID_COUNTER = 0
+
+    def __init__(self, input_ports: Union[str, list, dict], output_ports: Union[str, list, dict], node: dict) -> None:
+        self.id = PipeNode.ID_COUNTER
+        PipeNode.ID_COUNTER += 1
+        self.input_ports = input_ports
+        self.output_ports = output_ports
+        self.node = node
+
+    def in_ports(self):
+        return self.ports_as_list(self.input_ports)
+
+    def out_ports(self):
+        return self.ports_as_list(self.output_ports)
+
+    def ports_as_list(self, v: any):
+        if v is None:
+            return []
+        elif isinstance(v, str):
+            return [v]
+        elif isinstance(v, list):
+            return v
+        elif isinstance(v, dict):
+            return list(v.values())
+        else:
+            raise NotImplementedError(f'{type(v)}')
+
+    def __repr__(self) -> str:
+        return str(self.id)
+
+
+class DataNode(object):
+
+    def __init__(self, name: str) -> None:
+        self.name = name
+
+    def __repr__(self) -> str:
+        return self.name
+
+    def __hash__(self) -> int:
+        return hash(self.name)
+
+    def __eq__(self, o: object) -> bool:
+        return self.name == o.name
+
+
+pipe = {
+    'input': {
+        'GOOD': None,
+        'BAD': None
+    },
+    'modules': [
+        {
+            'input': None,
+            'output': 'GOOD',
+            'node': {
+                'type': 'Source',
+                'data': sample_dataset('good_', 100)
+            }
+        },
+        {
+            'input': None,
+            'output': 'BAD',
+            'node': {
+                'type': 'Source',
+                'data': sample_dataset('bad_', 100)
+            }
+        },
+        # {
+        #     'input': None,
+        #     'output': 'SUPER',
+        #     'node': {
+        #         'type': 'Source',
+        #         'data': sample_dataset('super_', 100)
+        #     }
+        # },
+        {
+            'input': 'BAD',
+            'output': {
+                'a': 'BAD_0',
+                'b': 'BAD_1'
+            },
+            'node': {
+                'type': 'SplitsOp',
+                'options': {
+                    'split_map': {
+                        'a': 0.5,
+                        'b': 0.5
+                    }
+                }
+            }
+        },
+        {
+            'input': 'GOOD',
+            'output': {
+                'a': 'GOOD_0',
+                'b': 'GOOD_1'
+            },
+            'node': {
+                'type': 'SplitsOp',
+                'options': {
+                    'split_map': {
+                        'a': 0.5,
+                        'b': 0.5
+                    }
+                }
+            }
+        },
+        {
+            'input': ['BAD_0', 'GOOD_0'],
+            'output': 'train',
+            'node': {
+                'type': 'AddOp',
+                'options': {}
+            }
+        },
+        {
+            'input': ['BAD_1', 'GOOD_1'],  # 'SUPER'],
+            'output': 'testval',
+            'node': {
+                'type': 'AddOp',
+                'options': {}
+            }
+        },
+        {
+            'input': 'testval',
+            'output': '_t',
+            'node': {
+                'type': 'ShuffleOp',
+                'options': {
+                    'seed': -1
+                }
+            }
+        },
+        {
+            'input': '_t',
+            'output': '_t2',
+            'node': {
+                'type': 'SubsampleOp',
+                'options': {
+                    'factor': 2
+                }
+            }
+        },
+        {
+            'input': '_t2',
+            'output': {
+                'a': 'test',
+                'b': 'val'
+            },
+            'node': {
+                'type': 'SplitsOp',
+                'options': {
+                    'split_map': {
+                        'a': 0.5,
+                        'b': 0.5
+                    }
+                }
+            }
+        },
+        # {
+        #     'input': 's',
+        #     'output': 's',
+        #     'node': {
+        #         'type': 'SubsampleOp',
+        #         'options': {
+        #             'factor': 0.5
+        #         }
+        #     }
+        # },
+        # {
+        #     'input': 's',
+        #     'output': {
+        #         'a': 'train',
+        #         'b': 'val',
+        #         'c': 'test',
+        #     },
+        #     'node': {
+        #         'type': 'SplitsOp',
+        #         'options': {
+        #             'split_map': {
+        #                 'a': 0.8,
+        #                 'b': 0.1,
+        #                 'c': 0.1
+        #             }
+        #         }
+        #     }
+        # }
+    ]
+}
+
+
+g = nx.DiGraph()
+
+modules = pipe['modules']
+random.shuffle(modules)
+nodes = []
+for m in pipe['modules']:
+
+    node = PipeNode(
+        input_ports=m['input'],
+        output_ports=m['output'],
+        node=m['node']
+    )
+    nodes.append(node)
+
+    print("NODE", node.id, node.in_ports(), node.out_ports())
+
+    [g.add_edge(DataNode(x), node) for x in node.in_ports()]
+    [g.add_edge(node, DataNode(x)) for x in node.out_ports()]
+
+
+# nx.draw_spring(g, with_labels=True)
+# plt.show()
+
+def prepare_in_data(node: PipeNode, cache: dict):
+    v = node.input_ports
+    if v is None:
+        return None
+    elif isinstance(v, str):
+        return cache[v]
+    elif isinstance(v, list):
+        return [cache[x] for x in v]
+    elif isinstance(v, dict):
+        return {k: cache[v] for k, v in v.items()}
+    else:
+        raise NotImplementedError(f'{type(v)}')
+
+
+def prepare_out_data(o: Union[SamplesSequence, list, dict], node: PipeNode, cache: dict):
+    v = node.output_ports
+    if v is None:
+        pass
+    elif isinstance(v, str):
+        cache[v] = o
+    elif isinstance(v, list):
+        for idx, name in enumerate(v):
+            cache[name] = o[idx]
+    elif isinstance(v, dict):
+        for k, name in v.items():
+            cache[name] = o[k]
+    else:
+        raise NotImplementedError(f'{type(v)}')
+
+
+_cache = {}
+A = nx.topological_sort(g)
+# for a in A:
+#     print("N: ", a)
+for a in A:
+    print("N: ", a)
+    if isinstance(a, PipeNode):
+        if a.node['type'] != 'Source':
+            op = SequenceOpFactory.create(a.node)
+            data = prepare_in_data(a, _cache)
+            out = op(data)
+            prepare_out_data(out, a, _cache)
+        else:
+            _cache[a.output_ports] = a.node['data']
+
+
+for k, v in _cache.items():
+    print(k, len(v))
+
+A = to_agraph(g)
+print(A)
+A.layout('dot')
+fname = f'{tempfile.NamedTemporaryFile().name}.png'
+print(fname)
+A.draw(fname)
+os.system(f'open {fname}')
diff --git a/pipelime/__init__.py b/pipelime/__init__.py
index 6e0b1f4..4042d27 100644
--- a/pipelime/__init__.py
+++ b/pipelime/__init__.py
@@ -3,3 +3,7 @@
 __author__ = """daniele de gregorio"""
 __email__ = 'daniele.degregorio@eyecan.ai'
 __version__ = '0.0.1'
+
+
+# Import transforms in order to register classes for serialization
+from pipelime.augmentations.transforms import *
diff --git a/pipelime/augmentations/__init__.py b/pipelime/augmentations/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/pipelime/augmentations/factory.py b/pipelime/augmentations/factory.py
new file mode 100644
index 0000000..650aad9
--- /dev/null
+++ b/pipelime/augmentations/factory.py
@@ -0,0 +1,24 @@
+
+from choixe.configurations import XConfig
+from schema import Optional, Schema
+import albumentations as A
+
+
+class AugmentationsFactory(object):
+
+    AUGMENTATIONS_CFG_SCHEMA = Schema({
+        Optional('__version__'): str,
+        'transform': dict
+    })
+
+    @classmethod
+    def build_from_file(cls, filename: str):
+        cfg = XConfig(filename=filename)
+        return cls.build_from_dict(cfg.to_dict())
+
+    @classmethod
+    def build_from_dict(cls, cfg: dict):
+        cfg = XConfig.from_dict(cfg)
+        cfg.set_schema(cls.AUGMENTATIONS_CFG_SCHEMA)
+        cfg.validate()
+        return A.from_dict(cfg.to_dict(discard_private_qualifiers=False))
diff --git a/examples/albumentations_test.py b/pipelime/augmentations/transforms.py
similarity index 77%
rename from examples/albumentations_test.py
rename to pipelime/augmentations/transforms.py
index 8214b21..69fe444 100644
--- a/examples/albumentations_test.py
+++ b/pipelime/augmentations/transforms.py
@@ -1,8 +1,6 @@
-from albumentations.core.serialization import SerializableMeta
-from albumentations.core.six import add_metaclass
-import rich
-import albumentations as A
+
 import cv2
+import albumentations as A
 
 
 class PadIfNeededV2(A.PadIfNeeded):
@@ -75,34 +73,3 @@ def update_params(self, params, **kwargs):
 
     def get_transform_init_args_names(self):
         return ("min_height", "min_width", "border_mode", "value", "mask_value", "row_align", "col_align")
-
-
-print(PadIfNeededV2.get_class_fullname())
-# A.core.serialization.SerializableMeta.__new__(PadIfNeededV2, 'PadIfNeededV2', )
-
-
-transform = A.Compose([
-    A.RandomCrop(width=450, height=450),
-    A.HorizontalFlip(p=0.5),
-    A.RandomBrightnessContrast(p=0.2),
-], bbox_params=A.BboxParams(format='coco'))
-#
-#transform = A.RGBShift()
-
-
-d = A.to_dict(transform)
-rich.print(d)
-a = {
-    'transform': {
-        '__class_fullname__': '__main__.PadIfNeededV2',
-        'min_height': 70,
-        'min_width': 70
-    }
-}
-
-
-t2 = A.from_dict(a)
-rich.print(t2)
-
-# d = PadIfNeededV2()
-# print(d)
diff --git a/pipelime/cli/__init__.py b/pipelime/cli/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/pipelime/cli/datasets/__init__.py b/pipelime/cli/datasets/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/pipelime/cli/datasets/generate_toy_dataset.py b/pipelime/cli/datasets/generate_toy_dataset.py
new file mode 100644
index 0000000..533422a
--- /dev/null
+++ b/pipelime/cli/datasets/generate_toy_dataset.py
@@ -0,0 +1,27 @@
+
+import numpy as np
+import cv2
+import click
+from pathlib import Path
+import yaml
+import json
+from pipelime.tools.toydataset import ToyDatasetGenerator
+
+
+@click.command("generate_toy_dataset")
+@click.option('--output_folder', type=str, required=True, help="Output folder")
+@click.option('--size', type=int, required=True, help="Number of output samples")
+@click.option('--image_size', type=int, default=256, help="Output image size")
+@click.option('--zfill', type=int, default=5, help="Zero padding digits")
+def generate_toy_dataset(output_folder, size, image_size, zfill):
+
+    ToyDatasetGenerator.generate_toy_dataset(
+        output_folder=output_folder,
+        size=size,
+        image_size=image_size,
+        zfill=zfill
+    )
+
+
+if __name__ == "__main__":
+    generate_toy_dataset()
diff --git a/pipelime/filesystem/__init__.py b/pipelime/filesystem/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/pipelime/filesystem/toolkit.py b/pipelime/filesystem/toolkit.py
new file mode 100644
index 0000000..460fb64
--- /dev/null
+++ b/pipelime/filesystem/toolkit.py
@@ -0,0 +1,122 @@
+from pathlib import Path
+import imghdr
+from typing import Union
+import numpy as np
+import imageio
+import yaml
+import json
+from collections import defaultdict
+
+
+class FSToolkit(object):
+
+    # Declare TREE structure
+    @classmethod
+    def tree(cls):
+        return defaultdict(cls.tree)
+
+    @classmethod
+    def tree_from_underscore_notation_files(cls, folder: str):
+        """Walk through files in folder generating a tree based on Underscore notation.
+        Leafs of abovementioned tree are string representing filenames, inner nodes represent
+        keys hierarchy.
+
+        :param folder: target folder
+        :type folder: str
+        :param skip_hidden_files: TRUE to skip files starting with '.', defaults to True
+        :type skip_hidden_files: bool, optional
+        :return: dictionary representing multilevel tree
+        :rtype: dict
+        """
+
+        keys_tree = cls.tree()
+        folder = Path(folder)
+        files = list(sorted(folder.glob('*')))
+        for f in files:
+            f: Path
+
+            if f.is_dir():
+                continue
+
+            name = f.stem
+
+            if name.startswith('.'):
+                continue
+
+            chunks = name.split('_', maxsplit=1)
+            if len(chunks) == 1:
+                continue
+
+            p = keys_tree
+            for index, chunk in enumerate(chunks):
+                if index < len(chunks) - 1:
+                    p = p[chunk]
+                else:
+                    p[chunk] = str(f)
+
+        return dict(keys_tree)
+
+    @classmethod
+    def get_file_extension(cls, filename, with_dot=False):
+        ext = Path(filename).suffix.lower()
+        if not with_dot and len(ext) > 0:
+            ext = ext[1:]
+        return ext
+
+    @classmethod
+    def is_metadata_file(cls, filename: str) -> bool:
+        ext = cls.get_file_extension(filename)
+        return ext in ['yml', 'json', 'toml', 'tml']
+
+    @classmethod
+    def is_file_image(cls, filename: str) -> bool:
+        return imghdr.what(filename) is not None
+
+    @classmethod
+    def is_file_numpy_array(cls, filename: str) -> bool:
+        ext = cls.get_file_extension(filename)
+        if ext in ['txt', 'data']:
+            try:
+                np.loadtxt(filename)
+                return True
+            except Exception:
+                return False
+        if ext in ['npy', 'npz']:
+            try:
+                np.load(filename)
+                return True
+            except Exception:
+                return False
+        return False
+
+    @classmethod
+    def load_data(cls, filename: str) -> Union[None, np.ndarray, dict]:
+        """ Load data from file based on its extension
+
+        :param filename: target filename
+        :type filename: str
+        :return: Loaded data as array or dict. May return NONE
+        :rtype: Union[None, np.ndarray, dict]
+        """
+
+        extension = cls.get_file_extension(filename)
+        data = None
+
+        if cls.is_file_image(filename):
+            data = np.array(imageio.imread(filename))
+
+        if cls.is_file_numpy_array(filename):
+            if extension in ['txt']:
+                data = np.loadtxt(filename)
+            elif extension in ['npy', 'npz']:
+                data = np.load(filename)
+            if data is not None:
+                data = np.atleast_2d(data)
+
+        if cls.is_metadata_file(filename):
+            if extension in ['yml']:
+                data = yaml.safe_load(open(filename, 'r'))
+            elif extension in ['json']:
+                data = json.load(open(filename))
+
+        return data
diff --git a/pipelime/lib.py b/pipelime/lib.py
deleted file mode 100644
index 9e38a9c..0000000
--- a/pipelime/lib.py
+++ /dev/null
@@ -1,814 +0,0 @@
-import rich
-from schema import Or, Schema
-import collections
-import dictquery as dq
-from dataclasses import dataclass
-import json
-import yaml
-from collections.abc import MutableMapping
-from abc import ABC,  abstractmethod
-from pathlib import Path
-from collections import defaultdict
-import imghdr
-from typing import Any, Dict, Sequence, Tuple, Union
-import numpy as np
-import imageio
-import random
-
-
-class FSToolkit(object):
-
-    @classmethod
-    def tree_from_underscore_notation_files(cls, folder, skip_hidden_files=True):
-        """Walk through files in folder generating a tree based on Underscore notation.
-        Leafs of abovementioned tree are string representing filenames, inner nodes represent
-        keys hierarchy.
-
-        :param folder: target folder
-        :type folder: str
-        :param skip_hidden_files: TRUE to skip files starting with '.', defaults to True
-        :type skip_hidden_files: bool, optional
-        :return: dictionary representing multilevel tree
-        :rtype: dict
-        """
-
-        # Declare TREE structure
-        def tree():
-            return defaultdict(tree)
-
-        keys_tree = tree()
-        folder = Path(folder)
-        files = list(sorted(folder.glob('*')))
-        for f in files:
-            name = f.stem
-
-            if skip_hidden_files:
-                if name.startswith('.'):
-                    continue
-
-            chunks = name.split('_', maxsplit=1)
-            if len(chunks) == 1:
-                chunks.append('none')
-            p = keys_tree
-            for index, chunk in enumerate(chunks):
-                if index < len(chunks) - 1:
-                    p = p[chunk]
-                else:
-                    p[chunk] = str(f)
-
-        return dict(keys_tree)
-
-    @classmethod
-    def get_file_extension(cls, filename, with_dot=False):
-        ext = Path(filename).suffix.lower()
-        if not with_dot and len(ext) > 0:
-            ext = ext[1:]
-        return ext
-
-    @classmethod
-    def is_metadata_file(cls, filename: str) -> bool:
-        ext = cls.get_file_extension(filename)
-        return ext in ['yml', 'json', 'toml', 'tml']
-
-    @classmethod
-    def is_file_image(cls, filename: str) -> bool:
-        return imghdr.what(filename) is not None
-
-    @classmethod
-    def is_file_numpy_array(cls, filename: str) -> bool:
-        ext = cls.get_file_extension(filename)
-        if ext in ['txt', 'data']:
-            try:
-                np.loadtxt(filename)
-                return True
-            except Exception:
-                return False
-        if ext in ['npy', 'npz']:
-            try:
-                np.load(filename)
-                return True
-            except Exception:
-                return False
-        return False
-
-    @classmethod
-    def load_data(cls, filename: str) -> Union[None, np.ndarray, dict]:
-        """ Load data from file based on its extension
-
-        :param filename: target filename
-        :type filename: str
-        :return: Loaded data as array or dict. May return NONE
-        :rtype: Union[None, np.ndarray, dict]
-        """
-
-        extension = cls.get_file_extension(filename)
-        data = None
-
-        if cls.is_file_image(filename):
-            data = np.array(imageio.imread(filename))
-
-        if cls.is_file_numpy_array(filename):
-            if extension in ['txt']:
-                data = np.loadtxt(filename)
-            elif extension in ['npy', 'npz']:
-                data = np.load(filename)
-            if data is not None:
-                data = np.atleast_2d(data)
-
-        if cls.is_metadata_file(filename):
-            if extension in ['yml']:
-                data = yaml.safe_load(open(filename, 'r'))
-            elif extension in ['json']:
-                data = json.load(open(filename))
-
-        return data
-
-
-@dataclass
-class MetaItem(object):
-    source: Any
-    source_type: Any
-
-
-class Sample(MutableMapping):
-
-    def __init__(self) -> None:
-        super().__init__()
-
-    def __init__(self, *args, **kwargs):
-        pass
-
-    @abstractmethod
-    def copy(self):
-        pass
-
-    @abstractmethod
-    def metaitem(self, key: any) -> MetaItem:
-        pass
-
-
-class PlainSample(Sample):
-
-    def __init__(self, data: dict = None):
-        super().__init__()
-        self._data = data if data is not None else {}
-
-    def __getitem__(self, key):
-        return self._data[key]
-
-    def __setitem__(self, key, value):
-        self._data[key] = value
-
-    def __delitem__(self, key):
-        del self._data[key]
-
-    def __iter__(self):
-        return iter(self._data.keys())
-
-    def __len__(self):
-        return len(self._data)
-
-    def __repr__(self) -> str:
-        return str(self._data)
-
-    def copy(self):
-        return PlainSample(data=self._data.copy())
-
-    def metaitem(self, key: any):
-        return MetaItem(
-            source=None,
-            source_type=None
-        )
-
-
-class FileSystemSample(Sample):
-
-    def __init__(self, data_map: dict, lazy: bool = True):
-        """ Creates a FileSystemSample starting from a key/filename map
-
-        :param data_map: key/filename map
-        :type data_map: dict
-        :param lazy: FALSE to preload data (slow), defaults to False
-        :type lazy: bool, optional
-        """
-        super().__init__()
-        self._filesmap = data_map
-        self._cached = {}
-        if not lazy:
-            for k in self.keys():
-                d = self[k]
-
-    def __getitem__(self, key):
-        if key not in self._cached:
-            self._cached[key] = FSToolkit.load_data(self._filesmap[key])
-        return self._cached[key]
-
-    def __setitem__(self, key, value):
-        self._cached[key] = value
-
-    def __delitem__(self, key):
-        del self._cached[key]
-
-    def __iter__(self):
-        return iter(set.union(set(self._filesmap.keys()), set(self._cached.keys())))
-
-    def __len__(self):
-        return len(self._filesmap)
-
-    def copy(self):
-        newsample = FileSystemSample(self._filesmap)
-        newsample._cached = self._cached.copy()
-        return newsample
-
-    def metaitem(self, key: any):
-        return MetaItem(
-            source=Path(self._filesmap[key]),
-            source_type=Path
-        )
-
-    # class XXX(dict):
-
-    #     def __init__(self, database: 'GenericSample'):
-    #         """ Creates a LazySample, aka a dict with items loaded when called
-
-    #         :param database: [description]
-    #         :type database: GenericSample
-    #         """
-    #         super().__init__()
-    #         self._database = database
-    #         self._cached = {}
-    #         self._keys_map = {}
-
-    #     def copy(self):
-    #         newsample = type(self)(database=self._database)
-    #         newsample._cached = self._cached.copy()
-    #         newsample._keys_map = self._keys_map.copy()
-    #         return newsample
-
-    #     def add_key(self, key: Any, reference: str):
-    #         self._keys_map[key] = reference
-
-    #     def __contains__(self, key: Any):
-    #         return key in self._keys_map
-
-    #     def get_path(self, key: Any) -> Path:
-    #         return Path(self._keys_map[key])
-
-    #     def keys(self):
-    #         return self._keys_map.keys()
-
-    #     def items(self):
-    #         return [(k, self[k]) for k in self.keys()]
-
-    #     def __getitem__(self, key: Any):
-
-    #         if key not in self._keys_map:
-    #             raise KeyError
-
-    #         if key not in self._cached:
-    #             self._cached[key] = self._database.load_data(self._keys_map[key])
-
-    #         return self._cached[key]
-
-    #     def pop(self, key):
-    #         k = self._keys_map[key]
-    #         del self._keys_map[key]
-    #         return k
-
-    #     def __setitem__(self, key, value):
-    #         self._keys_map[key] = value
-
-    #     def __delitem__(self, key):
-    #         if key in self._cached:
-    #             del self._cached[key]
-    #         if key in self._keys_map:
-    #             del self._keys_map[key]
-
-    #     def __iter__(self):
-    #         return iter(self._cached)
-
-    #     def __len__(self):
-    #         return len(self._keys_map)
-
-    #     def _keytransform(self, key):
-    #         return key
-
-
-# class SamplesSequenceAdder(object):
-
-#     def __add__(self, o: Any):
-#         if isinstance(o, SamplesSequence) or isinstance(o, type(self)):
-#             return type(self)(samples=self._samples + o._samples)
-#         else:
-#             raise NotImplementedError(f'Cannot apply __add__ to {type(o)}')
-
-
-class SamplesSequence(Sequence):
-
-    def __init__(self, samples: Sequence[Sample]):
-        self._samples = samples
-
-    @property
-    def samples(self):
-        return self._samples
-
-    def __len__(self) -> int:
-        return len(self._samples)
-
-    def __getitem__(self, idx: int):
-        if idx >= len(self):
-            raise IndexError
-
-        return self._samples[idx]
-
-    # def filter_by_query(self, query: str):
-    #     filtered_samples = []
-    #     for sample in self._samples:
-    #         if dq.match(sample, query):
-    #             filtered_samples.append(sample)
-    #     return SamplesSequence(samples=filtered_samples)
-
-    # def downsample(self, factor: int):
-    #     return SamplesSequence(samples=self._samples[::factor])
-
-    # def fraction(self, percentage: float):
-    #     new_size = int(len(self) * min(max(percentage, 0), 1.0))
-    #     return SamplesSequence(samples=self._samples[:new_size])
-
-    # def splits(self, percentages: Sequence = [0.8, 0.1, 0.1]):
-    #     """Splits sequence in N objects based on a percentage list
-
-    #     :param percentages: percentages list, defaults to [0.8, 0.1, 0.1]
-    #     :type percentages: list, optional
-    #     :return: list of PandasDatabase
-    #     :rtype: list
-    #     """
-    #     assert np.array(percentages).sum() <= 1.0, "Percentages sum must be <= 1.0"
-    #     sizes = []
-    #     for p in percentages:
-    #         sizes.append(int(len(self) * p))
-
-    #     sizes[-1] += len(self) - np.array(sizes).sum()
-
-    #     chunks = []
-    #     current_index = 0
-    #     for s in sizes:
-    #         _samples = self._samples[current_index:current_index + s]
-    #         chunks.append(SamplesSequence(samples=_samples))
-    #         current_index += s
-    #     return tuple(chunks)
-
-    # def splits_as_dict(self, percentages_dictionary: dict = {'train': 0.9, 'test': 0.1}):
-    #     """Splits PandasDatabase in N objects based on a percentage dictionary name/percentage
-
-    #     :param percentages_dictionary: percentages dictionary, defaults to {'train': 0.9, 'test': 0.1}
-    #     :type percentages_dictionary: dict, optional
-    #     :return: dict of name/PandasDatabase pairs
-    #     :rtype: dict
-    #     """
-    #     names = list(percentages_dictionary.keys())
-    #     percentages = list(percentages_dictionary.values())
-    #     chunks = self.splits(percentages=percentages)
-    #     output = {}
-    #     assert len(names) == len(chunks), "Len of chunks is different from percentage names number"
-    #     for idx in range(len(names)):
-    #         output[names[idx]] = chunks[idx]
-    #     return output
-
-    # def shuffle(self, seed=-1):
-    #     """Produces a shuffled copy of the original sequence
-
-    #     :param seed: controlled random seed , defaults to -1
-    #     :type seed: int, optional
-    #     :return: Shuffled copy of the original sequence
-    #     :rtype: SamplesSequence
-    #     """
-    #     new_data = self._samples.copy()
-    #     random.seed(seed)
-    #     random.shuffle(new_data)
-    #     return SamplesSequence(samples=new_data)
-
-    # def __add__(self, o: Any):
-    #     if isinstance(o, SamplesSequence):
-    #         return SamplesSequence(samples=self._samples + o._samples)
-    #     else:
-    #         raise NotImplementedError(f'Cannot apply __add__ to {type(o)}')
-
-    # def __mod__(self, o: Any):
-    #     if isinstance(o, int):
-    #         return self.downsample(o)
-    #     elif isinstance(o, float):
-    #         return self.fraction(o)
-    #     elif isinstance(o, str):
-    #         return self.filter_by_query(o)
-    #     else:
-    #         raise NotImplementedError(f'Cannot apply __mod__ to {type(o)}')
-
-    # def __truediv__(self, o: Any):
-    #     if isinstance(o, Tuple):
-    #         return self.splits(percentages=o)
-    #     elif isinstance(o, Dict):
-    #         return self.splits_as_dict(percentages_dictionary=o)
-    #     else:
-    #         raise NotImplementedError(f'Cannot apply __truediv__ to {type(o)}')
-
-
-# class SamplesSequenceOp(SamplesSequence, SamplesSequenceAdder):
-
-#     def __init__(self, samples: Sequence[Sample]):
-#         super().__init__(samples)
-
-
-class OpPort(object):
-
-    def __init__(self, schema):
-        self._schema = Schema(schema)
-
-    def match(self, o: 'OpPort'):
-        return self._schema.json_schema(0) == o._schema.json_schema(0)
-
-    def is_valid_data(self, o: Any):
-        return self._schema.validate(o)
-
-    def __repr__(self):
-        return str(self._schema)
-
-
-class SequenceOp(ABC):
-
-    @abstractmethod
-    def input_port(self) -> OpPort:
-        raise NotImplementedError()
-
-    @abstractmethod
-    def output_port(self) -> OpPort:
-        raise NotImplementedError()
-
-    @abstractmethod
-    def __call__(self, x: Any) -> Any:
-        p = self.input_port()
-        assert p.is_valid_data(x)
-
-    def match(self, x: 'SequenceOp'):
-        p0 = self.output_port()
-        p1 = x.input_port()
-        return p0.match(p1)
-
-    def print(self):
-        rich.print({
-            'name': self.__class__.__name__,
-            'input': self.input_port(),
-            'output': self.output_port()
-        })
-
-    @classmethod
-    @abstractmethod
-    def op_name(cls) -> str:
-        raise NotImplementedError()
-
-    @classmethod
-    @abstractmethod
-    def factory_schema(cls) -> Schema:
-        raise NotImplementedError()
-
-    @classmethod
-    def build_from_dict(cls, d: dict):
-        cls.factory_schema().validate(d)
-
-
-class SequenceOpFactory(object):
-
-    FACTORY_MAP: Dict[str, SequenceOp] = {}
-
-    @classmethod
-    def generic_op_schema(cls) -> Schema:
-        return Schema({
-            'type': str,
-            'options': dict
-        })
-
-    @classmethod
-    def register_op(cls, op: SequenceOp):
-        cls.FACTORY_MAP[op.__name__] = op
-
-    @classmethod
-    def create(cls, cfg: dict) -> SequenceOp:
-        cls.generic_op_schema().validate(cfg)
-        _t = cls.FACTORY_MAP[cfg['type']]
-        return _t.build_from_dict(cfg)
-
-
-def op_factory(o):
-    SequenceOpFactory.register_op(o)
-    return o
-
-
-@op_factory
-class AddOp(SequenceOp):
-
-    def input_port(self) -> OpPort:
-        return OpPort([SamplesSequence])
-
-    def output_port(self) -> OpPort:
-        return OpPort(SamplesSequence)
-
-    def __call__(self, x: Sequence[SamplesSequence]) -> SamplesSequence:
-        super().__call__(x)
-        new_samples = []
-        for s in x:
-            new_samples += s.samples
-        return SamplesSequence(samples=new_samples)
-
-    @classmethod
-    def op_name(cls) -> str:
-        return AddOp.__name__
-
-    @classmethod
-    def factory_schema(cls) -> Schema:
-        return Schema({
-            'type': cls.op_name(),
-            'options': dict
-        })
-
-    @classmethod
-    def build_from_dict(cls, d: dict):
-        super().build_from_dict(d)
-        return AddOp()
-
-
-@op_factory
-class SubsampleOp(SequenceOp):
-    def __init__(self, factor: Union[int, float]) -> None:
-        super().__init__()
-        self._factor = factor
-
-    def input_port(self) -> OpPort:
-        return OpPort(SamplesSequence)
-
-    def output_port(self) -> OpPort:
-        return OpPort(SamplesSequence)
-
-    def __call__(self, x: SamplesSequence) -> SamplesSequence:
-        super().__call__(x)
-
-        new_samples = x.samples.copy()
-        if isinstance(self._factor, int):
-            new_samples = new_samples[::self._factor]
-        elif isinstance(self._factor, float):
-            new_size = int(len(new_samples) * min(max(self._factor, 0), 1.0))
-            new_samples = new_samples[:new_size]
-
-        return SamplesSequence(samples=new_samples)
-
-    @classmethod
-    def op_name(cls) -> str:
-        return SubsampleOp.__name__
-
-    @classmethod
-    def factory_schema(cls) -> Schema:
-        return Schema({
-            'type': cls.op_name(),
-            'options': {
-                'factor': Or(float, int)
-            }
-        })
-
-    @classmethod
-    def build_from_dict(cls, d: dict):
-        super().build_from_dict(d)
-        return SubsampleOp(d['options']['factor'])
-
-
-@op_factory
-class ShuffleOp(SequenceOp):
-
-    def __init__(self, seed=-1) -> None:
-        super().__init__()
-        self._seed = seed
-
-    def input_port(self):
-        return OpPort(SamplesSequence)
-
-    def output_port(self):
-        return OpPort(SamplesSequence)
-
-    def __call__(self, x: SamplesSequence) -> Any:
-        super().__call__(x)
-        new_data = x.samples.copy()
-        random.seed(self._seed if self._seed >= 0 else None)
-        random.shuffle(new_data)
-        return SamplesSequence(samples=new_data)
-
-    @classmethod
-    def op_name(cls) -> str:
-        return ShuffleOp.__name__
-
-    @classmethod
-    def factory_schema(cls) -> Schema:
-        return Schema({
-            'type': cls.op_name(),
-            'options': {
-                'seed': int
-            }
-        })
-
-    @classmethod
-    def build_from_dict(cls, d: dict):
-        super().build_from_dict(d)
-        return ShuffleOp(d['options']['seed'])
-
-
-@op_factory
-class SplitsOp(SequenceOp):
-
-    def __init__(self, split_map: Dict[str, float]) -> None:
-        super().__init__()
-        self._split_map = split_map
-
-    def input_port(self):
-        return OpPort(SamplesSequence)
-
-    def output_port(self):
-        return OpPort({str: SamplesSequence})
-
-    def _splits(self, x: SamplesSequence, percentages: Sequence = [0.8, 0.1, 0.1]):
-        """Splits sequence in N objects based on a percentage list
-
-        :param percentages: percentages list, defaults to [0.8, 0.1, 0.1]
-        :type percentages: list, optional
-        :return: list of PandasDatabase
-        :rtype: list
-        """
-        assert np.array(percentages).sum() <= 1.0, "Percentages sum must be <= 1.0"
-        sizes = []
-        for p in percentages:
-            sizes.append(int(len(x) * p))
-
-        sizes[-1] += len(x) - np.array(sizes).sum()
-
-        chunks = []
-        current_index = 0
-        for s in sizes:
-            _samples = x.samples[current_index:current_index + s]
-            chunks.append(SamplesSequence(samples=_samples))
-            current_index += s
-        return tuple(chunks)
-
-    def _splits_as_dict(self, x: SamplesSequence, percentages_dictionary: dict = {'train': 0.9, 'test': 0.1}):
-        """Splits PandasDatabase in N objects based on a percentage dictionary name/percentage
-
-        :param percentages_dictionary: percentages dictionary, defaults to {'train': 0.9, 'test': 0.1}
-        :type percentages_dictionary: dict, optional
-        :return: dict of name/PandasDatabase pairs
-        :rtype: dict
-        """
-        names = list(percentages_dictionary.keys())
-        percentages = list(percentages_dictionary.values())
-        chunks = self._splits(x, percentages=percentages)
-        output = {}
-        assert len(names) == len(chunks), "Len of chunks is different from percentage names number"
-        for idx in range(len(names)):
-            output[names[idx]] = chunks[idx]
-        return output
-
-    def __call__(self, x: SamplesSequence) -> Dict[str, SamplesSequence]:
-        super().__call__(x)
-        return self._splits_as_dict(x, self._split_map)
-
-    @classmethod
-    def op_name(cls) -> str:
-        return SplitsOp.__name__
-
-    @classmethod
-    def factory_schema(cls) -> Schema:
-        return Schema({
-            'type': cls.op_name(),
-            'options': {
-                'split_map': {str: float}
-            }
-        })
-
-    @classmethod
-    def build_from_dict(cls, d: dict):
-        super().build_from_dict(d)
-        return SplitsOp(d['options']['split_map'])
-
-
-@op_factory
-class Dict2ListOp(SequenceOp):
-
-    def input_port(self):
-        return OpPort({str: SamplesSequence})
-
-    def output_port(self):
-        return OpPort([SamplesSequence])
-
-    def __call__(self, x: Dict[str, SamplesSequence]) -> Sequence[SamplesSequence]:
-        super().__call__(x)
-        return list(x.values())
-
-    @classmethod
-    def op_name(cls) -> str:
-        return Dict2ListOp.__name__
-
-    @classmethod
-    def factory_schema(cls) -> Schema:
-        return Schema({
-            'type': cls.op_name(),
-            'options': dict
-        })
-
-    @classmethod
-    def build_from_dict(cls, d: dict):
-        super().build_from_dict(d)
-        return Dict2ListOp()
-
-
-@op_factory
-class FilterByQueryOp(SequenceOp):
-
-    def __init__(self, query: str) -> None:
-        self._query = query
-
-    def input_port(self):
-        return OpPort(SamplesSequence)
-
-    def output_port(self):
-        return OpPort(SamplesSequence)
-
-    def __call__(self, x: SamplesSequence) -> SamplesSequence:
-        super().__call__(x)
-        filtered_samples = []
-        for sample in x.samples:
-            if dq.match(sample, self._query):
-                filtered_samples.append(sample)
-        return SamplesSequence(samples=filtered_samples)
-
-    @classmethod
-    def op_name(cls) -> str:
-        return FilterByQueryOp.__name__
-
-    @classmethod
-    def factory_schema(cls) -> Schema:
-        return Schema({
-            'type': cls.op_name(),
-            'options': {
-                'query': str
-            }
-        })
-
-    @classmethod
-    def build_from_dict(cls, d: dict):
-        super().build_from_dict(d)
-        return FilterByQueryOp(query=d['options']['query'])
-
-
-@op_factory
-class SplitByQueryOp(SequenceOp):
-
-    def __init__(self, query: str) -> None:
-        self._query = query
-
-    def input_port(self):
-        return OpPort(SamplesSequence)
-
-    def output_port(self):
-        return OpPort([SamplesSequence])
-
-    def __call__(self, x: SamplesSequence) -> Sequence[SamplesSequence]:
-        super().__call__(x)
-        a = []
-        b = []
-        for sample in x.samples:
-            if dq.match(sample, self._query):
-                a.append(sample)
-            else:
-                b.append(sample)
-
-        return (
-            SamplesSequence(samples=a),
-            SamplesSequence(samples=b)
-        )
-
-    @classmethod
-    def op_name(cls) -> str:
-        return SplitByQueryOp.__name__
-
-    @classmethod
-    def factory_schema(cls) -> Schema:
-        return Schema({
-            'type': cls.op_name(),
-            'options': {
-                'query': str
-            }
-        })
-
-    @classmethod
-    def build_from_dict(cls, d: dict):
-        super().build_from_dict(d)
-        return SplitByQueryOp(query=d['options']['query'])
-
-
-# SequenceOpFactory.register_op(AddOp)
-# SequenceOpFactory.register_op(SubsampleOp)
-# SequenceOpFactory.register_op(FilterByQueryOp)
diff --git a/pipelime/sequences/__init__.py b/pipelime/sequences/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/pipelime/sequences/operations.py b/pipelime/sequences/operations.py
new file mode 100644
index 0000000..1334bb9
--- /dev/null
+++ b/pipelime/sequences/operations.py
@@ -0,0 +1,628 @@
+import pydash as py_
+import dictquery as dq
+import numpy as np
+import random
+from typing import Any, Dict, Sequence, Union
+from pipelime.sequences.samples import GroupedSample, SamplesSequence
+from abc import ABC, abstractmethod
+from schema import Or, Schema
+import rich
+
+
+class OperationPort(object):
+
+    def __init__(self, schema: dict):
+        """ Creates an Operation Port with an associated schema (as dict)
+
+        :param schema: port data schema
+        :type schema: dict
+        """
+        self._schema = Schema(schema)
+
+    def match(self, o: 'OperationPort'):
+        return self._schema.json_schema(0) == o._schema.json_schema(0)
+
+    def is_valid_data(self, o: Any):
+        return self._schema.validate(o)
+
+    def __repr__(self):
+        return str(self._schema)
+
+
+class SequenceOperation(ABC):
+    """Object representing a generic pipeline operation on a sequence
+    """
+
+    def __init__(self) -> None:
+        pass
+
+    @abstractmethod
+    def input_port(self) -> OperationPort:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def output_port(self) -> OperationPort:
+        raise NotImplementedError()
+
+    @abstractmethod
+    def __call__(self, x: Any) -> Any:
+        p = self.input_port()
+        assert p.is_valid_data(x)
+
+    def match(self, x: 'SequenceOperation'):
+        p0 = self.output_port()
+        p1 = x.input_port()
+        return p0.match(p1)
+
+    def print(self):
+        rich.print({
+            'name': self.__class__.__name__,
+            'input': self.input_port(),
+            'output': self.output_port()
+        })
+
+    @classmethod
+    @abstractmethod
+    def op_name(cls) -> str:
+        raise NotImplementedError()
+
+    @classmethod
+    @abstractmethod
+    def factory_schema(cls) -> Schema:
+        raise NotImplementedError()
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        cls.factory_schema().validate(d)
+
+    @abstractmethod
+    def to_dict(self) -> dict:
+        pass
+
+
+class SequenceOperationFactory(object):
+
+    FACTORY_MAP: Dict[str, SequenceOperation] = {}
+
+    @classmethod
+    def generic_op_schema(cls) -> Schema:
+        return Schema({
+            'type': str,
+            'options': dict
+        })
+
+    @classmethod
+    def register_op(cls, op: SequenceOperation):
+        cls.FACTORY_MAP[op.__name__] = op
+
+    @classmethod
+    def create(cls, cfg: dict) -> SequenceOperation:
+        cls.generic_op_schema().validate(cfg)
+        _t = cls.FACTORY_MAP[cfg['type']]
+        return _t.build_from_dict(cfg)
+
+
+def register_operation_factory(o: SequenceOperation) -> SequenceOperation:
+    """ Register a SequenceOperation to the factory
+
+    :param o: operation to register
+    :type o: SequenceOperation
+    :return: the same operation. It is intent to be used as decorator
+    :rtype: SequenceOperation
+    """
+    SequenceOperationFactory.register_op(o)
+    return o
+
+
+@register_operation_factory
+class OperationSum(SequenceOperation):
+
+    def __init__(self) -> None:
+        """ Concatenatas multiple sequences
+        """
+        super().__init__()
+
+    def input_port(self) -> OperationPort:
+        return OperationPort([SamplesSequence])
+
+    def output_port(self) -> OperationPort:
+        return OperationPort(SamplesSequence)
+
+    def __call__(self, x: Sequence[SamplesSequence]) -> SamplesSequence:
+        super().__call__(x)
+        new_samples = []
+        for s in x:
+            new_samples += s.samples
+        return SamplesSequence(samples=new_samples)
+
+    @classmethod
+    def op_name(cls) -> str:
+        return OperationSum.__name__
+
+    @classmethod
+    def factory_schema(cls) -> Schema:
+        return Schema({
+            'type': cls.op_name(),
+            'options': dict
+        })
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        super().build_from_dict(d)
+        return OperationSum()
+
+    def to_dict(self):
+        return {
+            'type': self.op_name(),
+            'options': {}
+        }
+
+
+@register_operation_factory
+class OperationIdentity(SequenceOperation):
+
+    def __init__(self) -> None:
+        """ No Op
+        """
+        super().__init__()
+
+    def input_port(self) -> OperationPort:
+        return OperationPort(any)
+
+    def output_port(self) -> OperationPort:
+        return OperationPort(any)
+
+    def __call__(self, x: any) -> any:
+        super().__call__(x)
+        return x
+
+    @classmethod
+    def op_name(cls) -> str:
+        return OperationIdentity.__name__
+
+    @classmethod
+    def factory_schema(cls) -> Schema:
+        return Schema({
+            'type': cls.op_name(),
+            'options': dict
+        })
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        super().build_from_dict(d)
+        return OperationIdentity()
+
+    def to_dict(self):
+        return {
+            'type': self.op_name(),
+            'options': {}
+        }
+
+
+@register_operation_factory
+class OperationSubsample(SequenceOperation):
+    def __init__(self, factor: Union[int, float]) -> None:
+        """ Subsample an input sequence elements
+
+        :param factor: if INT is the subsampling factor, if FLOAT is the percentage of input elements to preserve
+        :type factor: Union[int, float]
+        """
+        super().__init__()
+        self._factor = factor
+
+    def input_port(self) -> OperationPort:
+        return OperationPort(SamplesSequence)
+
+    def output_port(self) -> OperationPort:
+        return OperationPort(SamplesSequence)
+
+    def __call__(self, x: SamplesSequence) -> SamplesSequence:
+        super().__call__(x)
+
+        new_samples = x.samples.copy()
+        if isinstance(self._factor, int):
+            new_samples = new_samples[::self._factor]
+        elif isinstance(self._factor, float):
+            new_size = int(len(new_samples) * min(max(self._factor, 0), 1.0))
+            new_samples = new_samples[:new_size]
+
+        return SamplesSequence(samples=new_samples)
+
+    @classmethod
+    def op_name(cls) -> str:
+        return OperationSubsample.__name__
+
+    @classmethod
+    def factory_schema(cls) -> Schema:
+        return Schema({
+            'type': cls.op_name(),
+            'options': {
+                'factor': Or(float, int)
+            }
+        })
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        super().build_from_dict(d)
+        return OperationSubsample(d['options']['factor'])
+
+    def to_dict(self) -> dict:
+        return {
+            'type': self.op_name(),
+            'options': {
+                'factor': self._factor
+            }
+        }
+
+
+@register_operation_factory
+class OperationShuffle(SequenceOperation):
+
+    def __init__(self, seed=-1) -> None:
+        """ Shuffle input sequence elements
+
+        :param seed: random seed (-1 for current system time), defaults to -1
+        :type seed: int, optional
+        """
+        super().__init__()
+        self._seed = seed
+
+    def input_port(self):
+        return OperationPort(SamplesSequence)
+
+    def output_port(self):
+        return OperationPort(SamplesSequence)
+
+    def __call__(self, x: SamplesSequence) -> Any:
+        super().__call__(x)
+        new_data = x.samples.copy()
+        random.seed(self._seed if self._seed >= 0 else None)
+        random.shuffle(new_data)
+        return SamplesSequence(samples=new_data)
+
+    @classmethod
+    def op_name(cls) -> str:
+        return OperationShuffle.__name__
+
+    @classmethod
+    def factory_schema(cls) -> Schema:
+        return Schema({
+            'type': cls.op_name(),
+            'options': {
+                'seed': int
+            }
+        })
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        super().build_from_dict(d)
+        return OperationShuffle(d['options']['seed'])
+
+    def to_dict(self) -> dict:
+        return {
+            'type': self.op_name(),
+            'options': {
+                'seed': self._seed
+            }
+        }
+
+
+@register_operation_factory
+class OperationSplits(SequenceOperation):
+
+    def __init__(self, split_map: Dict[str, float]) -> None:
+        """ Splits an input sequence in multiple sub-sequences in a key/sequence map
+
+        :param split_map: key/percengate map used as split map. Sum of percentages must be 1.0
+        :type split_map: Dict[str, float]
+        """
+        super().__init__()
+        self._split_map = split_map
+
+    def input_port(self):
+        return OperationPort(SamplesSequence)
+
+    def output_port(self):
+        return OperationPort({str: SamplesSequence})
+
+    def _splits(self, x: SamplesSequence, percentages: Sequence = [0.8, 0.1, 0.1]):
+        """Splits sequence in N objects based on a percentage list
+
+        :param percentages: percentages list, defaults to [0.8, 0.1, 0.1]
+        :type percentages: list, optional
+        :return: list of PandasDatabase
+        :rtype: list
+        """
+        assert np.array(percentages).sum() <= 1.0, "Percentages sum must be <= 1.0"
+        sizes = []
+        for p in percentages:
+            sizes.append(int(len(x) * p))
+
+        sizes[-1] += len(x) - np.array(sizes).sum()
+
+        chunks = []
+        current_index = 0
+        for s in sizes:
+            _samples = x.samples[current_index:current_index + s]
+            chunks.append(SamplesSequence(samples=_samples))
+            current_index += s
+        return tuple(chunks)
+
+    def _splits_as_dict(self, x: SamplesSequence, percentages_dictionary: dict = {'train': 0.9, 'test': 0.1}):
+        """Splits PandasDatabase in N objects based on a percentage dictionary name/percentage
+
+        :param percentages_dictionary: percentages dictionary, defaults to {'train': 0.9, 'test': 0.1}
+        :type percentages_dictionary: dict, optional
+        :return: dict of name/PandasDatabase pairs
+        :rtype: dict
+        """
+        names = list(percentages_dictionary.keys())
+        percentages = list(percentages_dictionary.values())
+        chunks = self._splits(x, percentages=percentages)
+        output = {}
+        assert len(names) == len(chunks), "Len of chunks is different from percentage names number"
+        for idx in range(len(names)):
+            output[names[idx]] = chunks[idx]
+        return output
+
+    def __call__(self, x: SamplesSequence) -> Dict[str, SamplesSequence]:
+        super().__call__(x)
+        return self._splits_as_dict(x, self._split_map)
+
+    @classmethod
+    def op_name(cls) -> str:
+        return OperationSplits.__name__
+
+    @classmethod
+    def factory_schema(cls) -> Schema:
+        return Schema({
+            'type': cls.op_name(),
+            'options': {
+                'split_map': {str: float}
+            }
+        })
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        super().build_from_dict(d)
+        return OperationSplits(d['options']['split_map'])
+
+    def to_dict(self) -> dict:
+        return {
+            'type': self.op_name(),
+            'options': {
+                'split_map': self._split_map
+            }
+        }
+
+
+@register_operation_factory
+class OperationDict2List(SequenceOperation):
+
+    def __init__(self) -> None:
+        """ Converts a Dict of sequences into a play list of sequences
+        """
+        super().__init__()
+
+    def input_port(self):
+        return OperationPort({str: SamplesSequence})
+
+    def output_port(self):
+        return OperationPort([SamplesSequence])
+
+    def __call__(self, x: Dict[str, SamplesSequence]) -> Sequence[SamplesSequence]:
+        super().__call__(x)
+        return list(x.values())
+
+    @classmethod
+    def op_name(cls) -> str:
+        return OperationDict2List.__name__
+
+    @classmethod
+    def factory_schema(cls) -> Schema:
+        return Schema({
+            'type': cls.op_name(),
+            'options': dict
+        })
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        super().build_from_dict(d)
+        return OperationDict2List()
+
+    def to_dict(self) -> dict:
+        return {
+            'type': self.op_name(),
+            'options': {}
+        }
+
+
+@register_operation_factory
+class OperationFilterByQuery(SequenceOperation):
+
+    def __init__(self, query: str) -> None:
+        """ Filter sequence elements based on query string. If sample contains a 'metadata' item
+        storing a dict like {'metadata':{'num': 10}}, a query string could be something like
+        query = '`metadata.num` > 10' .
+
+        :param query: query string (@see dictquery)
+        :type query: str
+        """
+        self._query = query
+
+    def input_port(self):
+        return OperationPort(SamplesSequence)
+
+    def output_port(self):
+        return OperationPort(SamplesSequence)
+
+    def __call__(self, x: SamplesSequence) -> SamplesSequence:
+        super().__call__(x)
+        filtered_samples = []
+        for sample in x.samples:
+            if dq.match(sample, self._query):
+                filtered_samples.append(sample)
+        return SamplesSequence(samples=filtered_samples)
+
+    @classmethod
+    def op_name(cls) -> str:
+        return OperationFilterByQuery.__name__
+
+    @classmethod
+    def factory_schema(cls) -> Schema:
+        return Schema({
+            'type': cls.op_name(),
+            'options': {
+                'query': str
+            }
+        })
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        super().build_from_dict(d)
+        return OperationFilterByQuery(query=d['options']['query'])
+
+    def to_dict(self) -> dict:
+        return {
+            'type': self.op_name(),
+            'options': {
+                'query': self._query
+            }
+        }
+
+
+@register_operation_factory
+class OperationSplitByQuery(SequenceOperation):
+
+    def __init__(self, query: str) -> None:
+        """ Splits sequence elements in two sub-sequences based on an input query. The first list
+        will contains elements with positve query matches, the second list the negative ones.
+
+        :param query: query string (@see dictquery)
+        :type query: str
+        """
+        self._query = query
+
+    def input_port(self):
+        return OperationPort(SamplesSequence)
+
+    def output_port(self):
+        return OperationPort([SamplesSequence])
+
+    def __call__(self, x: SamplesSequence) -> Sequence[SamplesSequence]:
+        super().__call__(x)
+        a = []
+        b = []
+        for sample in x.samples:
+            if dq.match(sample, self._query):
+                a.append(sample)
+            else:
+                b.append(sample)
+
+        return (
+            SamplesSequence(samples=a),
+            SamplesSequence(samples=b)
+        )
+
+    @classmethod
+    def op_name(cls) -> str:
+        return OperationSplitByQuery.__name__
+
+    @classmethod
+    def factory_schema(cls) -> Schema:
+        return Schema({
+            'type': cls.op_name(),
+            'options': {
+                'query': str
+            }
+        })
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        super().build_from_dict(d)
+        return OperationSplitByQuery(query=d['options']['query'])
+
+    def to_dict(self) -> dict:
+        return {
+            'type': self.op_name(),
+            'options': {
+                'query': self._query
+            }
+        }
+
+
+@register_operation_factory
+class OperationGroupBy(SequenceOperation):
+
+    def __init__(self, field: str, ungrouped: bool = False) -> None:
+        """ Groups sequence elements accoring to specific field
+
+        :param query: field string (@see pydash deep path notation)
+        :type query: str
+        :param ungrouped: TRUE to propagate ungrouped samples as separate group, if field is wrong
+        and ungrouped is TRUE, a single sample will be propagated. If some samples are lacking in the selected field,
+        they will be put in this special group.
+        :type ungrouped: bool
+        """
+        self._field = field.replace('`', '')
+        self._ungrouped = ungrouped
+
+    def input_port(self):
+        return OperationPort(SamplesSequence)
+
+    def output_port(self):
+        return OperationPort(SamplesSequence)
+
+    def __call__(self, x: SamplesSequence) -> Sequence[SamplesSequence]:
+        super().__call__(x)
+
+        groups_map = {}
+        none_group = []
+        for sample in x.samples:
+            value = py_.get(sample, self._field)
+            if value is not None:
+                if value not in groups_map:
+                    groups_map[value] = []
+                groups_map[value].append(sample)
+            else:
+                none_group.append(sample)
+
+        out_samples = []
+        for k, samples in groups_map.items():
+            g = GroupedSample(samples=samples)
+            g['__groupbyvalue__'] = k
+            out_samples.append(g)
+        if len(none_group) > 0 and self._ungrouped:
+            g = GroupedSample(samples=none_group)
+            g['__groupbyvalue__'] = None
+            out_samples.append(g)
+
+        return SamplesSequence(samples=out_samples)
+
+    @classmethod
+    def op_name(cls) -> str:
+        return OperationGroupBy.__name__
+
+    @classmethod
+    def factory_schema(cls) -> Schema:
+        return Schema({
+            'type': cls.op_name(),
+            'options': {
+                'field': str,
+                'ungrouped': bool
+            }
+        })
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        super().build_from_dict(d)
+        return OperationGroupBy(field=d['options']['field'], ungrouped=d['options']['ungrouped'])
+
+    def to_dict(self) -> dict:
+        return {
+            'type': self.op_name(),
+            'options': {
+                'field': self._field,
+                'ungrouped': self._ungrouped
+            }
+        }
diff --git a/pipelime/sequences/readers/__init__.py b/pipelime/sequences/readers/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/pipelime/sequences/readers/filesystem.py b/pipelime/sequences/readers/filesystem.py
new file mode 100644
index 0000000..79bcfcf
--- /dev/null
+++ b/pipelime/sequences/readers/filesystem.py
@@ -0,0 +1,32 @@
+from pathlib import Path
+from pipelime.sequences.samples import FileSystemSample, SamplesSequence
+from pipelime.filesystem.toolkit import FSToolkit
+import rich
+
+
+class UnderfolderReader(SamplesSequence):
+    DATA_SUBFOLDER = 'data'
+
+    def __init__(self, folder: str, copy_root_files: bool = True) -> None:
+        self._folder = Path(folder)
+        self._datafolder = self._folder / self.DATA_SUBFOLDER
+        assert self._datafolder.exists(), f'No data folder found: "{self._datafolder}"'
+
+        # builds tree from subfolder with underscore notation
+        self._tree = FSToolkit.tree_from_underscore_notation_files(self._datafolder)
+        self._ids = list(sorted(self._tree.keys()))
+        self._root_files = [x for x in Path(self._folder).glob('*') if x.is_file()]
+        self._root_data = {}
+        for f in self._root_files:
+            self._root_data[f.stem] = str(f)  # FSToolkit.load_data(f)
+
+        samples = []
+        for idx in range(len(self._ids)):
+            data = dict(self._tree[self._ids[idx]])
+            if copy_root_files:
+                data.update(self._root_data)
+
+            sample = FileSystemSample(data_map=data, lazy=True, id=self._ids[idx])
+            samples.append(sample)
+
+        super().__init__(samples=samples)
diff --git a/pipelime/sequences/samples.py b/pipelime/sequences/samples.py
new file mode 100644
index 0000000..33761a4
--- /dev/null
+++ b/pipelime/sequences/samples.py
@@ -0,0 +1,240 @@
+import uuid
+from pipelime.filesystem.toolkit import FSToolkit
+from dataclasses import dataclass
+from collections.abc import MutableMapping
+from abc import abstractmethod
+from pathlib import Path
+from typing import Any, Hashable, Sequence
+
+
+@dataclass
+class MetaItem(object):
+
+    def __init__(self) -> None:
+        pass
+
+    @abstractmethod
+    def source(self) -> any:
+        pass
+
+
+class MemoryItem(MetaItem):
+
+    def __init__(self) -> None:
+        super().__init__()
+
+    def source(self) -> any:
+        return None
+
+
+class FilesystemItem(MetaItem):
+
+    def __init__(self, path: str) -> None:
+        super().__init__()
+        self._path = Path(path)
+
+    def source(self) -> any:
+        return self._path
+
+
+class Sample(MutableMapping):
+
+    def __init__(self, id: Hashable = None) -> None:
+        self._id = id if id is not None else str(uuid.uuid1())
+
+    @property
+    def id(self):
+        return self._id
+
+    @abstractmethod
+    def rename(self, old_key: str, new_key: str):
+        pass
+
+    @abstractmethod
+    def copy(self):
+        pass
+
+    @abstractmethod
+    def metaitem(self, key: any) -> MetaItem:
+        pass
+
+
+class GroupedSample(Sample):
+
+    def __init__(self, samples: Sequence[Sample], id: Hashable = None) -> None:
+        """ Sample representing a group of basic samples
+
+        :param samples: list of samples to group
+        :type samples: Sequence[Sample]
+        :param id: hashable value used as id, defaults to None
+        :type id: Hashable, optional
+        """
+        super().__init__(id=id)
+        self._samples = samples
+
+    def _merge_dicts(self, ds: Sequence[dict]):
+        d = {}
+        if len(ds) > 0:
+            for k in ds[0].keys():
+                d[k] = tuple(d[k] for d in ds)
+        return d
+
+    def __getitem__(self, key):
+        if len(self._samples) > 0:
+            d = [x[key] for x in self._samples]
+            if isinstance(d[0], dict):
+                d = self._merge_dicts(d)
+            return d
+        return None
+
+    def __setitem__(self, key, value):
+        for x in self._samples:
+            x[key] = value
+
+    def __delitem__(self, key):
+        for x in self._samples:
+            del x[key]
+
+    def __iter__(self):
+        for x in self._samples:
+            return iter(x.keys())
+        return None
+
+    def __len__(self):
+        for x in self._samples:
+            return len(x)
+        return 0
+
+    def __repr__(self) -> str:
+        return str(self._samples)
+
+    def copy(self):
+        return GroupedSample(samples=self._samples)
+
+    def rename(self, old_key: str, new_key: str):
+        for x in self._samples:
+            x.rename(old_key=old_key, new_key=new_key)
+
+    def metaitem(self, key: any):
+        for x in self._samples:
+            return x.metaitem(key)
+        return None
+
+
+class PlainSample(Sample):
+
+    def __init__(self, data: dict = None, id: Hashable = None):
+        """ Plain sample (aka a dict wrapper)
+
+        :param data: dictionary data, defaults to None
+        :type data: dict, optional
+        :param id: hashable value used as id, defaults to None
+        :type id: Hashable, optional
+        """
+        super().__init__(id=id)
+        self._data = data if data is not None else {}
+
+    def __getitem__(self, key):
+        return self._data[key]
+
+    def __setitem__(self, key, value):
+        self._data[key] = value
+
+    def __delitem__(self, key):
+        del self._data[key]
+
+    def __iter__(self):
+        return iter(self._data.keys())
+
+    def __len__(self):
+        return len(self._data)
+
+    def __repr__(self) -> str:
+        return str(self._data)
+
+    def copy(self):
+        return PlainSample(data=self._data.copy())
+
+    def rename(self, old_key: str, new_key: str):
+        if new_key not in self._data and old_key in self._data:
+            self._data[new_key] = self._data[old_key]
+            del self._data[old_key]
+
+    def metaitem(self, key: any):
+        return MemoryItem()
+
+
+class FileSystemSample(Sample):
+
+    def __init__(self, data_map: dict, lazy: bool = True, id: Hashable = None):
+        """ Creates a FileSystemSample based on a key/filename map
+
+        :param data_map: key/filename map
+        :type data_map: dict
+        :param lazy: FALSE to preload data (slow), defaults to False
+        :type lazy: bool, optional
+        :param id: hashable value used as id
+        :type id: Hashable, optional
+        """
+        super().__init__(id=id)
+        self._filesmap = data_map
+        self._cached = {}
+        if not lazy:
+            for k in self.keys():
+                d = self[k]
+
+    def __getitem__(self, key):
+        if key not in self._cached:
+            self._cached[key] = FSToolkit.load_data(self._filesmap[key])
+        return self._cached[key]
+
+    def __setitem__(self, key, value):
+        self._cached[key] = value
+
+    def __delitem__(self, key):
+        if key in self._cached:
+            del self._cached[key]
+        if key in self._filesmap:
+            del self._filesmap[key]
+
+    def __iter__(self):
+        return iter(set.union(set(self._filesmap.keys()), set(self._cached.keys())))
+
+    def __len__(self):
+        return len(self._filesmap)
+
+    def copy(self):
+        newsample = FileSystemSample(self._filesmap)
+        newsample._cached = self._cached.copy()
+        return newsample
+
+    def rename(self, old_key: str, new_key: str):
+        if new_key not in self._filesmap and old_key in self._filesmap:
+            self._filesmap[new_key] = self._filesmap.pop(old_key)
+            if old_key in self._cached:
+                self._cached[new_key] = self._cached.pop(old_key)
+
+    def metaitem(self, key: any):
+        if key in self._filesmap:
+            return FilesystemItem(self._filesmap[key])
+        else:
+            return MemoryItem()
+
+
+class SamplesSequence(Sequence):
+
+    def __init__(self, samples: Sequence[Sample]):
+        self._samples = samples
+
+    @property
+    def samples(self):
+        return self._samples
+
+    def __len__(self) -> int:
+        return len(self._samples)
+
+    def __getitem__(self, idx: int):
+        if idx >= len(self):
+            raise IndexError
+
+        return self._samples[idx]
diff --git a/pipelime/sequences/stages.py b/pipelime/sequences/stages.py
new file mode 100644
index 0000000..3a5c9d6
--- /dev/null
+++ b/pipelime/sequences/stages.py
@@ -0,0 +1,231 @@
+import albumentations as A
+from abc import ABC, abstractmethod
+from typing import Dict, Hashable, Sequence, Union
+
+from schema import Schema
+from pipelime.sequences.samples import Sample
+
+
+class SampleStage(ABC):
+
+    def __init__(self):
+        pass
+
+    @abstractmethod
+    def __call__(self, x:
+                 Union[
+                     Sample,
+                     Sequence[Sample],
+                     Dict[Hashable, Sample]
+                 ]) -> Union[
+                     Sample,
+                     Sequence[Sample],
+                     Dict[Hashable, Sample]
+    ]:
+        pass
+
+    @classmethod
+    @abstractmethod
+    def stage_name(cls) -> str:
+        raise NotImplementedError()
+
+    @classmethod
+    @abstractmethod
+    def factory_schema(cls) -> Schema:
+        raise NotImplementedError()
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        cls.factory_schema().validate(d)
+
+    @abstractmethod
+    def to_dict(self) -> dict:
+        pass
+
+
+class SampleStagesFactory(object):
+
+    FACTORY_MAP: Dict[str, SampleStage] = {}
+
+    @classmethod
+    def generic_op_schema(cls) -> Schema:
+        return Schema({
+            'type': str,
+            'options': dict
+        })
+
+    @classmethod
+    def register_stage(cls, stage: SampleStage):
+        cls.FACTORY_MAP[stage.__name__] = stage
+
+    @classmethod
+    def create(cls, cfg: dict) -> SampleStage:
+        cls.generic_op_schema().validate(cfg)
+        _t = cls.FACTORY_MAP[cfg['type']]
+        return _t.build_from_dict(cfg)
+
+
+def register_stage_factory(s: SampleStage) -> SampleStage:
+    """ Register a SampleStage to the factory
+
+    :param s: stage to register
+    :type s: SampleStage
+    :return: the same stage. It is intent to be used as decorator
+    :rtype: SampleStage
+    """
+    SampleStagesFactory.register_stage(s)
+    return s
+
+
+@register_stage_factory
+class StageIdentity(SampleStage):
+
+    def __init__(self):
+        super().__init__()
+
+    def __call__(self, x: any) -> any:
+        return x
+
+    @classmethod
+    def stage_name(cls) -> str:
+        return StageIdentity.__name__
+
+    @classmethod
+    def factory_schema(cls) -> Schema:
+        return Schema({
+            'type': cls.stage_name(),
+            'options': {}
+        })
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        super().build_from_dict(d)
+        return StageIdentity()
+
+    def to_dict(self) -> dict:
+        return {
+            'type': self.stage_name(),
+            'options': {}
+        }
+
+
+@register_stage_factory
+class StageRemap(SampleStage):
+
+    def __init__(self, remap: dict, remove_missing: bool = True):
+        """ Remaps keys in sample
+
+        :param remap: old_key:new_key dictionary remap
+        :type remap: dict
+        :param remove_missing: if TRUE missing keys in remap will be removed in the output sample, defaults to True
+        :type remove_missing: bool, optional
+        """
+        super().__init__()
+        self._remap = remap
+        self._remove_missing = remove_missing
+
+    def __call__(self, x: Sample) -> Sample:
+
+        out: Sample = x.copy()
+        for k in x.keys():
+            # for k, v in x.items():
+            if k in self._remap:
+                out.rename(k, self._remap[k])
+            else:
+                if self._remove_missing:
+                    del out[k]
+        return out
+
+    @classmethod
+    def stage_name(cls) -> str:
+        return StageRemap.__name__
+
+    @classmethod
+    def factory_schema(cls) -> Schema:
+        return Schema({
+            'type': cls.stage_name(),
+            'options': {
+                'remap': dict,
+                'remove_missing': bool
+            }
+        })
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        super().build_from_dict(d)
+        return StageRemap(
+            remap=d['options']['remap'],
+            remove_missing=d['options']['remove_missing']
+        )
+
+    def to_dict(self) -> dict:
+        return {
+            'type': self.stage_name(),
+            'options': {
+                'remap': self._remap,
+                'remove_missing': self._remove_missing
+            }
+        }
+
+
+@register_stage_factory
+class StageAugmentations(SampleStage):
+
+    def __init__(self, transform_cfg: dict, targets: dict):
+        super().__init__()
+        self._transform_cfg = transform_cfg
+        self._targets = targets
+        self._transform: A.Compose = A.from_dict(transform_cfg)
+        self._transform.add_targets(self._purge_targets(self._targets))
+
+    def _purge_targets(self, targets: dict):
+        # TODO: could it be wrong if targets also contains 'image' or 'mask' (aka default target)?
+        return targets
+
+    def __call__(self, x: Sample) -> Sample:
+
+        try:
+            out = x.copy()
+            to_transform = {}
+            for k, data in x.items():
+                if k in self._targets:
+                    to_transform[k] = data
+
+            _transformed = self._transform(**to_transform)
+            for k in self._targets.keys():
+                out[k] = _transformed[k]
+
+            return out
+        except Exception as e:
+            raise Exception(f'Stage[{self.__class__.__name__}] -> {e}')
+
+    @classmethod
+    def stage_name(cls) -> str:
+        return StageAugmentations.__name__
+
+    @classmethod
+    def factory_schema(cls) -> Schema:
+        return Schema({
+            'type': cls.stage_name(),
+            'options': {
+                'transform_cfg': dict,
+                'targets': dict
+            }
+        })
+
+    @classmethod
+    def build_from_dict(cls, d: dict):
+        super().build_from_dict(d)
+        return StageAugmentations(
+            transform_cfg=d['options']['transform_cfg'],
+            targets=d['options']['targets']
+        )
+
+    def to_dict(self) -> dict:
+        return {
+            'type': self.stage_name(),
+            'options': {
+                'transform_cfg': self._transform_cfg,
+                'targets': self._targets
+            }
+        }
diff --git a/pipelime/tools/__init__.py b/pipelime/tools/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/pipelime/tools/toydataset.py b/pipelime/tools/toydataset.py
new file mode 100644
index 0000000..d395fee
--- /dev/null
+++ b/pipelime/tools/toydataset.py
@@ -0,0 +1,165 @@
+import yaml
+import json
+from pathlib import Path
+from typing import Sequence
+import numpy as np
+from PIL import Image, ImageDraw
+import uuid
+import imageio
+
+
+class ToyDatasetGenerator(object):
+
+    def __init__(self, palette=None):
+        pass
+
+    def generate_random_background(self, size, channels=3):
+        return np.random.uniform(0, 255, (size[1], size[0], channels)).astype(np.uint8)
+
+    def generate_random_object_2D(self, size, max_label=5):
+        size = np.array(size)
+        center = np.random.uniform(0.25 * size[0], 0.75 * size[0], (2,))
+        random_size = np.random.uniform(size * 0.05, size * 0.24, (2,))
+        top_left = np.array([center[0] - random_size[0] * 0.5, center[1] - random_size[1] * 0.5])
+        bottom_right = np.array([center[0] + random_size[0] * 0.5, center[1] + random_size[1] * 0.5])
+        diag = bottom_right - top_left
+
+        kp0 = np.array([center[0], center[1] - random_size[1] * 0.5])
+        kp1 = np.array([center[0], center[1] + random_size[1] * 0.5])
+
+        width = random_size[0]
+        height = random_size[1]
+        label = np.random.randint(0, max_label + 1)
+
+        return {
+            'center': center,
+            'size': random_size,
+            'tl': top_left,
+            'br': bottom_right,
+            'diag': diag,
+            'kp0': kp0,
+            'kp1': kp1,
+            'w': width,
+            'h': height,
+            'label': label
+        }
+
+    def generate_2d_object_bbox(self, size, obj):
+        center = obj['center']
+        w, h = size
+        return (
+            center[0] - obj['w'] * 0.5,
+            center[1] - obj['h'] * 0.5,
+            center[0] + obj['w'] * 0.5,
+            center[1] + obj['h'] * 0.5,
+            obj['label']
+        )
+
+    def generate_2d_object_keypoints(self, size, obj):
+
+        center = obj['center']
+        tl = obj['tl']
+        br = obj['br']
+        diag = obj['diag']
+        obj_size = obj['size']
+        w, h = size
+        label = obj['label']
+
+        orientation = np.arctan2(diag[1], diag[0])
+        orientation2 = np.arctan2(-diag[1], -diag[0])
+
+        scale = 0.5 * np.linalg.norm(diag)
+
+        # kp0 = (label, center[0] / w, center[1] / h, orientation, scale, 0)
+        kp1 = (tl[0], tl[1], -orientation, scale, label, 1)
+        kp2 = (br[0], br[1], -orientation2, scale, label, 2)
+
+        return [kp1, kp2]
+
+    def generate_2d_objects_images(self, size, objects):
+
+        image = Image.fromarray(self.generate_random_background(size))
+        mask = Image.new('L', size)
+        instances = Image.new('L', size)
+
+        for index, obj in enumerate(objects):
+            label = obj['label']
+            coords = tuple(obj['tl']), tuple(obj['br'])
+            color = tuple(np.random.randint(0, 255, (3,)))
+
+            # ImageDraw.Draw(image).ellipse(coords, fill=color)
+            # ImageDraw.Draw(mask).ellipse(coords, fill=label + 1)
+            # ImageDraw.Draw(instances).ellipse(coords, fill=index + 1)
+
+            ImageDraw.Draw(image).rectangle(coords, fill=color)
+            ImageDraw.Draw(mask).rectangle(coords, fill=label + 1)
+            ImageDraw.Draw(instances).rectangle(coords, fill=index + 1)
+
+        return {
+            'rgb': np.array(image),
+            'mask': np.array(mask),
+            'instances': np.array(instances)
+        }
+
+    def generate_image_sample(self, size, max_label=5, objects_number_range=[1, 5]):
+
+        objects = []
+        bboxes = []
+        keypoints = []
+        objects_number = np.random.randint(objects_number_range[0], objects_number_range[1])
+        for n in range(objects_number):
+            obj = self.generate_random_object_2D(size, max_label=max_label)
+            box = self.generate_2d_object_bbox(size, obj)
+            kps = self.generate_2d_object_keypoints(size, obj)
+            objects.append(obj)
+            bboxes.append(box)
+            keypoints.extend(kps)
+
+        data = self.generate_2d_objects_images(size, objects)
+        data.update({
+            'bboxes': bboxes,
+            'keypoints': keypoints,
+            'label': np.random.randint(max_label + 1),
+            'id': str(uuid.uuid1())
+        })
+        return data
+
+    @classmethod
+    def generate_toy_dataset(cls, output_folder: str, size: int, image_size: Sequence[int], zfill: int = 5):
+
+        output_folder = Path(output_folder)
+        output_folder.mkdir(parents=True, exist_ok=True)
+
+        generator = ToyDatasetGenerator()
+
+        for idx in range(size):
+
+            # Generate sample
+            sample = generator.generate_image_sample([image_size, image_size])
+
+            # Extracts metadata
+            metadata = {
+                'bboxes': sample['bboxes'],
+                'keypoints': sample['keypoints'],
+                'label': sample['label'],
+                'id': sample['id']
+            }
+
+            metadata = json.loads(json.dumps(metadata))
+
+            # Naming
+            name = str(idx).zfill(zfill)
+            image_name = f'{name}_image.jpg'
+            mask_name = f'{name}_mask.png'
+            instances_name = f'{name}_inst.png'
+            metadata_name = f'{name}_metadata.yml'
+            keypoints_name = f'{name}_keypoints.txt'
+            bboxes_name = f'{name}_bboxes.npy'
+
+            imageio.imwrite(str(output_folder / image_name), sample['rgb'])
+            imageio.imwrite(str(output_folder / mask_name), sample['mask'])
+            imageio.imwrite(str(output_folder / instances_name), sample['instances'])
+            np.savetxt(str(output_folder / keypoints_name), sample['keypoints'])
+            np.save(str(output_folder / bboxes_name), sample['bboxes'])
+
+            yaml.safe_dump(metadata, open(str(output_folder / metadata_name), 'w'))
diff --git a/requirements.txt b/requirements.txt
index b287601..9dbc08f 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -2,4 +2,8 @@ click>=7.1.2
 rich>=9.9.0
 dictquery
 schema
-albumentations>=0.5.2
\ No newline at end of file
+albumentations>=0.5.2
+choixe
+imageio
+pydash
+Pillow
\ No newline at end of file


Test patch:
diff --git a/examples/test_cfg.yml b/examples/test_cfg.yml
new file mode 100644
index 0000000..1e6dd6e
--- /dev/null
+++ b/examples/test_cfg.yml
@@ -0,0 +1,35 @@
+__version__: 0.5.1
+transform:
+  __class_fullname__: albumentations.core.composition.Compose
+  additional_targets: {}
+  bbox_params:
+    { 
+      check_each_transform: false,
+      format: yolo,
+      label_fields: null,
+      min_area: 0.0,
+      min_visibility: 0.0,
+    }
+  keypoint_params: null
+  p: 1.0
+  transforms:
+  - __class_fullname__: albumentations.augmentations.transforms.RandomCrop
+    always_apply: false
+    height: 450
+    p: 1.0
+    width: 450
+  - __class_fullname__: albumentations.augmentations.transforms.HorizontalFlip
+    always_apply: false
+    p: 0.5
+  - 
+    {
+      __class_fullname__: albumentations.augmentations.transforms.RandomBrightnessContrast,
+      always_apply: false,
+      brightness_by_max: true,
+      brightness_limit: [-0.2, 0.2],
+      contrast_limit: [-0.2, 0.2],
+      p: 0.2,
+    }
+  - __class_fullname__: pipelime.augmentations.transforms.PadIfNeededV2
+    min_height: 70
+    min_width: 70
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..ca1f096
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,36 @@
+from pipelime.tools.toydataset import ToyDatasetGenerator
+import pytest
+import os
+from pathlib import Path
+
+
+@pytest.fixture(scope='session')
+def data_folder():
+    dirname = os.path.dirname(__file__)
+    return os.path.join(dirname, 'sample_data')
+
+
+@pytest.fixture(scope='function')
+def toy_dataset_small(tmpdir):
+    folder = Path(tmpdir)
+    datafolder = folder / 'data'
+    size = 32
+    image_size = 256
+    zfill = 5
+    ToyDatasetGenerator.generate_toy_dataset(
+        datafolder,
+        size=size,
+        image_size=image_size,
+        zfill=zfill
+    )
+
+    return {
+        'folder': folder,
+        'data_folder': datafolder,
+        'size': size,
+        'image_size': image_size,
+        'zfill': zfill,
+        'keypoints_format': 'xyas',
+        'bboxes_format': 'pascal_voc',
+        'expected_keys': ['image', 'mask', 'inst', 'keypoints', 'bboxes']
+    }
diff --git a/tests/pipelime/__init__.py b/tests/pipelime/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/pipelime/augmentations/__init__.py b/tests/pipelime/augmentations/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/pipelime/augmentations/conftest.py b/tests/pipelime/augmentations/conftest.py
new file mode 100644
index 0000000..bdf039f
--- /dev/null
+++ b/tests/pipelime/augmentations/conftest.py
@@ -0,0 +1,12 @@
+import pytest
+from pathlib import Path
+
+
+@pytest.fixture(scope='session')
+def augmentations_test_configurations(data_folder):
+    configurations_folder = Path(data_folder) / 'augmentations'
+    return [
+        {
+            'filename': configurations_folder / 'full_custom_augmentations.yml'
+        },
+    ]
diff --git a/tests/pipelime/augmentations/test_factory.py b/tests/pipelime/augmentations/test_factory.py
new file mode 100644
index 0000000..8b8d6fa
--- /dev/null
+++ b/tests/pipelime/augmentations/test_factory.py
@@ -0,0 +1,32 @@
+
+from pipelime.augmentations.factory import AugmentationsFactory
+import albumentations as A
+import numpy as np
+from choixe.configurations import XConfig
+
+
+class TestAugmentationsFactory(object):
+
+    def _transform_test_image(self, transform):
+
+        img = np.random.uniform(0., 1., (256, 256, 3))
+        img = transform(image=img)
+
+    def test_creation_from_file(self, augmentations_test_configurations):
+
+        for cfg in augmentations_test_configurations:
+
+            transform = AugmentationsFactory.build_from_file(cfg['filename'])
+            assert transform is not None
+
+            self._transform_test_image(transform)
+
+    def test_creation_from_cfg(self, augmentations_test_configurations):
+
+        for cfg in augmentations_test_configurations:
+
+            xc = XConfig(filename=cfg['filename']).to_dict()
+            transform = AugmentationsFactory.build_from_dict(xc)
+            assert transform is not None
+
+            self._transform_test_image(transform)
diff --git a/tests/pipelime/sequences/__init__.py b/tests/pipelime/sequences/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/pipelime/sequences/conftest.py b/tests/pipelime/sequences/conftest.py
new file mode 100644
index 0000000..9ab921e
--- /dev/null
+++ b/tests/pipelime/sequences/conftest.py
@@ -0,0 +1,51 @@
+from pathlib import Path
+from pipelime.tools.toydataset import ToyDatasetGenerator
+from pipelime.sequences.samples import PlainSample, SamplesSequence
+import pytest
+import numpy as np
+
+
+@pytest.fixture(scope='session')
+def plain_samples_generator():
+    def generate_plain_samples(
+            namespace: str,
+            N: int,
+            group_each: int = 5,
+            heavy_data: bool = True
+    ):
+
+        samples = []
+        for i in range(N):
+            data = {
+                'idx': f'{namespace}{i}',
+                'number': i,
+                'fraction': i / 1000.,
+                'odd': i % 2 == 1,
+                'data0': np.random.uniform(0.0, 1.0, (64, 64, 3)) if heavy_data else 1.64643,
+                'data1': np.random.uniform(0, 255, (8, 8)).astype(np.uint8) if heavy_data else 25588,
+                'metadata': {
+                    'even': i % 2 == 0,
+                    'name': f'{namespace}{i}',
+                    'N': i,
+                    'deep': {
+                        'super_deep': 0,
+                        'groupby_field': int(i / group_each)
+                    }
+                }
+            }
+            samples.append(PlainSample(data=data))
+
+        return samples
+    return generate_plain_samples
+
+
+@pytest.fixture(scope='session')
+def plain_samples_sequence_generator(plain_samples_generator):
+    def generate_plain_samples_sequence(
+        namespace: str,
+        N: int,
+        group_each: int = 5,
+        heavy_data: bool = True
+    ):
+        return SamplesSequence(samples=plain_samples_generator(namespace, N, group_each, heavy_data))
+    return generate_plain_samples_sequence
diff --git a/tests/pipelime/sequences/readers/__init__.py b/tests/pipelime/sequences/readers/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/pipelime/sequences/readers/test_filesystem.py b/tests/pipelime/sequences/readers/test_filesystem.py
new file mode 100644
index 0000000..6becbb9
--- /dev/null
+++ b/tests/pipelime/sequences/readers/test_filesystem.py
@@ -0,0 +1,25 @@
+
+
+from pipelime.sequences.samples import FileSystemSample, FilesystemItem, Sample
+from pipelime.sequences.readers.filesystem import UnderfolderReader
+
+
+class TestUnderfolder(object):
+
+    def test_reader(self, toy_dataset_small):
+        print("OK", toy_dataset_small)
+
+        size = toy_dataset_small['size']
+        folder = toy_dataset_small['folder']
+        keys = toy_dataset_small['expected_keys']
+
+        reader = UnderfolderReader(folder=folder)
+        assert len(reader) == size
+
+        for sample in reader:
+            assert isinstance(sample, Sample)
+            assert isinstance(sample, FileSystemSample)
+
+            for key in keys:
+                assert key in sample
+                assert isinstance(sample.metaitem(key), FilesystemItem)
diff --git a/tests/pipelime/sequences/test_operations.py b/tests/pipelime/sequences/test_operations.py
new file mode 100644
index 0000000..23166c9
--- /dev/null
+++ b/tests/pipelime/sequences/test_operations.py
@@ -0,0 +1,331 @@
+import pydash
+import functools
+import hashlib
+from typing import Dict, Sequence
+import pytest
+import rich
+from schema import Schema
+from pipelime.sequences.samples import PlainSample, SamplesSequence
+from pipelime.sequences.operations import OperationDict2List, OperationFilterByQuery, OperationGroupBy, OperationIdentity, OperationPort, OperationShuffle, OperationSplitByQuery, OperationSplits, OperationSubsample, OperationSum, SequenceOperation, SequenceOperationFactory
+
+
+def _plug_test(op: SequenceOperation):
+    """ Test what a generic SequenceOperation should do
+
+    :param op: input SequenceOperation
+    :type op: SequenceOperation
+    """
+
+    assert isinstance(op, SequenceOperation)
+    assert isinstance(op.input_port(), OperationPort)
+    assert isinstance(op.output_port(), OperationPort)
+    assert op.input_port().match(op.input_port())
+    assert op.output_port().match(op.output_port())
+
+    reop = op.build_from_dict(op.to_dict())
+    assert isinstance(reop, SequenceOperation)
+    assert isinstance(op.factory_schema(), Schema)
+    op.print()
+
+    factored = SequenceOperationFactory.create(op.to_dict())
+    assert isinstance(factored, SequenceOperation)
+
+
+class TestOperationSum(object):
+
+    def test_sum(self, plain_samples_sequence_generator):
+
+        pairs = [
+            (32, 5),
+            (32, 1),
+            (8, 32),
+            (1, 24),
+            (0, 64),
+            (48, 0)
+        ]
+
+        for N, D in pairs:
+            datasets = []
+            for idx in range(D):
+                datasets.append(plain_samples_sequence_generator('d{idx}_', N))
+
+            op = OperationSum()
+            _plug_test(op)
+
+            if D > 0:
+                out = op(datasets)
+
+                assert isinstance(out, SamplesSequence)
+                assert N * D == len(out)
+            else:
+                with pytest.raises(AssertionError):
+                    out = op(datasets)
+
+
+class TestOperationSubsample(object):
+
+    def test_subsample(self, plain_samples_sequence_generator):
+
+        sizes = [32, 10, 128, 16, 1]
+        factors = [2, 0.1, -0.1, 10.2, 10, 20, 1.0, 1]
+
+        for F in factors:
+            for N in sizes:
+                dataset = plain_samples_sequence_generator('d0_', N)
+
+                op = OperationSubsample(factor=F)
+                _plug_test(op)
+
+                if N > 0:
+                    out = op(dataset)
+
+                    assert isinstance(out, SamplesSequence)
+
+                    if isinstance(F, int):
+                        if F > 1 and N > 1:
+                            assert len(out) < N
+                        if F == 1:
+                            assert len(out) == N
+
+                    if isinstance(F, float):
+                        if 0.0 < F < 1.0:
+                            assert len(out) < N
+
+                        if F == 1.0:
+                            assert len(out) == N
+
+
+class TestOperationIdentity(object):
+
+    def test_subsample(self, plain_samples_sequence_generator):
+
+        sizes = [32, 10, 128, 16, 1]
+
+        for N in sizes:
+            dataset = plain_samples_sequence_generator('d0_', N)
+
+            op = OperationIdentity()
+            _plug_test(op)
+
+            out = op(dataset)
+
+            assert out == dataset
+
+
+class TestOperationShuffle(object):
+
+    def _sign(self, dataset: SamplesSequence):
+        names = [x['idx'] for x in dataset.samples]
+        return hashlib.md5(bytes('_'.join(names), encoding='utf-8')).hexdigest()
+
+    def test_shuffle(self, plain_samples_sequence_generator):
+
+        sizes = [100, 10, 128, 20]
+
+        for N in sizes:
+            dataset = plain_samples_sequence_generator('d0_', N)
+
+            sign = self._sign(dataset)
+
+            op = OperationShuffle(seed=10)
+            _plug_test(op)
+
+            out = op(dataset)
+            out_sign = self._sign(out)
+
+            assert sign != out_sign, 'for a series of unfortunate events did the two hashes collide?'
+
+            rich.print(sign, "!=", out_sign)
+
+            assert len(dataset) == len(out)
+
+
+class TestOperationSplits(object):
+
+    def test_splits(self, plain_samples_sequence_generator):
+
+        N = 128
+        splits_cfgs = [
+            {'split_map': {'train': 0.5, 'test': 0.2, 'val': 0.3}, 'good': True},
+            {'split_map': {'a': 0.4, 'b': 0.2, 'c': 0.2, 'd': 0.2}, 'good': True},
+            {'split_map': {'a': 0.8, 'b': 0.}, 'good': True},
+            {'split_map': {'a': 1.0}, 'good': True},
+            {'split_map': {'a': 0.7}, 'good': True},
+            {'split_map': {'a': 1.7}, 'good': False},
+            {'split_map': {}, 'good': False},
+        ]
+
+        for cfg in splits_cfgs:
+
+            good = cfg['good']
+            split_map = cfg['split_map']
+            expected_splits = len(split_map)
+            dataset = plain_samples_sequence_generator('d0_', N)
+
+            if good:
+                op = OperationSplits(split_map=split_map)
+                _plug_test(op)
+                out_dict = op(dataset)
+                assert expected_splits == len(out_dict)
+                assert len(set(out_dict.keys()).difference(set(split_map.keys()))) == 0
+
+                sumup = 0
+                for k, d in out_dict.items():
+                    sumup += len(d)
+                    print("XX", len(d))
+
+                print("SUMP", split_map.keys(), sumup, N)
+                assert sumup == N
+
+                idx_maps: Dict[str, set] = {}
+                for k, d in out_dict.items():
+                    idxs = set([x['idx'] for x in d])
+                    idx_maps[k] = idxs
+
+                for k0, _ in idx_maps.items():
+                    for k1, _ in idx_maps.items():
+                        if k0 != k1:
+                            intersection = idx_maps[k0].intersection(idx_maps[k1])
+                            assert len(intersection) == 0
+            else:
+                with pytest.raises(Exception):
+                    op = OperationSplits(split_map=split_map)
+                    _plug_test(op)
+                    out_dict = op(dataset)
+
+
+class TestOperationDict2List(object):
+
+    def test_shuffle(self, plain_samples_sequence_generator):
+
+        pairs = [
+            (10, 5),
+            (10, 1),
+            (1, 1),
+            (1, 10),
+            (1, 0),
+        ]
+
+        for N, D in pairs:
+            datasets = []
+            datasets_map = {}
+            for idx in range(D):
+                d = plain_samples_sequence_generator('d{idx}_', N)
+                datasets.append(d)
+                datasets_map[str(idx)] = d
+
+            op = OperationDict2List()
+            _plug_test(op)
+
+            if D > 0:
+                out = op(datasets_map)
+
+                assert isinstance(out, list)
+            else:
+                with pytest.raises(Exception):
+                    out = op(datasets_map)
+
+
+class TestOperationFilterByQuery(object):
+
+    def test_filter_by_query(self, plain_samples_sequence_generator):
+
+        N = 128
+        dataset = plain_samples_sequence_generator('d0_', N)
+
+        queries_items = [
+            {'query': '`metadata.name` CONTAINS "d0_"', 'expected': N},
+            {'query': '`metadata.name` CONTAINS "@IMPOSSIBLE@!S_TRING12"', 'expected': 0},
+            {'query': f'`metadata.N` < {N/2}', 'expected': N / 2},
+            {'query': '`metadata.name` like "d?_*"', 'expected': N},
+
+        ]
+
+        for item in queries_items:
+            query = item['query']
+            expected = item['expected']
+            op = OperationFilterByQuery(query=query)
+            _plug_test(op)
+
+            out = op(dataset)
+
+            assert len(out) == expected
+
+
+class TestOperationSplitByQuery(object):
+
+    def test_split_by_query(self, plain_samples_sequence_generator):
+
+        N = 128
+        dataset = plain_samples_sequence_generator('d0_', N)
+
+        queries_items = [
+            {'query': '`metadata.name` CONTAINS "d0_"', 'expected': [N, 0]},
+            {'query': '`metadata.name` CONTAINS "@IMPOSSIBLE@!S_TRING12"', 'expected': [0, N]},
+            {'query': f'`metadata.N` < {N/2}', 'expected': [N / 2, N / 2]},
+            {'query': '`metadata.name` like "d?_*"', 'expected': [N, 0]},
+
+        ]
+
+        for item in queries_items:
+            query = item['query']
+            expecteds = item['expected']
+            op = OperationSplitByQuery(query=query)
+            _plug_test(op)
+
+            out = op(dataset)
+
+            assert isinstance(out, Sequence)
+            for idx in range(len(out)):
+                assert len(out[idx]) == expecteds[idx]
+
+            sumup = functools.reduce(lambda a, b: len(a) + len(b), out)
+            sumup_exp = functools.reduce(lambda a, b: a + b, expecteds)
+
+            assert sumup == sumup_exp
+
+
+class TestOperationGroupBy(object):
+
+    def test_group_by(self, plain_samples_sequence_generator):
+
+        N = 128
+        dataset = plain_samples_sequence_generator('d0_', N, heavy_data=False)
+
+        fields_items = [
+            {'field': '`metadata.deep.groupby_field`', 'valid': True},
+            {'field': '`metadata.even`', 'valid': True},
+            {'field': '`odd`', 'valid': True},
+            {'field': '`z`', 'valid': False},
+        ]
+
+        for item in fields_items:
+            field = item['field']
+            valid = item['valid']
+
+            for ungrouped in [True, False]:
+                op = OperationGroupBy(field=field, ungrouped=ungrouped)
+                _plug_test(op)
+
+                out = op(dataset)
+                if valid:
+                    assert len(out) < len(dataset)
+
+                    counters = {}
+                    for sample in out:
+                        for k, v in sample.items():
+                            rich.print(k, v)
+                        for key in sample.keys():
+                            if not isinstance(sample[key], dict):
+                                if key not in counters:
+                                    counters[key] = 0
+                                counters[key] += len(sample[key])
+
+                    for k, c in counters.items():
+                        print(k, c)
+                        assert c == N
+                else:
+                    if ungrouped:
+                        assert len(out) == 1
+                    else:
+                        assert len(out) == 0
diff --git a/tests/pipelime/sequences/test_samples.py b/tests/pipelime/sequences/test_samples.py
new file mode 100644
index 0000000..0c1abd2
--- /dev/null
+++ b/tests/pipelime/sequences/test_samples.py
@@ -0,0 +1,18 @@
+import rich
+from pipelime.sequences.samples import GroupedSample, PlainSample, SamplesSequence
+
+
+class TestGroupedSamples(object):
+
+    def test_groupby_samples(self, plain_samples_generator):
+
+        dataset = plain_samples_generator('d0_', 10)
+
+        g = GroupedSample(samples=dataset[::2])
+
+        for sample in dataset:
+            for key in sample.keys():
+                print(key, sample[key])
+
+        for key in g.keys():
+            rich.print(g[key])
diff --git a/tests/pipelime/sequences/test_stages.py b/tests/pipelime/sequences/test_stages.py
new file mode 100644
index 0000000..83efc1f
--- /dev/null
+++ b/tests/pipelime/sequences/test_stages.py
@@ -0,0 +1,106 @@
+from pipelime.sequences.readers.filesystem import UnderfolderReader
+from pipelime.sequences.samples import PlainSample
+from schema import Schema
+from pipelime.sequences.stages import SampleStage, SampleStagesFactory, StageAugmentations, StageIdentity, StageRemap
+
+
+def _plug_test(stage: SampleStage):
+    """ Test what a generic SampleStage should do
+
+    :param stage: input SampleStage
+    :type stage: SampleStage
+    """
+
+    assert isinstance(stage, SampleStage)
+
+    restage = stage.build_from_dict(stage.to_dict())
+    assert isinstance(restage, SampleStage)
+    assert isinstance(restage.factory_schema(), Schema)
+
+    factored = SampleStagesFactory.create(stage.to_dict())
+    assert isinstance(factored, SampleStage)
+
+
+class TestStageIdentity(object):
+
+    def test_identity(self):
+
+        s = PlainSample(data={'name': 'sample', 'idx': 111})
+
+        stage = StageIdentity()
+        _plug_test(stage)
+
+        out = stage(s)
+
+        assert out == s
+
+
+class TestStageRemap(object):
+
+    def test_remap(self):
+
+        s = PlainSample(data={'name': 'sample', 'idx': 111})
+
+        stage = StageRemap(remap={'name': 'a', 'idx': 'b', 'not_present': 'x'})
+        _plug_test(stage)
+
+        out = stage(s)
+
+        assert 'a' in out
+        assert 'b' in out
+        assert 'name' not in out
+        assert 'idx' not in out
+        assert 'x' not in out
+        assert 'not_present' not in out
+
+    def test_remap_remove(self):
+
+        s = PlainSample(data={'name': 'sample', 'idx': 111})
+
+        stage = StageRemap(remap={'name': 'a'}, remove_missing=True)
+        _plug_test(stage)
+
+        out = stage(s)
+
+        assert 'a' in out
+        assert 'idx' not in out
+
+
+class TestStageAugmentations(object):
+
+    def test_augmentations(self, toy_dataset_small):
+
+        import albumentations as A
+
+        folder = toy_dataset_small['folder']
+        keypoints_format = toy_dataset_small['keypoints_format']
+        bboxes_format = toy_dataset_small['bboxes_format']
+
+        reader = UnderfolderReader(folder=folder)
+
+        transform = A.Compose([
+            A.HorizontalFlip(p=1),
+            A.ShiftScaleRotate(shift_limit=0.2, scale_limit=0.6, rotate_limit=180, p=1.0),
+            A.RandomBrightnessContrast(p=1),
+        ],
+            keypoint_params=A.KeypointParams(format=keypoints_format, remove_invisible=False, angle_in_degrees=False),
+            bbox_params=A.BboxParams(format=bboxes_format)
+        )
+
+        stage = StageAugmentations(
+            transform_cfg=A.to_dict(transform),
+            targets={
+                'image': 'image',
+                'mask': 'mask',
+                'inst': 'inst',
+                'keypoints': 'keypoints',
+                'bboxes': 'bboxes'
+            }
+        )
+        _plug_test(stage)
+
+        for sample in reader:
+            out = stage(sample)
+
+            for key in sample.keys():
+                assert key in out
diff --git a/tests/sample_data/augmentations/full_custom_augmentations.yml b/tests/sample_data/augmentations/full_custom_augmentations.yml
new file mode 100644
index 0000000..0ddfe14
--- /dev/null
+++ b/tests/sample_data/augmentations/full_custom_augmentations.yml
@@ -0,0 +1,11 @@
+__version__: 0.5.1
+transform:
+  __class_fullname__: albumentations.core.composition.Compose
+  additional_targets: {}
+  bbox_params: null
+  keypoint_params: null
+  p: 1.0
+  transforms:
+  - __class_fullname__: pipelime.augmentations.transforms.PadIfNeededV2
+    min_height: 700
+    min_width: 700
diff --git a/tests/sample_data/augmentations/full_custom_augmentations_bboxes_keypoints.yml b/tests/sample_data/augmentations/full_custom_augmentations_bboxes_keypoints.yml
new file mode 100644
index 0000000..8460bea
--- /dev/null
+++ b/tests/sample_data/augmentations/full_custom_augmentations_bboxes_keypoints.yml
@@ -0,0 +1,23 @@
+__version__: 0.5.1
+transform:
+  __class_fullname__: albumentations.core.composition.Compose
+  additional_targets: {}
+  bbox_params:
+    { 
+      check_each_transform: false,
+      format: albumentations,
+      label_fields: null,
+      min_area: 0.0,
+      min_visibility: 0.0,
+    }
+  keypoint_params:
+    angle_in_degrees: true
+    check_each_transform: true
+    format: xy
+    label_fields: null
+    remove_invisible: true
+  p: 1.0
+  transforms:
+  - __class_fullname__: pipelime.augmentations.transforms.PadIfNeededV2
+    min_height: 700
+    min_width: 700
diff --git a/tests/test_pipelime.py b/tests/test_pipelime.py
index 7b05065..5ffbb5a 100644
--- a/tests/test_pipelime.py
+++ b/tests/test_pipelime.py
@@ -26,12 +26,18 @@ def test_content(response):
     # assert 'GitHub' in BeautifulSoup(response.content).title.string
 
 
-def test_command_line_interface():
-    """Test the CLI."""
-    runner = CliRunner()
-    result = runner.invoke(cli.main)
-    assert result.exit_code == 0
-    assert 'pipelime.cli.main' in result.output
-    help_result = runner.invoke(cli.main, ['--help'])
-    assert help_result.exit_code == 0
-    assert '--help  Show this message and exit.' in help_result.output
+# def test_command_line_interface():
+#     """Test the CLI."""
+#     runner = CliRunner()
+#     result = runner.invoke(cli.main)
+#     assert result.exit_code == 0
+#     assert 'pipelime.cli.main' in result.output
+#     help_result = runner.invoke(cli.main, ['--help'])
+#     assert help_result.exit_code == 0
+#     assert '--help  Show this message and exit.' in help_result.output
+
+
+def test_toy_dataset_small(toy_dataset_small):
+
+    for x in ['folder', 'data_folder', 'size', 'image_size', 'zfill', 'expected_keys', 'keypoints_format']:
+        assert x in toy_dataset_small