Skip to content
test_dataset.py 8.47 KiB
Newer Older
Christoph Knote's avatar
Aye
Christoph Knote committed
import unittest
import os
import io
import sys
import icartt

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

Christoph Knote's avatar
Aye
Christoph Knote committed
    for i in range(len(input)):
        inline  = input[i].strip().replace(" ", "")
        outline = output[i].strip().replace(" ", "")
        if not inline == outline:
Christoph Knote's avatar
Christoph Knote committed
            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("Line {:d} differs:".format(i))
                print("  input: {:s}".format(inline))
                print(" output: {:s}".format(outline))
                
                return False
Christoph Knote's avatar
Aye
Christoph Knote committed
    
    return True

Christoph Knote's avatar
Christoph Knote committed
class Simple1001TestCase(unittest.TestCase):
    def setUp(self):
        self.fn         = 'tests/examples/NOx_RHBrown_20040830_R0.ict'
        self.nheader    = 41
Christoph Knote's avatar
Aye
Christoph Knote committed

Christoph Knote's avatar
Christoph Knote committed
    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 )
Christoph Knote's avatar
Aye
Christoph Knote committed

Christoph Knote's avatar
Christoph Knote committed
    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.name,     "Start_UTC" )
        self.assertEqual( ict.IVAR.units,    "seconds" )
        self.assertEqual( ict.IVAR.longname, "number_of_seconds_from_0000_UTC" )
        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)
Christoph Knote's avatar
Aye
Christoph Knote committed
        
Christoph Knote's avatar
Christoph Knote committed
        self.assertEqual( [ DVAR.name 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.longname for DVAR in ict.DVARS.values() ], 
                          [ "seconds", "seconds", "deg_N", "deg_E", "meters", "ppbv", "ppbv", "ppbv", "ppbv" ] )

        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'],     "325 Broadway, Boulder, CO 80305; 303-497-3226; email:eric.j.williams@noaa.gov")
        self.assertEqual( ict.NCOM.keywords['PLATFORM'],            "NOAA research vessel Ronald H. Brown")
        self.assertEqual( ict.NCOM.keywords['LOCATION'],            "Latitude, longitude and elevation data are included in the data records")
        self.assertEqual( ict.NCOM.keywords['ASSOCIATED_DATA'],     "N/A")
        self.assertEqual( ict.NCOM.keywords['INSTRUMENT_INFO'],     "NO: chemiluminescence; NO2: narrow-band photolysis/chemiluminescence")
        self.assertEqual( ict.NCOM.keywords['DATA_INFO'],           "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'],         "included in the data records as variables with a _1sig suffix")
        self.assertEqual( ict.NCOM.keywords['ULOD_FLAG'],           "-7777")
        self.assertEqual( ict.NCOM.keywords['ULOD_VALUE'],          "N/A")
        self.assertEqual( ict.NCOM.keywords['LLOD_FLAG'],           "-8888")
        self.assertEqual( ict.NCOM.keywords['LLOD_VALUE'],          "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'],     "N/A")
        self.assertEqual( ict.NCOM.keywords['PROJECT_INFO'],        "ICARTT study; 1 July-15 August 2004; Gulf of Maine and North Atlantic Ocean")
        self.assertEqual( ict.NCOM.keywords['STIPULATIONS_ON_USE'], "Use of these data requires PRIOR OK from the PI")
        self.assertEqual( ict.NCOM.keywords['OTHER_COMMENTS'],      "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)
Christoph Knote's avatar
Aye
Christoph Knote committed
        str_out = io.StringIO()
        
Christoph Knote's avatar
Christoph Knote committed
        ict.write_header(str_out)
Christoph Knote's avatar
Aye
Christoph Knote committed
        
Christoph Knote's avatar
Christoph Knote committed
        self.assertTrue( compare_files(str_in, str_out, nlines=self.nheader) )
Christoph Knote's avatar
Aye
Christoph Knote committed

Christoph Knote's avatar
Christoph Knote committed
    def test_write_data(self):
        ict      = icartt.Dataset(self.fn, loadData=True)

        str_in  = open(self.fn)
Christoph Knote's avatar
Aye
Christoph Knote committed
        str_out = io.StringIO()
        
        ict.write(str_out)
        
Christoph Knote's avatar
Christoph Knote committed
        self.assertTrue( compare_files(str_in, str_out, skiplines=self.nheader) )
Christoph Knote's avatar
Aye
Christoph Knote committed

Christoph Knote's avatar
Christoph Knote committed
    def test_write(self):
        ict      = icartt.Dataset(self.fn, loadData=True)

        str_in  = open(self.fn)
Christoph Knote's avatar
Aye
Christoph Knote committed
        str_out = io.StringIO()
        
        ict.write(str_out)
        
Christoph Knote's avatar
Christoph Knote committed
        self.assertTrue( compare_files(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(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(str_in, str_out) )

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