Class MyTestCase

tests/cdslib/data/test_chunked_memmap.py:29–222

class MyTestCase(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        if os.path.exists('tmp_test_chunked_memmap'):
            raise RuntimeError('make sure tmp_test_chunked_memmap is empty to run the test')

    def test_write(self):

        # create random array
        samples_shape = [20, 10, 4]
        working_dir = 'tmp_test_chunked_memmap'
        samples = np.random.randn(*samples_shape).astype('float32')

        # determine chunk_idxs
        total_chunks = 3
        chunk_idxs = np.random.randint(total_chunks, size=samples_shape[0])

        # create chunked_memmap
        chunked_memmap = ChunkedMemmap(working_dir, remove_exist=True)
        chunked_memmap.add_all_samples(samples, chunk_idxs, dtype='float32')

        # check if the same
        arr = np.zeros(samples_shape)
        for i in range(samples_shape[0]):
            arr[i] = chunked_memmap[i]

        assert np.allclose(samples, arr)


    def test_multithread_write(self):

        # create random array
        samples_shape = [2000, 10, 4]
        working_dir = 'tmp_test_chunked_memmap'
        samples = np.random.randn(*samples_shape).astype('float32')
        num_workers = 5

        # determine chunk_idxs
        total_chunks = 3
        chunk_idxs = np.random.randint(total_chunks, size=samples_shape[0])

        # create chunked_memmap
        chunked_memmap = ChunkedMemmap(working_dir, remove_exist=True)
        chunked_memmap.add_all_samples(samples, chunk_idxs, dtype='float32', num_workers=num_workers)

        # check if the same
        arr = np.zeros(samples_shape)
        for i in range(samples_shape[0]):
            arr[i] = chunked_memmap[i]

        assert np.allclose(samples, arr)


    def test_load_existing(self):

        # create random array
        samples_shape = [1000, 10, 4]