CrocoDash.forcing package

CrocoDash.forcing package#

Submodules#

CrocoDash.forcing.base module#

class CrocoDash.forcing.base.BaseConfigurator(**kwargs)#

Bases: ABC

Base class for all CrocoDash configurators.

allowed_compsets: List[str] = []#

classmethod check_input_params_synced()#: Make sure the init args exactly match the input param names. This check is only run in testing

classmethod check_output_params_exist()#

abstractmethod configure()#: Bind input values to parameters and files.

depends_on_outputs: Dict[str, List[str]] = {}#

classmethod deserialize(data: Dict[str, Any])#

forbidden_compsets: List[str] = []#

get_input_param(name: str) → OutputParam#

get_input_param_object(name: str) → OutputParam#

get_output_filepaths(ocn_ice_directory)#: Get output files from the output parameters

get_output_param(name: str) → OutputParam#

get_output_param_object(name: str) → OutputParam#

input_params: List[Param]#

classmethod inspect(caseroot)#: Return an instance of the configurator with placeholder values for input and correct output params from case

property is_non_local: bool#

Whether this configurator’s case is non-local (CIME’s –non-local).

Sourced from the live Case via the registry rather than a declared input param, so an XMLConfigParam output doesn’t require its configurator to accept/thread a case_is_non_local ctor arg just to reach apply() – it’s always False when there’s no live Case (e.g. direct construction in tests, or deserialize()).

classmethod is_required(compset: str) → bool#

make_serializable(obj)#

name: str = ''#

output_params: List[OutputParam]#

process_components: Dict[str, str] = {}#

registry: ForcingConfigRegistry | None = None#

required_for_compsets: List[str] = []#

serialize() → Dict[str, Any]#

set_input_param(name: str, value)#

set_output_param(name: str, value)#

validate_args(**kwargs)#: Validate provided inputs against declared input_params.

classmethod validate_compset_compatibility(compset: str) → bool#

validate_output_filepaths(ocn_ice_directory)#

class CrocoDash.forcing.base.ConfigOutputParam(name: str, comment: str | None = None, is_file: bool = False)#

Bases: OutputParam

Derived value with no case-side effect (not written to user_nl or xmlchange’d).

Exists purely so set_output_param() + the generic serialize() pick it up under a configurator’s outputs, for values consumed only via config.json (e.g. by a sibling configurator’s process_* method).

apply()#: Apply the configuration change.

inspect(caseroot)#: Inspect the current value of this parameter in the case located at caseroot.

class CrocoDash.forcing.base.ForcingConfigRegistry(compset, inputs: dict, case=None)#

Bases: object

classmethod all_process_flags()#: Every process-component flag name any registered configurator can answer to, regardless of whether it’s active for a given case – used to build cli.py’s argparse flags generically.

classmethod available_process_flags(config: dict)#: Every process-component flag name available given which forcing types are present in a config.json dict – looked up from each entry’s declared class only (no full deserialize), so this works even when an entry’s inputs/outputs aren’t fully valid yet. Used by resolve_components/cli.py to answer “does this flag exist” without needing working configurator instances the way actual dispatch does.

find_active_configurators(compset, inputs: dict)#

classmethod find_required_configurators(compset)#: Returns the required configurations based on the compset in a list

classmethod find_valid_configurators(compset)#: Returns the valid configurations based on the compset in a list

get_active_configurators()#

classmethod get_configurator(obj_dict)#

classmethod get_configurator_from_name(name)#

classmethod get_configurator_output(config: dict, configurator_name: str, output_name: str)#: Look up another configurator’s serialized output value from config.json – for a process_*() method that depends on a sibling’s output but (unlike configure()) can’t rely on a live registry/Case, since process() may run in a different process. Raises a clear, named error instead of a bare KeyError several dict levels deep if the dependency isn’t there (e.g. the sibling hasn’t run yet, or never configured this case).

classmethod get_ctor_signature(configurator_cls)#

classmethod get_user_args(configurator_cls)#

classmethod instantiate_configurator(configurator_cls, inputs)#

is_active(name: str) → bool#: Return True if a configurator with this name is active.

classmethod register(configurator_cls: type)#

registered_types: List[type] = [<class 'CrocoDash.forcing.bgc.BGCConfigurator'>, <class 'CrocoDash.forcing.bgc.CICEConfigurator'>, <class 'CrocoDash.forcing.bgc.BGCICConfigurator'>, <class 'CrocoDash.forcing.bgc.BGCIronForcingConfigurator'>, <class 'CrocoDash.forcing.bgc.BGCRiverNutrientsConfigurator'>]#

classmethod resolve_process_targets(config: dict)#: Deserialize every configurator present in a config.json dict and return {flag_name: (configurator_instance, method_name)} covering every process component any of them declare. Used by driver.py to dispatch generically instead of a hand-maintained if-ladder.

classmethod return_missing_inputs(configurator_cls, inputs)#

run_configurators(config_path)#

class CrocoDash.forcing.base.InputFileParam(name: str, comment: str | None = None)#

Bases: InputParam

Base class for a single file parameter in our forcing configurations.

set_item(filepath: str)#: Bind a runtime value to this parameter.

class CrocoDash.forcing.base.InputParam(name: str, comment: str | None = None)#: Bases: Param

class CrocoDash.forcing.base.InputValueParam(name: str, comment: str | None = None)#

Bases: InputParam

Base class for a single value parameter in our forcing configurations.

set_item(item)#: Bind a runtime value to this parameter.

class CrocoDash.forcing.base.OutputParam(name: str, comment: str | None = None, is_file: bool = False)#

Bases: Param

Base class for a single configuration parameter applied to a CESM/MOM6 case.

abstractmethod apply()#: Apply the configuration change.

abstractmethod inspect(caseroot)#: Inspect the current value of this parameter in the case located at caseroot.

set_item(value: Any)#: Bind a runtime value to this parameter.

class CrocoDash.forcing.base.Param(name: str, comment: str | None = None)#

Bases: ABC

Base class for a single parameter in our forcing configurations.

abstractmethod set_item(item: Any)#: Bind a runtime value to this parameter.

exception CrocoDash.forcing.base.UndeclaredParamError#

Bases: KeyError

Code referenced an input/output param name that isn’t declared on the class (a schema bug, e.g. a stale name after a rename). Kept as a KeyError subclass so existing except KeyError call sites are unaffected, but a distinct type so a param-consistency check can catch exactly this failure and not any other incidental KeyError raised by a configurator’s own business logic.

class CrocoDash.forcing.base.UserNLConfigParam(name: str, user_nl_name: str = 'mom', comment: str | None = None, is_file: bool = False)#

Bases: OutputParam

Parameter written to a user_nl_<component> file (default: user_nl_mom).

apply()#: Apply the configuration change.

inspect(caseroot)#: Inspect the current value of this parameter in the case located at caseroot.

class CrocoDash.forcing.base.WorkflowContext(inputdir: Path, supergrid_path: Path, vgrid_path: Path, topo_path: Path, raw_data_dir: Path, regridded_data_dir: Path, output_path: Path, config: dict, preview: bool = False)#

Bases: object

Workflow-level paths shared by every configurator’s process step.

Built once by driver.run_workflow() from config.json + case_state, and passed into every process_*(ctx) call. These are paths no single configurator owns (they’re not part of any configurator’s own inputs/ outputs) but that most process steps need – grid/topo/vgrid files and the raw/regridded/output directories for this case.

property grid#

property ocn_topo#

class CrocoDash.forcing.base.XMLConfigParam(name: str, is_non_local: bool = False, comment: str | None = None, is_file: bool = False)#

Bases: OutputParam

Parameter applied via xmlchange.

XML changes are permanent and do not save previous state.

apply()#: Apply the configuration change.

inspect(caseroot)#: Inspect the current value of this parameter in the case located at caseroot.

CrocoDash.forcing.base.calendar_as_dict(calendar)#

Store a Calendar in config.json as a plain dict, leaving anything else be.

A configurator is normally constructed from a live Calendar, but also from the serialized dict (BaseConfigurator.deserialize) and from empty placeholders (BaseConfigurator.inspect), so only the dataclass is converted here.

CrocoDash.forcing.base.is_serializable(v)#

CrocoDash.forcing.base.register(cls)#

Decorator: register a BaseConfigurator subclass with ForcingConfigRegistry.

Every module under CrocoDash.forcing that defines a configurator class applies this at class-definition time; CrocoDash.forcing/__init__.py’s auto-discovery ensures those modules (and hence this decorator) run for every file in the package, so adding a new forcing type is just adding a new file here – nothing else needs to import it by name.

CrocoDash.forcing.bgc module#

class CrocoDash.forcing.bgc.BGCConfigurator#

Bases: BaseConfigurator

Toggles the MARBL-tracer namelist default – no forcing data of its own to extract. BGCIC/BGCIronForcing/BGCRiverNutrients below are the configurators that actually produce BGC forcing files.

allowed_compsets: List[str] = ['MARBL']#

configure()#: Bind input values to parameters and files.

input_params: List[Param] = []#

name: str = 'BGC'#

output_params: List[OutputParam] = [UserNLConfigParam(name='MAX_FIELDS')]#

required_for_compsets: List[str] = ['MARBL']#

class CrocoDash.forcing.bgc.BGCICConfigurator(marbl_ic_filepath)#

Bases: BaseConfigurator

allowed_compsets: List[str] = ['MARBL']#

configure()#: Bind input values to parameters and files.

input_params: List[Param] = [InputFileParam(name='marbl_ic_filepath')]#

name: str = 'BGCIC'#

output_params: List[OutputParam] = [UserNLConfigParam(name='MARBL_TRACERS_IC_FILE')]#

process(ctx)#: Copy the MARBL initial condition file into place.

process_components: Dict[str, str] = {'bgcic': 'process'}#

required_for_compsets: List[str] = ['MARBL']#

class CrocoDash.forcing.bgc.BGCIronForcingConfigurator(case_session_id, case_grid_name)#

Bases: BaseConfigurator

allowed_compsets: List[str] = ['MARBL']#

configure()#: Bind input values to parameters and files.

input_params: List[Param] = [InputValueParam(name='case_session_id'), InputValueParam(name='case_grid_name')]#

name: str = 'BGCIronForcing'#

output_params: List[OutputParam] = [UserNLConfigParam(name='MARBL_FESEDFLUX_FILE'), UserNLConfigParam(name='MARBL_FEVENTFLUX_FILE'), UserNLConfigParam(name='MARBL_FESEDFLUXRED_FILE')]#

process(ctx)#: Create dummy iron forcing files for MARBL.

process_components: Dict[str, str] = {'bgcironforcing': 'process'}#

required_for_compsets: List[str] = ['MARBL']#

class CrocoDash.forcing.bgc.BGCRiverNutrientsConfigurator(global_river_nutrients_filepath, case_session_id, case_grid_name, case_forcing_product=None, calendar=None)#

Bases: BaseConfigurator

allowed_compsets: List[str] = ['MARBL', 'DROF']#

configure()#: Bind input values to parameters and files.

depends_on_outputs: Dict[str, List[str]] = {'runoff': ['ROF2OCN_LIQ_RMAPNAME']}#

input_params: List[Param] = [InputFileParam(name='global_river_nutrients_filepath'), InputValueParam(name='case_session_id'), InputValueParam(name='case_grid_name'), InputValueParam(name='calendar')]#

name: str = 'BGCRiverNutrients'#

output_params: List[OutputParam] = [UserNLConfigParam(name='READ_RIV_FLUXES'), UserNLConfigParam(name='RIV_FLUX_FILE')]#

process(ctx)#: Regrid global river nutrients onto the ocean grid via the runoff mapping file – requires RunoffConfigurator’s process step to have already produced that mapping file. See depends_on_outputs above: driver.py derives the required run-before ordering from it automatically.

process_components: Dict[str, str] = {'bgcrivernutrients': 'process'}#

validate_args(**kwargs)#: Validate provided inputs against declared input_params.

class CrocoDash.forcing.bgc.CICEConfigurator#

Bases: BaseConfigurator

allowed_compsets: List[str] = ['CICE']#

configure()#: Bind input values to parameters and files.

input_params: List[Param] = []#

name: str = 'CICE'#

output_params: List[OutputParam] = [UserNLConfigParam(name='ice_ic'), UserNLConfigParam(name='ns_boundary_type'), UserNLConfigParam(name='ew_boundary_type'), UserNLConfigParam(name='close_boundaries')]#

CrocoDash.forcing package

Contents

CrocoDash.forcing package#

Submodules#

CrocoDash.forcing.base module#

CrocoDash.forcing.bgc module#

CrocoDash.forcing.chl module#

CrocoDash.forcing.driver module#

CrocoDash.forcing.ic module#

CrocoDash.forcing.mom6 module#

CrocoDash.forcing.obc module#

CrocoDash.forcing.runoff module#

CrocoDash.forcing.tides module#

CrocoDash.forcing.utils module#

Module contents#