test_read_data.py

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