getMH1OceanColor_Sci.py

← Back to workflow

Science quality MODIS-Aqua ocean color retrieval (chlorophyll-a, nFLH, PAR, Kd490).

"""
Retrieve and publish science-quality MODIS-Aqua ocean color products.

This script queries the NASA OceanColor file search API for MODIS-Aqua
Level-3 mapped science-quality ocean color products, compares the returned
filenames against files already staged in the production Google Cloud Storage
(GCS) bucket, downloads only missing files to the Cloud Run container,
publishes them with send_to_servers(), and removes local temporary files.

Products
--------
- chlor_a -> chla
- nflh    -> nflh
- par     -> par
- Kd_490  -> kd490

Periods
-------
- DAY -> 1day
- 8D  -> 8day
- MO  -> mday

NASA query settings
-------------------
- sensor_id: 7 (MODIS-Aqua)
- dtid: 1043
- resolution_id: 4km
- stream: science quality / standard delayed

Output
------
Files are published under the MH1 GCS path family, with product- and
period-specific subdirectories. Science-quality paths do not use the _NRT
suffix.

Runtime requirements
--------------------
- ROYLIB_CONFIG points to the runtime config.yml file.
- Earthdata .netrc and URS cookies are available under /tmp.
- The Cloud Run service account can read and write the configured GCS bucket.

Usage
-----
python getMH1OceanColor_Sci.py
"""

if __name__ == "__main__":
    from datetime import datetime
    import os
    import subprocess
    from pathlib import Path
    from roylib import *

    # Validation
    for _k in ("ERDPROD_BUCKET", "HOME_DIR", "NASA_OCEANDATA_URL"):
        if not CFG.get(_k):
            raise KeyError(f"CFG['{_k}'] is required but not set")

    # Mapping logic for NASA API and GCS paths
    products = {
        "chlor_a": "chla",
        "nflh": "nflh",
        "par": "par",
        "Kd_490": "kd490"
    }
    
    # Map NASA period strings to your directory names and interval flags
    periods = {
        "DAY": {"dir": "1day", "flag": "1"},
        "8D":  {"dir": "8day", "flag": "8"},
        "MO":  {"dir": "mday", "flag": "m"}
    }

    now = datetime.now()
    start_date = f"{now.year}-01-01"
    end_date = now.strftime("%Y-%m-%d")
    bucket_name = CFG.get("ERDPROD_BUCKET")
    
    # Pathing Fix: Use config-driven HOME_DIR (should be /tmp in Cloud Run)
    home_dir = Path(CFG.get("HOME_DIR", "/tmp"))
    cookie_path = home_dir / ".urs_cookies"
    download_dir = home_dir / "data"
    download_dir.mkdir(parents=True, exist_ok=True)

    for prod_id, prod_label in products.items():
        for nasa_per, per_info in periods.items():
            
            # GCS Path: Science Quality standardly omits the _NRT suffix
            stage_dir = f"satellite/MH1/{prod_label}/{per_info['dir']}"
            print(f"\n[START] Processing Science Quality: {prod_id} {nasa_per} -> {stage_dir}")

            # 1. Check GCS
            raw_staged = list_bucket_content(bucket_name, stage_dir)
            staged_flist = [os.path.basename(f).strip() for f in raw_staged] if raw_staged else []

            # 2. Query NASA API
            # dtid=1102 targets the Standard Refined Science products
            query_params = (
                f"results_as_file=1&sensor_id=7&dtid=1043&subType=1&resolution_id=4km"
                f"&sdate={start_date} 00:00:00&edate={end_date} 23:59:59"
                f"&prod_id={prod_id}&period={nasa_per}"
            )
            
            query_url = f'{CFG["NASA_FILE_SEARCH_URL"]} --post-data="{query_params}"'
            query_out = home_dir / "resources" / f"sci_{prod_label}_{nasa_per}_list.txt"
            query_out.parent.mkdir(parents=True, exist_ok=True)

            subprocess.run(f"wget -4 --no-check-certificate -O {query_out} {query_url}", shell=True, capture_output=True)

            # 3. Filter and Download
            if not query_out.exists(): continue
            
            with open(query_out, "r") as f:
                for line in f:
                    fname = line.strip()
                    if not fname or "No Results Found" in fname or "<html" in fname.lower():
                        continue
                    
                    if fname not in staged_flist:
                        out_path = download_dir / fname
                        print(f"[DOWNLOAD] {fname}")
                        
                        wget_cmd = (
                            f'wget -4 --netrc --auth-no-challenge=on --keep-session-cookies '
                            f'--load-cookies {cookie_path} --save-cookies {cookie_path} '
                            f'--content-disposition --no-check-certificate '
                            f'-O "{out_path}" {CFG["NASA_GETFILE_URL"]}/{fname}'
                        )
                        
                        result = subprocess.run(wget_cmd, shell=True, capture_output=True)

                        if result.returncode == 0:
                            # Destination logic for Science products
                            dst_dir = f"MH1/{prod_label}"
                            try:
                                send_to_servers(str(out_path), dst_dir, per_info['flag'])
                            finally:
                                if out_path.exists(): os.remove(out_path)
            
    print("\n[COMPLETE] Science Quality Suite finished.")