Skip to content

Commit

Permalink
Use uploaded data
Browse files Browse the repository at this point in the history
  • Loading branch information
zmoon committed May 5, 2024
1 parent f4f4580 commit aeb9296
Show file tree
Hide file tree
Showing 2 changed files with 58 additions and 2 deletions.
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
docs/api/
tests/data/MOP*.he5
tests/data/TROPOMI*.nc
tests/data/OMPS-*.h5

# Default GitHub .gitignore for Python below:

Expand Down
59 changes: 57 additions & 2 deletions tests/test_omps_l3.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,67 @@
import shutil
import warnings
from pathlib import Path

import pandas as pd
import pytest
from filelock import FileLock

from monetio.sat._omps_l3_mm import open_dataset

HERE = Path(__file__).parent

def test_open_dataset():
ps = sorted(Path("~/Downloads/").expanduser().glob("OMPS-TO3-L3-DAILY_v2.1_*.h5"))
FNS = [
"OMPS-TO3-L3-DAILY_v2.1_20190905.h5",
"OMPS-TO3-L3-DAILY_v2.1_20190906.h5",
]


def retrieve_test_file(i):
fn = FNS[i]

# Download to tests/data if not already present
p = HERE / "data" / fn
if not p.is_file():
warnings.warn(f"Downloading test file {fn} for OMPS L3 test")
import requests

r = requests.get(
"https://csl.noaa.gov/groups/csl4/modeldata/melodies-monet/data/"
f"example_observation_data/satellite/{fn}",
stream=True,
)
r.raise_for_status()
with open(p, "wb") as f:
f.write(r.content)

return p


@pytest.fixture(scope="session")
def test_file_paths(tmp_path_factory, worker_id):
if worker_id == "master":
# Not executing with multiple workers;
# let pytest's fixture caching do its job
return [retrieve_test_file(i) for i in range(len(FNS))]

# Get the temp directory shared by all workers
root_tmp_dir = tmp_path_factory.getbasetemp().parent

# Copy to the shared test location
p_tests = []
for i, p in enumerate(FNS):
p_test = root_tmp_dir / f"omps_l3_test_{i}.he5"
with FileLock(p_test.as_posix() + ".lock"):
if not p_test.is_file():
p = retrieve_test_file(i)
shutil.copy(p, p_test)
p_tests.append(p_test)

return p_tests


def test_open_dataset(test_file_paths):
ps = test_file_paths
ds = open_dataset(ps)

assert ds.sizes["time"] == len(ps) == 2
Expand Down

0 comments on commit aeb9296

Please sign in to comment.