Forcing Processors

Forcing processors for NOS-OFS models.

Atmospheric:

GFSProcessor - Global Forecast System (0.25°, hourly) HRRRProcessor - High-Resolution Rapid Refresh (3km, hourly, CONUS) GEFSProcessor - Global Ensemble Forecast System (0.25/0.50°, 3-hourly)

Ocean Boundary:

RTOFSProcessor - Real-Time Ocean Forecast System (SSH, T/S/UV boundaries)

River:

NWMProcessor - National Water Model (streamflow → vsource/msource) RiverClimProcessor - USGS daily climatology (when NWM/BUFR unavailable)

Tidal:

TidalProcessor - Tidal constituents (bctides.in generation)

Nudging:

NudgingProcessor - T/S interior nudging (TEM_nu.nc, SAL_nu.nc)

Model Config:

ParamNmlProcessor - Generate/patch SCHISM param.nml HotstartProcessor - Find and validate restart files PartitionProcessor - Generate partition.prop for MPI decomposition

UFS-Coastal:

ESMFMeshProcessor - ESMF mesh file for DATM coupling BlenderProcessor - HRRR+GFS Delaunay blending for DATM

Writers:

SfluxWriter - SCHISM sflux NetCDF output (nws=2) DATMWriter - UFS-Coastal DATM NetCDF output (nws=4)

class nos_utils.forcing.ForcingProcessor(config: ForcingConfig, input_path: Path, output_path: Path)

Bases: ABC

Abstract base class for atmospheric forcing processors.

Subclasses implement source-specific file discovery, GRIB2 extraction, variable conversion, and output writing.

SOURCE_NAME: str = 'Unknown'

MIN_FILE_SIZE: int = 0

abstractmethod process() → ForcingResult

Process forcing data and generate output files.

abstractmethod find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

validate_file_size(path: Path, min_bytes: int = 0) → bool

QC: reject files smaller than threshold.

create_output_dir() → None

Create output directory if needed.

write_files_used(files: List[Path], output_dir: Path, source: str, phase: str) → Path

Write met_files_used log matching Fortran format.

Format: filepath on one line, then “YYYY MM DD CYC FHR” on the next. Output filename: met_files_used_{phase}_{cyc:02d}_{source}.dat

class nos_utils.forcing.ForcingResult(success: bool, source: str, output_files: List[Path] = <factory>, errors: List[str] = <factory>, warnings: List[str] = <factory>, metadata: Dict[str, ~typing.Any]=<factory>)

Bases: object

Result from forcing data processing.

success: bool

source: str

output_files: List[Path]

errors: List[str]

warnings: List[str]

metadata: Dict[str, Any]

class nos_utils.forcing.GFSProcessor(config: ForcingConfig, input_path: Path, output_path: Path, variables: List[str] | None = None, resolution: str = '0p25', extractor: GRIBExtractor | None = None, phase: str = 'nowcast', time_hotstart: datetime | None = None)

Bases: ForcingProcessor

GFS atmospheric forcing processor.

Extracts meteorological variables from GFS GRIB2 files and writes SCHISM-compatible sflux NetCDF files or DATM forcing for UFS-Coastal.

SOURCE_NAME: str = 'GFS'

MIN_FILE_SIZE_BY_RES = {'0p25': 400000000, '0p50': 40000000, 'sflux': 10000000}

GRIB2_VARIABLES = {'apcp': ('APCP', 'surface'), 'dlwrf': ('DLWRF', 'surface'), 'dswrf': ('DSWRF', 'surface'), 'evp': ('EVP', 'surface'), 'lhtfl': ('LHTFL', 'surface'), 'prate': ('PRATE', 'surface'), 'prmsl': ('PRMSL', 'mean sea level'), 'rh': ('RH', '2 m above ground'), 'shtfl': ('SHTFL', 'surface'), 'spfh': ('SPFH', '2 m above ground'), 'stmp': ('TMP', '2 m above ground'), 'tcdc': ('TCDC', 'entire atmosphere'), 'tdair': ('DPT', '2 m above ground'), 'ulwrf': ('ULWRF', 'surface'), 'uswrf': ('USWRF', 'surface'), 'uwind': ('UGRD', '10 m above ground'), 'vwind': ('VGRD', '10 m above ground'), 'wtmp': ('TMP', 'surface')}

DEFAULT_VARIABLES = ['uwind', 'vwind', 'prmsl', 'stmp', 'spfh', 'dlwrf', 'dswrf', 'prate']

MIN_FILE_SIZE: int = 40000000

property extractor: GRIBExtractor

process() → ForcingResult

Process GFS forcing data.

Pipeline: discover files -> extract GRIB2 -> filter to time window -> write sflux or DATM

find_input_files() → List[Path]

Config-driven GFS file discovery with multi-cycle fallback.

Searches multiple GFS cycles to cover the full nowcast+forecast window. If the primary list is incomplete, supplements with backup files.

class nos_utils.forcing.HRRRProcessor(config: ForcingConfig, input_path: Path, output_path: Path, variables: List[str] | None = None, regrid_dx: float = 0.03, extractor: GRIBExtractor | None = None, phase: str = 'nowcast', time_hotstart: datetime | None = None)

Bases: ForcingProcessor

HRRR atmospheric forcing processor (secondary source).

HRRR failure is always non-fatal — returns success=True with warnings. GFS provides primary coverage; HRRR is an enhancement where available.

SOURCE_NAME: str = 'HRRR'

MIN_FILE_SIZE: int = 0

MAX_FORECAST_HOURS = 48

GRIB2_VARIABLES = {'dlwrf': ('DLWRF', 'surface'), 'dswrf': ('DSWRF', 'surface'), 'prate': ('PRATE', 'surface'), 'prmsl': ('MSLMA', 'mean sea level'), 'spfh': ('SPFH', '2 m above ground'), 'stmp': ('TMP', '2 m above ground'), 'uwind': ('UGRD', '10 m above ground'), 'vwind': ('VGRD', '10 m above ground')}

DEFAULT_VARIABLES = ['uwind', 'vwind', 'prmsl', 'stmp', 'spfh', 'dlwrf', 'dswrf', 'prate']

property extractor: GRIBExtractor

process() → ForcingResult

Process HRRR forcing data.

HRRR is optional — all failures return success=True with warnings.

find_input_files() → List[Path]

Find HRRR GRIB2 files for nowcast + forecast window.

Nowcast: hourly analysis files (f01) from multiple cycles Forecast: hourly forecast files (f01-f48) from current cycle

class nos_utils.forcing.GEFSProcessor(config: ForcingConfig, input_path: Path, output_path: Path, member: str = 'c00', product: str = 'pgrb2sp25', resolution: str = '0p25', extractor: GRIBExtractor | None = None)

Bases: ForcingProcessor

GEFS ensemble atmospheric forcing processor.

Processes one ensemble member at a time. For a full ensemble run, create multiple GEFSProcessor instances with different member IDs.

SOURCE_NAME: str = 'GEFS'

MIN_FILE_SIZE: int = 5000000

GRIB2_VARIABLES = {'apcp': ('APCP', 'surface'), 'dlwrf': ('DLWRF', 'surface'), 'dswrf': ('DSWRF', 'surface'), 'pres': ('PRES', 'surface'), 'prmsl': ('PRMSL', 'mean sea level'), 'rh': ('RH', '2 m above ground'), 'stmp': ('TMP', '2 m above ground'), 'uwind': ('UGRD', '10 m above ground'), 'vwind': ('VGRD', '10 m above ground')}

EXTRACT_VARIABLES = ['uwind', 'vwind', 'prmsl', 'stmp', 'rh', 'apcp', 'pres', 'dlwrf', 'dswrf']

OUTPUT_VARIABLES = ['uwind', 'vwind', 'prmsl', 'stmp', 'spfh', 'prate', 'dlwrf', 'dswrf']

property extractor: GRIBExtractor

process() → ForcingResult

Process GEFS forcing data for one ensemble member.

Pipeline: discover -> extract GRIB2 -> convert RH->SPFH, APCP->PRATE -> write sflux

find_input_files() → List[Path]

3-hourly GEFS file discovery for a specific ensemble member.

Primary list (covering nowcast + forecast):

yest t06z f006 + yest t12z f003-f006 + yest t18z f003-f006 + today t00z f003-f006 + today t06z f003-f006 + today t12z f003-f{max_fhr}

Backup: yest t12z f003-f{max_fhr}

Target: ~53 files for 5.5-day run, minimum ~35.

static convert_rh_to_spfh(rh: numpy.ndarray, temp: numpy.ndarray, pres: numpy.ndarray) → numpy.ndarray

Convert relative humidity to specific humidity using Tetens formula.

es = 611.2 * exp(17.67 * (T - 273.15) / (T - 29.65)) [saturation vapor pressure, Pa] e = (RH / 100.0) * es [actual vapor pressure, Pa] SPFH = 0.622 * e / (P - 0.378 * e) [specific humidity, kg/kg]

Parameters:

• rh – Relative humidity (%)

• temp – Temperature (K)

• pres – Surface pressure (Pa)

Returns:

Specific humidity (kg/kg)

static convert_apcp_to_prate(apcp: numpy.ndarray, dt_seconds: float = 10800.0) → numpy.ndarray

Convert accumulated precipitation to precipitation rate.

PRATE = APCP / dt_seconds

Parameters:

• apcp – Accumulated precipitation (kg/m²) over dt_seconds

• dt_seconds – Accumulation period (default: 10800s = 3 hours for GEFS)

Returns:

Precipitation rate (kg/m²/s)

class nos_utils.forcing.RTOFSProcessor(config: ForcingConfig, input_path: Path, output_path: Path, grid_file: Path | None = None, obc_ctl_file: Path | None = None, vgrid_file: Path | None = None, phase: str | None = None, time_hotstart: datetime | None = None)

Bases: ForcingProcessor

RTOFS ocean boundary condition processor for SCHISM.

Interpolates RTOFS data to SCHISM open boundary nodes from hgrid.ll.

SOURCE_NAME: str = 'RTOFS'

MIN_FILE_SIZE_2D = 150000000

MIN_FILE_SIZE_3D = 200000000

property is_stofs_mode: bool

True if using STOFS-style ROI-based processing.

process() → ForcingResult

Process forcing data and generate output files.

find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

find_input_files_by_type() → Tuple[List[Path], List[Path]]

Find RTOFS 2D and 3D files, sorted by valid time and deduplicated.

class nos_utils.forcing.NWMProcessor(config: ForcingConfig, input_path: Path, output_path: Path, river_config: RiverConfig | None = None, phase: str | None = None, time_hotstart: datetime | None = None)

Bases: ForcingProcessor

NWM river forcing processor for SCHISM.

Extracts streamflow for configured river reaches from NWM output, maps to SCHISM mesh nodes, and creates vsource.th/msource.th files.

SOURCE_NAME: str = 'NWM'

MIN_FILE_SIZE: int = 0

PRODUCTS = ['analysis_assim', 'short_range', 'medium_range']

property river_config: RiverConfig | None

property is_stofs_mode: bool

True if using STOFS-style aggregated sources (from_sources_json).

process() → ForcingResult

Process NWM river forcing data.

Pipeline: load config → find NWM files → extract streamflow → write output Falls back to climatology if NWM data is unavailable.

find_input_files() → List[Path]

Find NWM channel_rt files for the run window.

class nos_utils.forcing.RiverConfig(feature_ids: List[int], node_indices: List[int], clim_flows: List[float], names: List[str] | None = None, feature_id_groups: List[List[int]] | None = None)

Bases: object

River configuration: maps NWM reach IDs to SCHISM mesh nodes.

classmethod from_text(filepath: Path) → RiverConfig

Load river config from text file.

Supports three formats:

Format 1 (NWM reach file — secofs.nwm.reach.dat):

REACH_ID FLAG (header) 2 (count) 20104159 1 (feature_id flag) 9643431 1

Format 2 (full river config):

feature_id node_index river_name clim_flow

Format 3 (Fortran river.ctl — secofs.river.ctl):

Section 1: USGS station info (Q_mean per station) Section 2: Grid node mappings (NODE_ID, Q_Scale, RiverID)

classmethod from_json(filepath: Path) → RiverConfig

Load river config from JSON file.

classmethod from_sources_json(filepath: Path) → RiverConfig

Load STOFS sources.json: {element_id: [fid1, fid2, …]}.

STOFS-3D-ATL uses VIMS gen_sourcesink.py format where each source element maps to one or more NWM feature_ids. Flow is summed across all feature_ids belonging to each source element.

Parameters:

filepath – Path to sources.json (e.g. stofs_3d_atl_river_sources_conus.json)

class nos_utils.forcing.RiverClimProcessor(config: ForcingConfig, input_path: Path, output_path: Path, river_ctl_path: Path | None = None, clim_path: Path | None = None, phase: str = 'nowcast', time_hotstart: datetime | None = None)

Bases: ForcingProcessor

USGS climatology-based river forcing processor.

For OFS systems where NWM and real-time USGS data are unavailable, generates river forcing from the daily climatology file.

SOURCE_NAME: str = 'RIVER_CLIM'

find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

process() → ForcingResult

Generate river forcing from USGS climatology.

class nos_utils.forcing.TidalProcessor(config: ForcingConfig, input_path: Path, output_path: Path, phase: str = 'nowcast', time_hotstart: datetime | None = None)

Bases: ForcingProcessor

Tidal forcing processor for SCHISM.

Generates bctides.in with harmonic constituents and nodal corrections.

SOURCE_NAME: str = 'TIDAL'

process() → ForcingResult

Generate bctides.in tidal boundary file.

Tries in order: 0. Fortran mode: call tide_fac executable with template (most accurate) 1. Template mode: update bctides.in_template with Python nodal corrections 2. Copy mode: copy static bctides.in from input_path 3. Python mode: generate minimal bctides.in from scratch

find_input_files() → List[Path]

Find bctides template or static files.

nos_utils.forcing.compute_nodal_corrections(start_time: datetime, constituents: List[str], run_days: float = 0.25) → Dict[str, Dict[str, float]]

Compute tidal nodal factors and equilibrium arguments (V0+u).

Exact port of the NOAA/SCHISM Fortran tide_fac program (Schureman 1958). Nodal factors computed at mid-run, equilibrium arguments at start.

Parameters:

• start_time – Model start datetime

• constituents – List of constituent names

• run_days – Run length in days (nodal factors computed at midpoint)

Returns:

nodal_factor, “v0_plus_u”: deg}

Return type:

Dict mapping constituent name -> {“f”

class nos_utils.forcing.NudgingProcessor(config: ForcingConfig, input_path: Path, output_path: Path, nudge_weight_file: Path | None = None, rtofs_input_path: Path | None = None)

Bases: ForcingProcessor

Generate T/S interior nudging fields from RTOFS data.

Nudging relaxes interior model T/S toward observed values with a configurable timescale. Only nodes identified in the nudging weight files (TEM_nudge.gr3, SAL_nudge.gr3) receive nudging.

SOURCE_NAME: str = 'NUDGING'

property is_stofs_mode: bool

True if using STOFS-style ROI-based nudging with Fortran exe.

process() → ForcingResult

Generate nudging fields.

find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

class nos_utils.forcing.ParamNmlProcessor(config: ForcingConfig, input_path: Path, output_path: Path, template_name: str = 'param.nml', phase: str = 'nowcast', time_hotstart: datetime | None = None)

Bases: ForcingProcessor

Generate SCHISM param.nml from template with runtime substitutions.

Usage:

proc = ParamNmlProcessor(config, template_path, output_path) result = proc.process()

SOURCE_NAME: str = 'PARAM_NML'

process() → ForcingResult

Generate param.nml from template.

find_input_files()

Discover input GRIB2 files for the run window.

static patch_param(param_path: Path, **kwargs) → None

Patch specific values in an existing param.nml file.

Usage:

ParamNmlProcessor.patch_param(path, rnday=2.0, ihot=1, nws=4)

class nos_utils.forcing.HotstartProcessor(config: ForcingConfig, input_path: Path, output_path: Path, run_name: str = 'secofs', max_lookback_days: int = 3)

Bases: ForcingProcessor

Find and validate SCHISM hotstart files.

Searches for hotstart.nc from previous cycles, extracts timing info, and copies/links to the working directory.

SOURCE_NAME: str = 'HOTSTART'

MIN_HOTSTART_SIZE = 1000

HOTSTART_PATTERNS = ['hotstart.nc', 'hotstart_it=*.nc', '{run}.hotstart.nc', '{run}.t{cyc:02d}z.hotstart.nc']

process() → ForcingResult

Find and validate hotstart file.

Returns HotstartInfo in result.metadata[“hotstart_info”].

find_input_files() → List[Path]

Find all candidate hotstart files.

class nos_utils.forcing.HotstartInfo(filepath: Path, time_seconds: float, iths: int, n_nodes: int, n_levels: int)

Bases: object

Information extracted from a SCHISM hotstart.nc file.

property time_days: float

class nos_utils.forcing.PartitionProcessor(config: ForcingConfig, input_path: Path, output_path: Path, nprocs: int = 1, grid_file: Path | None = None, method: str = 'round_robin')

Bases: ForcingProcessor

Generate partition.prop for SCHISM MPI domain decomposition.

SOURCE_NAME: str = 'PARTITION'

process() → ForcingResult

Generate partition.prop.

find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

class nos_utils.forcing.ESMFMeshProcessor(config: ForcingConfig, input_path: Path, output_path: Path, forcing_file: Path | None = None)

Bases: ForcingProcessor

Generate ESMF mesh file from a regular lat/lon forcing grid.

The mesh file defines the spatial grid for CDEPS DATM component in UFS-Coastal coupled simulations.

SOURCE_NAME: str = 'ESMF_MESH'

process() → ForcingResult

Generate ESMF mesh from forcing grid or config domain.

find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

class nos_utils.forcing.BlenderProcessor(config: ForcingConfig, input_path: Path, output_path: Path, target_dx: float = 0.025)

Bases: ForcingProcessor

Blend HRRR + GFS sflux files into a single DATM forcing file.

Uses Delaunay triangulation for HRRR (Lambert Conformal) to regular lat/lon interpolation. HRRR overrides GFS where coverage exists.

SOURCE_NAME: str = 'BLENDER'

process() → ForcingResult

Blend GFS + HRRR sflux files into datm_forcing.nc.

find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

class nos_utils.forcing.SfluxWriter(output_dir: Path, source_index: int = 1, single_file: bool = True)

Bases: object

Creates SCHISM sflux NetCDF files for any atmospheric source.

Usage:

writer = SfluxWriter(output_dir=Path(“/data/sflux”), source_index=1) files = writer.write_all(data, times, lons, lats, base_date) writer.write_sflux_inputs(met_num=2)

write_all(data: Dict[str, List[numpy.ndarray]], times: List[datetime], lons: numpy.ndarray, lats: numpy.ndarray, base_date: datetime) → List[Path]

Write all sflux files.

In single_file mode (default, COMF convention): writes all timesteps into sflux_{type}_{source}.1.nc — one file per type.

In multi-file mode: splits by calendar day into .1.nc, .2.nc, etc.

Parameters:

• data – Dict mapping variable names to lists of 2D arrays (one per time step)

• times – List of datetime objects (one per time step)

• lons – 1D longitude array

• lats – 1D latitude array

• base_date – Reference datetime for time axis

Returns:

List of created file paths

write_day(data: Dict[str, List[numpy.ndarray]], times: List[datetime], lons: numpy.ndarray, lats: numpy.ndarray, base_date: datetime, day_num: int) → List[Path]

Write sflux_air, sflux_rad, sflux_prc for a single day.

Returns:

List of created file paths (up to 3)

write_sflux_inputs(met_num: int = 1) → Path

Write sflux_inputs.txt namelist for SCHISM.

In single_file mode (COMF convention): writes minimal defaults &sflux_inputs followed by /. SCHISM applies the same defaults internally, so the result is functionally identical.

In multi-file mode: writes explicit weight/window parameters (legacy behavior).

Parameters:

met_num – Number of met sources (1 or 2)

Returns:

Path to sflux_inputs.txt

class nos_utils.forcing.DATMWriter

Bases: object

Creates datm_forcing.nc for UFS-Coastal CDEPS DATM component.

Usage:

writer = DATMWriter() path = writer.write(data, times, lons, lats, output_path)

# With GFS+HRRR blending: path = writer.write_blended(gfs_data, hrrr_data, …, output_path)

write(data: Dict[str, List[numpy.ndarray]], times: List[datetime], lons: numpy.ndarray, lats: numpy.ndarray, output_path: Path, epoch: datetime | None = None) → Path

Write datm_forcing.nc from extracted arrays (single source, no blending).

Parameters:

• data – Dict mapping sflux variable names to lists of 2D arrays

• times – List of datetime objects

• lons – 1D longitude array

• lats – 1D latitude array

• output_path – Path for output file

• epoch – Reference datetime for time axis (default: 1970-01-01)

Returns:

Path to created file

write_blended(gfs_data: Dict[str, List[numpy.ndarray]], hrrr_data: Dict[str, List[numpy.ndarray]] | None, gfs_times: List[datetime], hrrr_times: List[datetime] | None, target_lons: numpy.ndarray, target_lats: numpy.ndarray, hrrr_lons_2d: numpy.ndarray | None, hrrr_lats_2d: numpy.ndarray | None, output_path: Path, epoch: datetime | None = None) → Path

Write datm_forcing.nc with GFS+HRRR blending.

HRRR overrides GFS where HRRR has spatial coverage. Uses Delaunay triangulation for HRRR LCC -> regular grid interpolation.

Parameters:

• gfs_data – GFS extracted data (on regular lat/lon grid)

• hrrr_data – HRRR extracted data (on LCC grid, or None)

• gfs_times – GFS time steps

• hrrr_times – HRRR time steps (or None)

• target_lons – 1D target longitude array

• target_lats – 1D target latitude array

• hrrr_lons_2d – 2D HRRR longitude array (or None)

• hrrr_lats_2d – 2D HRRR latitude array (or None)

• output_path – Path for output file

• epoch – Reference datetime

Returns:

Path to created file

static get_grid_dims(lons: numpy.ndarray, lats: numpy.ndarray) → Tuple[int, int]

Return (nx_global, ny_global) for datm_in configuration.

CRITICAL: These must match the actual file dimensions (lesson #19).

Base classes

Base classes for forcing data processors.

All forcing processors (GFS, HRRR, GEFS) inherit from ForcingProcessor and implement the process() and find_input_files() methods.

class nos_utils.forcing.base.ForcingResult(success: bool, source: str, output_files: List[Path] = <factory>, errors: List[str] = <factory>, warnings: List[str] = <factory>, metadata: Dict[str, ~typing.Any]=<factory>)

Bases: object

Result from forcing data processing.

success: bool

source: str

output_files: List[Path]

errors: List[str]

warnings: List[str]

metadata: Dict[str, Any]

class nos_utils.forcing.base.ForcingProcessor(config: ForcingConfig, input_path: Path, output_path: Path)

Bases: ABC

Abstract base class for atmospheric forcing processors.

Subclasses implement source-specific file discovery, GRIB2 extraction, variable conversion, and output writing.

SOURCE_NAME: str = 'Unknown'

MIN_FILE_SIZE: int = 0

abstractmethod process() → ForcingResult

Process forcing data and generate output files.

abstractmethod find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

validate_file_size(path: Path, min_bytes: int = 0) → bool

QC: reject files smaller than threshold.

create_output_dir() → None

Create output directory if needed.

write_files_used(files: List[Path], output_dir: Path, source: str, phase: str) → Path

Write met_files_used log matching Fortran format.

Format: filepath on one line, then “YYYY MM DD CYC FHR” on the next. Output filename: met_files_used_{phase}_{cyc:02d}_{source}.dat

Atmospheric

GFS

GFS (Global Forecast System) forcing processor.

Processes GFS 0.25° GRIB2 data and creates SCHISM sflux or DATM forcing files.

Input: GFS GRIB2 files from COMINgfs

Pattern: gfs.YYYYMMDD/HH/atmos/gfs.tHHz.pgrb2.0p25.fHHH

gfs.YYYYMMDD/HH/atmos/gfs.tHHz.sfluxgrbfHHH.grib2

Resolution: “0p25” (hourly, ~500MB), “0p50” (3-hourly, ~60MB),

“sflux” (hourly, ~30MB — surface fields only, native ~0.25°)

Output:

sflux mode (nws=2): sflux_air_1.N.nc, sflux_rad_1.N.nc, sflux_prc_1.N.nc DATM mode (nws=4): datm_forcing.nc

class nos_utils.forcing.gfs.GFSProcessor(config: ForcingConfig, input_path: Path, output_path: Path, variables: List[str] | None = None, resolution: str = '0p25', extractor: GRIBExtractor | None = None, phase: str = 'nowcast', time_hotstart: datetime | None = None)

Bases: ForcingProcessor

GFS atmospheric forcing processor.

Extracts meteorological variables from GFS GRIB2 files and writes SCHISM-compatible sflux NetCDF files or DATM forcing for UFS-Coastal.

SOURCE_NAME: str = 'GFS'

MIN_FILE_SIZE_BY_RES = {'0p25': 400000000, '0p50': 40000000, 'sflux': 10000000}

GRIB2_VARIABLES = {'apcp': ('APCP', 'surface'), 'dlwrf': ('DLWRF', 'surface'), 'dswrf': ('DSWRF', 'surface'), 'evp': ('EVP', 'surface'), 'lhtfl': ('LHTFL', 'surface'), 'prate': ('PRATE', 'surface'), 'prmsl': ('PRMSL', 'mean sea level'), 'rh': ('RH', '2 m above ground'), 'shtfl': ('SHTFL', 'surface'), 'spfh': ('SPFH', '2 m above ground'), 'stmp': ('TMP', '2 m above ground'), 'tcdc': ('TCDC', 'entire atmosphere'), 'tdair': ('DPT', '2 m above ground'), 'ulwrf': ('ULWRF', 'surface'), 'uswrf': ('USWRF', 'surface'), 'uwind': ('UGRD', '10 m above ground'), 'vwind': ('VGRD', '10 m above ground'), 'wtmp': ('TMP', 'surface')}

DEFAULT_VARIABLES = ['uwind', 'vwind', 'prmsl', 'stmp', 'spfh', 'dlwrf', 'dswrf', 'prate']

MIN_FILE_SIZE: int = 40000000

property extractor: GRIBExtractor

process() → ForcingResult

Process GFS forcing data.

Pipeline: discover files -> extract GRIB2 -> filter to time window -> write sflux or DATM

find_input_files() → List[Path]

Config-driven GFS file discovery with multi-cycle fallback.

Searches multiple GFS cycles to cover the full nowcast+forecast window. If the primary list is incomplete, supplements with backup files.

HRRR

HRRR (High-Resolution Rapid Refresh) forcing processor.

Processes HRRR 3km GRIB2 data as a secondary atmospheric forcing source. HRRR provides higher resolution over CONUS but has limited domain and forecast length.

Input: HRRR GRIB2 files from COMINhrrr

Pattern: hrrr.YYYYMMDD/conus/hrrr.tHHz.wrfsfcfHH.grib2 Resolution: 3km (Lambert Conformal projection) Max forecast: 48 hours

Key differences from GFS:

• Uses MSLMA instead of PRMSL for sea level pressure

• Lambert Conformal grid — MUST regrid to regular lat/lon (lesson #14)

• Optional source — failure is non-fatal

• Writes to source_index=2 (secondary) for SCHISM sflux blending

Output:

sflux mode: sflux_air_2.N.nc, sflux_rad_2.N.nc, sflux_prc_2.N.nc

class nos_utils.forcing.hrrr.HRRRProcessor(config: ForcingConfig, input_path: Path, output_path: Path, variables: List[str] | None = None, regrid_dx: float = 0.03, extractor: GRIBExtractor | None = None, phase: str = 'nowcast', time_hotstart: datetime | None = None)

Bases: ForcingProcessor

HRRR atmospheric forcing processor (secondary source).

HRRR failure is always non-fatal — returns success=True with warnings. GFS provides primary coverage; HRRR is an enhancement where available.

SOURCE_NAME: str = 'HRRR'

MIN_FILE_SIZE: int = 0

MAX_FORECAST_HOURS = 48

GRIB2_VARIABLES = {'dlwrf': ('DLWRF', 'surface'), 'dswrf': ('DSWRF', 'surface'), 'prate': ('PRATE', 'surface'), 'prmsl': ('MSLMA', 'mean sea level'), 'spfh': ('SPFH', '2 m above ground'), 'stmp': ('TMP', '2 m above ground'), 'uwind': ('UGRD', '10 m above ground'), 'vwind': ('VGRD', '10 m above ground')}

DEFAULT_VARIABLES = ['uwind', 'vwind', 'prmsl', 'stmp', 'spfh', 'dlwrf', 'dswrf', 'prate']

property extractor: GRIBExtractor

process() → ForcingResult

Process HRRR forcing data.

HRRR is optional — all failures return success=True with warnings.

find_input_files() → List[Path]

Find HRRR GRIB2 files for nowcast + forecast window.

Nowcast: hourly analysis files (f01) from multiple cycles Forecast: hourly forecast files (f01-f48) from current cycle

GEFS

GEFS (Global Ensemble Forecast System) forcing processor.

Processes GEFS GRIB2 data for ensemble atmospheric forcing. Each ensemble member is processed independently, producing its own set of sflux files.

Input: GEFS GRIB2 files from COMINgefs

Pattern: gefs.YYYYMMDD/HH/atmos/{product}/{prefix}.tHHz.{file_product}.{res}.fHHH prefix: gec00 (control), gep01-gep30 (perturbation) Resolution: 0.25° (pgrb2sp25) or 0.50° (pgrb2ap5) Temporal: 3-hourly (f003, f006, …, f132) Min file size: 5 MB

Key differences from GFS:

• 3-hourly (not hourly)

• Has RH instead of SPFH -> requires RH->SPFH conversion (Tetens formula)

• Has APCP instead of PRATE -> requires APCP/10800 conversion

• Needs PRES:surface for RH->SPFH conversion

• Ensemble: 30 perturbation + 1 control member

class nos_utils.forcing.gefs.GEFSProcessor(config: ForcingConfig, input_path: Path, output_path: Path, member: str = 'c00', product: str = 'pgrb2sp25', resolution: str = '0p25', extractor: GRIBExtractor | None = None)

Bases: ForcingProcessor

GEFS ensemble atmospheric forcing processor.

Processes one ensemble member at a time. For a full ensemble run, create multiple GEFSProcessor instances with different member IDs.

SOURCE_NAME: str = 'GEFS'

MIN_FILE_SIZE: int = 5000000

GRIB2_VARIABLES = {'apcp': ('APCP', 'surface'), 'dlwrf': ('DLWRF', 'surface'), 'dswrf': ('DSWRF', 'surface'), 'pres': ('PRES', 'surface'), 'prmsl': ('PRMSL', 'mean sea level'), 'rh': ('RH', '2 m above ground'), 'stmp': ('TMP', '2 m above ground'), 'uwind': ('UGRD', '10 m above ground'), 'vwind': ('VGRD', '10 m above ground')}

EXTRACT_VARIABLES = ['uwind', 'vwind', 'prmsl', 'stmp', 'rh', 'apcp', 'pres', 'dlwrf', 'dswrf']

OUTPUT_VARIABLES = ['uwind', 'vwind', 'prmsl', 'stmp', 'spfh', 'prate', 'dlwrf', 'dswrf']

property extractor: GRIBExtractor

process() → ForcingResult

Process GEFS forcing data for one ensemble member.

Pipeline: discover -> extract GRIB2 -> convert RH->SPFH, APCP->PRATE -> write sflux

find_input_files() → List[Path]

3-hourly GEFS file discovery for a specific ensemble member.

Primary list (covering nowcast + forecast):

yest t06z f006 + yest t12z f003-f006 + yest t18z f003-f006 + today t00z f003-f006 + today t06z f003-f006 + today t12z f003-f{max_fhr}

Backup: yest t12z f003-f{max_fhr}

Target: ~53 files for 5.5-day run, minimum ~35.

static convert_rh_to_spfh(rh: numpy.ndarray, temp: numpy.ndarray, pres: numpy.ndarray) → numpy.ndarray

Convert relative humidity to specific humidity using Tetens formula.

es = 611.2 * exp(17.67 * (T - 273.15) / (T - 29.65)) [saturation vapor pressure, Pa] e = (RH / 100.0) * es [actual vapor pressure, Pa] SPFH = 0.622 * e / (P - 0.378 * e) [specific humidity, kg/kg]

Parameters:

• rh – Relative humidity (%)

• temp – Temperature (K)

• pres – Surface pressure (Pa)

Returns:

Specific humidity (kg/kg)

static convert_apcp_to_prate(apcp: numpy.ndarray, dt_seconds: float = 10800.0) → numpy.ndarray

Convert accumulated precipitation to precipitation rate.

PRATE = APCP / dt_seconds

Parameters:

• apcp – Accumulated precipitation (kg/m²) over dt_seconds

• dt_seconds – Accumulation period (default: 10800s = 3 hours for GEFS)

Returns:

Precipitation rate (kg/m²/s)

Ocean boundary

RTOFS

RTOFS ocean boundary condition processor.

Creates SCHISM boundary time-history files by interpolating RTOFS global ocean model output to SCHISM open boundary nodes.

Output:

• elev2D.th.nc — SSH at boundary nodes (time, nOpenBndNodes)

• TEM_3D.th.nc — Temperature at boundary nodes (time, nOpenBndNodes, nLevels)

• SAL_3D.th.nc — Salinity at boundary nodes (time, nOpenBndNodes, nLevels)

• uv3D.th.nc — Velocity at boundary nodes (time, nOpenBndNodes, nLevels, 2)

Reads SCHISM boundary node locations from hgrid.ll (or obc.ctl for exact Fortran-matching node list) and interpolates RTOFS data to those nodes.

IMPORTANT: SSH bias correction

The Fortran gen_3Dth_from_hycom applies a station-based bias correction:

1. Reads real-time tide gauge observations (NOSBUFR)

2. Computes AVGERR = mean(obs - RTOFS) per station (~1.25m for SECOFS)

3. Applies WLOBC += weight * (AVGERR + obs_subtidal) per boundary node

This Python processor does NOT apply this correction — the ~1.25m SSH offset relative to Fortran output is expected. For production runs, use hybrid mode (Fortran OBC) until Python has access to real-time tide gauge data.

Works with any SCHISM-based OFS (SECOFS, STOFS-3D-ATL, CREOFS, etc.)

class nos_utils.forcing.rtofs.RTOFSProcessor(config: ForcingConfig, input_path: Path, output_path: Path, grid_file: Path | None = None, obc_ctl_file: Path | None = None, vgrid_file: Path | None = None, phase: str | None = None, time_hotstart: datetime | None = None)

Bases: ForcingProcessor

RTOFS ocean boundary condition processor for SCHISM.

Interpolates RTOFS data to SCHISM open boundary nodes from hgrid.ll.

SOURCE_NAME: str = 'RTOFS'

MIN_FILE_SIZE_2D = 150000000

MIN_FILE_SIZE_3D = 200000000

property is_stofs_mode: bool

True if using STOFS-style ROI-based processing.

process() → ForcingResult

Process forcing data and generate output files.

find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

find_input_files_by_type() → Tuple[List[Path], List[Path]]

Find RTOFS 2D and 3D files, sorted by valid time and deduplicated.

ADT satellite SSH blending

ADT (Absolute Dynamic Topography) satellite SSH blender.

Blends CMEMS satellite ADT observations with RTOFS SSH to improve boundary condition accuracy for STOFS-3D-ATL.

The core formula (from stofs_3d_atl_create_obc_3d_th.sh lines 580-610):

SSH_final = SSH_rtofs - SSH_rtofs(t=0) + ADT(t=0)

This removes the RTOFS bias at t=0 and replaces it with the satellite-observed absolute dynamic topography, preserving RTOFS temporal variability.

Input:

• SSH_1.nc — RTOFS SSH prepared by RTOFSProcessor._stofs_prepare_ssh()

• CMEMS ADT: nrt_global_allsat_phy_l4_YYYYMMDD_YYYYMMDD.nc

• Weight file: stofs_3d_atl_adt_weight.nc (regridding weights)

Output:

• SSH_1.nc updated with ADT-blended surf_el values

Graceful fallback: returns None if ADT data is unavailable (RTOFS-only SSH used).

class nos_utils.forcing.adt.ADTBlender(config: ForcingConfig, input_path: Path)

Bases: object

Blend CMEMS ADT satellite SSH with RTOFS SSH.

blend_ssh(ssh_path: Path, work_dir: Path) → Path | None

Blend ADT into RTOFS SSH_1.nc.

Parameters:

• ssh_path – Path to SSH_1.nc (RTOFS-only)

• work_dir – Working directory for intermediate files

Returns:

Path to updated SSH_1.nc with ADT blending, or None if ADT unavailable.

River

NWM

NWM (National Water Model) river forcing processor.

Creates SCHISM river forcing files from NWM streamflow data:

• vsource.th — Volume source time history (m³/s per river)

• msource.th — Mass source (temperature, salinity per river)

• source_sink.in — SCHISM source/sink node configuration

Input: NWM NetCDF channel_rt files from COMINnwm

Pattern: nwm.YYYYMMDD/nwm.tHHz.{product}.channel_rt.f*.conus.nc Products: analysis_assim, short_range, medium_range

Fallback: Monthly climatology when NWM data is unavailable.

class nos_utils.forcing.nwm.RiverConfig(feature_ids: List[int], node_indices: List[int], clim_flows: List[float], names: List[str] | None = None, feature_id_groups: List[List[int]] | None = None)

Bases: object

River configuration: maps NWM reach IDs to SCHISM mesh nodes.

classmethod from_text(filepath: Path) → RiverConfig

Load river config from text file.

Supports three formats:

Format 1 (NWM reach file — secofs.nwm.reach.dat):

REACH_ID FLAG (header) 2 (count) 20104159 1 (feature_id flag) 9643431 1

Format 2 (full river config):

feature_id node_index river_name clim_flow

Format 3 (Fortran river.ctl — secofs.river.ctl):

Section 1: USGS station info (Q_mean per station) Section 2: Grid node mappings (NODE_ID, Q_Scale, RiverID)

classmethod from_json(filepath: Path) → RiverConfig

Load river config from JSON file.

classmethod from_sources_json(filepath: Path) → RiverConfig

Load STOFS sources.json: {element_id: [fid1, fid2, …]}.

STOFS-3D-ATL uses VIMS gen_sourcesink.py format where each source element maps to one or more NWM feature_ids. Flow is summed across all feature_ids belonging to each source element.

Parameters:

filepath – Path to sources.json (e.g. stofs_3d_atl_river_sources_conus.json)

class nos_utils.forcing.nwm.NWMProcessor(config: ForcingConfig, input_path: Path, output_path: Path, river_config: RiverConfig | None = None, phase: str | None = None, time_hotstart: datetime | None = None)

Bases: ForcingProcessor

NWM river forcing processor for SCHISM.

Extracts streamflow for configured river reaches from NWM output, maps to SCHISM mesh nodes, and creates vsource.th/msource.th files.

SOURCE_NAME: str = 'NWM'

MIN_FILE_SIZE: int = 0

PRODUCTS = ['analysis_assim', 'short_range', 'medium_range']

property river_config: RiverConfig | None

property is_stofs_mode: bool

True if using STOFS-style aggregated sources (from_sources_json).

process() → ForcingResult

Process NWM river forcing data.

Pipeline: load config → find NWM files → extract streamflow → write output Falls back to climatology if NWM data is unavailable.

find_input_files() → List[Path]

Find NWM channel_rt files for the run window.

River climatology

USGS climatology-based river forcing processor.

For OFS systems (like SECOFS) where NWM has no matching reaches and real-time USGS BUFR data is unavailable, river forcing is generated entirely from the USGS daily climatology file (nosofs.river.clim.usgs.nc).

Reads:

• {OFS}.river.ctl — River node mappings, station IDs, scaling factors

• nosofs.river.clim.usgs.nc — Daily climatological discharge, temperature, salinity

Writes:

• vsource.th — Volume source (hourly, positive m³/s per node)

• msource.th — Temperature + salinity per node (hourly)

• source_sink.in — SCHISM source/sink node configuration

• schism_flux.th — Volume flux at model dt (negative = inflow)

• schism_temp.th — River temperature at model dt

• schism_salt.th — River salinity at model dt

• Tar archive of schism_flux/temp/salt.th

class nos_utils.forcing.river_clim.RiverCTL(n_nodes: int, n_stations: int, dt_hours: float, station_ids: List[str], q_flags: List[int], q_means: List[float], t_means: List[float], node_ids: List[int], river_id_q: List[int], q_scales: List[float], river_id_t: List[int], t_scales: List[float], river_flags: List[int], river_names: List[str])

Bases: object

Parsed river control file.

n_nodes: int

n_stations: int

dt_hours: float

station_ids: List[str]

q_flags: List[int]

q_means: List[float]

t_means: List[float]

node_ids: List[int]

river_id_q: List[int]

q_scales: List[float]

river_id_t: List[int]

t_scales: List[float]

river_flags: List[int]

river_names: List[str]

nos_utils.forcing.river_clim.parse_river_ctl(filepath: Path) → RiverCTL

Parse COMF river control file.

Format:

Section 1: USGS station definitions

N_NODES N_STATIONS DELT Header line StationID … Q_Flag TS_Flag “Name”

Section 2: Grid node mappings

Header line GRID_ID NODE_ID ELE_ID DIR FLAG RiverID_Q Q_Scale RiverID_T T_Scale “Name”

nos_utils.forcing.river_clim.load_usgs_climatology(clim_path: Path, station_id: str) → Tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray, numpy.ndarray, numpy.ndarray]

Load daily climatological Q, T, S for a USGS station.

Returns:

(months, days, discharge, temperature, salinity) — each shape (366,)

nos_utils.forcing.river_clim.interp_clim_to_times(months: numpy.ndarray, days: numpy.ndarray, values: numpy.ndarray, times_dt: List[datetime]) → numpy.ndarray

Interpolate daily climatology to arbitrary datetimes.

Linear interpolation between the two surrounding daily clim values. Wraps around Dec 31 -> Jan 1.

class nos_utils.forcing.river_clim.RiverClimProcessor(config: ForcingConfig, input_path: Path, output_path: Path, river_ctl_path: Path | None = None, clim_path: Path | None = None, phase: str = 'nowcast', time_hotstart: datetime | None = None)

Bases: ForcingProcessor

USGS climatology-based river forcing processor.

For OFS systems where NWM and real-time USGS data are unavailable, generates river forcing from the daily climatology file.

SOURCE_NAME: str = 'RIVER_CLIM'

find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

process() → ForcingResult

Generate river forcing from USGS climatology.

Tidal

Tidal forcing processor.

Generates SCHISM tidal boundary condition file (bctides.in) with:

• 8 major tidal constituents (M2, S2, N2, K2, K1, O1, P1, Q1)

• Nodal corrections (f, u) computed for the model start time

• Accounts for the 18.6-year lunar nodal cycle

Three modes:

1. Template-based: Update pre-computed bctides.in_template with new start time

2. Python-native: Compute nodal corrections and generate bctides.in from scratch

3. Copy-only: Use static bctides.in from FIX directory

Output: bctides.in

class nos_utils.forcing.tidal.TidalProcessor(config: ForcingConfig, input_path: Path, output_path: Path, phase: str = 'nowcast', time_hotstart: datetime | None = None)

Bases: ForcingProcessor

Tidal forcing processor for SCHISM.

Generates bctides.in with harmonic constituents and nodal corrections.

SOURCE_NAME: str = 'TIDAL'

process() → ForcingResult

Generate bctides.in tidal boundary file.

Tries in order: 0. Fortran mode: call tide_fac executable with template (most accurate) 1. Template mode: update bctides.in_template with Python nodal corrections 2. Copy mode: copy static bctides.in from input_path 3. Python mode: generate minimal bctides.in from scratch

find_input_files() → List[Path]

Find bctides template or static files.

nos_utils.forcing.tidal.compute_nodal_corrections(start_time: datetime, constituents: List[str], run_days: float = 0.25) → Dict[str, Dict[str, float]]

Compute tidal nodal factors and equilibrium arguments (V0+u).

Exact port of the NOAA/SCHISM Fortran tide_fac program (Schureman 1958). Nodal factors computed at mid-run, equilibrium arguments at start.

Parameters:

• start_time – Model start datetime

• constituents – List of constituent names

• run_days – Run length in days (nodal factors computed at midpoint)

Returns:

nodal_factor, “v0_plus_u”: deg}

Return type:

Dict mapping constituent name -> {“f”

Nudging

T/S interior nudging field generator.

Creates SAL_nu.nc and TEM_nu.nc from RTOFS 3D temperature/salinity fields. These files apply interior relaxation (nudging) toward observed T/S values at specified nodes with a configurable timescale.

Input: RTOFS 3D NetCDF files (temperature, salinity on depth levels)

Output:

• TEM_nu.nc — temperature nudging field

• SAL_nu.nc — salinity nudging field

Replaces: nudging portion of nos_ofs_create_forcing_obc / gen_hycom_3Dth_nudge.py

class nos_utils.forcing.nudging.NudgingProcessor(config: ForcingConfig, input_path: Path, output_path: Path, nudge_weight_file: Path | None = None, rtofs_input_path: Path | None = None)

Bases: ForcingProcessor

Generate T/S interior nudging fields from RTOFS data.

Nudging relaxes interior model T/S toward observed values with a configurable timescale. Only nodes identified in the nudging weight files (TEM_nudge.gr3, SAL_nudge.gr3) receive nudging.

SOURCE_NAME: str = 'NUDGING'

property is_stofs_mode: bool

True if using STOFS-style ROI-based nudging with Fortran exe.

process() → ForcingResult

Generate nudging fields.

find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

Model configuration

param.nml

SCHISM param.nml generator.

Reads a param.nml template and substitutes runtime parameters:

• rnday: run duration in days

• start_year/month/day/hour: model start time

• ihot: hotstart mode (0=cold, 1=hotstart with time reset)

• nws: atmospheric forcing mode (0=none, 2=sflux, 4=DATM/UFS)

Replaces: nos_ofs_prep_schism_ctl.sh (sed-based substitution)

Template placeholders (Fortran namelist format):

rnday = rnday_value start_year = start_year_value start_month = start_month_value start_day = start_day_value start_hour = start_hour_value

class nos_utils.forcing.param_nml.ParamNmlProcessor(config: ForcingConfig, input_path: Path, output_path: Path, template_name: str = 'param.nml', phase: str = 'nowcast', time_hotstart: datetime | None = None)

Bases: ForcingProcessor

Generate SCHISM param.nml from template with runtime substitutions.

Usage:

proc = ParamNmlProcessor(config, template_path, output_path) result = proc.process()

SOURCE_NAME: str = 'PARAM_NML'

process() → ForcingResult

Generate param.nml from template.

find_input_files()

Discover input GRIB2 files for the run window.

static patch_param(param_path: Path, **kwargs) → None

Patch specific values in an existing param.nml file.

Usage:

ParamNmlProcessor.patch_param(path, rnday=2.0, ihot=1, nws=4)

Hotstart

SCHISM hotstart/restart file processor.

Reads SCHISM hotstart.nc to extract:

• Model time (for computing rnday and time_hotstart)

• Time step counter (iths)

• Basic validation (file size, key variables present)

Searches for hotstart files from previous cycles with automatic date fallback.

Replaces: nos_ofs_read_restart (Fortran executable)

Key SCHISM hotstart.nc variables:

time — scalar: model time in seconds iths — scalar: time step counter eta2 — [node]: surface elevation tr_nd — [node, nVert, ntracers]: tracers at nodes

class nos_utils.forcing.hotstart.HotstartInfo(filepath: Path, time_seconds: float, iths: int, n_nodes: int, n_levels: int)

Bases: object

Information extracted from a SCHISM hotstart.nc file.

property time_days: float

class nos_utils.forcing.hotstart.HotstartProcessor(config: ForcingConfig, input_path: Path, output_path: Path, run_name: str = 'secofs', max_lookback_days: int = 3)

Bases: ForcingProcessor

Find and validate SCHISM hotstart files.

Searches for hotstart.nc from previous cycles, extracts timing info, and copies/links to the working directory.

SOURCE_NAME: str = 'HOTSTART'

MIN_HOTSTART_SIZE = 1000

HOTSTART_PATTERNS = ['hotstart.nc', 'hotstart_it=*.nc', '{run}.hotstart.nc', '{run}.t{cyc:02d}z.hotstart.nc']

process() → ForcingResult

Find and validate hotstart file.

Returns HotstartInfo in result.metadata[“hotstart_info”].

find_input_files() → List[Path]

Find all candidate hotstart files.

Partition

SCHISM domain partition generator.

Creates partition.prop assigning mesh elements to MPI compute ranks.

Two methods:

1. Round-robin (default): Simple element_id % nprocs assignment

2. Contiguous blocks: Sequential block assignment

Input: hgrid.gr3 (SCHISM grid file) — reads n_elements from line 2 Output: partition.prop (one rank per line, n_elements lines)

Note: Production uses ParMETIS for load-balanced partitioning. Round-robin is adequate for testing and containers.

class nos_utils.forcing.partition.PartitionProcessor(config: ForcingConfig, input_path: Path, output_path: Path, nprocs: int = 1, grid_file: Path | None = None, method: str = 'round_robin')

Bases: ForcingProcessor

Generate partition.prop for SCHISM MPI domain decomposition.

SOURCE_NAME: str = 'PARTITION'

process() → ForcingResult

Generate partition.prop.

find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

UFS-Coastal

ESMF mesh

ESMF mesh file generator for UFS-Coastal DATM coupling.

Creates an ESMF unstructured mesh file from a regular lat/lon forcing grid. Used by CDEPS DATM to interpolate atmospheric forcing to the ocean mesh.

CRITICAL: elementMask must be set to 1 (active), NOT 0 (masked).

elementMask=0 causes CMEPS to mask out ALL elements, resulting in zero atmospheric forcing passed to SCHISM (lesson #18).

Input: Regular lat/lon grid (from datm_forcing.nc or sflux files) Output: esmf_mesh.nc (ESMF unstructured mesh format)

class nos_utils.forcing.esmf_mesh.ESMFMeshProcessor(config: ForcingConfig, input_path: Path, output_path: Path, forcing_file: Path | None = None)

Bases: ForcingProcessor

Generate ESMF mesh file from a regular lat/lon forcing grid.

The mesh file defines the spatial grid for CDEPS DATM component in UFS-Coastal coupled simulations.

SOURCE_NAME: str = 'ESMF_MESH'

process() → ForcingResult

Generate ESMF mesh from forcing grid or config domain.

find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

Blender

HRRR + GFS atmospheric forcing blender.

Blends high-resolution HRRR (3km, CONUS) with global GFS (0.25°) on a regular lat/lon grid via Delaunay triangulation. HRRR overrides GFS where HRRR has spatial coverage.

Input:

• GFS sflux files (sflux_air_1.*.nc) — primary, global coverage

• HRRR sflux files (sflux_air_2.*.nc) — secondary, CONUS only

Output:

• datm_forcing.nc — blended forcing on regular grid for UFS-Coastal DATM

Wind rotation: HRRR Lambert Conformal winds are grid-relative and must be rotated to earth-relative. Formula uses LoV and LaD from the LC projection.

class nos_utils.forcing.blender.BlenderProcessor(config: ForcingConfig, input_path: Path, output_path: Path, target_dx: float = 0.025)

Bases: ForcingProcessor

Blend HRRR + GFS sflux files into a single DATM forcing file.

Uses Delaunay triangulation for HRRR (Lambert Conformal) to regular lat/lon interpolation. HRRR overrides GFS where coverage exists.

SOURCE_NAME: str = 'BLENDER'

process() → ForcingResult

Blend GFS + HRRR sflux files into datm_forcing.nc.

find_input_files() → List[Path]

Discover input GRIB2 files for the run window.

nos_utils.forcing.blender.rotate_winds_lcc(u_grid, v_grid, lon_2d, lov=-97.5, lad=38.5)

Rotate grid-relative winds to earth-relative for Lambert Conformal.

HRRR GRIB2 winds are relative to the LC grid. DATM/SCHISM need earth-relative (true north) winds.

Parameters:

• u_grid – Grid-relative wind components

• v_grid – Grid-relative wind components

• lon_2d – 2D longitude array (degrees)

• lov – Longitude of vertical (degrees)

• lad – Latitude of tangency (degrees)

Returns:

(u_earth, v_earth) rotated to true north

Writers

Sflux writer

SCHISM sflux NetCDF writer.

Creates sflux_air, sflux_rad, sflux_prc files in SCHISM-compatible format. Shared by GFS, HRRR, and GEFS processors — eliminates duplicate output code.

File naming convention: sflux_{type}_{source_index}.{file_num}.nc

• type: air, rad, prc

• source_index: 1 = primary (GFS/GEFS), 2 = secondary (HRRR)

• file_num: always 1 in single_file mode (COMF default);

  1, 2, 3, … per day in multi-file mode

Dimensions: (ntime, ny_grid, nx_grid) Time: days since base_date, must be monotonically increasing.

COMF convention (single_file=True, the default):

All timesteps go into one file per type: sflux_air_1.1.nc. sflux_inputs.txt is minimal (&sflux_inputs / /).

Multi-file mode (single_file=False):

Timesteps are split by calendar day relative to base_date. sflux_inputs.txt includes explicit weight/window parameters.

class nos_utils.forcing.sflux_writer.SfluxWriter(output_dir: Path, source_index: int = 1, single_file: bool = True)

Bases: object

Creates SCHISM sflux NetCDF files for any atmospheric source.

Usage:

writer = SfluxWriter(output_dir=Path(“/data/sflux”), source_index=1) files = writer.write_all(data, times, lons, lats, base_date) writer.write_sflux_inputs(met_num=2)

write_all(data: Dict[str, List[numpy.ndarray]], times: List[datetime], lons: numpy.ndarray, lats: numpy.ndarray, base_date: datetime) → List[Path]

Write all sflux files.

In single_file mode (default, COMF convention): writes all timesteps into sflux_{type}_{source}.1.nc — one file per type.

In multi-file mode: splits by calendar day into .1.nc, .2.nc, etc.

Parameters:

• data – Dict mapping variable names to lists of 2D arrays (one per time step)

• times – List of datetime objects (one per time step)

• lons – 1D longitude array

• lats – 1D latitude array

• base_date – Reference datetime for time axis

Returns:

List of created file paths

write_day(data: Dict[str, List[numpy.ndarray]], times: List[datetime], lons: numpy.ndarray, lats: numpy.ndarray, base_date: datetime, day_num: int) → List[Path]

Write sflux_air, sflux_rad, sflux_prc for a single day.

Returns:

List of created file paths (up to 3)

write_sflux_inputs(met_num: int = 1) → Path

Write sflux_inputs.txt namelist for SCHISM.

In single_file mode (COMF convention): writes minimal defaults &sflux_inputs followed by /. SCHISM applies the same defaults internally, so the result is functionally identical.

In multi-file mode: writes explicit weight/window parameters (legacy behavior).

Parameters:

met_num – Number of met sources (1 or 2)

Returns:

Path to sflux_inputs.txt

DATM writer

DATM (Data Atmosphere) NetCDF writer for UFS-Coastal.

Creates datm_forcing.nc for CDEPS DATM in-memory coupling with SCHISM (nws=4). Supports both direct write from extracted arrays and blended GFS+HRRR output.

Output dimensions: (time, latitude, longitude) on a regular lat/lon grid. CRITICAL: nx_global/ny_global in datm_in must match actual file dimensions (lesson #19).

class nos_utils.forcing.datm_writer.DATMWriter

Bases: object

Creates datm_forcing.nc for UFS-Coastal CDEPS DATM component.

Usage:

writer = DATMWriter() path = writer.write(data, times, lons, lats, output_path)

# With GFS+HRRR blending: path = writer.write_blended(gfs_data, hrrr_data, …, output_path)

write(data: Dict[str, List[numpy.ndarray]], times: List[datetime], lons: numpy.ndarray, lats: numpy.ndarray, output_path: Path, epoch: datetime | None = None) → Path

Write datm_forcing.nc from extracted arrays (single source, no blending).

Parameters:

• data – Dict mapping sflux variable names to lists of 2D arrays

• times – List of datetime objects

• lons – 1D longitude array

• lats – 1D latitude array

• output_path – Path for output file

• epoch – Reference datetime for time axis (default: 1970-01-01)

Returns:

Path to created file

write_blended(gfs_data: Dict[str, List[numpy.ndarray]], hrrr_data: Dict[str, List[numpy.ndarray]] | None, gfs_times: List[datetime], hrrr_times: List[datetime] | None, target_lons: numpy.ndarray, target_lats: numpy.ndarray, hrrr_lons_2d: numpy.ndarray | None, hrrr_lats_2d: numpy.ndarray | None, output_path: Path, epoch: datetime | None = None) → Path

Write datm_forcing.nc with GFS+HRRR blending.

HRRR overrides GFS where HRRR has spatial coverage. Uses Delaunay triangulation for HRRR LCC -> regular grid interpolation.

Parameters:

• gfs_data – GFS extracted data (on regular lat/lon grid)

• hrrr_data – HRRR extracted data (on LCC grid, or None)

• gfs_times – GFS time steps

• hrrr_times – HRRR time steps (or None)

• target_lons – 1D target longitude array

• target_lats – 1D target latitude array

• hrrr_lons_2d – 2D HRRR longitude array (or None)

• hrrr_lats_2d – 2D HRRR latitude array (or None)

• output_path – Path for output file

• epoch – Reference datetime

Returns:

Path to created file

static get_grid_dims(lons: numpy.ndarray, lats: numpy.ndarray) → Tuple[int, int]

Return (nx_global, ny_global) for datm_in configuration.

CRITICAL: These must match the actual file dimensions (lesson #19).