test_dataset.py

import unittest
import pathlib
import io
import datetime
import numpy as np
import icartt

# working directory, example files
wd = pathlib.Path(__file__).parent
fns = (wd/'examples').glob('*.ict')

def compareFiles(fn, strIn, strOut, skiplines=0, nlines=-1): # pragma: no cover
    strOut.seek(0)
    strIn.seek(0)
    input  = strIn.readlines()
    output = strOut.readlines()
    strIn.close()
    strOut.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         = wd/'examples/NOx_RHBrown_20040830_R0.ict'
        self.nHeader    = 41

    def tearDown(self):
        pass

    def testOpen(self):
        ict      = icartt.Dataset(self.fn, loadData=False)
        self.assertEqual( type(ict), icartt.Dataset )

    def testFormat(self):
        ict      = icartt.Dataset(self.fn, loadData=False)
        self.assertEqual( ict.format, icartt.Formats.FFI1001 )

    def testN(self):
        ict      = icartt.Dataset(self.fn, loadData=False)
        self.assertEqual( ict.nHeader,     self.nHeader )
        self.assertEqual( len(ict.dependentVariables),  9 )
        self.assertEqual( len(ict.normalComments),  18 )
        self.assertEqual( len(ict.specialComments),   0 )

    def testIvar(self):
        ict      = icartt.Dataset(self.fn, loadData=False)
        self.assertEqual( ict.independentVariable.shortname,       "Start_UTC" )
        self.assertEqual( ict.independentVariable.units,           "seconds" )
        self.assertEqual( ict.independentVariable.standardname,    "number_of_seconds_from_0000_UTC" )
        self.assertEqual( ict.independentVariable.longname,        None )
        self.assertEqual( ict.independentVariable.scale,           1.0 )
        self.assertEqual( ict.independentVariable.miss,            -99999.0 )

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

        self.assertEqual( [ DVAR.shortname for DVAR in ict.dependentVariables.values() ],
                          [ "Stop_UTC", "Mid_UTC", "DLat", "DLon", "Elev", "NO_ppbv", "NO_1sig", "NO2_ppbv", "NO2_1sig" ] )

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

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

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

        self.assertEqual( [ DVAR.scale for DVAR in ict.dependentVariables.values() ],
                          [ '1', '1', '1', '1', '1', '1', '1', '1', '1' ] )

        self.assertEqual( [ DVAR.miss for DVAR in ict.dependentVariables.values() ],
                          [ '-9999', '-9999', '-9999', '-9999', '-9999', '-9999', '-9999', '-9999', '-9999' ] )

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

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

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

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

        strIn  = open(self.fn)
        strOut = io.StringIO()

        ict.writeHeader(f=strOut)

        self.assertTrue( compareFiles(self.fn, strIn, strOut, nlines=self.nHeader) )

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

        strIn  = open(self.fn)
        strOut = io.StringIO()

        ict.write(f=strOut)

        self.assertTrue( compareFiles(self.fn, strIn, strOut, skiplines=self.nHeader) )

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

        strIn  = open(self.fn)
        strOut = io.StringIO()

        ict.write(f=strOut)

        self.assertTrue( compareFiles(self.fn, strIn, strOut) )

class Create1001TestCase(unittest.TestCase):
    def testCreateDs(self):
        ict                                    = icartt.Dataset(format=icartt.Formats.FFI1001)

        ict.PIName                             = 'Knote, Christoph'
        ict.PIAffiliation                      = 'Faculty of Medicine, University Augsburg, Germany'
        ict.dataSourceDescription              = 'Example data'
        ict.missionName                        = 'MBEES'
        ict.dateOfCollection                   = datetime.datetime.today()
        ict.dateOfRevision                     = datetime.datetime.today()

        ict.dataIntervalCode                   = [ 0 ]

        ict.independentVariable                = icartt.Variable(  'Time_Start',
                                                                    'seconds_from_0_hours_on_valid_date',
                                                                    'Time_Start',
                                                                    'Time_Start',
                                                                    vartype=icartt.VariableType.IndependentVariable,
                                                                    scale=1.0, miss=-9999999)

        ict.dependentVariables['Time_Stop']    = icartt.Variable(   'Time_Stop',
                                                                    'seconds_from_0_hours_on_valid_date',
                                                                    'Time_Stop',
                                                                    'Time_Stop',
                                                                    scale=1.0, miss=-9999999)

        ict.dependentVariables['Payload']      = icartt.Variable(   'Payload',
                                                                    'some_units',
                                                                    'Payload',
                                                                    'Payload',
                                                                    scale=1.0, miss=-9999999)

        ict.specialComments.append("Some comments on this dataset:")
        ict.specialComments.append("They are just examples!")
        ict.specialComments.append("Adapt as needed.")

        ict.endDefineMode()

        ict.data.add( Time_Start = 12.3, Time_Stop = 12.5, Payload = 23789423.2e5 )

        mydict = { 'Time_Start': 12.6, 'Time_Stop': 13.1, 'Payload': 324235644.1e5 }
        ict.data.add( **mydict )

        data = np.array( [ (13.4, 14.0, 2348925e5), (14.1, 14.9, 23425634e5) ] )
        ict.data.addBulk( data )

        strOut = io.StringIO()

        ict.write(f=strOut)

        return True

# 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 testOpen(self):
        for fn in fns:
            with self.subTest(msg=f"Opening test file {str(fn)}"):
                ict      = icartt.Dataset(fn, loadData=False)
                self.assertEqual( type(ict), icartt.Dataset )

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

    def testWriteHeader(self):
        for fn in fns:
            with self.subTest(msg=f"Writing header for test file {str(fn)}"):
                ict      = icartt.Dataset(fn, loadData=False)

                strIn  = open(fn)
                strOut = io.StringIO()

                ict.writeHeader(f=strOut)

                self.assertTrue( compareFiles(fn, strIn, strOut, nlines=ict.nHeader) )

    def testWrite(self):
        for fn in fns:
            with self.subTest(msg=f"Writing data for test file {str(fn)}"):
                ict      = icartt.Dataset(fn, loadData=True)

                strIn  = open(fn)
                strOut = io.StringIO()

                ict.write(f=strOut)

                self.assertTrue( compareFiles(fn, strIn, strOut) )

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