From a70834b485bac0b99c7638fc17fc06103c6f1a85 Mon Sep 17 00:00:00 2001 From: Kimberly Hyde Date: Tue, 18 Aug 2026 12:50:11 -0400 Subject: [PATCH] Changed subset to map_subset to be consistent with the processing functions --- src/pipelines/pp_pipeline.py | 4 +- src/pipelines/psc_pipeline.py | 4 +- .../satprocessing_parallel_pipeline.py | 49 ++++++++++++++----- src/pipelines/stats_pipeline.py | 4 +- 4 files changed, 42 insertions(+), 19 deletions(-) diff --git a/src/pipelines/pp_pipeline.py b/src/pipelines/pp_pipeline.py index d59123e..6a5be1a 100644 --- a/src/pipelines/pp_pipeline.py +++ b/src/pipelines/pp_pipeline.py @@ -30,7 +30,7 @@ def parse_args(): parser.add_argument("--chl_dataset", type=str, required=False, help="Name of CHL dataset") parser.add_argument("--sst_dataset", type=str, required=False, help="Name of SST dataset") parser.add_argument("--par_dataset", type=str, required=False, help="Name of PAR dataset") - parser.add_argument("--subset", type=str, required=False, help="Name of the region (e.g. NES, NWA) to subset the data to") + parser.add_argument("--map_subset", type=str, required=False, help="Name of the map region (e.g. NES, NWA) to subset the data to") parser.add_argument("--daterange", type=str, help="Flexible date input: year(s), date(s), or range") parser.add_argument("--logfile", type=str, help="Optional path to log file for this run") # ← Add this line args = parser.parse_args() @@ -46,7 +46,7 @@ def parse_args(): if __name__ == "__main__": args = parse_args() - run_pp_pipeline(chl_dataset=args.chl_dataset, sst_dataset=args.sst_dataset, par_dataset=args.par_dataset, subset=args.subset, daterange=args.daterange, logfile=args.logfile) + run_pp_pipeline(chl_dataset=args.chl_dataset, sst_dataset=args.sst_dataset, par_dataset=args.par_dataset, map_subset=args.map_subset, daterange=args.daterange, logfile=args.logfile) # Example terminal window command # python3 run_pp_pipeline.py --sst_dataset CORALSST --daterange 1998 diff --git a/src/pipelines/psc_pipeline.py b/src/pipelines/psc_pipeline.py index 6185e44..75393f2 100644 --- a/src/pipelines/psc_pipeline.py +++ b/src/pipelines/psc_pipeline.py @@ -29,7 +29,7 @@ def parse_args(): parser = argparse.ArgumentParser(description="Run phytoplankton size class pipeline") parser.add_argument("--chl_dataset", type=str, required=False, help="Name of CHL dataset") parser.add_argument("--sst_dataset", type=str, required=False, help="Name of SST dataset") - parser.add_argument("--subset", type=str, required=False, help="Name of the region (e.g. NES, NWA) to subset the data to") + parser.add_argument("--map_subset", type=str, required=False, help="Name of the map region (e.g. NES, NWA) to subset the data to") parser.add_argument("--daterange", type=str, help="Flexible date input: year(s), date(s), or range") parser.add_argument("--logfile", type=str, help="Optional path to log file for this run") # ← Add this line parser.add_argument("--verbose", action="store_true", help="Print detailed progress logs") @@ -50,7 +50,7 @@ def parse_args(): if __name__ == "__main__": args = parse_args() - run_psc_pipeline(chl_dataset=args.chl_dataset, sst_dataset=args.sst_dataset, subset=args.subset, daterange=args.daterange, logfile=args.logfile, verbose=args.verbose) + run_psc_pipeline(chl_dataset=args.chl_dataset, sst_dataset=args.sst_dataset, map_subset=args.map_subset, daterange=args.daterange, logfile=args.logfile, verbose=args.verbose) # Example terminal window command # python3 run_psc_pipeline.py --sst_dataset CORALSST --daterange 1998 diff --git a/src/pipelines/satprocessing_parallel_pipeline.py b/src/pipelines/satprocessing_parallel_pipeline.py index e31963a..dffb618 100644 --- a/src/pipelines/satprocessing_parallel_pipeline.py +++ b/src/pipelines/satprocessing_parallel_pipeline.py @@ -8,6 +8,29 @@ import time from datetime import datetime +""" +Command Line Usage Examples: + +1. Basic run (PSC pipeline): + Uses defaults for the psc pipeline and default batch_size (4). + $ python3 run_psc_parallel.py --pipeline psc + +2. PSC Pipeline with keywords + Uses the default chl dataset, but specifies the sst dataset, map_subset, the batch_size and includes the verbose keyword + $ python3 run_psc_parallel.py --pipeline psc --sst_dataset CORALSST --map_subset NES --verbose --batch_size 6 + +2. Run Primary Production (PP) pipeline based on SST years: + Changes the main product to SST, subsets to the NWA region, and increases parallel jobs to 6. + $ python3 run_psc_parallel.py --pipeline pp --main_prod sst --sst_dataset modis_aqua_sst --map_subset NWA --batch_size 6 + +3. Fully loaded run with multiple datasets and verbose output: + Runs the PSC pipeline, discovers years based on the RRS dataset, limits the region to NES, runs 8 years concurrently, and prints verbose logs. + $ python3 run_psc_parallel.py --pipeline psc --main_prod rrs --rrs_dataset viirs_snpp_rrs --chl_dataset viirs_snpp_chl --sst_dataset viirs_snpp_sst --subset NES --batch_size 8 --verbose + +4. Dry run / testing (if implemented downstream): + $ python3 run_psc_parallel.py --pipeline psc --chl_dataset test_chl --dry_run --verbose +""" + # Add project root to sys.path project_root = Path(__file__).resolve().parent.parent.parent.parent @@ -16,13 +39,13 @@ from utilities.bootstrap.environment import bootstrap_environment env = bootstrap_environment(verbose=False) -from utilities.src.utilities.product_utilities import get_prod_files +from utilities.src.utilities.file_utilities import get_prod_files from utilities.src.utilities.date_utilities import get_source_file_dates import time def run_rolling(pipeline, prod, dataset, chl_dataset=None, sst_dataset=None, par_dataset=None, rrs_dataset=None, - subset=None, max_parallel=5, verbose=False, log_dir=None, summary_log=None, timestamp=None): + map_subset=None, max_parallel=5, verbose=False, log_dir=None, summary_log=None, timestamp=None): dataset = resolve_dataset(prod, chl_dataset, sst_dataset, par_dataset, rrs_dataset) if verbose and not dataset: @@ -51,7 +74,7 @@ def run_rolling(pipeline, prod, dataset, chl_dataset=None, sst_dataset=None, par chl_dataset=chl_dataset, par_dataset=par_dataset, rrs_dataset=rrs_dataset, - subset=subset, + map_subset=map_subset, log_dir=log_dir, timestamp=timestamp ) @@ -98,7 +121,7 @@ def chunk_years(years, batch_size): return [years[i:i + batch_size] for i in range(0, len(years), batch_size)] def build_command(pipeline, year, chl_dataset=None, sst_dataset=None, par_dataset=None, rrs_dataset=None, - subset=None, log_dir=None, timestamp=None): + map_subset=None, log_dir=None, timestamp=None): pipeline = pipeline.lower().strip() script_map = { "psc": "psc_pipeline.py", @@ -118,8 +141,8 @@ def build_command(pipeline, year, chl_dataset=None, sst_dataset=None, par_datase cmd += ["--par_dataset", par_dataset] if rrs_dataset: cmd += ["--rrs_dataset", rrs_dataset] - if subset: - cmd += ["--subset", subset] + if map_subset: + cmd += ["--map_subset", map_subset] cmd += ["--daterange", str(year)] if log_dir: @@ -128,7 +151,7 @@ def build_command(pipeline, year, chl_dataset=None, sst_dataset=None, par_datase return cmd -def run_year(pipeline, year, chl_dataset=None, sst_dataset=None, par_dataset=None, rrs_dataset=None, subset=None, log_dir=None, timestamp=None): +def run_year(pipeline, year, chl_dataset=None, sst_dataset=None, par_dataset=None, rrs_dataset=None, map_subset=None, log_dir=None, timestamp=None): cmd = build_command( pipeline=pipeline, year=year, @@ -136,7 +159,7 @@ def run_year(pipeline, year, chl_dataset=None, sst_dataset=None, par_dataset=Non sst_dataset=sst_dataset, par_dataset=par_dataset, rrs_dataset=rrs_dataset, - subset=subset, + map_subset=map_subset, log_dir=log_dir, timestamp=timestamp ) @@ -146,7 +169,7 @@ def run_year(pipeline, year, chl_dataset=None, sst_dataset=None, par_dataset=Non import time def run_batches(pipeline, prod, dataset, chl_dataset=None, sst_dataset=None, par_dataset=None, rrs_dataset=None, - subset=None, batch_size=5, verbose=False, log_dir=None, summary_log=None, timestamp=None): + map_subset=None, batch_size=5, verbose=False, log_dir=None, summary_log=None, timestamp=None): dataset = resolve_dataset(prod, chl_dataset, sst_dataset, par_dataset, rrs_dataset) if verbose and not dataset: @@ -174,7 +197,7 @@ def run_batches(pipeline, prod, dataset, chl_dataset=None, sst_dataset=None, par chl_dataset=chl_dataset, par_dataset=par_dataset, rrs_dataset=rrs_dataset, - subset=subset, + map_subset=map_subset, log_dir=log_dir, timestamp=timestamp ) @@ -204,7 +227,7 @@ def run_batches(pipeline, prod, dataset, chl_dataset=None, sst_dataset=None, par parser.add_argument("--sst_dataset", type=str, help="Name of SST dataset") parser.add_argument("--par_dataset", type=str, help="Name of PAR dataset") parser.add_argument("--rrs_dataset", type=str, help="Name of RRS dataset") - parser.add_argument("--subset", type=str, help="Region subset (e.g. NES, NWA)") + parser.add_argument("--map_subset", type=str, help="Name of map subset (e.g. NES, NWA) region to subset the global data to") parser.add_argument("--logfile", type=str, help="Optional path to log file for this run") parser.add_argument("--batch_size", type=int, default=4, help="Number of years to run in parallel") parser.add_argument("--verbose", action="store_true", help="Enable verbose output") @@ -228,7 +251,7 @@ def run_batches(pipeline, prod, dataset, chl_dataset=None, sst_dataset=None, par sst_dataset=args.sst_dataset, par_dataset=args.par_dataset, rrs_dataset=args.rrs_dataset, - subset=args.subset, + map_subset=args.map_subset, batch_size=args.batch_size, verbose=args.verbose, log_dir=log_dir, @@ -244,7 +267,7 @@ def run_batches(pipeline, prod, dataset, chl_dataset=None, sst_dataset=None, par sst_dataset=args.sst_dataset, par_dataset=args.par_dataset, rrs_dataset=args.rrs_dataset, - subset=args.subset, + map_subset=args.map_subset, max_parallel=args.batch_size, # reuse batch_size as parallel limit verbose=args.verbose, log_dir=log_dir, diff --git a/src/pipelines/stats_pipeline.py b/src/pipelines/stats_pipeline.py index ff25389..e4179c6 100644 --- a/src/pipelines/stats_pipeline.py +++ b/src/pipelines/stats_pipeline.py @@ -32,7 +32,7 @@ def parse_args(): parser.add_argument("--periods", type=str, nargs='+', required=True, help="Period code(s) of the output statistics (e.g. W, M, A)") parser.add_argument("--dataset", type=str, required=False, help="Name of product dataset (e.g. CHL, SST, PSC)") parser.add_argument("--version", type=str, required=False, help="Name of dataset version (e.g. V6.0)") - parser.add_argument("--subset", type=str, required=False, help="Name of the region (e.g. NES, NWA) to subset the data to") + parser.add_argument("--map_subset", type=str, required=False, help="Name of the map region (e.g. NES, NWA) to subset the data to") parser.add_argument("--daterange", type=str, help="Flexible date input: year(s), date(s), or range") parser.add_argument("--climatology_range", type=str, help="The year range for the climatology (e.g. 1991,2020)") parser.add_argument("--parallel_runs", type=int, default=1, help="The number of stats jobs to run in parallel") @@ -57,7 +57,7 @@ def parse_args(): periods=args.periods, dataset=args.dataset, dataset_version=args.version, - subset=args.subset, + map_subset=args.map_subset, daterange=args.daterange, climatology_range=args.climatology_range, parallel_runs=args.parallel_runs,