Simple Obs db

.flake8

@@ -0,0 +1,4 @@
+[flake8]
+max-line-length = 135
+exclude = .git,__pycache__
+ignore = E203, W503

.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

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

ush/python/pyobsforge/obsdb/__init__.py

@@ -0,0 +1 @@
+from .obsdb import BaseDatabase  # noqa

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}")

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}")

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