Create PPE

CreatePPEConfig dataclass

CreatePPEConfig(
    verbose: int = 0,
    log_dir: Path | str = "",
    log_mode: str = "w",
    *,
    simulation_setup_path: Path,
    build_base_only: bool = False,
    build_only: bool = False,
    frozen_base_case: bool = False,
    keepexe: bool = False,
    overwrite_base_case: bool = False,
    overwrite_ppe: bool = True
)

Bases: BaseConfig

get_checked_and_derived_config

get_checked_and_derived_config() -> CheckedCreatePPEConfig

Check and handle arguments for the PPE configuration.

Source code in tinkertool/scripts/create_ppe/config.py

def get_checked_and_derived_config(self) -> 'CheckedCreatePPEConfig':
    """Check and handle arguments for the PPE configuration."""
    time_str = time.strftime("%Y%m%d-%H%M%S")
    log_file = Path(self.log_dir).joinpath(f'tinkertool_{time_str}.log')

    # derived fields - we unpack the simulation setup file
    simulation_setup: configparser.ConfigParser = read_config(self.simulation_setup_path)
    # - ppe_settings
    baseroot = Path(simulation_setup['ppe_settings'].get('baseroot',vars=os.environ)).resolve()
    basecasename = simulation_setup['ppe_settings']['basecasename']
    ## - paramfile
    pdim: str = simulation_setup['ppe_settings']['pdim']
    paramfile_path: Path = Path(simulation_setup['ppe_settings'].get('paramfile',vars=os.environ)).resolve()
    validate_file(paramfile_path, ".nc", "paramfile", new_file=False)
    paramfile: xr.Dataset = xr.open_dataset(paramfile_path)
    if pdim not in paramfile.dims:
        raise SystemExit(f"ERROR: {pdim} is not a valid dimension in {paramfile_path}. \nParamfile dimensions are: {list(paramfile.dims.keys())}")
    paramDataset: xr.Dataset = paramfile
    componentdict: dict = {}
    logging.debug(f"Processing paramfile {paramfile_path} with parameters: {list(paramDataset.variables.keys())}")
    for param in [ param for param in paramDataset.variables.keys() if param != pdim ]:
        esm_component = paramDataset.variables[param].attrs.get('esm_component', None)
        if esm_component is None:
            err_msg = f"Parameter {param} in paramfile {paramfile_path} does not have an 'esm_component' attribute."
            logging.error(err_msg)
            raise ValueError(err_msg)
        componentdict[param] = esm_component
    num_sims = paramfile.sizes[pdim]
    num_vars = len(paramfile.variables.keys())-1
    ensemble_num = paramfile[pdim][:].values

    namelist_collection_dict = {}
    for component_nl_name in simulation_setup.options('namelist_control'):
        control_nl = simulation_setup['namelist_control'].get(component_nl_name,vars=os.environ)
        if control_nl is not None:
            control_nl = Path(control_nl).resolve()
            validate_file(control_nl, ".ini", f"namelist control file {control_nl.name}.ini", new_file=False)
            namelist_collection_dict[component_nl_name] = read_config(control_nl)
        else:
            logging.warning(f"Control namelist is None for {control_nl.name}, using model default")
    # - create_case
    simulation_setup['create_case']['cesmroot'] = simulation_setup['create_case'].get('cesmroot',vars=os.environ)
    cesmroot = Path(simulation_setup['create_case']['cesmroot']).resolve()
    validate_directory(cesmroot, "CESM root directory")
    if os.environ.get('CESMROOT') != str(cesmroot):
        logging.warning(f"CESMROOT environment variable is set to {os.environ.get('CESMROOT')}, but the simulation setup file specifies {cesmroot}.")
        logging.warning("This may cause issues with CIME paths. Consider choosing one cesmroot.")

    add_CIME_paths_and_import(cesmroot)
    if self.__dict__.get('log_file', log_file) is not None:
        log_file = self.__dict__.get('log_file', log_file)
    # remove log_file from __dict__ to avoid duplication
    if 'log_file' in self.__dict__:
        del self.__dict__['log_file']
    return CheckedCreatePPEConfig(
        **self.__dict__,
        log_file=log_file,
        simulation_setup=simulation_setup,
        baseroot=baseroot,
        basecasename=basecasename,
        paramfile_path=paramfile_path,
        pdim=pdim,
        paramDataset=paramDataset,
        componentdict=componentdict,
        num_sims=num_sims,
        num_vars=num_vars,
        ensemble_num=ensemble_num,
        namelist_collection_dict=namelist_collection_dict,
        cesmroot=cesmroot
    )

SubmitPPEConfig dataclass

SubmitPPEConfig(
    verbose: int = 0,
    log_dir: Path | str = "",
    log_mode: str = "w",
    *,
    cases: Union[str, Path, list[str], list[Path]]
)

Bases: BaseConfig

get_checked_and_derived_config

get_checked_and_derived_config() -> CheckedSubmitPPEConfig

Check and handle arguments for the PPE configuration.

Source code in tinkertool/scripts/create_ppe/config.py

def get_checked_and_derived_config(self) -> 'CheckedSubmitPPEConfig':
    """Check and handle arguments for the PPE configuration."""
    # Create log file path (from parent class logic)
    time_str = time.strftime("%Y%m%d-%H%M%S")
    log_file = Path(self.log_dir).joinpath(f'tinkertool_{time_str}.log')
    if self.__dict__.get('log_file') is None:
        self.__dict__['log_file'] = log_file
    return CheckedSubmitPPEConfig(**self.__dict__)

CheckBuildConfig dataclass

CheckBuildConfig(
    verbose: int = 0,
    log_dir: Path | str = "",
    log_mode: str = "w",
    *,
    cases: Union[str, Path, list[str], list[Path]]
)

Bases: SubmitPPEConfig

Configuration for checking build status of PPE cases.

Functionally identical to SubmitPPEConfig but with a name that clearly indicates its purpose: checking if ensemble member cases have been built successfully.

get_checked_and_derived_config

get_checked_and_derived_config() -> CheckedCheckBuildConfig

Check and handle arguments for the build check configuration.

Source code in tinkertool/scripts/create_ppe/config.py

def get_checked_and_derived_config(self) -> 'CheckedCheckBuildConfig':
    """Check and handle arguments for the build check configuration."""
    # Get the parent checked config
    parent_config = super().get_checked_and_derived_config()
    # Return as CheckedCheckBuildConfig
    return CheckedCheckBuildConfig(**parent_config.__dict__)

CheckedCheckBuildConfig dataclass

CheckedCheckBuildConfig(
    verbose: int = 0,
    log_dir: Path | str = "",
    log_mode: str = "w",
    *,
    log_file: Path = (
        lambda: joinpath(
            "output",
            f"tinkertool.{strftime('%Y%m%d-%H%M%S')}.log",
        )
    )(),
    cases: list[Path] = list(),
)

Bases: CheckedSubmitPPEConfig

Checked configuration for checking build status of PPE cases.

The validated and processed version of CheckBuildConfig with all fields verified and cases converted to Path objects.

PrestageEnsembleConfig dataclass

PrestageEnsembleConfig(
    verbose: int = 0,
    log_dir: Path | str = "",
    log_mode: str = "w",
    *,
    cases: Union[str, Path, list[str], list[Path]]
)

Bases: SubmitPPEConfig

Configuration for prestaging PPE ensemble cases.

Functionally identical to SubmitPPEConfig but with a name that clearly indicates its purpose: preparing ensemble member cases by copying restart files and setting up the run environment before job submission.

get_checked_and_derived_config

get_checked_and_derived_config() -> (
    CheckedPrestageEnsembleConfig
)

Check and handle arguments for the prestage configuration.

Source code in tinkertool/scripts/create_ppe/config.py

def get_checked_and_derived_config(self) -> 'CheckedPrestageEnsembleConfig':
    """Check and handle arguments for the prestage configuration."""
    # Get the parent checked config
    parent_config = super().get_checked_and_derived_config()
    # Return as CheckedPrestageEnsembleConfig
    return CheckedPrestageEnsembleConfig(**parent_config.__dict__)

CheckedPrestageEnsembleConfig dataclass

CheckedPrestageEnsembleConfig(
    verbose: int = 0,
    log_dir: Path | str = "",
    log_mode: str = "w",
    *,
    log_file: Path = (
        lambda: joinpath(
            "output",
            f"tinkertool.{strftime('%Y%m%d-%H%M%S')}.log",
        )
    )(),
    cases: list[Path] = list(),
)

Bases: CheckedSubmitPPEConfig

Checked configuration for prestaging PPE ensemble cases.

The validated and processed version of PrestageEnsembleConfig with all fields verified and cases converted to Path objects.

check_build

check_build(config: CheckBuildConfig) -> bool

Check if the build was successful by checking for case.build success in each case directories CaseStatus

Parameters

config : SubmitPPEConfig Configuration object containing the cases to check and logging settings. Returns

---

bool True if all cases have a successful build, False otherwise.

Source code in tinkertool/scripts/create_ppe/create_ppe.py

def check_build(config: CheckBuildConfig) -> bool:
    """Check if the build was successful by checking for case.build success in each case directories CaseStatus

    Parameters
    ----------
    config : SubmitPPEConfig
        Configuration object containing the cases to check and logging settings.
    Returns
    -------
    bool
        True if all cases have a successful build, False otherwise.
    """

    # check if CheckBuildConfig is valid
    checked_config: CheckedCheckBuildConfig = config.get_checked_and_derived_config()

    # set up logging if not already set
    if not logging.getLogger('tinkertool_log').handlers:
        setup_logging(checked_config.verbose, checked_config.log_file, checked_config.log_mode, 'tinkertool_log')
        log_info_detailed('tinkertool_log', 'tinkertool_log logger set up')


    all_build_success = True
    for case in checked_config.cases:
        case_status_file = os.path.join(case, 'CaseStatus')
        if not os.path.exists(case_status_file):
            logging.error(f"CaseStatus file not found in {case}.")
            all_build_success = False
            continue

        with open(case_status_file, 'r') as file:
            found = False
            for line in file:
                if 'case.build success' in line:
                    found = True
                    break

        if found:
            log_info_detailed("tinkertool_log", f"Build successful for case {case}.")
        else:
            logging.error(f"Build failed for case {case}. cat {case.joinpath('CaseStatus')} for details.")
            all_build_success = False

    return all_build_success

prestage_ensemble

prestage_ensemble(config: PrestageEnsembleConfig) -> bool

Prestage the ensemble members by copying the base case input files to each member's input directory.

Parameters

config : PrestageEnsembleConfig Configuration object containing the cases to prestage and logging settings.

Source code in tinkertool/scripts/create_ppe/create_ppe.py

def prestage_ensemble(config: PrestageEnsembleConfig) -> bool:
    """Prestage the ensemble members by copying the base case input files to each member's input directory.

    Parameters
    ----------
    config : PrestageEnsembleConfig
        Configuration object containing the cases to prestage and logging settings.
    """

    # check if PrestageEnsembleConfig is valid
    checked_config: CheckedPrestageEnsembleConfig = config.get_checked_and_derived_config()

    # set up logging if not already set
    if not logging.getLogger('tinkertool_log').handlers:
        setup_logging(checked_config.verbose, checked_config.log_file, checked_config.log_mode, 'tinkertool_log')
        log_info_detailed('tinkertool_log', 'tinkertool_log logger set up')

    logging.info(">> Starting PPE prestaging")

    rsync_time_cap_restart_files = 600 # sec
    rsync_time_cap_rpointer_files = 120 # sec

    all_prestage_success = True
    for caseroot in checked_config.cases:
        with Case(caseroot, read_only=False) as case:

            rundir = Path(case.get_value('RUNDIR')).resolve()
            run_type = case.get_value('RUN_TYPE')
            run_refdir = Path(case.get_value('RUN_REFDIR')).resolve()
            if case.get_value('GET_REFCASE') == 'TRUE':
                logging.warning(
                    f"Case {case} has \n"\
                    "> 'GET_REFCASE'='TRUE' with \n"\
                    f"> 'RUN_REFDIR'={run_refdir} \n"\
                    "skipping manual prestaging."\
                    "Note that this might cause a crash if cases are submitted simultaneously."
                )
            else:
                if not run_refdir.exists():
                    logging.error(f"Reference directory {run_refdir} does not exist for case {caseroot}. Skipping prestaging.")
                    all_prestage_success = False
                    continue
                logging.debug(f"Prestaging case {caseroot} from reference directory {run_refdir} to run directory {rundir}")
                # check if 'rest' is in RUN_REFDIR path
                # if not throw a warning and ask user to confirm prestaging
                rest_dir = True
                if 'rest' not in str(run_refdir).lower():
                    logging.warning(
                        f"RUN_REFDIR {run_refdir} for case {caseroot} does not contain 'rest' in its path. \n"
                        "This might indicate that the reference directory is not a restart directory. \n"
                        "You will be prompted to confirm prestaging. \n"
                    )
                    rest_dir = False
                # copy the netcdf files from the 'RUN_REFDIR' to the 'RUNDIR'
                # and set 'RUN_REFDIR'='RUNDIR'. This is needed to ensure that the
                # cases can run independently and do not interfere with each other.
                ref_netcdf_files = list(run_refdir.glob('*.nc'))
                if ref_netcdf_files:
                    if not rest_dir:
                        input_with_timer(
                            prompt=f"Press Enter to confirm prestaging of {len(ref_netcdf_files)} netCDF files from {run_refdir} to {rundir} for case {caseroot}, or Ctrl+C to abort. (Auto-continue in 30 seconds)\n",
                            timeout=30,
                            default=''
                        )
                    # Use rsync with shell expansion for glob pattern and progress display
                    cmd_str = f"rsync --archive --progress '{run_refdir}'/*.nc '{rundir}'/."
                    log_info_detailed('tinkertool_log', f"Copying {len(ref_netcdf_files)} netCDF files with rsync from {run_refdir} to {rundir}")
                    logging.debug(f"Command: {cmd_str}")

                    try:
                        result = subprocess.run(
                            cmd_str,
                            cwd=caseroot,
                            check=True,
                            shell=True,
                            capture_output=True,
                            text=True,
                            timeout=rsync_time_cap_restart_files
                        )
                        log_info_detailed('tinkertool_log', f"rsync completed successfully")
                        if result.stdout:
                            log_info_detailed('tinkertool_log', f"rsync stdout: {result.stdout}")
                    except subprocess.TimeoutExpired:
                        error_msg = f"rsync timeout after {rsync_time_cap_restart_files/60} minutes for case {caseroot}."
                        logging.error(error_msg)
                        all_prestage_success = False
                        continue
                    except subprocess.CalledProcessError as e:
                        error_msg = f"Failed to prestage ref_netcdf_files files for case {caseroot}."
                        logging.error(error_msg)
                        if e.stderr:
                            logging.error(f"rsync stderr: {e.stderr}")
                        all_prestage_success = False
                        continue
                    set_value_with_status_update(case, 'RUN_REFDIR', str(rundir), kill_on_error=False)
                else:
                    logging.warning(f"No netcdf files found in {run_refdir}. Skipping prestaging for case {caseroot}.")
                    all_prestage_success = False

                if run_type == 'branch':
                    # in a branch run, we need to prestage the rpointer files as well
                    # copy the rpointer files from the original run_refdir to the rundir
                    # and set 'DRV_RESTART_POINTER' to the rpointer file name with the correct date and time
                    rpointer_files = list(run_refdir.glob('rpointer*'))
                    if rpointer_files:
                        if not rest_dir:
                            input_with_timer(
                                prompt=f"Press Enter to confirm prestaging of {len(rpointer_files)} rpointer files from {run_refdir} to {rundir} for case {caseroot}, or Ctrl+C to abort. (Auto-continue in 30 seconds)\n",
                                timeout=30,
                                default=''
                            )
                        cmd_str = f"rsync --archive '{run_refdir}'/rpointer* '{str(rundir)}'/."
                        log_info_detailed('tinkertool_log', f"Copying {len(rpointer_files)} rpointer files with rsync from {run_refdir} to {rundir}")
                        logging.debug(f"Command: {cmd_str}")
                        try:
                            result = subprocess.run(
                                cmd_str,
                                cwd=caseroot,
                                check=True,
                                shell=True,
                                capture_output=True,
                                text=True,
                                timeout=rsync_time_cap_rpointer_files
                            )
                            log_info_detailed('tinkertool_log', f"rpointer rsync completed successfully")
                            if result.stdout:
                                log_info_detailed('tinkertool_log', f"rsync stdout: {result.stdout}")
                        except subprocess.TimeoutExpired:
                            error_msg = f"rpointer rsync timeout after {rsync_time_cap_rpointer_files/60} minute for case {caseroot}."
                            logging.error(error_msg)
                            all_prestage_success = False
                            continue
                        except subprocess.CalledProcessError as e:
                            error_msg = f"Failed to prestage rpointer files for case {caseroot}."
                            logging.error(error_msg)
                            if e.stderr:
                                logging.error(f"rsync stderr: {e.stderr}")
                            all_prestage_success = False
                            continue

                        set_value_with_status_update(case, 'DRV_RESTART_POINTER', f"rpointer.cpl.{case.get_value('RUN_REFDATE')}-{case.get_value('RUN_REFTOD')}", kill_on_error=False)
                    else:
                        logging.warning(f"No rpointer files found in {run_refdir}. Skipping prestaging for case {caseroot}.")
                        all_prestage_success = False

    logging.info(f">> {len(checked_config.cases)} cases prestaged successfully.")
    return all_prestage_success

bulk_xmlchange

bulk_xmlchange(
    cases: list[Path] | list[str],
    xml_changes: (
        dict[str, str | dict[str, str]]
        | list[dict[str, str | dict[str, str]]]
    ),
) -> None

Apply bulk XML changes to a list of cases.

Parameters

cases : list[Path] List of case directories to apply the XML changes to. xml_changes : dict[str, str | dict[str, str]] | list[dict[str, str | dict[str, str]]] Dictionary (or list of dictionaries) of XML changes to apply. Keys are XML variable names. Values can either be a string (for global changes) or a dictionary with subgroups as keys and values to apply.

Source code in tinkertool/scripts/create_ppe/create_ppe.py

def bulk_xmlchange(
    cases: list[Path] | list[str],
    xml_changes: dict[str, str | dict[str, str]] | list[dict[str, str | dict[str, str]]]
) -> None:
    """Apply bulk XML changes to a list of cases.

    Parameters
    ----------
    cases : list[Path]
        List of case directories to apply the XML changes to.
    xml_changes : dict[str, str | dict[str, str]] | list[dict[str, str | dict[str, str]]]
        Dictionary (or list of dictionaries) of XML changes to apply. Keys are XML variable names.
        Values can either be a string (for global changes) or a dictionary with subgroups as keys and values to apply.
    """
    logging.info(">> Starting bulk XML changes for PPE cases")

    # Handle single dictionary input
    if isinstance(xml_changes, dict):
        xml_changes = [xml_changes]

    for caseroot in cases:
        with Case(str(caseroot), read_only=False) as case:
            log_info_detailed('tinkertool_log', f"Applying XML changes to case {caseroot.name}")
            for change in xml_changes:
                for var, value in change.items():
                    if isinstance(value, dict):  # If value is a dictionary, it means subgroup-specific changes
                        for subgroup, sub_value in value.items():
                            old_value = case.get_value(var, subgroup=subgroup)
                            set_value_with_status_update(case, var, sub_value, subgroup=subgroup, kill_on_error=False)
                            logging.debug(
                                f"Case {caseroot.name}: Changed XML variable '{var}' in subgroup '{subgroup}' "
                                f"from '{old_value}' to '{sub_value}'"
                            )
                    else:  # If value is a simple string, apply to the default
                        old_value = case.get_value(var)
                        set_value_with_status_update(case, var, value, kill_on_error=False)
                        logging.debug(
                            f"Case {caseroot.name}: Changed XML variable '{var}' from '{old_value}' to '{value}'"
                        )

    logging.info(">> Finished bulk XML changes for PPE cases")