<commit_before>#!/usr/bin/env python

'''
@file prototype/hdf/test/test_hdf_array_iterator.py
@author <NAME>
@test prototype.hdf.test.test_hdf_array_iterator.py test suite for hdf_array_iterator.py
'''

from prototype.hdf.hdf_array_iterator import acquire_data


@attr('INT', group='dm')
class HDFArrayIteratorTest(IonIntegrationTestCase):

    def setUp(self):

        import h5py, numpy

        #--------------------------------------------------------------------
        # Create an hdf file for testing
        #--------------------------------------------------------------------

        a = numpy.arange(50)

        file = h5py.File('data.hdf5', 'w')

        dset = file.create_dataset("salinity", data=a)

        file.close()

        #--------------------------------------------------------------------
        # Create another file for testing
        #--------------------------------------------------------------------

        a = numpy.arange(30)

        b = numpy.arange(100)

        file = h5py.File('measurements.hdf5', 'w')

        dset = file.create_dataset("temperature", data=a)
        dset2 = file.create_dataset("conductivity", data = b)

        file.close()

    def tearDown(self):
        """
        Cleanup. Delete Subscription, Stream, Process Definition
        """
        self._stop_container()

    def test_acquire_data_from_multiple_files(self):
        """
        Test whether data can be acquired from multiple hdf5 files
        """

        generator = acquire_data(hdf_files = ['data.hdf5','measurements.hdf5'], var_name = None, buffer_size = 50, slice_= (slice(1,100)), concatenate_block_size = 12  )

        out = generator.next()

        # assert that the dataset 'salinity' in the first hdf5 file has been opened

        self.assertTrue('salinity' in out[4])

        out = generator.next()

        # assert that the second hdf5 file has been opened and one of its datasets has been opened
        self.assertTrue(('temperature' in out[4]) or ('conductivity' in out[4]) )


    def test_acquire_data_from_multiple_datasets(self):
        """
        Test whether data can be acquired from multiple datasets from an hdf5 file
        """

        generator = acquire_data(hdf_files = ['measurements.hdf5'], var_name = None, buffer_size = 50, slice_= (slice(1,100)), concatenate_block_size = 12  )

        out = generator.next()

        # assert that the dataset 'salinity' in the first hdf5 file has been opened

        self.assertTrue(('temperature' in out[4]) and ('conductivity' in out[4]))

    def test_acquire_data_with_var_name(self):
        """
        Test whether supplying a var_name confines the selection to be of only that var_name
        """

        generator = acquire_data(hdf_files = ['measurements.hdf5'], var_name = 'conductivity', buffer_size = 3, slice_= (slice(1,100)), concatenate_block_size = 12  )

        out = generator.next()

        # assert that the dataset 'salinity' in the first hdf5 file has been opened

        self.assertTrue('conductivity' in out[4])

        self.assertTrue(not ('temperature' in out[4]))

    def test_buffer_size(self):
        """
        Test that the chunk of data that is read from the hdf file is of the size buffer_size
        """

        buffer_size = 3
<commit_msg>Added test for hdf_array_iterator code.<commit_after>#!/usr/bin/env python

'''
@file prototype/hdf/test/test_hdf_array_iterator.py
@author <NAME>
@test prototype.hdf.test.test_hdf_array_iterator.py test suite for hdf_array_iterator.py
'''

import os
from prototype.hdf.hdf_array_iterator import acquire_data

from nose.plugins.attrib import attr
from pyon.util.int_test import IonIntegrationTestCase
import unittest

@attr('INT', group='dm')
class HDFArrayIteratorTest(IonIntegrationTestCase):

    def setUp(self):

        import h5py, numpy

        #--------------------------------------------------------------------
        # Create an hdf file for testing
        #--------------------------------------------------------------------

        a = numpy.arange(50)

        file = h5py.File('data.hdf5', 'w')

        dset = file.create_dataset("salinity", data=a)

        file.close()

        #--------------------------------------------------------------------
        # Create another file for testing
        #--------------------------------------------------------------------

        a = numpy.arange(30)

        b = numpy.arange(100)

        file = h5py.File('measurements.hdf5', 'w')

        dset = file.create_dataset("temperature", data=a)
        dset2 = file.create_dataset("conductivity", data = b)

        file.close()

    def tearDown(self):
        """
        Cleanup. Delete Subscription, Stream, Process Definition
        """

        os.remove('measurements.hdf5')
        os.remove('data.hdf5')

    def test_acquire_data_from_multiple_files(self):
        """
        Test whether data can be acquired from multiple hdf5 files
        """

        generator = acquire_data(hdf_files = ['data.hdf5','measurements.hdf5'], var_name = None, buffer_size = 50, slice_= (slice(1,100)), concatenate_block_size = 12  )

        out = generator.next()

        # assert that the dataset 'salinity' in the first hdf5 file has been opened

        self.assertTrue('salinity' in out[4])

        out = generator.next()

        # assert that the second hdf5 file has been opened and one of its datasets has been opened
        self.assertTrue(('temperature' in out[4]) or ('conductivity' in out[4]) )


    def test_acquire_data_from_multiple_datasets(self):
        """
        Test whether data can be acquired from multiple datasets from an hdf5 file
        """

        generator = acquire_data(hdf_files = ['measurements.hdf5'], var_name = None, buffer_size = 50, slice_= (slice(1,100)), concatenate_block_size = 12  )

        out = generator.next() # the first time next() is called loads up the temperature data.

        # now that the temperature data has been exhausted since we chose a very large buffer_size,
        # calling generator.next() will load up the conductivity data
        out = generator.next()

        # assert that the dataset 'salinity' in the first hdf5 file has been opened

        print ("arrays_out: %s" % out[4])

        self.assertTrue(('temperature' in out[4]) and ('conductivity' in out[4]))

    def test_acquire_data_with_var_name(self):
        """
        Test whether supplying a var_name confines the selection to be of only that var_name
        """

        generator = acquire_data(hdf_files = ['measurements.hdf5'], var_name = 'conductivity', buffer_size = 3, slice_= (slice(1,100)), concatenate_block_size = 12  )

        out = generator.next()

        # assert that the dataset 'salinity' in the first hdf5 file has been opened

        self.assertTrue('conductivity' in out[4])

        self.assertTrue(not ('temperature' in out[4]))

    def test_buffer_size(self):
        """
        Test that the chunk of data that is read from the hdf file is of the size buffer_size
        """

        buffer_size = 3

{'fraction_non_alphanumeric': 0.17115086631919987, 'fraction_numerical': 0.01563600507113678, 'mean_word_length': 2.093681917211329, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5098742', 'n_tokens_mistral': 2058, 'n_tokens_neox': 1928, 'n_words': 567}