Skip to content
Snippets Groups Projects

introduction of pathlib.Path

Merged introduction of pathlib.Path
(removed):master into master
1 unresolved thread
Merged Florian Obersteiner requested to merge (removed):master into master
1 unresolved thread
Viewing commit d7538c91
Prev
Show latest version
1 file
+ 24
− 24
Compare changes
  • Side-by-side
  • Inline
tests/test_dataset.py
+ 265
− 170
import unittest
import os
import pathlib
import io
import datetime
import numpy as np
import icartt
def compareFiles(fn, strIn, strOut, skiplines=0, nlines=-1): # pragma: no cover
# 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()
content_in = strIn.readlines()
content_out = strOut.readlines()
strIn.close()
strOut.close()
if nlines > 0:
input = input[skiplines:(skiplines+nlines)]
output = output[skiplines:(skiplines+nlines)]
content_in = content_in[skiplines : (skiplines + nlines)]
content_out = content_out[skiplines : (skiplines + nlines)]
else:
input = input[skiplines:]
output = output[skiplines:]
if not len(input) == len(output):
content_in = content_in[skiplines:]
content_out = content_out[skiplines:]
if not len(content_in) == len(content_out):
return False
for i in range(len(input)):
inline = input[i].strip().replace(" ", "")
outline = output[i].strip().replace(" ", "")
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
# 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
# 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))
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 = 'tests/examples/NOx_RHBrown_20040830_R0.ict'
self.nHeader = 41
self.fn = wd / "examples/NOx_RHBrown_20040830_R0.ict"
self.nHeader = 41
def tearDown(self):
a = 1
pass
def testOpen(self):
ict = icartt.Dataset(self.fn, loadData=False)
self.assertEqual( type(ict), icartt.Dataset )
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 )
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 )
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 )
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' ] )
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" ])
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 )
ict = icartt.Dataset(self.fn, loadData=True)
self.assertEqual(type(ict), icartt.Dataset)
def testWriteHeader(self):
ict = icartt.Dataset(self.fn, loadData=False)
ict = icartt.Dataset(self.fn, loadData=False)
strIn = open(self.fn)
strIn = open(self.fn)
strOut = io.StringIO()
ict.writeHeader(f=strOut)
self.assertTrue( compareFiles(self.fn, strIn, strOut, nlines=self.nHeader) )
self.assertTrue(compareFiles(self.fn, strIn, strOut, nlines=self.nHeader))
def testWriteData(self):
ict = icartt.Dataset(self.fn, loadData=True)
ict = icartt.Dataset(self.fn, loadData=True)
strIn = open(self.fn)
strIn = open(self.fn)
strOut = io.StringIO()
ict.write(f=strOut)
self.assertTrue( compareFiles(self.fn, strIn, strOut, skiplines=self.nHeader) )
self.assertTrue(compareFiles(self.fn, strIn, strOut, skiplines=self.nHeader))
def testWrite(self):
ict = icartt.Dataset(self.fn, loadData=True)
ict = icartt.Dataset(self.fn, loadData=True)
strIn = open(self.fn)
strIn = open(self.fn)
strOut = io.StringIO()
ict.write(f=strOut)
self.assertTrue( compareFiles(self.fn, strIn, 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 = 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!")
@@ -202,59 +294,62 @@ class Create1001TestCase(unittest.TestCase):
ict.endDefineMode()
ict.data.add( Time_Start = 12.3, Time_Stop = 12.5, Payload = 23789423.2e5 )
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 )
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 )
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" ]
# 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="Opening test file {:s}".format(fn)):
ict = icartt.Dataset(fn, loadData=False)
self.assertEqual( type(ict), icartt.Dataset )
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="Reading data from test file {:s}".format(fn)):
ict = icartt.Dataset(fn, loadData=True)
self.assertEqual( type(ict), icartt.Dataset )
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="Writing header for test file {:s}".format(fn)):
ict = icartt.Dataset(fn, loadData=False)
strIn = open(fn)
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) )
self.assertTrue(compareFiles(fn, strIn, strOut, nlines=ict.nHeader))
def testWrite(self):
for fn in fns:
with self.subTest(msg="Writing data for test file {:s}".format(fn)):
ict = icartt.Dataset(fn, loadData=True)
strIn = open(fn)
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()
self.assertTrue(compareFiles(fn, strIn, strOut))
if __name__ == "__main__": # pragma: no cover
unittest.main()
\ No newline at end of file