t_holography_dataset_class.py

import logging
import unittest
import tempfile
import h5py
import numpy
import os
from lofar.calibration.common.datacontainers import HolographyDataset, HolographySpecification
from lofar.calibration.common.datacontainers import HolographyObservation


logger = logging.getLogger('t_holography_dataset_class')
# READ doc/Holography_Data_Set.md!  It contains the location from which the
# test data must be downloaded.
path_to_test_data = '/var/tmp/holography'
path_to_test_dataset = path_to_test_data + '/CS001HBA0.hdf5'

class TestHolographyDatasetClass(unittest.TestCase):
    def test_create_hds(self):
        '''
        Reads a Measurement Set from a file and converts it to an HDS.
        ''' 
        # Read the MS into memory.
        holist = HolographyObservation.list_observations_in_path(path_to_test_data)
        hbsflist = HolographySpecification.list_bsf_files_in_path(path_to_test_data)
        for hbsf in hbsflist:
            hbsf.read_file()
        ho_per_ms = [(hbsf, ho) for hbsf, ho in zip(hbsflist, holist)]
  
        # Now create the Holography Data Set in memory from the MS data.
        holography_dataset = HolographyDataset()
        holography_dataset.load_from_beam_specification_and_ms(
            'CS013HBA0', ho_per_ms)

        # Make sure the data in memory is OK.
        self.assertEqual(holography_dataset.source_name, '3C 147')

    def test_store_and_read_from_hdf(self):
        test_dict = dict(CS001=dict(BEAM0=1, BEAM1=2, BEAM3=dict(FR1='str2'), ATTRIBUTES=dict(a=1)), CS003='str')

        with tempfile.NamedTemporaryFile(suffix='.hdf5') as tfile:
            tfile.close()

            h5file = h5py.File(tfile.name)
            HolographyDataset._store_grouped_data(h5file, '/test', test_dict)
            h5file.close()

            h5file = h5py.File(tfile.name)
            read_dict = HolographyDataset._read_grouped_data(h5file, '/test')
            self.assertDictEqual(test_dict, read_dict)


if __name__ == '__main__':
    logging.basicConfig(format='%(name)s : %(message)s')
    unittest.main()