test_2110.py

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

import icartt

try:
    from _utils import compareFiles  # we're executing from the directory of this script
except ImportError:
    from ._utils import compareFiles  # we're executing from another directory

# working directory, example files
wd = pathlib.Path(__file__).parent


class Simple2110TestCase(unittest.TestCase):
    def setUp(self):
        self.fn = wd / "example_data" / "expect_warn" / "PAVE-AR_DC8_20050203_R0.ict"
        self.nHeader = 55

    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.FFI2110)

    def testN(self):
        ict = icartt.Dataset(self.fn, loadData=False)
        self.assertEqual(ict.nHeader, self.nHeader)
        self.assertEqual(len(ict.auxiliaryVariables), 11)
        self.assertEqual(len(ict.dependentVariables), 7)
        self.assertEqual(len(ict.normalComments), 18)
        self.assertEqual(len(ict.specialComments), 1)

    def testIvar(self):
        ict = icartt.Dataset(self.fn, loadData=False)
        self.assertEqual(ict.independentVariable.shortname, "UTC")
        self.assertEqual(ict.independentVariable.units, "seconds")
        self.assertEqual(ict.independentVariable.standardname, "Time_Start")
        self.assertEqual(
            ict.independentVariable.longname, "number of seconds from 00:00 UTC"
        )
        self.assertEqual(ict.independentVariable.scale, 1.0)
        self.assertEqual(ict.independentVariable.miss, -99999.0)

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

        self.assertEqual(
            [AUXVAR.shortname for AUXVAR in ict.auxiliaryVariables.values()],
            [
                "NumAlts",
                "Year",
                "Month",
                "Day",
                "AvgTime",
                "Lat",
                "Lon",
                "PAlt",
                "GPSAlt",
                "SAT",
                "SZA",
            ],
        )

        self.assertEqual(
            [AUXVAR.units for AUXVAR in ict.auxiliaryVariables.values()],
            [
                "#",
                "yyyy",
                "mm",
                "dd",
                "minutes",
                "degrees",
                "degrees",
                "meters",
                "meters",
                "K",
                "degrees",
            ],
        )

        self.assertEqual(
            [AUXVAR.standardname for AUXVAR in ict.auxiliaryVariables.values()],
            [
                "Number_of_altitudes",
                "Year_UTC",
                "Month_UTC",
                "Day_UTC",
                "Averaging_time",
                "Latitude",
                "Longitude",
                "pressure_altitude",
                "GPS_altitude",
                "Static_air_temperature",
                "Sun_Zenith_Angle",
            ],
        )

        self.assertEqual(
            [AUXVAR.longname for AUXVAR in ict.auxiliaryVariables.values()],
            [
                "Number_of_altitudes_reported",
                "Year_UTC",
                "Month_UTC",
                "Day_UTC",
                "Averaging_time_of_presented_data xxx.x_minutes",
                None,
                None,
                None,
                None,
                None,
                None,
            ],
        )

        self.assertEqual(
            [AUXVAR.scale for AUXVAR in ict.auxiliaryVariables.values()],
            ["1", "1", "1", "1", "1", "1", "1", "1", "1", "1", "1"],
        )

        self.assertEqual(
            [AUXVAR.miss for AUXVAR in ict.auxiliaryVariables.values()],
            [
                "-9999",
                "-9999",
                "-9999",
                "-9999",
                "-9999",
                "-9999",
                "-9999",
                "-9999",
                "-9999",
                "-9999",
                "-9999",
            ],
        )

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

        self.assertEqual(
            [DVAR.shortname for DVAR in ict.dependentVariables.values()],
            [
                "TempK[]",
                "Log10_NumDensity[]",
                "TempK_Err[]",
                "AerKlet[]",
                "Log10_O3NumDensity[]",
                "O3_MR[]",
                "Log10_O3NumDensity_Err[]",
            ],
        )

        self.assertEqual(
            [DVAR.units for DVAR in ict.dependentVariables.values()],
            [
                "K",
                "part/cc",
                "K",
                "Klet",
                "part/cc",
                "ppb",
                "part/cc",
            ],
        )

        self.assertEqual(
            [DVAR.standardname for DVAR in ict.dependentVariables.values()],
            [
                "Temperature",
                "Log10_NumDensity",
                "Temperature_Error",
                "Aerosol",
                "Log10_O3NumDensity",
                "Ozone_mixing_ratio",
                "Log10_O3NumDensity_Error",
            ],
        )

        self.assertEqual(
            [DVAR.longname for DVAR in ict.dependentVariables.values()],
            [
                "Temperature_array",
                "Log10_NumDensity_array",
                "Temperature_error_array",
                "Aerosol_array",
                "Log10_Ozone_NumDensity_array",
                "Ozone_mixing_ratio_array",
                "Log10_NumDensity_error_array",
            ],
        )

        self.assertEqual(
            [DVAR.scale for DVAR in ict.dependentVariables.values()],
            [
                "0.1",
                "0.0001",
                "0.1",
                "0.01",
                "0.0001",
                "0.1",
                "0.0001",
            ],
        )

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

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

        self.assertEqual(
            ict.normalComments.keywords["PI_CONTACT_INFO"].data,
            ["Enter PI Address here"],
        )
        self.assertEqual(
            ict.normalComments.keywords["PLATFORM"].data,
            ["NASA DC8"],
        )
        self.assertEqual(
            ict.normalComments.keywords["LOCATION"].data,
            ["Lat, Lon, and Alt included in the data records"],
        )
        self.assertEqual(ict.normalComments.keywords["ASSOCIATED_DATA"].data, ["N/A"])
        self.assertEqual(
            ict.normalComments.keywords["INSTRUMENT_INFO"].data,
            ["N/A"],
        )
        self.assertEqual(
            ict.normalComments.keywords["DATA_INFO"].data,
            ["N/A"],
        )
        self.assertEqual(
            ict.normalComments.keywords["UNCERTAINTY"].data,
            ["Contact PI"],
        )
        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"],
        )
        self.assertEqual(
            ict.normalComments.keywords["DM_CONTACT_INFO"].data,
            ["Enter Data Manager Info here"],
        )
        self.assertEqual(
            ict.normalComments.keywords["PROJECT_INFO"].data,
            ["PAVE MISSION: Jan-Feb 2005"],
        )
        self.assertEqual(
            ict.normalComments.keywords["STIPULATIONS_ON_USE"].data,
            ["Use of these data should be done in consultation with the PI"],
        )
        self.assertEqual(ict.normalComments.keywords["OTHER_COMMENTS"].data, ["N/A"])
        # TODO test revision information

    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 Create2110TestCase(unittest.TestCase):
    def testCreateDs(self):
        now = datetime.datetime.today()

        ict = icartt.Dataset(format=icartt.Formats.FFI2110)

        ict.PIName = "Knote, Christoph"
        ict.PIAffiliation = "Faculty of Medicine, University Augsburg, Germany"
        ict.dataSourceDescription = "Example data"
        ict.missionName = "MBEES"
        ict.dateOfCollection = now.timetuple()[:3]
        ict.dateOfRevision = now.timetuple()[:3]

        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.independentBoundedVariable = icartt.Variable(
            "Altitude",
            "altitude_above_ground_in_meters",
            "Altitude",
            "Altitude",
            vartype=icartt.VariableType.IndependentBoundedVariable,
            scale=1.0,
            miss=-9999999,
        )

        # ICARTT convention: first aux variable contains number of dependent elements
        ict.auxiliaryVariables["nAltitudes"] = icartt.Variable(
            "nAltitudes",
            "number_of_dependent_variable_items",
            "variable",
            "nAltitudes",
            scale=1.0,
            miss=-9999999,
        )

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

        ict.auxiliaryVariables["Longitude"] = icartt.Variable(
            "Longitude",
            "longitude_in_degrees",
            "Longitude",
            "Longitude",
            scale=1.0,
            miss=-9999999,
        )

        ict.auxiliaryVariables["Latitude"] = icartt.Variable(
            "Latitude",
            "latitude_in_degrees",
            "Latitude",
            "Latitude",
            scale=1.0,
            miss=-9999999,
        )

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

        ict.dependentVariables["Payload2"] = icartt.Variable(
            "Payload2", "some_units", "Payload2", "Payload2", 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.")

        # we can just use len of the list to check number of comments
        self.assertEqual(len(ict.specialComments), 3)

        # let's define some normal comments... 21 lines
        ncom = {
            "PI_CONTACT_INFO": "PI1 pi-email@mail.com\nPI2 more-email@what.com",
            "PLATFORM": "a platform",
            "LOCATION": "somewhere",
            "ASSOCIATED_DATA": "met sensor data",
            "INSTRUMENT_INFO": "super cool instrument",
            "DATA_INFO": f"icartt Python package version: {icartt.__version__}",
            "UNCERTAINTY": "not much",
            "ULOD_FLAG": "-7777",
            "ULOD_VALUE": "N/A",
            "LLOD_FLAG": "-8888",
            "LLOD_VALUE": "N/A",
            "DM_CONTACT_INFO": "datamanager@mail.edu",
            "PROJECT_INFO": "the campaign",
            "STIPULATIONS_ON_USE": "no",
            "OTHER_COMMENTS": "a lot more info\non multiple lines",
            "REVISION": (
                "R1\n"
                "R1: revised time synchronization.\n"
                "R0: initial, preliminary version."
            ),
        }
        for k, v in ncom.items():
            ict.normalComments.keywords[k].append(v)

        # we can check if nlines method of normalComments class works
        self.assertEqual(ict.normalComments.nlines, 21)

        ict.normalComments.freeform.append("free comment line 1")
        ict.normalComments.freeform.append("free comment line 2")
        self.assertEqual(ict.normalComments.nlines, 23)

        ict.endDefineMode()

        # and times must be NaT
        self.assertTrue(np.isnat(ict.times))

        # note, the second variable ('4') is the number of dependent lines to follow
        #                       ivar, ndepvar, auxvar1, auxvar2, auxvar3
        auxData = np.array([(12.3, 4, 12.5, 48.21, 10.3)])
        #                   ibvar,  dvar1,  dvar2
        depData = np.array(
            [(0, 123, 8.4e4), (100, 122, 9.1e4), (250, 115, 9.3e4), (500, 106, 9.8e4)]
        )

        ict.data.add(auxData, depData)

        # ... and so forth
        auxData = np.array([(12.4, 2, 12.8, 48.41, 12.1)])

        #                   ibvar,  dvar1,  dvar2
        depData = np.array([(0, 153, 7.3e4), (270, 172, 8.9e4)])

        ict.data.add(auxData, depData)

        #        import pdb; pdb.set_trace()

        # elements of the time array must be equal to our input
        t0 = np.datetime64(datetime.datetime(*now.timetuple()[:3]), "ns")
        for have, want in zip(ict.times, (12.3, 12.4)):
            self.assertEqual(int(have - t0), int(want * 10**9))

        strOut = io.StringIO()

        ict.write(f=strOut)

        return True


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