import unittest import pathlib import io import datetime import numpy as np import icartt # working directory, example files wd = pathlib.Path(__file__).parent fns_pass = (wd / "example_data").glob("*.ict") fns_fail = (wd / "example_data" / "will_fail").glob("*.ict") def compareFiles(fn, strIn, strOut, skiplines=0, nlines=-1): # pragma: no cover strOut.seek(0) strIn.seek(0) content_in = strIn.readlines() content_out = strOut.readlines() strIn.close() strOut.close() if nlines > 0: content_in = content_in[skiplines : (skiplines + nlines)] content_out = content_out[skiplines : (skiplines + nlines)] else: content_in = content_in[skiplines:] content_out = content_out[skiplines:] if not len(content_in) == len(content_out): return False for inline, outline in zip(content_in, content_out): inline = inline.strip().replace(" ", "") outline = outline.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(f"{str(fn)}: line {i:d} differs:") print(f" input: {inline}") print(f" output: {outline}") return False return True class Simple1001TestCase(unittest.TestCase): def setUp(self): self.fn = wd / "example_data" / "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): now = datetime.datetime.today() 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 = 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.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() # we have not added data yet, so data must be None self.assertIsNone(ict.data[:]) # and times must be NaT self.assertTrue(np.isnat(ict.times)) 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) # 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.6, 13.4, 14.1)): 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()