Skip to content
Snippets Groups Projects
test_read_data.py 4.12 KiB
Newer Older
import numpy as np
import xarray as xr
import os

import pytest

Paolo Veglio's avatar
Paolo Veglio committed
import mvcm.read_data as rd


@pytest.fixture
def fixturepath():
    return os.path.join(os.path.dirname(__file__), 'fixtures')


@pytest.fixture
Paolo Veglio's avatar
Paolo Veglio committed
def l1b_file(fixturepath):
    return os.path.join(fixturepath, 'VNP02MOD.A2022173.1312.001.2022174011547.uwssec.nc')
@pytest.fixture
Paolo Veglio's avatar
Paolo Veglio committed
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()
Paolo Veglio's avatar
Paolo Veglio committed
    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']
Paolo Veglio's avatar
Paolo Veglio committed
    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