netCDF Processing Error test

netcdf_mem_file_test.py

import requests
import netCDF4
import gzip
import tempfile
import traceback
import os.path
import glob


def test_cdf_bytes(cdf_compressed_bytes: bytes):
    cdf_bytes = gzip.decompress(cdf_compressed_bytes)

    print("Loading from Memory")
    station_names = None
    try:
        with netCDF4.Dataset('filename.nc', memory=cdf_bytes, encoding_errors='strict') as cdf_ds:
            station_names = cdf_ds['stationName'][:]
            cdf_ds['latitude'][:]
            print('ok')
    except:
        print("Failed")
        # traceback.print_exc()

    cdf_bytes = gzip.decompress(cdf_compressed_bytes)

    print("Loading from File")
    with tempfile.TemporaryDirectory() as td:
        cdf_path = os.path.join(td, "file.nc")
        with open(cdf_path, "wb") as f:
            f.write(cdf_bytes)

        try:
            with netCDF4.Dataset(cdf_path, encoding_errors='strict') as cdf_ds:
                if station_names is not None:
                    assert cdf_ds['stationName'][:] == station_names
                cdf_ds['latitude'][:]
                print('ok')
        except:
            print("Failed")
            traceback.print_exc()


def main():
    # response = requests.get("https://madis-data.ncep.noaa.gov/madisPublic1/data/archive/2002/05/19/point/metar/netcdf/20020519_0600.gz")
    # test_cdf_bytes(response.content)

    for file in glob.glob("/Users/amohr/Downloads/2*0.gz"):
        print(f"Testing: {file}")
        with open(file, 'rb') as f:
            test_cdf_bytes(f.read())


if __name__ == '__main__':
    main()