merged with master

Author: Maren Pielka

src/data_stack/dataset/reporting.py

@@ -19,7 +19,6 @@ class DatasetIteratorReport:
     sample_pos: int
     target_pos: int
     tag_pos: int
-    sample_shape: List[int]
     target_dist: Dict[Union[str, int], int]
     iteration_speed: float
     sub_reports: List["DatasetIteratorReport"]
@@ -38,9 +37,9 @@ def generate_report(iterator: InformedDatasetIterator, report_format: ReportForm
         target_dist = {k: v for k, v in sorted(Counter([row[meta.target_pos] for row in iterator]).items())}
         iteration_speed = DatasetIteratorReportGenerator.measure_iteration_speed(iterator)
         # generate report
+
         report = DatasetIteratorReport(meta.identifier, meta.dataset_name, meta.dataset_tag, len(iterator), meta.sample_pos,
-                                       meta.target_pos, meta.tag_pos, list(iterator[0][meta.sample_pos].shape), target_dist,
-                                       iteration_speed, sub_reports)
+                                       meta.target_pos, meta.tag_pos, target_dist, iteration_speed, sub_reports)
         # format report
         if report_format == DatasetIteratorReportGenerator.ReportFormat.JSON:
             return DatasetIteratorReportGenerator._to_json(report)

src/data_stack/dataset/splitter.py

@@ -158,17 +158,21 @@ def __init__(self,
 
     def split(self, dataset_iterator: DatasetIteratorIF) -> Tuple[List[DatasetIteratorIF], List[List[DatasetIteratorIF]]]:
         # create outer loop folds
-        targets = [sample[self.target_pos] for sample in dataset_iterator]
-        folds_indices = [fold[1] for fold in self.outer_splitter.split(X=np.zeros(len(targets)), y=targets)]
-        outer_folds = [DatasetIteratorView(dataset_iterator, fold_indices) for fold_indices in folds_indices]
+        targets = np.array([sample[self.target_pos] for sample in dataset_iterator])
+        outer_folds_indices = [fold[1] for fold in self.outer_splitter.split(X=np.zeros(len(targets)), y=targets)]
+        outer_fold_iterators = [DatasetIteratorView(dataset_iterator, fold_indices) for fold_indices in outer_folds_indices]
         # create inner loop folds
-        inner_folds_list = []  # contains [inner folds of outer_fold_1, inner folds of outer_fold_2 ...]
-        for iterator in outer_folds:
-            targets = [sample[self.target_pos] for sample in iterator]
-            folds_indices = [fold[1] for fold in self.inner_splitter.split(X=np.zeros(len(targets)), y=targets)]
-            inner_folds = [DatasetIteratorView(iterator, fold_indices) for fold_indices in folds_indices]
-            inner_folds_list.append(inner_folds)
-        return outer_folds, inner_folds_list
+        inner_folds_iterators_list = []  # contains [inner folds of outer_fold_1, inner folds of outer_fold_2 ...]
+        for outer_fold_id in range(len(outer_fold_iterators)):
+            # concat the indices of the splits which belong to the train splits
+            train_split_ids = [i for i in range(len(outer_folds_indices)) if i != outer_fold_id]
+            outer_train_fold_indices = np.array([indice for i in train_split_ids for indice in outer_folds_indices[i]])
+            inner_targets = targets[outer_train_fold_indices]
+            inner_folds_indices = [outer_train_fold_indices[inner_fold[1]]
+                                   for inner_fold in self.inner_splitter.split(X=np.zeros(len(inner_targets)), y=inner_targets)]
+            inner_folds = [DatasetIteratorView(dataset_iterator, fold_indices) for fold_indices in inner_folds_indices]
+            inner_folds_iterators_list.append(inner_folds)
+        return outer_fold_iterators, inner_folds_iterators_list
 
     def get_indices(self, dataset_iterator: DatasetIteratorIF) -> Tuple[List[List[int]], List[List[int]]]:
         outer_folds, inner_folds_list = self.split(dataset_iterator)

src/setup.py

@@ -5,7 +5,7 @@
 
 setup(
     name='datastack',
-    version='0.0.9',
+    version='0.0.10',
     author='Max Luebbering',
     description="DataStack, a stream based solution for machine learning dataset retrieval and storage",
     long_description=long_description,

unittests/dataset/test_reporting.py

@@ -1,13 +1,11 @@
 import pytest
-from typing import List
-from data_stack.mnist.factory import MNISTFactory
 from data_stack.io.storage_connectors import StorageConnector, StorageConnectorFactory
 from data_stack.dataset.reporting import DatasetIteratorReportGenerator
 import tempfile
 import shutil
-from data_stack.dataset.iterator import InformedDatasetIterator
-from data_stack.dataset.meta import MetaFactory
 from data_stack.dataset.factory import InformedDatasetFactory
+from data_stack.dataset.meta import DatasetMeta, MetaFactory
+from data_stack.dataset.iterator import DatasetIteratorIF, SequenceDatasetIterator, InformedDatasetIterator
 
 
 class TestReporting:
@@ -22,36 +20,38 @@ def tmp_folder_path(self) -> str:
     def storage_connector(self, tmp_folder_path: str) -> StorageConnector:
         return StorageConnectorFactory.get_file_storage_connector(tmp_folder_path)
 
-    @pytest.fixture(scope="session")
-    def mnist_factory(self, storage_connector) -> List[int]:
-        mnist_factory = MNISTFactory(storage_connector)
-        return mnist_factory
-
-    def test_plain_iterator_reporting(self, mnist_factory):
-        iterator, iterator_meta = mnist_factory.get_dataset_iterator(config={"split": "train"})
-        dataset_meta = MetaFactory.get_dataset_meta(identifier="id x", dataset_name="MNIST",
-                                                    dataset_tag="train", iterator_meta=iterator_meta)
-
-        informed_iterator = InformedDatasetIterator(iterator, dataset_meta)
-        report = DatasetIteratorReportGenerator.generate_report(informed_iterator)
+    # @pytest.fixture(scope="session")
+    # def mnist_factory(self, storage_connector) -> List[int]:
+    #     mnist_factory = MNISTFactory(storage_connector)
+    #     return mnist_factory
+
+    @pytest.fixture
+    def dataset_meta(self) -> DatasetMeta:
+        iterator_meta = MetaFactory.get_iterator_meta(sample_pos=0, target_pos=1, tag_pos=2)
+        return MetaFactory.get_dataset_meta(identifier="identifier_1",
+                                            dataset_name="TEST DATASET",
+                                            dataset_tag="train",
+                                            iterator_meta=iterator_meta)
+
+    @pytest.fixture
+    def dataset_iterator(self) -> DatasetIteratorIF:
+        targets = [j for i in range(10) for j in range(9)] + [10]*1000
+        samples = [0]*len(targets)
+        return SequenceDatasetIterator(dataset_sequences=[samples, targets])
+
+    @pytest.fixture
+    def informed_dataset_iterator(self, dataset_iterator, dataset_meta) -> DatasetIteratorIF:
+        return InformedDatasetFactory.get_dataset_iterator(dataset_iterator, dataset_meta)
+
+    def test_plain_iterator_reporting(self, informed_dataset_iterator):
+        report = DatasetIteratorReportGenerator.generate_report(informed_dataset_iterator)
         print(report)
-        assert report.length == 60000 and not report.sub_reports
-
-    def test_combined_iterator_reporting(self, mnist_factory):
-        
-        iterator_train, iterator_train_meta = mnist_factory.get_dataset_iterator(config={"split": "train"})
-        iterator_test, iterator_test_meta = mnist_factory.get_dataset_iterator(config={"split": "test"})
-        meta_train = MetaFactory.get_dataset_meta(identifier="id x", dataset_name="MNIST",
-                                                  dataset_tag="train", iterator_meta=iterator_train_meta)
-        meta_test = MetaFactory.get_dataset_meta(identifier="id x", dataset_name="MNIST",
-                                                 dataset_tag="train", iterator_meta=iterator_test_meta)
-
-        informed_iterator_train = InformedDatasetFactory.get_dataset_iterator(iterator_train, meta_train)
-        informed_iterator_test = InformedDatasetFactory.get_dataset_iterator(iterator_test, meta_test)
-
-        meta_combined = MetaFactory.get_dataset_meta_from_existing(informed_iterator_train.dataset_meta, dataset_tag="full")
+        assert report.length == 1090 and not report.sub_reports
 
-        iterator = InformedDatasetFactory.get_combined_dataset_iterator([informed_iterator_train, informed_iterator_test], meta_combined)
+    def test_combined_iterator_reporting(self, informed_dataset_iterator):
+        meta_combined = MetaFactory.get_dataset_meta_from_existing(informed_dataset_iterator.dataset_meta, dataset_tag="full")
+        iterator = InformedDatasetFactory.get_combined_dataset_iterator(
+            [informed_dataset_iterator, informed_dataset_iterator], meta_combined)
         report = DatasetIteratorReportGenerator.generate_report(iterator)
-        assert report.length == 70000 and report.sub_reports[0].length == 60000 and report.sub_reports[1].length == 10000
+        assert report.length == 2180 and report.sub_reports[0].length == 1090 and report.sub_reports[1].length == 1090
         assert not report.sub_reports[0].sub_reports and not report.sub_reports[1].sub_reports

unittests/dataset/test_splitter.py

@@ -1,8 +1,9 @@
 import pytest
 import numpy as np
+import collections
 from data_stack.dataset.iterator import DatasetIteratorIF, SequenceDatasetIterator
 from typing import List, Dict
-from data_stack.dataset.splitter import RandomSplitterImpl, StratifiedSplitterImpl, Splitter
+from data_stack.dataset.splitter import RandomSplitterImpl, StratifiedSplitterImpl, Splitter, NestedCVSplitterImpl
 from data_stack.dataset.meta import DatasetMeta, MetaFactory
 
 
@@ -27,6 +28,12 @@ def dataset_meta(self) -> DatasetMeta:
     def dataset_iterator(self) -> DatasetIteratorIF:
         return SequenceDatasetIterator(dataset_sequences=[list(range(10)), list(range(10))])
 
+    @pytest.fixture
+    def big_dataset_iterator(self) -> DatasetIteratorIF:
+        targets = [j for i in range(10) for j in range(9)] + [10] * 1000
+        samples = [0] * len(targets)
+        return SequenceDatasetIterator(dataset_sequences=[samples, targets])
+
     @pytest.fixture
     def dataset_iterator_stratifiable(self) -> DatasetIteratorIF:
         return SequenceDatasetIterator(dataset_sequences=[list(range(20)), list(np.ones(8, dtype=int))+
@@ -56,6 +63,55 @@ def test_stratification(self, split_config: Dict[str, int], dataset_iterator_str
         assert(sum([sample[1] for sample in iterator_splits[1]]) == 2)
         assert(sum([sample[1] for sample in iterator_splits[2]]) == 2)
 
+    @pytest.mark.parametrize(
+        "num_outer_loop_folds, num_inner_loop_folds, inner_stratification, outer_stratification, shuffle",
+        [(5, 2, True, True, False), (5, 2, True, True, True), (5, 2, False, False, True),
+         (5, 2, False, False, False)],
+    )
+    def test_nested_cv_splitter(self, num_outer_loop_folds: int, num_inner_loop_folds: int,
+                                inner_stratification: bool,
+                                outer_stratification: bool, shuffle: bool, big_dataset_iterator: DatasetIteratorIF):
+        splitter_impl = NestedCVSplitterImpl(num_outer_loop_folds=num_outer_loop_folds,
+                                             num_inner_loop_folds=num_inner_loop_folds,
+                                             inner_stratification=inner_stratification,
+                                             outer_stratification=outer_stratification,
+                                             shuffle=shuffle)
+        splitter = Splitter(splitter_impl)
+        outer_folds, inner_folds = splitter.split(big_dataset_iterator)
+        # make sure that outer folds have no intersection
+        for i in range(len(outer_folds)):
+            for j in range(len(outer_folds)):
+                if i != j:
+                    # makes sure there is no intersection
+                    assert len(set(outer_folds[i].indices).intersection(set(outer_folds[j].indices))) == 0
+        # make sure that inner folds have no intersection
+        for i in range(len(inner_folds)):
+            for j in range(len(inner_folds[i])):
+                for k in range(len(inner_folds[i])):
+                    if j != k:
+                        # makes sure there is no intersection
+                        assert len(set(inner_folds[i][j].indices).intersection(set(inner_folds[i][k].indices))) == 0
+        # test stratification
+        if outer_stratification:
+            class_counts = dict(collections.Counter([t for _, t in big_dataset_iterator]))
+            class_counts_per_fold = {target_class: int(count / num_outer_loop_folds) for target_class, count in
+                                     class_counts.items()}
+            for fold in outer_folds:
+                fold_class_counts = dict(collections.Counter([t for _, t in fold]))
+                for key in list(class_counts_per_fold.keys()) + list(fold_class_counts.keys()):
+                    assert class_counts_per_fold[key] == fold_class_counts[key]
+
+        if inner_stratification:
+            for i in range(len(inner_folds)):
+                class_counts = dict(collections.Counter([t for _, t in outer_folds[i]]))
+                class_counts_per_fold = {
+                    target_class: int(count * (num_outer_loop_folds - 1) / num_inner_loop_folds) for
+                    target_class, count in class_counts.items()}
+                for fold in inner_folds[i]:
+                    fold_class_counts = dict(collections.Counter([t for _, t in fold]))
+                    for key in list(class_counts_per_fold.keys()) + list(fold_class_counts.keys()):
+                        assert class_counts_per_fold[key] == fold_class_counts[key]
+
     def test_seeding(self):
         ratios = [0.4, 0.6]
         dataset_length = 100

unittests/io/test_retriever.py

@@ -61,19 +61,19 @@ def file_retriever(self, storage_connector: StorageConnector):
     def http_retriever_impl(self, storage_connector: StorageConnector):
         return HTTPRetrieverImpl(storage_connector)
 
-    def test_http_retriever_retrieve(self, http_retriever: Retriever, http_retrieval_job: ResourceDefinition):
-        http_retriever.retrieve([http_retrieval_job])
-        storage_connector = http_retriever.retriever_impl.storage_connector
-        resource = storage_connector.get_resource(http_retrieval_job.identifier)
-        assert TestBaseRetriever.get_md5(resource) == http_retrieval_job.md5_sum
-
-    def test_http_retriever_impl_download_file(self, http_retriever_impl: HTTPRetrieverImpl, http_retrieval_job: ResourceDefinition, tmp_folder_path: str):
-        file_path = http_retriever_impl._download_file(url=http_retrieval_job.source,
-                                                       dest_folder=tmp_folder_path,
-                                                       md5=http_retrieval_job.md5_sum)
-        with open(file_path, "rb") as fd:
-            md5_sum = TestBaseRetriever.get_md5(fd)
-        return md5_sum == http_retrieval_job.md5_sum
+    # def test_http_retriever_retrieve(self, http_retriever: Retriever, http_retrieval_job: ResourceDefinition):
+    #     http_retriever.retrieve([http_retrieval_job])
+    #     storage_connector = http_retriever.retriever_impl.storage_connector
+    #     resource = storage_connector.get_resource(http_retrieval_job.identifier)
+    #     assert TestBaseRetriever.get_md5(resource) == http_retrieval_job.md5_sum
+
+    # def test_http_retriever_impl_download_file(self, http_retriever_impl: HTTPRetrieverImpl, http_retrieval_job: ResourceDefinition, tmp_folder_path: str):
+    #     file_path = http_retriever_impl._download_file(url=http_retrieval_job.source,
+    #                                                    dest_folder=tmp_folder_path,
+    #                                                    md5=http_retrieval_job.md5_sum)
+    #     with open(file_path, "rb") as fd:
+    #         md5_sum = TestBaseRetriever.get_md5(fd)
+    #     return md5_sum == http_retrieval_job.md5_sum
 
     def test_file_retriever_retrieve(self, file_retriever: Retriever, file_retrieval_job: ResourceDefinition):
         file_retriever.retrieve([file_retrieval_job])