test_dataset.py

import unittest
import os
import io
import icartt

def compare_files(fn, str_in, str_out, skiplines=0, nlines=-1): # pragma: no cover
    str_out.seek(0)
    str_in.seek(0)
    input  = str_in.readlines()
    output = str_out.readlines()
    str_in.close()
    str_out.close()
    
    if nlines > 0:
        input  = input[skiplines:(skiplines+nlines)]
        output = output[skiplines:(skiplines+nlines)]
    else:
        input  = input[skiplines:]
        output = output[skiplines:]
    
    if not len(input) == len(output):
        return False

    for i in range(len(input)):
        inline  = input[i].strip().replace(" ", "")
        outline = output[i].strip().replace(" ", "")
        if not inline == outline:
            valid_data_line = False
            # maybe this is a data line in which we only have different number formatting?
            # compare as floats
            try:
                insteps  = [ float(x) for x in inline.split(",")  ]
                outsteps = [ float(x) for x in outline.split(",") ]
                if ( len(insteps) == len(outsteps) ):
                    valid_data_line = True
                    for i in range(len(insteps)):
                        valid_data_line = valid_data_line and insteps[i] == outsteps[i]
            except:
                pass
            
            valid_var_line = False
            try:
                insteps  = [ x.strip() for x in inline.split(",")  ]
                outsteps = [ x.strip() for x in outline.split(",") ]
                if ( len(insteps) == 2 and len(outsteps) == 3 ):
                    valid_var_line = ( insteps[0] == outsteps[0] and insteps[1] == outsteps[1] and insteps[1] == outsteps[2] )
            except:
                pass
            
            if not valid_data_line and not valid_var_line:
                print("{:s}: line {:d} differs:".format(fn, i))
                print("  input: {:s}".format(inline))
                print(" output: {:s}".format(outline))
                
                return False
    
    return True

class Simple1001TestCase(unittest.TestCase):
    def setUp(self):
        self.fn         = 'tests/examples/NOx_RHBrown_20040830_R0.ict'
        self.nheader    = 41

    def tearDown(self):
        a = 1
    
    def test_open(self):
        ict      = icartt.Dataset(self.fn, loadData=False)
        self.assertEqual( type(ict), icartt.Dataset )

    def test_format(self):
        ict      = icartt.Dataset(self.fn, loadData=False)
        self.assertEqual( ict.format, 1001 )

    def test_n(self):
        ict      = icartt.Dataset(self.fn, loadData=False)
        self.assertEqual( ict.nheader,     self.nheader )
        self.assertEqual( len(ict.DVARS),  9 )
        self.assertEqual( len(ict.NCOM),  18 )
        self.assertEqual( len(ict.SCOM),   0 )
    
    def test_ivar(self):
        ict      = icartt.Dataset(self.fn, loadData=False)
        self.assertEqual( ict.IVAR.shortname,       "Start_UTC" )
        self.assertEqual( ict.IVAR.units,           "seconds" )
        self.assertEqual( ict.IVAR.standardname,    "number_of_seconds_from_0000_UTC" )
        self.assertEqual( ict.IVAR.longname,        None )
        self.assertEqual( ict.IVAR.scale,           1.0 )
        self.assertEqual( ict.IVAR.miss,            -99999.0 )

    def test_dvar(self):
        ict      = icartt.Dataset(self.fn, loadData=False)
        
        self.assertEqual( [ DVAR.shortname for DVAR in ict.DVARS.values() ], 
                          [ "Stop_UTC", "Mid_UTC", "DLat", "DLon", "Elev", "NO_ppbv", "NO_1sig", "NO2_ppbv", "NO2_1sig" ] )

        self.assertEqual( [ DVAR.units for DVAR in ict.DVARS.values() ], 
                          [ "seconds", "seconds", "deg_N", "deg_E", "meters", "ppbv", "ppbv", "ppbv", "ppbv" ] )

        self.assertEqual( [ DVAR.standardname for DVAR in ict.DVARS.values() ], 
                          [ None, None, None, None, None, None, None, None, None])

        self.assertEqual( [ DVAR.longname for DVAR in ict.DVARS.values() ], 
                          [ None, None, None, None, None, None, None, None, None])

        self.assertEqual( [ DVAR.scale for DVAR in ict.DVARS.values() ], 
                          [ 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0 ] )

        self.assertEqual( [ DVAR.miss for DVAR in ict.DVARS.values() ], 
                          [ -9999.0, -9999.0, -9999.0, -9999.0, -9999.0, -9999.0, -9999.0, -9999.0, -9999.0 ] )

    def test_NCOM(self):
        ict      = icartt.Dataset(self.fn, loadData=False)
        
        self.assertEqual( ict.NCOM.keywords['PI_CONTACT_INFO'].data,     [ "325 Broadway, Boulder, CO 80305; 303-497-3226; email:eric.j.williams@noaa.gov" ])
        self.assertEqual( ict.NCOM.keywords['PLATFORM'].data,            [ "NOAA research vessel Ronald H. Brown" ])
        self.assertEqual( ict.NCOM.keywords['LOCATION'].data,            [ "Latitude, longitude and elevation data are included in the data records" ])
        self.assertEqual( ict.NCOM.keywords['ASSOCIATED_DATA'].data,     [ "N/A" ])
        self.assertEqual( ict.NCOM.keywords['INSTRUMENT_INFO'].data,     [ "NO: chemiluminescence; NO2: narrow-band photolysis/chemiluminescence" ])
        self.assertEqual( ict.NCOM.keywords['DATA_INFO'].data,           [ "All data with the exception of the location data are in ppbv. All oneminute averages contain at least 35 seconds of data, otherwise missing." ])
        self.assertEqual( ict.NCOM.keywords['UNCERTAINTY'].data,         [ "included in the data records as variables with a _1sig suffix" ])
        self.assertEqual( ict.NCOM.keywords['ULOD_FLAG'].data,           [ "-7777" ])
        self.assertEqual( ict.NCOM.keywords['ULOD_VALUE'].data,          [ "N/A" ])
        self.assertEqual( ict.NCOM.keywords['LLOD_FLAG'].data,           [ "-8888" ])
        self.assertEqual( ict.NCOM.keywords['LLOD_VALUE'].data,          [ "N/A, N/A, N/A, N/A, N/A, 0.005, N/A, 0.025, N/A" ])
        self.assertEqual( ict.NCOM.keywords['DM_CONTACT_INFO'].data,     [ "N/A" ])
        self.assertEqual( ict.NCOM.keywords['PROJECT_INFO'].data,        [ "ICARTT study; 1 July-15 August 2004; Gulf of Maine and North Atlantic Ocean" ])
        self.assertEqual( ict.NCOM.keywords['STIPULATIONS_ON_USE'].data, [ "Use of these data requires PRIOR OK from the PI" ])
        self.assertEqual( ict.NCOM.keywords['OTHER_COMMENTS'].data,      [ "N/A" ])

    def test_read_data(self):
        ict      = icartt.Dataset(self.fn, loadData=True)
        self.assertEqual( type(ict), icartt.Dataset )

    def test_write_header(self):
        ict      = icartt.Dataset(self.fn, loadData=False)

        str_in  = open(self.fn)
        str_out = io.StringIO()
        
        ict.write_header(str_out)
        
        self.assertTrue( compare_files(self.fn, str_in, str_out, nlines=self.nheader) )

    def test_write_data(self):
        ict      = icartt.Dataset(self.fn, loadData=True)

        str_in  = open(self.fn)
        str_out = io.StringIO()
        
        ict.write(str_out)
        
        self.assertTrue( compare_files(self.fn, str_in, str_out, skiplines=self.nheader) )

    def test_write(self):
        ict      = icartt.Dataset(self.fn, loadData=True)

        str_in  = open(self.fn)
        str_out = io.StringIO()
        
        ict.write(str_out)
        
        self.assertTrue( compare_files(self.fn, str_in, str_out) )


fns        = [ os.path.join("tests", "examples", fn) for fn in os.listdir(os.path.join("tests", "examples")) if fn.endswith(".ict")]
#fns = [ "tests/examples/AROTAL-RAY_DC8_20040715_R1.ict" ]

class BulkIOTestCase(unittest.TestCase):
    def test_open(self):
        for fn in fns:
            with self.subTest(msg="Opening test file {:s}".format(fn)):
                ict      = icartt.Dataset(fn, loadData=False)
                self.assertEqual( type(ict), icartt.Dataset )

    def test_read_data(self):
        for fn in fns:
            with self.subTest(msg="Reading data from test file {:s}".format(fn)):
                ict      = icartt.Dataset(fn, loadData=True)
                self.assertEqual( type(ict), icartt.Dataset )

    def test_write_header(self):
        for fn in fns:
            with self.subTest(msg="Writing header for test file {:s}".format(fn)):
                ict      = icartt.Dataset(fn, loadData=False)
        
                str_in  = open(fn)
                str_out = io.StringIO()
                
                ict.write_header(str_out)
                
                self.assertTrue( compare_files(fn, str_in, str_out, nlines=ict.nheader) )

    def test_write(self):
        for fn in fns:
            with self.subTest(msg="Writing data for test file {:s}".format(fn)):
                ict      = icartt.Dataset(fn, loadData=True)
        
                str_in  = open(fn)
                str_out = io.StringIO()
                
                ict.write(str_out)
                
                self.assertTrue( compare_files(fn, str_in, str_out) )

if __name__ == '__main__': # pragma: no cover
    unittest.main()