import os import numpy as np import pytest import xarray as xr import mvcm.read_data as rd @pytest.fixture def fixturepath(): return os.path.join(os.path.dirname(__file__), "fixtures") @pytest.fixture def l1b_file(fixturepath): return os.path.join( fixturepath, "VNP02MOD.A2022173.1312.001.2022174011547.uwssec.nc" ) @pytest.fixture def geo_file(fixturepath): return os.path.join( fixturepath, "VNP03MOD.A2022173.1312.001.2022174012746.uwssec.nc" ) # @pytest.fixture # def ancillary_dir(fixturepath): # return os.path.join(fixturepath, '') @pytest.fixture def sst_file(): return "oisst.20220622" @pytest.fixture def ndvi_file(): return "NDVI.FM.c004.v2.0.WS.00-04.177.hdf" @pytest.fixture def geos_file_1(): return "GEOS.fpit.asm.inst3_2d_asm_Nx.GEOS5124.20220622_1200.V01.nc4" @pytest.fixture def geos_file_2(): return "GEOS.fpit.asm.inst3_2d_asm_Nx.GEOS5124.20220622_1500.V01.nc4" @pytest.fixture def geos_land(): return "GEOS.fpit.asm.tavg1_2d_lnd_Nx.GEOS5124.20220622_1330.V01.nc4" @pytest.fixture def geos_ocean(): return "GEOS.fpit.asm.tavg1_2d_ocn_Nx.GEOS5124.20220622_1330.V01.nc4" @pytest.fixture def geos_constants(): return "GEOS.fp.asm.const_2d_asm_Nx.00000000_0000.V01.nc4" @pytest.fixture def ref_file(fixturepath): return os.path.join(fixturepath, "ref_ancillary.nc") # this is only temporary. At some point I'll write a better test def test_l1b(fixturepath, l1b_file, geo_file): viirs = rd.ReadData( satellite="snpp", sensor="viirs", file_name_l1b=l1b_file, file_name_geo=geo_file ) geo = viirs.read_viirs_geo() l1b = viirs.read_viirs_l1b(geo.solar_zenith.values) geo_vars = [ "latitude", "longitude", "height", "range", "solar_zenith", "solar_azimuth", "sensor_zenith", "sensor_azimuth", "land_water_mask", "quality_flag", "relative_azimuth", "sunglint_angle", "scattering_angle", ] assert list(geo).sort() == geo_vars.sort() assert "M11" in l1b.data_vars def test_sst(fixturepath, sst_file, ref_file): viirs = rd.ReadData(satellite="snpp", sensor="viirs") geo = viirs.read_viirs_geo() ancillary = rd.ReadAncillary( latitude=geo.latitude.values, longitude=geo.longitude.values, resolution=1, ancillary_dir=fixturepath, sst_file=sst_file, ) sst = ancillary.get_sst() check_differences(ref_file, sst, "sst") def test_ndvi(fixturepath, ndvi_file, ref_file): viirs = rd.ReadData(satellite="snpp", sensor="viirs") geo = viirs.read_viirs_geo() ancillary = rd.ReadAncillary( latitude=geo.latitude.values, longitude=geo.longitude.values, resolution=1, ancillary_dir=fixturepath, ndvi_file=ndvi_file, ) ndvi = ancillary.get_ndvi() check_differences(ref_file, ndvi, "ndvi") def test_geos( fixturepath, geos_file_1, geos_file_2, geos_land, geos_ocean, geos_constants, ref_file, ): viirs = rd.ReadData(satellite="snpp", sensor="viirs") geo = viirs.read_viirs_geo() ancillary = rd.ReadAncillary( latitude=geo.latitude.values, longitude=geo.longitude.values, resolution=1, ancillary_dir=fixturepath, geos_file_1=geos_file_1, geos_file_2=geos_file_2, geos_land=geos_land, geos_ocean=geos_ocean, geos_constants=geos_constants, ) geos_data = ancillary.get_geos() for var in list(geos_data.keys()): check_differences(ref_file, geos_data[var], f"geos_{var}") def check_differences(ref_file, test_data, var_name): ref_data = xr.open_dataset(ref_file) check = np.allclose(ref_data[var_name].values, test_data, equal_nan=True) assert check