Skip to content

Satdata #48

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
khyde merged 3 commits into from
Aug 18, 2026
Merged

Satdata #48

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
edb2a42
Merge pull request #45 from NEFSC/main
khyde Jul 31, 2026
a709e69
Merge pull request #47 from NEFSC/main
khyde Aug 3, 2026
a70834b
Changed subset to map_subset to be consistent with the processing fun…
khyde Aug 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/pipelines/pp_pipeline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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
4 changes: 2 additions & 2 deletions src/pipelines/psc_pipeline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand All @@ -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
49 changes: 36 additions & 13 deletions src/pipelines/satprocessing_parallel_pipeline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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:
Expand Down Expand Up @@ -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
)
Expand Down Expand Up @@ -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",
Expand All @@ -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:
Expand All @@ -128,15 +151,15 @@ 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,
chl_dataset=chl_dataset,
sst_dataset=sst_dataset,
par_dataset=par_dataset,
rrs_dataset=rrs_dataset,
subset=subset,
map_subset=map_subset,
log_dir=log_dir,
timestamp=timestamp
)
Expand All @@ -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:
Expand Down Expand Up @@ -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
)
Expand Down Expand Up @@ -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")
Expand All @@ -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,
Expand All @@ -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,
Expand Down
4 changes: 2 additions & 2 deletions src/pipelines/stats_pipeline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand All @@ -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,
Expand Down
Loading