Auto PR - develop → MAPL-v3 - Add SMOS preprocessing scripts.

CHANGELOG.md

@@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added Matlab and python readers for binary Tb scaling parameters files.
 - Added python reader for binary catparam files.
 - Added QC of SMAP L1C_TB using max value for Tb_error.
+- Added SMOS Tb preprocessing scripts.
 
 
 ### Changed

GEOSldas_App/util/inputs/obs_preproc/SMOS_preproc/.gitignore

@@ -0,0 +1,4 @@
+__pycache__/
+*.pyc
+out.log
+*.swp

GEOSldas_App/util/inputs/obs_preproc/SMOS_preproc/README.md

@@ -0,0 +1,65 @@
+# SMOS_preproc
+
+## 1. Purpose
+Preprocesses SMOS L1C brightness temperature data for near-real-time (NRT)
+ingestion into the MERRA21C-land pipeline:
+
+```
+EE .zip  --(smos-ee-to-nc.sh)-->  NetCDF  --(preprocess_nc)-->  EASEv2 M36 "REG" binaries  --(SCLF1C_reg2fit)-->  Tb40 "FIT" binaries
+```
+
+Normal operation is a **daily cron job** that processes the previous day's
+data. It also supports on-demand backfill of a single day or a date range.
+
+## 2. Requirements
+- Runs on Discover/NCCS.
+- Environment: `module load python/GEOSpyD` (provides numpy, scipy, netCDF4,
+  PyYAML — no other Python environment should be needed).
+- External script dependency: `smos-ee-to-nc.sh` (path set in `config.yaml`,
+  currently maintained under `/discover/nobackup/projects/gmao/smap/...` —
+  **not part of this repo**, 
+- Static input: `data/GEOSIT_to_EASEv2_M36.mat` — pre-generated EASEv2
+  regridding weights, link to the actual location on Discover
+- Static Aux files: `data/SM_OPER_AUX_GAL_SM_20050101T000000_20500101T000000_001_003_3`
+  link to the actual location on Discover. 
+
+## 3. Configuration (`config.yaml`)
+| Key                         | Meaning                                                                                         |
+|-----------------------------|-------------------------------------------------------------------------------------------------|
+| `ee_to_nc_script`           | Path to the external `smos-ee-to-nc.sh` converter.                                              |
+| `smos_base_path`            | Where incoming SMOS `SM_*_MIR_SCLF1C_*.zip` (EE) files land, organized `<base>/Y<yyyy>/M<mm>/`. |
+| `tmp_nc_path`               | Scratch directory for converted NetCDF files. A `to_delete/` subfolder is created here. Files therein are safe to delete after completion. |
+| `out_reg_path`              | Output root for REG binaries (`SMOS_reg_Tb_*.bin`, organized by `<out_reg_path>/<YYYYMM>/`). FIT binaries are written to a sibling directory: `_reg_` in this path is replaced by `_fit_` and further nested under `SMOS_fit_poly2/<YYYYMM>/`. |
+| `GEOSIT_path`               | Path to GEOS-IT data.                                                                           |
+| `GEOSIT_to_EASEv2_M36_file` | Path and name of file mapping from GEOS-IT output grid to EASEv2_M36 grid.                      |
+| `SM_OPER_AUX_GAL_SM_path`   | Path to SMOS galaxy correction files.                                                           |
+
+
+Current values are set for the production Discover paths under
+`/discover/nobackup/projects/gmao/smap/SMAP_Nature/SMOS/...` and
+`/discover/nobackup/dao_ops/SMOS/AOSMOS.4662/SCLF1C/`.
+
+## 4. Usage
+```
+module load python/GEOSpyD
+
+python SMOSproc_main.py                                    # NRT: process yesterday's data (normal cron mode)
+python SMOSproc_main.py --date  20260801                   # Backfill a single day
+python SMOSproc_main.py --start 20260801 --end 20260803    # Backfill an inclusive date range
+```
+`--date` and `--start`/`--end` are mutually exclusive; `--end` requires
+`--start`.
+
+## 5. Layout
+```
+SMOSproc_main.py          top-level driver 
+config.yaml               paths 
+data/                     static regridding weights (GEOSIT_to_EASEv2_M36.mat) and AUX 
+src/preprocess_nc.py      NetCDF -> EASEv2 M36 REG binaries
+src/SCLF1C_reg2fit.py     REG -> Tb40 FIT binaries
+src/readwrite/            I/O helpers (SMOS NetCDF, REG binary, GEOS-IT, aux Gal SM)
+src/helper/               grid/geometry/time utilities (EASEv2 indexing, h/v tile
+                          conversion, celestial angle calc, galactic+atmospheric correction)
+```
+
+

GEOSldas_App/util/inputs/obs_preproc/SMOS_preproc/SMOSproc_main.py

@@ -0,0 +1,164 @@
+import os
+import shutil
+import logging
+import multiprocessing
+import yaml
+
+import sys; sys.path.append('../../../shared/python/')
+
+from src import preprocess_nc, SCLF1C_reg2fit
+from src.helper.util import *
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    datefmt='%Y-%m-%d %H:%M:%S'
+)
+
+# ---------------------------------------------------------
+# Configuration 
+# ---------------------------------------------------------
+CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)),'config.yaml')
+with open(CONFIG_PATH) as f:
+    config = yaml.safe_load(f)['paths']
+
+EE_TO_NC_SCRIPT           = config['ee_to_nc_script']
+SMOS_BASE_PATH            = config['smos_base_path']
+TMP_NC_PATH               = config['tmp_nc_path']
+OUT_REG_PATH              = config['out_reg_path']
+
+def run_in_isolated_process(func, *args):
+    """
+    Runs a function in an isolated child process and waits for it to finish.
+    This prevents memory leaks or state changes in external libraries from 
+    affecting the main script.
+    """
+    p = multiprocessing.Process(target=func, args=args)
+    p.start()
+    p.join()
+
+    if p.exitcode != 0:
+        logging.error(f"Process running {func.__name__} failed with exit code {p.exitcode}")
+        raise RuntimeError(f"{func.__name__} failed during execution.")
+
+    logging.info(f"Successfully completed {func.__name__}.")
+
+def process_ee_to_nc(date_time: datetime) -> list:
+    """
+    Finds .zip ee files for the given date, converts them to .nc if needed,
+    and returns a list of resulting netcdf files.
+    """
+    # Build search pattern for EE files
+    year, month, day = date_time.strftime('Y%Y'), date_time.strftime('M%m'), date_time.strftime('%d')
+    date_str = date_time.strftime('%Y%m%d')
+
+    search_pattern = os.path.join(
+        SMOS_BASE_PATH, year, month,
+        f'SM_*_MIR_SCLF1C_{date_str}*_724_*zip'
+    )
+
+    eeflist = sorted(glob.glob(search_pattern))
+
+    logging.info(f"[{date_str}] Found {len(eeflist)} zip files to process.")
+
+    if not eeflist:
+        raise RuntimeError(f"No SMOS EE zip files found for {date_str}")
+
+    # Convert EE to NC
+    for fee in eeflist:
+        base_name = os.path.basename(fee)[:-3] + 'nc'
+        target_nc_file = os.path.join(TMP_NC_PATH, base_name)
+
+        if not os.path.isfile(target_nc_file):
+            logging.info(f"Running ee_to_nc conversion on: {fee}")
+            result = subprocess.run([
+                EE_TO_NC_SCRIPT,
+                '--target-directory', TMP_NC_PATH,
+                fee
+            ], capture_output=True, text=True)
+
+            if result.returncode != 0:
+                logging.error(f"Conversion script failed for {fee}:\n{result.stderr}")
+                raise RuntimeError("ee_to_nc conversion corrupted")
+
+            logging.info(f"Done ee_to_nc conversion: {fee}")
+
+    # Return list of resulting NC files
+    nc_search_pattern = os.path.join(TMP_NC_PATH, f'SM_*_MIR_SCLF1C_{date_str}*724*.nc')
+    ncflist = sorted(glob.glob(nc_search_pattern))
+
+    if len(eeflist) != len(ncflist):
+        logging.warning(f"File count mismatch: {len(eeflist)} zip files vs {len(ncflist)} nc files.")
+
+    return ncflist
+
+def main():
+    args = parse_args()
+    start_time, end_time = get_time_range(args)
+    logging.info(
+        f"Processing {start_time:%Y-%m-%d} through "
+        f"{(end_time - timedelta(days=1)):%Y-%m-%d}"
+    )
+
+    # Ensure necessary directories exist
+    os.makedirs(os.path.join(TMP_NC_PATH, 'to_delete'), exist_ok=True)
+    os.makedirs(OUT_REG_PATH, exist_ok=True)
+
+    current_date = start_time
+
+    while current_date < end_time:
+        date_str = current_date.strftime('%Y%m%d')
+        logging.info(f"=== Starting processing for {date_str} ===")
+
+        # Step 1: Convert files to NetCDF
+        try:
+            ncflist = process_ee_to_nc(current_date)
+        except RuntimeError as e:
+            logging.error(f"Halting processing for {date_str} due to error: {e}")
+            current_date += timedelta(days=1)
+            continue
+
+        # Step 2: Preprocess NetCDF files into REG 
+        for fnc in ncflist:
+            logging.info(f"Preprocessing NC file: {fnc}")
+            try:
+                run_in_isolated_process(preprocess_nc, fnc, config)
+            except RuntimeError as e:
+                logging.error(f"Skipping {fnc} due to error: {e}")
+                continue
+
+            # Move processed file to 'to_delete'
+            dest = os.path.join(TMP_NC_PATH, 'to_delete', os.path.basename(fnc))
+            shutil.move(fnc, dest)
+            logging.info(f"Moved {fnc} to cleanup directory.")
+
+        # Step 3: Run REG to FIT processing for Ascending and Descending
+        next_date = current_date + timedelta(days=1)
+
+        logging.info("Running SCLF1C_reg2fit for Ascending (_A)")
+        try:
+            run_in_isolated_process(SCLF1C_reg2fit, OUT_REG_PATH, current_date, next_date, '_A')
+        except RuntimeError as e:
+            logging.error(f"SCLF1C_reg2fit failed for Ascending (_A) on {date_str}: {e}")
+
+        logging.info("Running SCLF1C_reg2fit for Descending (_D)")
+        try:
+            run_in_isolated_process(SCLF1C_reg2fit, OUT_REG_PATH, current_date, next_date, '_D')
+        except RuntimeError as e:
+            logging.error(f"SCLF1C_reg2fit failed for Descending (_D) on {date_str}: {e}")
+
+        logging.info(f"=== Completed processing to Tb40 for {date_str} ===")
+
+        # Advance to the next day
+        current_date += timedelta(days=1)
+
+
+if __name__ == '__main__':
+    # Set the multiprocessing start method exactly once here
+    try:
+        multiprocessing.set_start_method('spawn')
+    except RuntimeError:
+        pass # Context was already set
+
+    main()

GEOSldas_App/util/inputs/obs_preproc/SMOS_preproc/config.yaml

@@ -0,0 +1,8 @@
+paths:
+   ee_to_nc_script:            /discover/nobackup/projects/gmao/smap/SMAP_Nature/SMOS/smos-ee-to-netcdf/smos-ee-to-nc.sh
+   smos_base_path:             /discover/nobackup/dao_ops/SMOS/AOSMOS.4662/SCLF1C/
+   tmp_nc_path:                /discover/nobackup/projects/gmao/smap/SMAP_Nature/SMOS/netcdf/
+   out_reg_path:               /discover/nobackup/projects/gmao/smap/SMAP_Nature/SMOS/EASEv2/ESA/SMOS_M36_SCLF1C_reg_nosky_noatm_v724_ESA_v102/
+   GEOSIT_path:                /discover/nobackup/projects/gmao/geos-it/dao_ops/archive/
+   GEOSIT_to_EASEv2_M36_file:  /discover/nobackup/projects/gmao/smap/SMAP_Nature/SMOS/Regrid_data/GEOSIT_to_EASEv2_M36.mat
+   SM_OPER_AUX_GAL_SM_path:    /discover/nobackup/projects/gmao/smap/SMAP_Nature/SMOS/SMOS_ESA/AUX_DATA/SM_OPER_AUX_GAL_SM_20050101T000000_20500101T000000_001_003_3/