Skip to content

crop module

Module to prepare crop parameters

WOFOSTCropParametersProvider (YAMLCropDataProvider)

A data provider for WOFOST crop parameters. This class extends the standard YAMLCropDataProvider to automatically locate and load crop-specific parameter files, injects missing model-specific parameters, and applies user overrides.

Parameters:

Name Type Description Default
crop_name str

The name of the crop (e.g., 'wheat', 'maize').

required
variety_name str

The specific variety of the crop (e.g., 'Winter_wheat_101').

required
model_name str

The WOFOST model version (e.g., 'Wofost73_PP').

required
crop_overrides dict

Dictionary of parameters to override standard values.

None
Source code in cropengine/crop.py
class WOFOSTCropParametersProvider(YAMLCropDataProvider):
    """
    A data provider for WOFOST crop parameters. This class extends the standard YAMLCropDataProvider
    to automatically locate and load crop-specific parameter files, injects missing model-specific
    parameters, and applies user overrides.

    Args:
        crop_name (str): The name of the crop (e.g., 'wheat', 'maize').
        variety_name (str): The specific variety of the crop (e.g., 'Winter_wheat_101').
        model_name (str): The WOFOST model version (e.g., 'Wofost73_PP').
        crop_overrides (dict, optional): Dictionary of parameters to override standard values.
    """

    def __init__(
        self,
        crop_name: str,
        variety_name: str,
        model_name: str,
        crop_overrides: dict = None,
    ):
        # 1. Get the directory path directly from the module
        config_path = list(wofost_crop_params.__path__)[0]

        # 2. Initialize the parent class with the string path
        super().__init__(fpath=str(config_path))

        # 3. Set the active crop
        self.set_active_crop(crop_name, variety_name)
        self.crop_name = crop_name
        self.variety_name = variety_name
        self.model_name = model_name
        self.crop_overrides = crop_overrides if crop_overrides else {}

        self.param_metadata = self._get_param_metadata()

    def _get_param_metadata(self) -> list[dict]:
        """
        Retrieves metadata for the current crop variety from the YAML configuration.
        """
        with pkg_resources.files(wofost_crop_params).joinpath(
            f"{self.crop_name}.yaml"
        ).open("r") as f:
            crop_config = yaml.safe_load(f)
            crop_variety_config = crop_config["CropParameters"]["Varieties"][
                self.variety_name
            ]

        param_metadata = []
        existing_params = set()

        for param, info in crop_variety_config.items():
            try:
                # 1. Determine Value (Override > Variety)
                if param in self.crop_overrides:
                    final_val = self.crop_overrides[param]
                else:
                    final_val = info[0]

                # 2. Update PCSE internal dictionary
                self[param] = final_val

                # 3. Build Metadata
                param_dict = {
                    "parameter": param,
                    "description": info[1],
                    "unit": info[-1][0] if len(info[-1]) == 1 else info[-1],
                    "required": False,
                    "value": final_val,
                }
                param_metadata.append(param_dict)
                existing_params.add(param)

            except (IndexError, TypeError, KeyError):
                continue

        try:
            with pkg_resources.files(configs).joinpath("crop_params.yaml").open(
                "r"
            ) as f:
                crop_config = yaml.safe_load(f)["wofost"]

                if self.model_name in crop_config["model_mapping"]:
                    profile_name = crop_config["model_mapping"][self.model_name]
                    profile_def = crop_config["profiles"][profile_name]

                    required_params = set(profile_def.get("required", []))
                    all_param_defs = crop_config["crop_params"]

                    for param in required_params:
                        if param in existing_params:
                            continue

                        if param in all_param_defs:
                            param_meta = all_param_defs[param]
                            default_val = param_meta["default"]

                            if param in self.crop_overrides:
                                final_val = self.crop_overrides[param]
                            else:
                                final_val = default_val
                                print(
                                    f"🚨 [WARN] Required crop parameter '{param}' missing for model '{self.model_name}'. "
                                    f"Using default value: {final_val}"
                                )

                            self[param] = final_val

                            param_dict = {
                                "parameter": param,
                                "description": param_meta["description"],
                                "unit": param_meta["unit"],
                                "required": True,
                                "value": final_val,
                            }
                            param_metadata.append(param_dict)
                            existing_params.add(param)

        except Exception as e:
            raise RuntimeError(f"Failed to load crop_params.yaml: {e}")

        return param_metadata

get_available_crop_varieties(model, crop)

Retrieves available varieties and their metadata for a specific crop.

Parameters:

Name Type Description Default
model str

The name of the simulation model.

required
crop str

The name of the crop (e.g., 'wheat').

required

Returns:

Type Description
dict | None

A dictionary where keys are variety identifiers and values are metadata strings (descriptions). Returns None if the model is not supported.

Source code in cropengine/crop.py
def get_available_crop_varieties(model: str, crop: str) -> dict | None:
    """
    Retrieves available varieties and their metadata for a specific crop.

    Args:
        model (str): The name of the simulation model.
        crop (str): The name of the crop (e.g., 'wheat').

    Returns:
        dict | None: A dictionary where keys are variety identifiers and values
        are metadata strings (descriptions). Returns None if the model is not supported.
    """
    if model.startswith("Wofost"):
        with pkg_resources.files(wofost_crop_params).joinpath(f"{crop}.yaml").open(
            "r"
        ) as f:
            crop_config = yaml.safe_load(f)

            all_crop_varieties = crop_config["CropParameters"]["Varieties"].keys()
            crop_varieties = {}

            for v in all_crop_varieties:
                # Extract the descriptive metadata for the variety
                meta = crop_config["CropParameters"]["Varieties"][v]["Metadata"]
                crop_varieties[v] = meta

        return crop_varieties

    else:
        return None

get_available_crops(model)

Retrieves a list of supported crops for a specific model type.

Parameters:

Name Type Description Default
model str

The name of the simulation model (e.g., "Wofost72_PP").

required

Returns:

Type Description
list[str]

A list of available crop names.

Source code in cropengine/crop.py
def get_available_crops(model: str) -> list[str]:
    """
    Retrieves a list of supported crops for a specific model type.

    Args:
        model (str): The name of the simulation model (e.g., "Wofost72_PP").

    Returns:
        list[str]: A list of available crop names.
    """
    if model.startswith("Wofost"):
        with pkg_resources.files(wofost_crop_params).joinpath("crops.yaml").open(
            "r"
        ) as f:
            available_crops = yaml.safe_load(f)["available_crops"]

        return available_crops

    else:
        return []