diff --git a/.flake8 b/.flake8 new file mode 100644 index 00000000..e74ea330 --- /dev/null +++ b/.flake8 @@ -0,0 +1,4 @@ +[flake8] +max-line-length = 135 +exclude = .git,__pycache__ +ignore = E203, W503 diff --git a/.github/workflows/pytest.yaml b/.github/workflows/pytest.yaml new file mode 100644 index 00000000..7fe80a07 --- /dev/null +++ b/.github/workflows/pytest.yaml @@ -0,0 +1,40 @@ +name: Run Tests on Pull Request + +on: + pull_request: + branches: + - main + +jobs: + test: + runs-on: ubuntu-latest + + steps: + - name: Checkout Code + uses: actions/checkout@v4 # Clones the repository + with: + submodules: recursive + + - name: Set Up Python + uses: actions/setup-python@v4 + with: + python-version: '3.9' + + - name: Install Dependencies + run: | + python -m venv obsdb + source obsdb/bin/activate + pip install --upgrade pip + pip install -r requirements.txt + + - name: Run Code Style Check + run: | + source obsdb/bin/activate + pip install flake8 + flake8 ush/python/pyobsforge/obsdb/ + flake8 ush/python/pyobsforge/tests/ + + - name: Run Pytest + run: | + source obsdb/bin/activate + pytest ush/python/pyobsforge/tests/ --disable-warnings -v diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 00000000..2ce041f1 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,33 @@ +certifi==2025.1.31 +charset-normalizer==3.4.1 +contourpy==1.3.0 +cycler==0.12.1 +exceptiongroup==1.2.2 +flake8==7.1.2 +fonttools==4.56.0 +idna==3.10 +iniconfig==2.0.0 +Jinja2==3.1.6 +kiwisolver==1.4.7 +MarkupSafe==3.0.2 +matplotlib +mccabe==0.7.0 +numpy +packaging==24.2 +panda==0.3.1 +pandas==2.2.3 +pillow==11.1.0 +pluggy==1.5.0 +pycodestyle==2.12.1 +pyflakes==3.2.0 +pyparsing==3.2.1 +pytest==8.3.5 +python-dateutil==2.9.0.post0 +pytz==2025.1 +PyYAML==6.0.2 +requests==2.32.3 +six==1.17.0 +tomli==2.2.1 +tzdata==2025.1 +urllib3==2.3.0 +pysqlite3 diff --git a/ush/python/pyobsforge/obsdb/__init__.py b/ush/python/pyobsforge/obsdb/__init__.py new file mode 100644 index 00000000..5f2a238e --- /dev/null +++ b/ush/python/pyobsforge/obsdb/__init__.py @@ -0,0 +1 @@ +from .obsdb import BaseDatabase # noqa diff --git a/ush/python/pyobsforge/obsdb/ghrsst_db.py b/ush/python/pyobsforge/obsdb/ghrsst_db.py new file mode 100644 index 00000000..aa0b3def --- /dev/null +++ b/ush/python/pyobsforge/obsdb/ghrsst_db.py @@ -0,0 +1,79 @@ +import os +import glob +from datetime import datetime +from pyobsforge.obsdb import BaseDatabase + + +class GhrSstDatabase(BaseDatabase): + """Class to manage an observation file database for data assimilation.""" + + def __init__(self, db_name="obs_files.db", + dcom_dir="/lfs/h1/ops/prod/dcom/", + obs_dir="sst"): + base_dir = os.path.join(dcom_dir, '*', obs_dir) + super().__init__(db_name, base_dir) + + def create_database(self): + """Create the SQLite database and observation files table.""" + query = """ + CREATE TABLE IF NOT EXISTS obs_files ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + filename TEXT UNIQUE, + obs_time TIMESTAMP, + ingest_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + instrument TEXT, + satellite TEXT, + obs_type TEXT + ) + """ + self.execute_query(query) + + def parse_filename(self, filename): + """Extract metadata from filenames matching the expected pattern.""" + parts = os.path.basename(filename).replace('_', '-').split('-') + if len(parts) >= 6 and parts[0].isdigit() and len(parts[0]) == 14: + obs_time = datetime.strptime(parts[0][0:12], "%Y%m%d%H%M") + obs_type = parts[4] if len(parts) > 2 else None + instrument = parts[5] if len(parts) > 3 else None + satellite = parts[6] if len(parts) > 4 else None + return filename, obs_time, instrument, satellite, obs_type + return None + + def ingest_files(self): + """Scan the directory for new observation files and insert them into the database.""" + obs_files = glob.glob(os.path.join(self.base_dir, "*.nc")) + print(f"Found {len(obs_files)} new files to ingest") + for file in obs_files: + parsed_data = self.parse_filename(file) + if parsed_data: + query = """ + INSERT INTO obs_files (filename, obs_time, instrument, satellite, obs_type) + VALUES (?, ?, ?, ?, ?) + """ + self.insert_record(query, parsed_data) + + +# Example Usage +if __name__ == "__main__": + db = GhrSstDatabase(db_name="sst_obs.db", + dcom_dir="/home/gvernier/Volumes/hera-s1/runs/realtimeobs/lfs/h1/ops/prod/dcom/", + obs_dir="sst") + + # Check for new files + db.ingest_files() + + # Query files for a given DA cycle + da_cycle = "20250316000000" + cutoff_delta = 4 + valid_files = db.get_valid_files(da_cycle, + instrument="VIIRS", + satellite="NPP", + obs_type="SSTsubskin", + cutoff_delta=cutoff_delta) + + print(f"Found {len(valid_files)} valid files for DA cycle {da_cycle}") + for valid_file in valid_files: + if os.path.exists(valid_file): + print(f"Valid file: {valid_file}") + else: + print(f"File does not exist: {valid_file}") diff --git a/ush/python/pyobsforge/obsdb/jrr_aod_db.py b/ush/python/pyobsforge/obsdb/jrr_aod_db.py new file mode 100644 index 00000000..1ee4ee14 --- /dev/null +++ b/ush/python/pyobsforge/obsdb/jrr_aod_db.py @@ -0,0 +1,73 @@ +import os +import glob +from datetime import datetime +from pyobsforge.obsdb import BaseDatabase + + +class JrrAodDatabase(BaseDatabase): + """Class to manage an observation file database for JRR-AOD data.""" + + def __init__(self, db_name="jrr_aod_obs.db", + dcom_dir="/lfs/h1/ops/prod/dcom/", + obs_dir="jrr_aod"): + base_dir = os.path.join(dcom_dir, '*', obs_dir) + super().__init__(db_name, base_dir) + + def create_database(self): + """Create the SQLite database and observation files table.""" + query = """ + CREATE TABLE IF NOT EXISTS obs_files ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + filename TEXT UNIQUE, + obs_time TIMESTAMP, + ingest_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """ + self.execute_query(query) + + def parse_filename(self, filename): + """Extract metadata from filenames matching the JRR-AOD pattern.""" + # Make sure the filename matches the expected pattern + # Pattern: JRR-AOD_v3r2_n21_sYYYYMMDDHHMMSS_eYYYYMMDDHHMMSS_cYYYYMMDDHHMMSS.nc + basename = os.path.basename(filename) + parts = basename.split('_') + try: + if len(parts) >= 4 and parts[0] == "JRR-AOD": + obs_time = datetime.strptime(parts[3][1:13], "%Y%m%d%H%M") + return filename, obs_time + except ValueError: + return None + + return None + + def ingest_files(self): + """Scan the directory for new JRR-AOD observation files and insert them into the database.""" + obs_files = glob.glob(os.path.join(self.base_dir, "*.nc")) + print(f"Found {len(obs_files)} new files to ingest") + for file in obs_files: + parsed_data = self.parse_filename(file) + if parsed_data: + query = """ + INSERT INTO obs_files (filename, obs_time) + VALUES (?, ?) + """ + self.insert_record(query, parsed_data) + + +if __name__ == "__main__": + db = JrrAodDatabase(dcom_dir="/home/gvernier/Volumes/hera-s1/runs/realtimeobs/lfs/h1/ops/prod/dcom/") + + # Check for new files + db.ingest_files() + + # Query files for a given DA cycle + da_cycle = "20250316120000" + cutoff_delta = 5 + valid_files = db.get_valid_files(da_cycle, cutoff_delta=cutoff_delta) + + print(f"Found {len(valid_files)} valid files for DA cycle {da_cycle}") + for valid_file in valid_files: + if os.path.exists(valid_file): + print(f"Valid file: {valid_file}") + else: + print(f"File does not exist: {valid_file}") diff --git a/ush/python/pyobsforge/obsdb/obsdb.py b/ush/python/pyobsforge/obsdb/obsdb.py new file mode 100644 index 00000000..3e0f290d --- /dev/null +++ b/ush/python/pyobsforge/obsdb/obsdb.py @@ -0,0 +1,95 @@ +import sqlite3 +from datetime import datetime, timedelta +from wxflow.sqlitedb import SQLiteDB + + +class BaseDatabase(SQLiteDB): + """Base class for managing different types of file-based databases.""" + + def __init__(self, db_name: str, base_dir: str) -> None: + """ + Initialize the database. + + :param db_name: Name of the SQLite database. + :param base_dir: Directory containing observation files. + """ + super().__init__(db_name) + self.base_dir = base_dir + self.create_database() + + def create_database(self): + """Create the SQLite database. Must be implemented by subclasses.""" + raise NotImplementedError("Subclasses must implement create_database method") + + def get_connection(self): + """Return the database connection.""" + return self.connection + + def parse_filename(self): + """Parse a filename and extract relevant metadata. Must be implemented by subclasses.""" + raise NotImplementedError("Subclasses must implement parse_filename method") + + def ingest_files(self): + """Scan the directory for new observation files and insert them into the database.""" + raise NotImplementedError("Subclasses must implement ingest_files method") + + def insert_record(self, query: str, params: tuple) -> None: + """Insert a record into the database.""" + self.connect() + cursor = self.connection.cursor() + try: + cursor.execute(query, params) + self.connection.commit() + except sqlite3.IntegrityError: + pass # Skip duplicates + finally: + self.disconnect() + + def execute_query(self, query: str, params: tuple = None) -> list: + """Execute a query and return the results.""" + self.connect() + cursor = self.connection.cursor() + cursor.execute(query, params or []) + results = cursor.fetchall() + self.disconnect() + return results + + def get_valid_files(self, + da_cycle: str, + window_hours: int = 3, + instrument: str = None, + satellite: str = None, + obs_type: str = None, + cutoff_delta: int = 0) -> list: + """ + Retrieve a list of observation files within a DA window, possibly filtered by instrument, + satellite, observation type, and cutoff delta (known latency to emulate the early cycle if needed). + """ + da_time = datetime.strptime(da_cycle, "%Y%m%d%H%M%S") + window = timedelta(hours=window_hours) + cutoff_delta = timedelta(hours=cutoff_delta) + window_begin = da_time - window + window_end = da_time + window - cutoff_delta + + query = """ + SELECT filename FROM obs_files + WHERE obs_time BETWEEN ? AND ? + """ + params = [window_begin, window_end] + + if instrument: + query += " AND instrument = ?" + params.append(instrument) + if satellite: + query += " AND satellite = ?" + params.append(satellite) + if obs_type: + query += " AND obs_type = ?" + params.append(obs_type) + + results = self.execute_query(query, tuple(params)) + valid_files = [] + for row in results: + valid_files.append(row[0]) + + return valid_files diff --git a/ush/python/pyobsforge/tests/__init__.py b/ush/python/pyobsforge/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ush/python/pyobsforge/tests/test_ghrsst_database.py b/ush/python/pyobsforge/tests/test_ghrsst_database.py new file mode 100644 index 00000000..02332466 --- /dev/null +++ b/ush/python/pyobsforge/tests/test_ghrsst_database.py @@ -0,0 +1,142 @@ +import os +import tempfile +import shutil +import sqlite3 +from datetime import datetime + +import pytest + +from pyobsforge.obsdb.ghrsst_db import GhrSstDatabase # Adjust as needed + + +@pytest.fixture +def temp_obs_dir(): + """Create a temp directory with mock GHRSST NetCDF files.""" + base_dir = tempfile.mkdtemp() + sub_dir = os.path.join(base_dir, "some_subdir", "sst") + os.makedirs(sub_dir) + + filenames = [ + "20250316100000-OSPO-L3U_GHRSST-SSTsubskin-AVHRRF_MB-ACSPO.nc", + "20250316120000-OSPO-L3U_GHRSST-SSTsubskin-AVHRRF_MB-ACSPO.nc", + "20250316150000-OSPO-L3U_GHRSST-SSTsubskin-AVHRRF_MB-ACSPO.nc", + "20250316100000-OSPO-L3U_GHRSST-SSTsubskin-AVHRRF_MC-ACSPO.nc", + "20250316120000-OSPO-L3U_GHRSST-SSTsubskin-AVHRRF_MC-ACSPO.nc", + "20250316150000-OSPO-L3U_GHRSST-SSTsubskin-AVHRRF_MC-ACSPO.nc", + "20250316100000-OSPO-L3U_GHRSST-SSTsubskin-VIIRS_NPP-ACSPO.nc", + "20250316120000-OSPO-L3U_GHRSST-SSTsubskin-VIIRS_NPP-ACSPO.nc", + "20250316150000-OSPO-L3U_GHRSST-SSTsubskin-VIIRS_NPP-ACSPO.nc", + "20250316100000-OSPO-L3U_GHRSST-SSTsubskin-VIIRS_N20-ACSPO.nc", + "20250316120000-OSPO-L3U_GHRSST-SSTsubskin-VIIRS_N20-ACSPO.nc", + "20250316150000-OSPO-L3U_GHRSST-SSTsubskin-VIIRS_N20-ACSPO.nc", + "20250316100000-OSPO-L3U_GHRSST-SSTsubskin-VIIRS_N21-ACSPO.nc", + "20250316120000-OSPO-L3U_GHRSST-SSTsubskin-VIIRS_N21-ACSPO.nc", + "20250316150000-OSPO-L3U_GHRSST-SSTsubskin-VIIRS_N21-ACSPO.nc", + "invalid_file.nc" + ] + for fname in filenames: + with open(os.path.join(sub_dir, fname), "w") as f: + f.write("fake content") + + yield base_dir + shutil.rmtree(base_dir) + + +@pytest.fixture +def db(temp_obs_dir): + """Initialize test database.""" + db_path = os.path.join(temp_obs_dir, "ghrsst_test.db") + return GhrSstDatabase(db_name=db_path, dcom_dir=temp_obs_dir, obs_dir="sst") + + +def test_create_database(db): + db.create_database() + conn = sqlite3.connect(db.db_name) + cursor = conn.cursor() + cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='obs_files'") + assert cursor.fetchone() is not None + conn.close() + + +def test_parse_valid_filename(db): + fname = "20250316123400-OSPO-L3U_GHRSST-SSTsubskin-AVHRRF_MB-ACSPO.nc" + parsed = db.parse_filename(fname) + assert parsed is not None + assert parsed[0] == fname + assert parsed[1] == datetime(2025, 3, 16, 12, 34) + assert parsed[2] == "AVHRRF" + assert parsed[3] == "MB" + assert parsed[4] == "SSTsubskin" + + +def test_parse_invalid_filename(db): + assert db.parse_filename("junk.nc") is None + assert db.parse_filename("20250316_invalid_filename.nc") is None + + +def test_ingest_files(db): + db.ingest_files() + conn = sqlite3.connect(db.db_name) + cursor = conn.cursor() + cursor.execute("SELECT COUNT(*) FROM obs_files") + count = cursor.fetchone()[0] + conn.close() + assert count == 15, "Should ingest 3 valid GHRSST files" + + +def test_get_valid_files(db): + db.ingest_files() + da_cycle = "20250316120000" + cutoff_delta = 3 # hours + + # Test for AVHRRF_MB + valid_files = db.get_valid_files(da_cycle, + instrument="AVHRRF", + satellite="MB", + obs_type="SSTsubskin", + cutoff_delta=cutoff_delta) + + # Files at 10:00 and 12:00 are within +/- 3h of 12:00 + assert any("202503161000" in f for f in valid_files) + assert any("202503161200" in f for f in valid_files) + assert all("202503161500" not in f for f in valid_files) + assert len(valid_files) == 2 + + # Test for VIIRS_NPP + valid_files = db.get_valid_files(da_cycle, + instrument="VIIRS", + satellite="NPP", + obs_type="SSTsubskin", + cutoff_delta=cutoff_delta) + + # Files at 10:00 and 12:00 are within +/- 3h of 12:00 + assert any("202503161000" in f for f in valid_files) + assert any("202503161200" in f for f in valid_files) + assert all("202503161500" not in f for f in valid_files) + assert len(valid_files) == 2 + + # Test for VIIRS_N20 + valid_files = db.get_valid_files(da_cycle, + instrument="VIIRS", + satellite="N20", + obs_type="SSTsubskin", + cutoff_delta=cutoff_delta) + + # Files at 10:00 and 12:00 are within +/- 3h of 12:00 + assert any("202503161000" in f for f in valid_files) + assert any("202503161200" in f for f in valid_files) + assert all("202503161500" not in f for f in valid_files) + assert len(valid_files) == 2 + + # Test for VIIRS_N21 + valid_files = db.get_valid_files(da_cycle, + instrument="VIIRS", + satellite="N21", + obs_type="SSTsubskin", + cutoff_delta=cutoff_delta) + + # Files at 10:00 and 12:00 are within +/- 3h of 12:00 + assert any("202503161000" in f for f in valid_files) + assert any("202503161200" in f for f in valid_files) + assert all("202503161500" not in f for f in valid_files) + assert len(valid_files) == 2 diff --git a/ush/python/pyobsforge/tests/test_jrr_aod_database.py b/ush/python/pyobsforge/tests/test_jrr_aod_database.py new file mode 100644 index 00000000..91a3329a --- /dev/null +++ b/ush/python/pyobsforge/tests/test_jrr_aod_database.py @@ -0,0 +1,176 @@ +import os +import tempfile +import shutil +import sqlite3 +from datetime import datetime + +import pytest + +from pyobsforge.obsdb.jrr_aod_db import JrrAodDatabase + + +@pytest.fixture +def temp_obs_dir(): + """Create a temporary directory with mock JRR-AOD NetCDF files.""" + base_dir = tempfile.mkdtemp() + sub_dir = os.path.join(base_dir, "some_subdir", "jrr_aod") + os.makedirs(sub_dir) + + # Create mock NetCDF files (content doesn't matter) + filenames = [ + "JRR-AOD_v3r2_n21_s202503161000000_e202503161030000_c202503161045000.nc", + "JRR-AOD_v3r2_n21_s202503161200000_e202503161230000_c202503161245000.nc", + "JRR-AOD_v3r2_n21_s202503161500000_e202503161530000_c202503161545000.nc", + "invalid_file.nc" + ] + for fname in filenames: + with open(os.path.join(sub_dir, fname), "w") as f: + f.write("dummy") + + yield base_dir + shutil.rmtree(base_dir) + + +@pytest.fixture +def db(temp_obs_dir): + """ + Create an instance of JrrAodDatabase using in-memory SQLite and + the temp_obs_dir, then initialize the database. + """ + db_path = os.path.join(temp_obs_dir, "test_jrr_aod.db") + database = JrrAodDatabase( + db_name=db_path, + dcom_dir=temp_obs_dir, + obs_dir="jrr_aod" + ) + return database + + +def test_create_database(db): + """ + Test the creation of the database and the 'obs_files' table. + + This test performs the following steps: + 1. Creates the database. + 2. Ingests files into the database. + 3. Connects to the database. + 4. Checks for the existence of the 'obs_files' table. + + Args: + db: The database object to be tested. + + Asserts: + - The 'obs_files' table is created in the database. + """ + db.create_database() + db.ingest_files() + db.connect() + conn = db.get_connection() + cursor = conn.cursor() + cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='obs_files'") + result = cursor.fetchall() + conn.close() + + assert result is not None, "obs_files table should be created" + + +def test_parse_valid_filename(db): + """ + Test the parsing of a valid filename in the database. + + This test performs the following steps: + 1. Creates the database. + 2. Ingests files into the database. + 3. Connects to the database. + 4. Checks the tables in the database. + 5. Parses a given filename and verifies the parsed output. + + Args: + db: The database object to be tested. + + Asserts: + - The parsed filename is not None. + - The first element of the parsed result matches the original filename. + - The second element of the parsed result matches the expected datetime. + """ + db.create_database() + db.ingest_files() + db.connect() + fname = "JRR-AOD_v3r2_n21_s202503161234567_e202503161300000_c202503161315000.nc" + conn = db.get_connection() + cursor = conn.cursor() + cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") + conn.close() + parsed = db.parse_filename(fname) + + assert parsed is not None + assert parsed[0] == fname + assert parsed[1] == datetime(2025, 3, 16, 12, 34) + + +def test_parse_invalid_filename(db): + """ + Test the `parse_filename` method of the database with invalid filenames. + + This test ensures that the `parse_filename` method returns `None` when provided + with filenames that do not conform to the expected format. + + Args: + db: The database object that contains the `parse_filename` method. + + Assertions: + - Asserts that `parse_filename` returns `None` for a completely invalid filename. + - Asserts that `parse_filename` returns `None` for a filename that partially + matches the expected format but contains invalid components. + """ + assert db.parse_filename("garbage.nc") is None + assert db.parse_filename("JRR-AOD_v3r2_n21_invalid.nc") is None + + +def test_ingest_files(db): + """ + Test the ingestion of files into the database. + + This test checks if the `ingest_files` method of the `db` object correctly ingests files into the database. + It connects to the database, queries the `obs_files` table to count the number of ingested files, and asserts + that the count is 3, indicating that 3 valid JRR-AOD files should be ingested. + """ + db.ingest_files() + conn = sqlite3.connect(db.db_name) + cursor = conn.cursor() + cursor.execute("SELECT COUNT(*) FROM obs_files") + count = cursor.fetchone()[0] + conn.close() + assert count == 3, "Should ingest 3 valid JRR-AOD files" + + +def test_get_valid_files(db): + """ + Test the `get_valid_files` method of the database. + + This test ingests files into the database and then retrieves the valid files + for a given data assimilation (DA) cycle time and cutoff delta. It checks that + only the files within the specified cutoff delta are returned as valid. + + Steps: + 1. Ingest files into the database. + 2. Define a DA cycle time (`da_cycle`) and a cutoff delta in hours (`cutoff_delta`). + 3. Retrieve the valid files using the `get_valid_files` method. + 4. Assert that files at 10:00 and 12:00 are within +/- 3 hours of 12:00. + 5. Assert that files at 15:00 are not within the valid range. + 6. Assert that the total number of valid files is 2. + + Args: + db: The database object to be tested. + """ + db.ingest_files() + da_cycle = "20250316120000" + cutoff_delta = 3 # hours + + valid_files = db.get_valid_files(da_cycle, cutoff_delta=cutoff_delta) + + # Only files at 10:00 and 12:00 should be within +/- 3 hours of 12:00 + assert any("202503161000" in f for f in valid_files) + assert any("202503161200" in f for f in valid_files) + assert all("202503161500" not in f for f in valid_files) + assert len(valid_files) == 2