Skip to content
Snippets Groups Projects
test_gtsdf.py 9.67 KiB
'''
Created on 12/09/2013

@author: mmpe
'''
from __future__ import division
from __future__ import unicode_literals
from __future__ import print_function
from __future__ import absolute_import
from builtins import super
from builtins import range
from future import standard_library
standard_library.install_aliases()

import h5py
import numpy as np
from wetb import gtsdf

import unittest
import os

tmp_path = os.path.dirname(__file__) + "/tmp/"
tfp = os.path.dirname(os.path.abspath(__file__)) + "/test_files/"

class Test_gsdf(unittest.TestCase):
    def setUp(self):
        unittest.TestCase.setUp(self)
        if not os.path.isdir(tmp_path):
            os.makedirs(tmp_path)

    @classmethod
    def tearDownClass(cls):
        super(Test_gsdf, cls).tearDownClass()
        #shutil.rmtree(tmp_path)

    def test_minimum_requirements (self):
        fn = tmp_path + "minimum.hdf5"
        f = h5py.File(fn, "w")
        #no type
        self.assertRaises(ValueError, gtsdf.load, fn)
        f.attrs["type"] = "General time series data format"

        #no no_blocks
        self.assertRaises(ValueError, gtsdf.load, fn)
        f.attrs["no_blocks"] = 0

        #no block0000
        self.assertRaises(ValueError, gtsdf.load, fn)
        b = f.create_group("block0000")

        #no data
        self.assertRaises(ValueError, gtsdf.load, fn)
        b.create_dataset("data", data=np.empty((0, 0)))
        f.close()
        gtsdf.load(fn)


    def test_save_no_hdf5_ext(self):
        fn = tmp_path + "no_hdf5_ext"
        gtsdf.save(fn, np.arange(12).reshape(4, 3))
        _, _, info = gtsdf.load(fn + ".hdf5")
        self.assertEqual(info['name'], 'no_hdf5_ext')

    def test_load_filename(self):
        fn = tmp_path + "filename.hdf5"
        gtsdf.save(fn, np.arange(12).reshape(4, 3))
        _, _, info = gtsdf.load(fn)
        self.assertEqual(info['name'], 'filename')


    def test_load_fileobject(self):
        fn = tmp_path + "fileobject.hdf5"
        gtsdf.save(fn, np.arange(12).reshape(4, 3))
        _, _, info = gtsdf.load(fn)
        self.assertEqual(info['name'], 'fileobject')

    def test_save_wrong_no_attr_info(self):
        fn = tmp_path + "wrong_no_attr_info.hdf5"
        self.assertRaises(AssertionError, gtsdf.save, fn, np.arange(12).reshape(4, 3), attribute_names=['Att1'])
        self.assertRaises(AssertionError, gtsdf.save, fn, np.arange(12).reshape(4, 3), attribute_units=['s'])
        self.assertRaises(AssertionError, gtsdf.save, fn, np.arange(12).reshape(4, 3), attribute_descriptions=['desc'])

    def test_info(self):
        fn = tmp_path + "info.hdf5"
        gtsdf.save(fn, np.arange(12).reshape(6, 2),
                   name='datasetname',
                   description='datasetdescription',
                   attribute_names=['att1', 'att2'],
                   attribute_units=['s', 'm/s'],
                   attribute_descriptions=['d1', 'd2'])
        _, _, info = gtsdf.load(fn)
        self.assertEqual(info['name'], "datasetname")
        self.assertEqual(info['type'], "General time series data format")
        self.assertEqual(info['description'], "datasetdescription")
        self.assertEqual(list(info['attribute_names']), ['att1', 'att2'])
        self.assertEqual(list(info['attribute_units']), ['s', 'm/s'])
        self.assertEqual(list(info['attribute_descriptions']), ['d1', 'd2'])

    def test_no_time(self):
        fn = tmp_path + 'time.hdf5'
        gtsdf.save(fn, np.arange(12).reshape(6, 2))
        time, _, _ = gtsdf.load(fn)
        np.testing.assert_array_equal(time, np.arange(6))

    def test_int_time(self):
        fn = tmp_path + 'time.hdf5'
        gtsdf.save(fn, np.arange(12).reshape(6, 2), time=range(4, 10))
        time, _, _ = gtsdf.load(fn)
        np.testing.assert_array_equal(time, range(4, 10))


    def test_time_offset(self):
        fn = tmp_path + 'time.hdf5'
        gtsdf.save(fn, np.arange(12).reshape(6, 2), time=range(6), time_start=4)
        time, _, _ = gtsdf.load(fn)
        np.testing.assert_array_equal(time, range(4, 10))


    def test_time_gain_offset(self):
        fn = tmp_path + 'time.hdf5'
        gtsdf.save(fn, np.arange(12).reshape(6, 2), time=range(6), time_step=1 / 4, time_start=4)
        time, _, _ = gtsdf.load(fn)
        np.testing.assert_array_equal(time, np.arange(4, 5.5, .25))

    def test_float_time(self):
        fn = tmp_path + 'time.hdf5'
        gtsdf.save(fn, np.arange(12).reshape(6, 2), time=np.arange(4, 5.5, .25))
        time, _, _ = gtsdf.load(fn)
        np.testing.assert_array_equal(time, np.arange(4, 5.5, .25))

    def test_data(self):
        fn = tmp_path + 'data.hdf5'
        d = np.arange(12).reshape(6, 2)
        gtsdf.save(fn, d)
        f = h5py.File(fn)
        self.assertEqual(f['block0000']['data'].dtype, np.uint16)
        f.close()
        _, data, _ = gtsdf.load(fn)
        np.testing.assert_array_almost_equal(data, np.arange(12).reshape(6, 2), 4)

    def test_data_float(self):
        fn = tmp_path + 'time.hdf5'
        d = np.arange(12).reshape(6, 2)
        gtsdf.save(fn, d, dtype=np.float32)
        f = h5py.File(fn)
        self.assertEqual(f['block0000']['data'].dtype, np.float32)
        f.close()
        _, data, _ = gtsdf.load(fn)
        np.testing.assert_array_equal(data, np.arange(12).reshape(6, 2))


    def test_all(self):
        fn = tmp_path + "all.hdf5"
        gtsdf.save(fn, np.arange(12).reshape(6, 2),
                   name='datasetname',
                   time=range(6), time_step=1 / 4, time_start=4,
                   description='datasetdescription',
                   attribute_names=['att1', 'att2'],
                   attribute_units=['s', 'm/s'],
                   attribute_descriptions=['d1', 'd2'])
        time, data, info = gtsdf.load(fn)
        self.assertEqual(info['name'], "datasetname")
        self.assertEqual(info['type'], "General time series data format")
        self.assertEqual(info['description'], "datasetdescription")
        self.assertEqual(list(info['attribute_names']), ['att1', 'att2'])
        self.assertEqual(list(info['attribute_units']), ['s', 'm/s'])
        self.assertEqual(list(info['attribute_descriptions']), ['d1', 'd2'])
        np.testing.assert_array_equal(time, np.arange(4, 5.5, .25))
        np.testing.assert_array_almost_equal(data, np.arange(12).reshape(6, 2), 4)

    def test_append(self):
        fn = tmp_path + 'append.hdf5'
        d = np.arange(48, dtype=np.float32).reshape(24, 2)
        d[2, 0] = np.nan
        gtsdf.save(fn, d)
        _, data, _ = gtsdf.load(fn)
        np.testing.assert_array_almost_equal(data, d, 3)
        gtsdf.append_block(fn, d)
        _, data, _ = gtsdf.load(fn)
        self.assertEqual(data.shape, (48, 2))
        np.testing.assert_array_almost_equal(data, np.append(d, d, 0), 3)
        f = h5py.File(fn)
        self.assertIn('gains', f['block0001'])
        f.close()

    def test_append_small_block(self):
        fn = tmp_path + 'append_small_block.hdf5'
        d = np.arange(12, dtype=np.float32).reshape(2, 6)
        gtsdf.save(fn, d)
        gtsdf.append_block(fn, d + 12)
        f = h5py.File(fn)
        self.assertNotIn('gains', f['block0001'])
        f.close()


    def test_nan_float(self):
        fn = tmp_path + 'nan.hdf5'
        d = np.arange(12, dtype=np.float32).reshape(6, 2)
        d[2, 0] = np.nan
        gtsdf.save(fn, d)
        _, data, _ = gtsdf.load(fn)
        np.testing.assert_array_almost_equal(data, d, 4)



    def test_outlier(self):
        fn = tmp_path + 'outlier.hdf5'
        d = np.arange(12, dtype=np.float32).reshape(6, 2)
        d[2, 0] = 10 ** 4
        d[3, 1] = 10 ** 4
        self.assertRaises(Warning, gtsdf.save, fn, d)
        _, data, _ = gtsdf.load(fn)

    def test_inf(self):
        fn = tmp_path + 'outlier.hdf5'
        d = np.arange(12, dtype=np.float32).reshape(6, 2)
        d[2, 0] = np.inf
        d[3, 1] = 10 ** 3
        self.assertRaises(ValueError, gtsdf.save, fn, d)

    def test_loadpandas(self):
        fn = tmp_path + "all.hdf5"
        gtsdf.save(fn, np.arange(12).reshape(6, 2),
                   name='datasetname',
                   time=range(6), time_step=1 / 4, time_start=4,
                   description='datasetdescription',
                   attribute_names=['att1', 'att2'],
                   attribute_units=['s', 'm/s'],
                   attribute_descriptions=['d1', 'd2'])
        df = gtsdf.load_pandas(fn)

    def test_loadtesthdf5(self):
        time, data, info = gtsdf.load(tfp + 'test.hdf5')
        self.assertEqual(time[1], 0.05)
        self.assertEqual(data[1, 1], 11.986652374267578)
        self.assertEqual(info['attribute_names'][1], "WSP gl. coo.,Vy")
    
    def test_loadhdf5File(self):
        f = h5py.File(tfp + 'test.hdf5')
        time, data, info = gtsdf.load(f)
        
        self.assertEqual(time[1], 0.05)
        self.assertEqual(data[1, 1], 11.986652374267578)
        self.assertEqual(info['attribute_names'][1], "WSP gl. coo.,Vy")
        
        
    def test_gtsdf_dataset(self):
        ds = gtsdf.Dataset(tfp+'test.hdf5')
        self.assertEqual(ds('Time')[1], 0.05)
        self.assertEqual(ds.Time[1], 0.05)
        self.assertRaisesRegex(AttributeError, "'Dataset' object has no attribute 'Time1'", lambda : ds.Time1)
        self.assertEqual(ds(2)[1], 11.986652374267578)
        n = ds.info['attribute_names'][2]
        self.assertEqual(n, "WSP gl. coo.,Vy")
        self.assertEqual(ds(n)[1], 11.986652374267578)
        
        
        






if __name__ == "__main__":
    #import sys;sys.argv = ['', 'Test.testName']
    unittest.main()