mindspore/tests/ut/python/dataset/test_apply.py

# Copyright 2019 Huawei Technologies Co., Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
import numpy as np

import mindspore.dataset as ds
import mindspore.dataset.vision.c_transforms as vision
from mindspore import log as logger

DATA_DIR = "../data/dataset/testPK/data"


# Generate 1d int numpy array from 0 - 64
def generator_1d():
    for i in range(64):
        yield (np.array([i]),)


def test_apply_generator_case():
    # apply dataset operations
    data1 = ds.GeneratorDataset(generator_1d, ["data"])
    data2 = ds.GeneratorDataset(generator_1d, ["data"])

    def dataset_fn(ds_):
        ds_ = ds_.repeat(2)
        return ds_.batch(4)

    data1 = data1.apply(dataset_fn)
    data2 = data2.repeat(2)
    data2 = data2.batch(4)

    for item1, item2 in zip(data1.create_dict_iterator(num_epochs=1, output_numpy=True),
                            data2.create_dict_iterator(num_epochs=1, output_numpy=True)):
        np.testing.assert_array_equal(item1["data"], item2["data"])


def test_apply_imagefolder_case():
    # apply dataset map operations
    data1 = ds.ImageFolderDataset(DATA_DIR, num_shards=4, shard_id=3)
    data2 = ds.ImageFolderDataset(DATA_DIR, num_shards=4, shard_id=3)

    decode_op = vision.Decode()
    normalize_op = vision.Normalize([121.0, 115.0, 100.0], [70.0, 68.0, 71.0])

    def dataset_fn(ds_):
        ds_ = ds_.map(operations=decode_op)
        ds_ = ds_.map(operations=normalize_op)
        ds_ = ds_.repeat(2)
        return ds_

    data1 = data1.apply(dataset_fn)
    data2 = data2.map(operations=decode_op)
    data2 = data2.map(operations=normalize_op)
    data2 = data2.repeat(2)

    for item1, item2 in zip(data1.create_dict_iterator(num_epochs=1, output_numpy=True),
                            data2.create_dict_iterator(num_epochs=1, output_numpy=True)):
        np.testing.assert_array_equal(item1["image"], item2["image"])


def test_apply_flow_case_0(id_=0):
    # apply control flow operations
    data1 = ds.GeneratorDataset(generator_1d, ["data"])

    def dataset_fn(ds_):
        if id_ == 0:
            ds_ = ds_.batch(4)
        elif id_ == 1:
            ds_ = ds_.repeat(2)
        elif id_ == 2:
            ds_ = ds_.batch(4)
            ds_ = ds_.repeat(2)
        else:
            ds_ = ds_.shuffle(buffer_size=4)
        return ds_

    data1 = data1.apply(dataset_fn)
    num_iter = 0
    for _ in data1.create_dict_iterator(num_epochs=1):
        num_iter = num_iter + 1

    if id_ == 0:
        assert num_iter == 16
    elif id_ == 1:
        assert num_iter == 128
    elif id_ == 2:
        assert num_iter == 32
    else:
        assert num_iter == 64


def test_apply_flow_case_1(id_=1):
    # apply control flow operations
    data1 = ds.GeneratorDataset(generator_1d, ["data"])

    def dataset_fn(ds_):
        if id_ == 0:
            ds_ = ds_.batch(4)
        elif id_ == 1:
            ds_ = ds_.repeat(2)
        elif id_ == 2:
            ds_ = ds_.batch(4)
            ds_ = ds_.repeat(2)
        else:
            ds_ = ds_.shuffle(buffer_size=4)
        return ds_

    data1 = data1.apply(dataset_fn)
    num_iter = 0
    for _ in data1.create_dict_iterator(num_epochs=1):
        num_iter = num_iter + 1

    if id_ == 0:
        assert num_iter == 16
    elif id_ == 1:
        assert num_iter == 128
    elif id_ == 2:
        assert num_iter == 32
    else:
        assert num_iter == 64


def test_apply_flow_case_2(id_=2):
    # apply control flow operations
    data1 = ds.GeneratorDataset(generator_1d, ["data"])

    def dataset_fn(ds_):
        if id_ == 0:
            ds_ = ds_.batch(4)
        elif id_ == 1:
            ds_ = ds_.repeat(2)
        elif id_ == 2:
            ds_ = ds_.batch(4)
            ds_ = ds_.repeat(2)
        else:
            ds_ = ds_.shuffle(buffer_size=4)
        return ds_

    data1 = data1.apply(dataset_fn)
    num_iter = 0
    for _ in data1.create_dict_iterator(num_epochs=1):
        num_iter = num_iter + 1

    if id_ == 0:
        assert num_iter == 16
    elif id_ == 1:
        assert num_iter == 128
    elif id_ == 2:
        assert num_iter == 32
    else:
        assert num_iter == 64


def test_apply_flow_case_3(id_=3):
    # apply control flow operations
    data1 = ds.GeneratorDataset(generator_1d, ["data"])

    def dataset_fn(ds_):
        if id_ == 0:
            ds_ = ds_.batch(4)
        elif id_ == 1:
            ds_ = ds_.repeat(2)
        elif id_ == 2:
            ds_ = ds_.batch(4)
            ds_ = ds_.repeat(2)
        else:
            ds_ = ds_.shuffle(buffer_size=4)
        return ds_

    data1 = data1.apply(dataset_fn)
    num_iter = 0
    for _ in data1.create_dict_iterator(num_epochs=1):
        num_iter = num_iter + 1

    if id_ == 0:
        assert num_iter == 16
    elif id_ == 1:
        assert num_iter == 128
    elif id_ == 2:
        assert num_iter == 32
    else:
        assert num_iter == 64


def test_apply_exception_case():
    # apply exception operations
    data1 = ds.GeneratorDataset(generator_1d, ["data"])

    def dataset_fn(ds_):
        ds_ = ds_.repeat(2)
        return ds_.batch(4)

    def exception_fn():
        return np.array([[0], [1], [3], [4], [5]])

    try:
        data1 = data1.apply("123")
        for _ in data1.create_dict_iterator(num_epochs=1):
            pass
        assert False
    except TypeError:
        pass

    try:
        data1 = data1.apply(exception_fn)
        for _ in data1.create_dict_iterator(num_epochs=1):
            pass
        assert False
    except TypeError:
        pass

    try:
        data2 = data1.apply(dataset_fn)
        _ = data1.apply(dataset_fn)
        for _, _ in zip(data1.create_dict_iterator(num_epochs=1), data2.create_dict_iterator(num_epochs=1)):
            pass
        assert False
    except ValueError as e:
        logger.info("Got an exception in DE: {}".format(str(e)))


if __name__ == '__main__':
    logger.info("Running test_apply.py test_apply_generator_case() function")
    test_apply_generator_case()

    logger.info("Running test_apply.py test_apply_imagefolder_case() function")
    test_apply_imagefolder_case()

    logger.info("Running test_apply.py test_apply_flow_case(id) function")
    test_apply_flow_case_0()
    test_apply_flow_case_1()
    test_apply_flow_case_2()
    test_apply_flow_case_3()

    logger.info("Running test_apply.py test_apply_exception_case() function")
    test_apply_exception_case()