utils

aquacrop.utils.data

get_data(filename, **kwargs)

get selected data file

Source code in aquacrop/utils/data.py

def get_data(filename, **kwargs):
    """
    get selected data file
    """
    filepath = os.path.join(data.__path__[0], filename)

    return np.genfromtxt(filepath, **kwargs)

get_filepath(filename)

get selected data file

Source code in aquacrop/utils/data.py

def get_filepath(filename):
    """
    get selected data file
    """
    filepath = os.path.join(data.__path__[0], filename)

    return filepath

list_data()

lists all built-in data files

Source code in aquacrop/utils/data.py

def list_data():
    """
    lists all built-in data files
    """
    path = data.__path__[0]

    return os.listdir(path)

aquacrop.utils.lars

AquaCropModel

This is the main class of the AquaCrop-OSPy model. It is in charge of executing all the operations.

Parameters:

sim_start_time (str): YYYY/MM/DD, Simulation start date

sim_end_time (str): date YYYY/MM/DD, Simulation end date

weather_df: daily weather data , created using prepare_weather

soil: Soil object contains paramaters and variables of the soil
        used in the simulation

crop: Crop object contains Paramaters and variables of the crop used
        in the simulation

initial_water_content: Defines water content at start of simulation

irrigation_management: Defines irrigation strategy

field_management: Defines field management options

fallow_field_management: Defines field management options during fallow period

groundwater: Stores information on water table parameters

co2_concentration: Defines CO2 concentrations

off_season: (True) simulate off-season or (False) skip ahead to start of 
            next growing season

Source code in aquacrop/core.py

class AquaCropModel:
    """
    This is the main class of the AquaCrop-OSPy model.
    It is in charge of executing all the operations.

    Parameters:

        sim_start_time (str): YYYY/MM/DD, Simulation start date

        sim_end_time (str): date YYYY/MM/DD, Simulation end date

        weather_df: daily weather data , created using prepare_weather

        soil: Soil object contains paramaters and variables of the soil
                used in the simulation

        crop: Crop object contains Paramaters and variables of the crop used
                in the simulation

        initial_water_content: Defines water content at start of simulation

        irrigation_management: Defines irrigation strategy

        field_management: Defines field management options

        fallow_field_management: Defines field management options during fallow period

        groundwater: Stores information on water table parameters

        co2_concentration: Defines CO2 concentrations

        off_season: (True) simulate off-season or (False) skip ahead to start of 
                    next growing season


    """

    # Model parameters
    __steps_are_finished: bool = False  # True if all steps of the simulation are done.
    __has_model_executed: bool = False  # Determines if the model has been run
    __has_model_finished: bool = False  # Determines if the model is finished
    __start_model_execution: float = 0.0  # Time when the execution start
    __end_model_execution: float = 0.0  # Time when the execution end
    # Attributes initialised later
    _clock_struct: "ClockStruct"
    _param_struct: "ParamStruct"
    _init_cond: "InitialCondition"
    _outputs: "Output"
    _weather: "DataFrame"

    def __init__(
        self,
        sim_start_time: str,
        sim_end_time: str,
        weather_df: "DataFrame",
        soil: "Soil",
        crop: "Crop",
        initial_water_content: "InitialWaterContent",
        irrigation_management: Optional["IrrigationManagement"] = None,
        field_management: Optional["FieldMngt"] = None,
        fallow_field_management: Optional["FieldMngt"] = None,
        groundwater: Optional["GroundWater"] = None,
        co2_concentration: Optional["CO2"] = None,
        off_season: bool=False,
    ) -> None:

        self.sim_start_time = sim_start_time
        self.sim_end_time = sim_end_time
        self.weather_df = weather_df
        self.soil = soil
        self.crop = crop
        self.initial_water_content = initial_water_content   
        self.co2_concentration = co2_concentration
        self.off_season = off_season

        self.irrigation_management = irrigation_management
        self.field_management = field_management
        self.fallow_field_management = fallow_field_management
        self.groundwater = groundwater

        if irrigation_management is None:
            self.irrigation_management = IrrigationManagement(irrigation_method=0)
        if field_management is None:
            self.field_management = FieldMngt()
        if fallow_field_management is None:
            self.fallow_field_management = FieldMngt()
        if groundwater is None:
            self.groundwater = GroundWater()
        if co2_concentration is None:
            self.co2_concentration = CO2()

    @property
    def sim_start_time(self) -> str:
        """
        Return sim start date
        """
        return self._sim_start_time

    @sim_start_time.setter
    def sim_start_time(self, value: str) -> None:
        """
        Check if sim start date is in a correct format.
        """

        if _sim_date_format_is_correct(value) is not False:
            self._sim_start_time = value
        else:
            raise ValueError("sim_start_time format must be 'YYYY/MM/DD'")

    @property
    def sim_end_time(self) -> str:
        """
        Return sim end date
        """
        return self._sim_end_time

    @sim_end_time.setter
    def sim_end_time(self, value: str) -> None:
        """
        Check if sim end date is in a correct format.
        """
        if _sim_date_format_is_correct(value) is not False:
            self._sim_end_time = value
        else:
            raise ValueError("sim_end_time format must be 'YYYY/MM/DD'")

    @property
    def weather_df(self) -> "DataFrame":
        """
        Return weather dataframe
        """
        return self._weather_df

    @weather_df.setter
    def weather_df(self, value: "DataFrame"):
        """
        Check if weather dataframe is in a correct format.
        """
        weather_df_columns = "Date MinTemp MaxTemp Precipitation ReferenceET".split(" ")
        if not all([column in value for column in weather_df_columns]):
            raise ValueError(
                "Error in weather_df format. Check if all the following columns exist "
                + "(Date MinTemp MaxTemp Precipitation ReferenceET)."
            )

        self._weather_df = value

    def _initialize(self) -> None:
        """
        Initialise all model variables
        """

        # Initialize ClockStruct object
        self._clock_struct = read_clock_parameters(
            self.sim_start_time, self.sim_end_time, self.off_season
        )

        # get _weather data
        self.weather_df = read_weather_inputs(self._clock_struct, self.weather_df)

        # read model params
        self._clock_struct, self._param_struct = read_model_parameters(
            self._clock_struct, self.soil, self.crop, self.weather_df
        )

        # read irrigation management
        self._param_struct = read_irrigation_management(
            self._param_struct, self.irrigation_management, self._clock_struct
        )

        # read field management
        self._param_struct = read_field_management(
            self._param_struct, self.field_management, self.fallow_field_management
        )

        # read groundwater table
        self._param_struct = read_groundwater_table(
            self._param_struct, self.groundwater, self._clock_struct
        )

        # Compute additional variables
        self._param_struct.CO2 = self.co2_concentration
        self._param_struct = compute_variables(
            self._param_struct, self.weather_df, self._clock_struct
        )

        # read, calculate inital conditions
        self._param_struct, self._init_cond = read_model_initial_conditions(
            self._param_struct, self._clock_struct, self.initial_water_content, self.crop
        )

        self._param_struct = create_soil_profile(self._param_struct)

        # Outputs results (water_flux, crop_growth, final_stats)
        self._outputs = Output(self._clock_struct.time_span, self._init_cond.th)

        # save model _weather to _init_cond
        self._weather = self.weather_df.values

    def run_model(
        self,
        num_steps: int = 1,
        till_termination: bool = False,
        initialize_model: bool = True,
        process_outputs: bool = False,
    ) -> bool:
        """
        This function is responsible for executing the model.

        Arguments:

            num_steps: Number of steps (Days) to be executed.

            till_termination: Run the simulation to completion

            initialize_model: Whether to initialize the model \
            (i.e., go back to beginning of season)

            process_outputs: process outputs into dataframe before \
                simulation is finished

        Returns:
            True if finished
        """

        if initialize_model:
            self._initialize()

        if till_termination:
            self.__start_model_execution = time.time()
            while self._clock_struct.model_is_finished is False:

                (
                    self._clock_struct,
                    self._init_cond,
                    self._param_struct,
                    self._outputs,
                ) = self._perform_timestep()
            self.__end_model_execution = time.time()
            self.__has_model_executed = True
            self.__has_model_finished = True
            return True
        else:
            if num_steps < 1:
                raise ValueError("num_steps must be equal to or greater than 1.")
            self.__start_model_execution = time.time()
            for i in range(num_steps):

                if (i == range(num_steps)[-1]) and (process_outputs is True):
                    self.__steps_are_finished = True

                (
                    self._clock_struct,
                    self._init_cond,
                    self._param_struct,
                    self._outputs,
                ) = self._perform_timestep()

                if self._clock_struct.model_is_finished:
                    self.__end_model_execution = time.time()
                    self.__has_model_executed = True
                    self.__has_model_finished = True
                    return True

            self.__end_model_execution = time.time()
            self.__has_model_executed = True
            self.__has_model_finished = False
            return True

    def _perform_timestep(
        self,
    ) -> Tuple["ClockStruct", "InitialCondition", "ParamStruct", "Output"]:

        """
        Function to run a single time-step (day) calculation of AquaCrop-OS
        """

        # extract _weather data for current timestep
        weather_step = _weather_data_current_timestep(
            self._weather, self._clock_struct.time_step_counter
        )

        # Get model solution_single_time_step
        new_cond, param_struct, outputs = solution_single_time_step(
            self._init_cond,
            self._param_struct,
            self._clock_struct,
            weather_step,
            self._outputs,
        )

        # Check model termination
        clock_struct = self._clock_struct
        clock_struct.model_is_finished = check_model_is_finished(
            self._clock_struct.step_end_time,
            self._clock_struct.simulation_end_date,
            self._clock_struct.model_is_finished,
            self._clock_struct.season_counter,
            self._clock_struct.n_seasons,
            new_cond.harvest_flag,
        )

        # Update time step
        clock_struct, _init_cond, param_struct = update_time(
            clock_struct, new_cond, param_struct, self._weather, self.crop
        )

        # Create  _outputsdataframes when model is finished
        final_water_flux_growth_outputs = outputs_when_model_is_finished(
            clock_struct.model_is_finished,
            outputs.water_flux,
            outputs.water_storage,
            outputs.crop_growth,
            self.__steps_are_finished,
        )

        if final_water_flux_growth_outputs is not False:
            (
                outputs.water_flux,
                outputs.water_storage,
                outputs.crop_growth,
            ) = final_water_flux_growth_outputs

        return clock_struct, _init_cond, param_struct, outputs

    def get_simulation_results(self):
        """
        Return all the simulation results
        """
        if self.__has_model_executed:
            if self.__has_model_finished:
                return self._outputs.final_stats
            else:
                return False  # If the model is not finished, the results are not generated.
        else:
            raise ValueError(
                "You cannot get results without running the model. "
                + "Please execute the run_model() method."
            )

    def get_water_storage(self):
        """
        Return water storage in soil results
        """
        if self.__has_model_executed:
            return self._outputs.water_storage
        else:
            raise ValueError(
                "You cannot get results without running the model. "
                + "Please execute the run_model() method."
            )

    def get_water_flux(self):
        """
        Return water flux results
        """
        if self.__has_model_executed:
            return self._outputs.water_flux
        else:
            raise ValueError(
                "You cannot get results without running the model. "
                + "Please execute the run_model() method."
            )

    def get_crop_growth(self):
        """
        Return crop growth results
        """
        if self.__has_model_executed:
            return self._outputs.crop_growth
        else:
            raise ValueError(
                "You cannot get results without running the model. "
                + "Please execute the run_model() method."
            )

    def get_additional_information(self) -> Dict[str, Union[bool, float]]:
        """
        Additional model information.

        Returns:
            dict: {has_model_finished,execution_time}

        """
        if self.__has_model_executed:
            return {
                "has_model_finished": self.__has_model_finished,
                "execution_time": self.__end_model_execution
                - self.__start_model_execution,
            }
        else:
            raise ValueError(
                "You cannot get results without running the model. "
                + "Please execute the run_model() method."
            )

sim_end_time: str property writable

Return sim end date

sim_start_time: str property writable

Return sim start date

weather_df: DataFrame property writable

Return weather dataframe

get_additional_information()

Additional model information.

Returns:

Name	Type	Description
dict	Dict[str, Union[bool, float]]	{has_model_finished,execution_time}

Source code in aquacrop/core.py

def get_additional_information(self) -> Dict[str, Union[bool, float]]:
    """
    Additional model information.

    Returns:
        dict: {has_model_finished,execution_time}

    """
    if self.__has_model_executed:
        return {
            "has_model_finished": self.__has_model_finished,
            "execution_time": self.__end_model_execution
            - self.__start_model_execution,
        }
    else:
        raise ValueError(
            "You cannot get results without running the model. "
            + "Please execute the run_model() method."
        )

get_crop_growth()

Return crop growth results

Source code in aquacrop/core.py

def get_crop_growth(self):
    """
    Return crop growth results
    """
    if self.__has_model_executed:
        return self._outputs.crop_growth
    else:
        raise ValueError(
            "You cannot get results without running the model. "
            + "Please execute the run_model() method."
        )

get_simulation_results()

Return all the simulation results

Source code in aquacrop/core.py

def get_simulation_results(self):
    """
    Return all the simulation results
    """
    if self.__has_model_executed:
        if self.__has_model_finished:
            return self._outputs.final_stats
        else:
            return False  # If the model is not finished, the results are not generated.
    else:
        raise ValueError(
            "You cannot get results without running the model. "
            + "Please execute the run_model() method."
        )

get_water_flux()

Return water flux results

Source code in aquacrop/core.py

def get_water_flux(self):
    """
    Return water flux results
    """
    if self.__has_model_executed:
        return self._outputs.water_flux
    else:
        raise ValueError(
            "You cannot get results without running the model. "
            + "Please execute the run_model() method."
        )

get_water_storage()

Return water storage in soil results

Source code in aquacrop/core.py

def get_water_storage(self):
    """
    Return water storage in soil results
    """
    if self.__has_model_executed:
        return self._outputs.water_storage
    else:
        raise ValueError(
            "You cannot get results without running the model. "
            + "Please execute the run_model() method."
        )

run_model(num_steps=1, till_termination=False, initialize_model=True, process_outputs=False)

This function is responsible for executing the model.

Arguments:

num_steps: Number of steps (Days) to be executed.

till_termination: Run the simulation to completion

initialize_model: Whether to initialize the model             (i.e., go back to beginning of season)

process_outputs: process outputs into dataframe before                 simulation is finished

Returns:

Type	Description
bool	True if finished

Source code in aquacrop/core.py

def run_model(
    self,
    num_steps: int = 1,
    till_termination: bool = False,
    initialize_model: bool = True,
    process_outputs: bool = False,
) -> bool:
    """
    This function is responsible for executing the model.

    Arguments:

        num_steps: Number of steps (Days) to be executed.

        till_termination: Run the simulation to completion

        initialize_model: Whether to initialize the model \
        (i.e., go back to beginning of season)

        process_outputs: process outputs into dataframe before \
            simulation is finished

    Returns:
        True if finished
    """

    if initialize_model:
        self._initialize()

    if till_termination:
        self.__start_model_execution = time.time()
        while self._clock_struct.model_is_finished is False:

            (
                self._clock_struct,
                self._init_cond,
                self._param_struct,
                self._outputs,
            ) = self._perform_timestep()
        self.__end_model_execution = time.time()
        self.__has_model_executed = True
        self.__has_model_finished = True
        return True
    else:
        if num_steps < 1:
            raise ValueError("num_steps must be equal to or greater than 1.")
        self.__start_model_execution = time.time()
        for i in range(num_steps):

            if (i == range(num_steps)[-1]) and (process_outputs is True):
                self.__steps_are_finished = True

            (
                self._clock_struct,
                self._init_cond,
                self._param_struct,
                self._outputs,
            ) = self._perform_timestep()

            if self._clock_struct.model_is_finished:
                self.__end_model_execution = time.time()
                self.__has_model_executed = True
                self.__has_model_finished = True
                return True

        self.__end_model_execution = time.time()
        self.__has_model_executed = True
        self.__has_model_finished = False
        return True

CO2

Bases: object

Attributes:

ref_concentration (float): reference CO2 concentration

current_concentration (float): current CO2 concentration (initialize if constant_conc=True)

constant_conc (bool): use constant conc every season

co2_data (DataFrame): CO2 timeseries (2 columns: 'year' and 'ppm')

Source code in aquacrop/entities/co2.py

class CO2(object):

    """

    Attributes:

        ref_concentration (float): reference CO2 concentration

        current_concentration (float): current CO2 concentration (initialize if constant_conc=True)

        constant_conc (bool): use constant conc every season

        co2_data (DataFrame): CO2 timeseries (2 columns: 'year' and 'ppm')

    """

    def __init__(
        self,
        ref_concentration=369.41,
        current_concentration=0.,
        constant_conc=False,
        co2_data=None,
    ):
        self.ref_concentration = ref_concentration
        self.current_concentration = current_concentration
        self.constant_conc = constant_conc
        if co2_data is not None:
            self.co2_data = co2_data
        else:
            self.co2_data = pd.read_csv(
                    f"{acfp}/data/MaunaLoaCO2.txt",
                    header=1,
                    delim_whitespace=True,
                    names=["year", "ppm"],
    )
        self.co2_data_processed = None

FieldMngt

Field Management Class containing mulches and bunds parameters

Attributes:

mulches (bool):  Soil surface covered by mulches (yield_ or N)

bunds (bool):  Surface bunds present (yield_ or N)

curve_number_adj (bool): Field conditions affect curve number (yield_ or N)

sr_inhb (bool): Management practices fully inhibit surface runoff (yield_ or N)

mulch_pct (float):  Area of soil surface covered by mulches (%)

f_mulch (float): Soil evaporation adjustment factor due to effect of mulches

z_bund (float): Bund height, user specifies in (m) but immediately converted to (mm) on initialisation for coherent calculations

bund_water (float): Initial water height in surface bunds (mm)

curve_number_adj_pct (float): Percentage change in curve number (positive or negative)

Source code in aquacrop/entities/fieldManagement.py

class FieldMngt:
    """
    Field Management Class containing mulches and bunds parameters

    Attributes:

        mulches (bool):  Soil surface covered by mulches (yield_ or N)

        bunds (bool):  Surface bunds present (yield_ or N)

        curve_number_adj (bool): Field conditions affect curve number (yield_ or N)

        sr_inhb (bool): Management practices fully inhibit surface runoff (yield_ or N)

        mulch_pct (float):  Area of soil surface covered by mulches (%)

        f_mulch (float): Soil evaporation adjustment factor due to effect of mulches

        z_bund (float): Bund height, user specifies in (m) but immediately converted to (mm) on initialisation for coherent calculations

        bund_water (float): Initial water height in surface bunds (mm)

        curve_number_adj_pct (float): Percentage change in curve number (positive or negative)

    """

    def __init__(
        self,
        mulches=False,
        bunds=False,
        curve_number_adj=False,
        sr_inhb=False,
        mulch_pct=50,
        f_mulch=0.5,
        z_bund=0,
        bund_water=0,
        curve_number_adj_pct=0,
    ):

        self.mulches = mulches  #  Soil surface covered by mulches (yield_ or N)
        self.bunds = bunds  #  Surface bunds present (yield_ or N)
        self.curve_number_adj = curve_number_adj  # Field conditions affect curve number (yield_ or N)
        self.sr_inhb = sr_inhb  # Management practices fully inhibit surface runoff (yield_ or N)

        self.mulch_pct = mulch_pct  #  Area of soil surface covered by mulches (%)
        self.f_mulch = f_mulch  # Soil evaporation adjustment factor due to effect of mulches
        self.z_bund = z_bund * 1000 # Bund height, user-specified as (m), here immediately converted to (mm)
        self.bund_water = bund_water  # Initial water height in surface bunds (mm)
        self.curve_number_adj_pct = curve_number_adj_pct  # Percentage change in curve number (positive or negative)

GroundWater

Ground Water Class stores information on water table params

Attributes:

water_table (str):  Water table considered (Y or N)

method (str):  Water table input data ('Constant' or 'Variable')

dates (list): water table observation dates

values (list): water table observation depths

Source code in aquacrop/entities/groundWater.py

class GroundWater:
    """
    Ground Water Class stores information on water table params

    Attributes:

        water_table (str):  Water table considered (Y or N)

        method (str):  Water table input data ('Constant' or 'Variable')

        dates (list): water table observation dates

        values (list): water table observation depths

    """

    def __init__(self, water_table="N", method="Constant", dates=[], values=[]):

        self.water_table = water_table
        self.method = method
        self.dates = dates
        self.values = values

IrrigationManagement

IrrigationManagement Class defines irrigation strategy

Attributes:

irrigation_method (int):  Irrigation method {0: rainfed, 1: soil moisture targets, 2: set time interval,
                                        3: predifined schedule, 4: net irrigation, 5: constant depth }

WetSurf (int): Soil surface wetted by irrigation (%)

AppEff (int): Irrigation application efficiency (%)

MaxIrr (float): Maximum depth (mm) that can be applied each day

SMT (list):  Soil moisture targets (%taw) to maintain in each growth stage (only used if irrigation method is equal to 1)

IrrInterval (int): Irrigation interval in days (only used if irrigation method is equal to 2)

Schedule (pandas.DataFrame): DataFrame containing dates and depths

NetIrrSMT (float): Net irrigation threshold moisture level (% of taw that will be maintained, for irrigation_method=4)

Depth (float): constant depth to apply on each day (mm)

Source code in aquacrop/entities/irrigationManagement.py

class IrrigationManagement:

    """
    IrrigationManagement Class defines irrigation strategy

    Attributes:


        irrigation_method (int):  Irrigation method {0: rainfed, 1: soil moisture targets, 2: set time interval,
                                                3: predifined schedule, 4: net irrigation, 5: constant depth }

        WetSurf (int): Soil surface wetted by irrigation (%)

        AppEff (int): Irrigation application efficiency (%)

        MaxIrr (float): Maximum depth (mm) that can be applied each day

        SMT (list):  Soil moisture targets (%taw) to maintain in each growth stage (only used if irrigation method is equal to 1)

        IrrInterval (int): Irrigation interval in days (only used if irrigation method is equal to 2)

        Schedule (pandas.DataFrame): DataFrame containing dates and depths

        NetIrrSMT (float): Net irrigation threshold moisture level (% of taw that will be maintained, for irrigation_method=4)

        Depth (float): constant depth to apply on each day (mm)

    """

    def __init__(self, irrigation_method, **kwargs):
        self.irrigation_method = irrigation_method

        self.WetSurf = 100.0
        self.AppEff = 100.0
        self.MaxIrr = 25.0
        self.MaxIrrSeason = 10_000.0
        self.SMT = np.zeros(4)
        self.IrrInterval = 0
        self.Schedule = []
        self.NetIrrSMT = 80.0
        self.depth = 0.0

        if irrigation_method == 1:
            self.SMT = [100] * 4

        if irrigation_method == 2:
            self.IrrInterval = 3

        if irrigation_method == 3:
            # wants a pandas dataframe with Date and Depth, pd.Datetime and float
            """
            dates = pd.DatetimeIndex(['20/10/1979','20/11/1979','20/12/1979'])
            depths = [25,25,25]
            irr=pd.DataFrame([dates,depths]).T
            irr.columns=['Date','Depth']
            """
            self.Schedule = pd.DataFrame(columns=["Date", "Depth"])

        if irrigation_method == 4:
            self.NetIrrSMT = 80

        if irrigation_method == 5:
            self.depth = 0

        allowed_keys = {
            "name",
            "WetSurf",
            "AppEff",
            "MaxIrr",
            "MaxIrrSeason",
            "SMT",
            "IrrInterval",
            "NetIrrSMT",
            "Schedule",
            "depth",
        }

        self.__dict__.update((k, v) for k, v in kwargs.items() if k in allowed_keys)

Output

Class to hold output data

During Simulation these are numpy arrays and are converted to pandas dataframes at the end of the simulation

Atributes:

water_flux (pandas.DataFrame, numpy.array): Daily water flux changes

water_storage (pandas.DataFrame, numpy array): daily water content of each soil compartment

crop_growth (pandas.DataFrame, numpy array): daily crop growth variables

final_stats (pandas.DataFrame, numpy array): final stats at end of each season

Source code in aquacrop/entities/output.py

class Output:
    """
    Class to hold output data

    During Simulation these are numpy arrays and are converted to pandas dataframes
    at the end of the simulation

    Atributes:

        water_flux (pandas.DataFrame, numpy.array): Daily water flux changes

        water_storage (pandas.DataFrame, numpy array): daily water content of each soil compartment

        crop_growth (pandas.DataFrame, numpy array): daily crop growth variables

        final_stats (pandas.DataFrame, numpy array): final stats at end of each season

    """

    def __init__(self, time_span, initial_th):

        self.water_storage = np.zeros((len(time_span), 3 + len(initial_th)))
        self.water_flux = np.zeros((len(time_span), 16))
        self.crop_growth = np.zeros((len(time_span), 15))
        self.final_stats = pd.DataFrame(
            columns=[
                "Season",
                "crop Type",
                "Harvest Date (YYYY/MM/DD)",
                "Harvest Date (Step)",
                "Dry yield (tonne/ha)",
                "Fresh yield (tonne/ha)",
                "Yield potential (tonne/ha)",
                "Seasonal irrigation (mm)",
            ]
        )

check_iwc_soil_match(iwc_layers, soil_layers)

This function checks if the number of soil layers is equivalent between the user-specified soil profile and initial water content.

Return

boolean: True if number of layers match

Source code in aquacrop/core.py

def check_iwc_soil_match(iwc_layers: int, soil_layers: int) -> bool:
    """
    This function checks if the number of soil layers is equivalent between the user-specified soil profile and initial water content.

    Arguments:
        iwc_layers
        soil_layers

    Return:
        boolean: True if number of layers match

    """
    if(iwc_layers == soil_layers):
        return True
    else:
        return False

check_model_is_finished(step_end_time, simulation_end_date, model_is_finished, season_counter, n_seasons, harvest_flag)

Function to check and declare model termination

Arguments:

step_end_time (str):  date of next step

simulation_end_date (str):  date of end of simulation

model_is_finished (bool):  is model finished

season_counter (int):  tracking the number of seasons simulated

n_seasons (int):  total number of seasons being simulated

harvest_flag (bool):  Has crop been harvested

Returns:

model_is_finished (bool): is simulation finished

Source code in aquacrop/timestep/check_if_model_is_finished.py

def check_model_is_finished(
    step_end_time: str,
    simulation_end_date: str,
    model_is_finished: bool,
    season_counter: int,
    n_seasons: int,
    harvest_flag: bool,
) -> bool:
    """
    Function to check and declare model termination


    Arguments:

        step_end_time (str):  date of next step

        simulation_end_date (str):  date of end of simulation

        model_is_finished (bool):  is model finished

        season_counter (int):  tracking the number of seasons simulated

        n_seasons (int):  total number of seasons being simulated

        harvest_flag (bool):  Has crop been harvested

    Returns:

        model_is_finished (bool): is simulation finished


    """

    # Check if current time-step is the last
    current_time = step_end_time
    if current_time < simulation_end_date:
        model_is_finished = False
    elif current_time >= simulation_end_date:
        model_is_finished = True

    # Check if at the end of last growing season ##
    # Allow model to exit early if crop has reached maturity or died, and in
    # the last simulated growing season
    if (harvest_flag is True) and (season_counter == n_seasons - 1):
        model_is_finished = True

    return model_is_finished

compile_all_AOT_files()

Numba AOT compile functions to improve speed

Source code in aquacrop/scripts/checkIfPackageIsCompiled.py

def compile_all_AOT_files():
    """
    Numba AOT compile functions to improve speed

    """
    try:
        from ..solution.solution_aeration_stress import aeration_stress
        from ..solution.solution_water_stress import water_stress
        from ..solution.solution_evap_layer_water_content import (
            evap_layer_water_content,
        )
        from ..solution.solution_root_zone_water import root_zone_water
        from ..solution.solution_cc_development import cc_development
        from ..solution.solution_update_CCx_CDC import update_CCx_CDC
        from ..solution.solution_cc_required_time import cc_required_time
        from ..solution.solution_temperature_stress import temperature_stress
        from ..solution.solution_HIadj_pre_anthesis import HIadj_pre_anthesis
        from ..solution.solution_HIadj_post_anthesis import HIadj_post_anthesis
        from ..solution.solution_HIadj_pollination import HIadj_pollination
        from ..solution.solution_growing_degree_day import growing_degree_day
        from ..solution.solution_drainage import drainage
        from ..solution.solution_rainfall_partition import rainfall_partition
        from ..solution.solution_check_groundwater_table import check_groundwater_table
        from ..solution.solution_soil_evaporation import soil_evaporation
        from ..solution.solution_root_development import root_development
        from ..solution.solution_infiltration import infiltration
        from ..solution.solution_HIref_current_day import HIref_current_day
        from ..solution.solution_biomass_accumulation import biomass_accumulation

    except:
        print("\033[1;32m Compiling modules... This could take some time.")
        print(
            " Note: The compilation is only necessary the first time that the library is used."
        )
        call(["python", "-m", "aquacrop.scripts.initiate_library"])

compute_variables(param_struct, weather_df, clock_struct, acfp=dirname(dirname(abspath(__file__))))

Function to compute additional variables needed to run the model eg. CO2 Creates cropstruct jit class objects

Arguments:

param_struct (ParamStruct):  Contains model paramaters

weather_df (DataFrame):  weather data

clock_struct (ClockStruct):  time params

acfp (Path):  path to aquacrop directory containing co2 data

Returns:

param_struct (ParamStruct):  updated model params

Source code in aquacrop/initialize/compute_variables.py

def compute_variables(
    param_struct: "ParamStruct",
    weather_df: "DataFrame",
    clock_struct: "ClockStruct",
    acfp: str = dirname(dirname(abspath(__file__))),
) -> "ParamStruct":
    """
    Function to compute additional variables needed to run the model eg. CO2
    Creates cropstruct jit class objects

    Arguments:

        param_struct (ParamStruct):  Contains model paramaters

        weather_df (DataFrame):  weather data

        clock_struct (ClockStruct):  time params

        acfp (Path):  path to aquacrop directory containing co2 data

    Returns:

        param_struct (ParamStruct):  updated model params


    """

    if param_struct.water_table == 1:

        param_struct.Soil.add_capillary_rise_params()

    # Calculate readily evaporable water in surface layer
    if param_struct.Soil.adj_rew == 0:
        param_struct.Soil.rew = round(
            (
                1000
                * (
                    param_struct.Soil.profile.th_fc.iloc[0]
                    - param_struct.Soil.profile.th_dry.iloc[0]
                )
                * param_struct.Soil.evap_z_surf
            ),
            2,
        )

    if param_struct.Soil.calc_cn == 1:
        # adjust curve number
        ksat = param_struct.Soil.profile.Ksat.iloc[0]
        if ksat > 864:
            param_struct.Soil.cn = 46
        elif ksat > 347:
            param_struct.Soil.cn = 61
        elif ksat > 36:
            param_struct.Soil.cn = 72
        elif ksat > 0:
            param_struct.Soil.cn = 77

        assert ksat > 0

    for i in range(param_struct.NCrops):

        crop = param_struct.CropList[i]
        # crop.calculate_additional_params()

        # Crop calander
        crop = compute_crop_calendar(
            crop,
            clock_struct.planting_dates,
            clock_struct.simulation_start_date,
            clock_struct.simulation_end_date,
            clock_struct.time_span,
            weather_df,
        )

        # Harvest index param_struct.Seasonal_Crop_List[clock_struct.season_counter].Paramsgrowth coefficient
        crop.HIGC = calculate_HIGC(
            crop.YldFormCD,
            crop.HI0,
            crop.HIini,
        )

        # Days to linear harvest_index switch point
        if crop.CropType == 3:
            # Determine linear switch point and HIGC rate for fruit/grain crops
            crop.tLinSwitch, crop.dHILinear = calculate_HI_linear(
                crop.YldFormCD, crop.HIini, crop.HI0, crop.HIGC
            )
        else:
            # No linear switch for leafy vegetable or root/tiber crops
            crop.tLinSwitch = 0
            crop.dHILinear = 0.0

        param_struct.CropList[i] = crop

    # Calculate WP adjustment factor for elevation in CO2 concentration
    # Load CO2 data
    co2Data = param_struct.CO2.co2_data

    # Years
    start_year, end_year = pd.DatetimeIndex(
        [clock_struct.simulation_start_date, clock_struct.simulation_end_date]
    ).year
    sim_years = np.arange(start_year, end_year + 1)

    # Interpolate data
    CO2conc_interp = np.interp(sim_years, co2Data.year, co2Data.ppm)

    # Store data
    param_struct.CO2.co2_data_processed = pd.Series(CO2conc_interp, index=sim_years)  # maybe get rid of this

    # Get CO2 concentration for first year
    CO2conc = param_struct.CO2.co2_data_processed.iloc[0]

    # param_struct.CO2 = param_struct.co2_concentration_adj

    # if user specified constant concentration
    if  param_struct.CO2.constant_conc is True:
        if param_struct.CO2.current_concentration > 0.:
            CO2conc = param_struct.CO2.current_concentration
        else:
            CO2conc = param_struct.CO2.co2_data_processed.iloc[0]

    param_struct.CO2.current_concentration = CO2conc

    CO2ref = param_struct.CO2.ref_concentration

    # Get CO2 weighting factor for first year
    if CO2conc <= CO2ref:
        fw = 0
    else:
        if CO2conc >= 550:
            fw = 1
        else:
            fw = 1 - ((550 - CO2conc) / (550 - CO2ref))

    # Determine adjustment for each crop in first year of simulation
    for i in range(param_struct.NCrops):
        crop = param_struct.CropList[i]
        # Determine initial adjustment
        fCO2old = (CO2conc / CO2ref) / (
            1
            + (CO2conc - CO2ref)
            * (
                (1 - fw) * crop.bsted
                + fw * ((crop.bsted * crop.fsink) + (crop.bface * (1 - crop.fsink)))
            )
        )
        # New adjusted correction coefficient for CO2 (version 7 of AquaCrop)
    if (CO2conc > CO2ref):
        # Calculate shape factor
        fshape = -4.61824 - 3.43831*crop.fsink - 5.32587*crop.fsink*crop.fsink
        # Determine adjustment for CO2
        if (CO2conc >= 2000):
            fCO2new = 1.58  # Maximum CO2 adjustment 
        else:
            CO2rel = (CO2conc-CO2ref)/(2000-CO2ref)
            fCO2new = 1 + 0.58 * ((np.exp(CO2rel*fshape) - 1)/(np.exp(fshape) - 1))


    # Select adjusted coefficient for CO2
    if (CO2conc <= CO2ref):
        fCO2 = fCO2old
    elif ((CO2conc <= 550) and (fCO2old < fCO2new)):
        fCO2 = fCO2old
    else:
        fCO2 = fCO2new

        # Consider crop type
    if crop.WP >= 40:
        # No correction for C4 crops
        ftype = 0
    elif crop.WP <= 20:
        # Full correction for C3 crops
        ftype = 1
    else:
        ftype = (40 - crop.WP) / (40 - 20)

        # Total adjustment
    crop.fCO2 = 1 + ftype * (fCO2 - 1)

    param_struct.CropList[i] = crop


    # change this later
    if param_struct.NCrops == 1:
        crop_list = [
            deepcopy(param_struct.CropList[0])
            for i in range(len(param_struct.CropChoices))
        ]
        # param_struct.Seasonal_Crop_List = [deepcopy(param_struct.CropList[0]) for i in range(len(param_struct.CropChoices))]

    else:
        crop_list = param_struct.CropList

    # add crop for out of growing season
    # param_struct.Fallow_Crop = deepcopy(param_struct.Seasonal_Crop_List[0])
    Fallow_Crop = deepcopy(crop_list[0])

    param_struct.Seasonal_Crop_List = []
    for crop in crop_list:
        crop_struct = CropStruct()
        for a, v in crop.__dict__.items():
            if hasattr(crop_struct, a):
                crop_struct.__setattr__(a, v)

        param_struct.Seasonal_Crop_List.append(crop_struct)

    fallow_struct = CropStruct()
    for a, v in Fallow_Crop.__dict__.items():
        if hasattr(fallow_struct, a):
            fallow_struct.__setattr__(a, v)

    param_struct.Fallow_Crop = fallow_struct

    return param_struct

create_soil_profile(param_struct)

funciton to create soil profile namedTuple to store soil info. Its much faster to access the info when its in a namedTuple compared to a dataframe

Arguments:

param_struct (ParamStruct):  Contains model crop and soil paramaters

Returns:

param_struct (ParamStruct):  updated with soil profile

Source code in aquacrop/initialize/create_soil_profile.py

def create_soil_profile(param_struct: "ParamStruct") -> "ParamStruct":
    """
    funciton to create soil profile namedTuple to store soil info.
    Its much faster to access the info when its in a namedTuple
    compared to a dataframe

    Arguments:

        param_struct (ParamStruct):  Contains model crop and soil paramaters

    Returns:

        param_struct (ParamStruct):  updated with soil profile


    """

    profile = SoilProfile(int(param_struct.Soil.profile.shape[0]))

    pdf = param_struct.Soil.profile.astype("float64")

    profile.dz = pdf.dz.values
    profile.dzsum = pdf.dzsum.values
    profile.zBot = pdf.zBot.values
    profile.z_top = pdf.z_top.values
    profile.zMid = pdf.zMid.values

    profile.Comp = np.int64(pdf.Comp.values)
    profile.Layer = np.int64(pdf.Layer.values)
    # profile.Layer_dz = pdf.Layer_dz.values
    profile.th_wp = pdf.th_wp.values
    profile.th_fc = pdf.th_fc.values
    profile.th_s = pdf.th_s.values

    profile.Ksat = pdf.Ksat.values
    profile.Penetrability = pdf.penetrability.values
    profile.th_dry = pdf.th_dry.values
    profile.tau = pdf.tau.values
    profile.th_fc_Adj = pdf.th_fc_Adj.values

    if param_struct.water_table == 1:
        profile.aCR = pdf.aCR.values
        profile.bCR = pdf.bCR.values
    else:
        profile.aCR = pdf.dz.values * 0.0
        profile.bCR = pdf.dz.values * 0.0

    # param_struct.Soil.profile = profile

    param_struct.Soil.Profile = SoilProfileNT(
        dz=profile.dz,
        dzsum=profile.dzsum,
        zBot=profile.zBot,
        z_top=profile.z_top,
        zMid=profile.zMid,
        Comp=profile.Comp,
        Layer=profile.Layer,
        th_wp=profile.th_wp,
        th_fc=profile.th_fc,
        th_s=profile.th_s,
        Ksat=profile.Ksat,
        Penetrability=profile.Penetrability,
        th_dry=profile.th_dry,
        tau=profile.tau,
        th_fc_Adj=profile.th_fc_Adj,
        aCR=profile.aCR,
        bCR=profile.bCR,
    )

    return param_struct

outputs_when_model_is_finished(model_is_finished, flux_output, water_output, growth_outputs, steps_are_finished)

Function that turns numpy array outputs into pandas dataframes

Arguments:

model_is_finished (bool):  is model finished

flux_output (numpy.array): water flux_output

water_output (numpy.array):  water storage in each compartment

growth_outputs (numpy.array):  crop growth variables

n_seasons (int):  total number of seasons being simulated

steps_are_finished (bool):  have the simulated num_steps finished

Returns:

flux_output (pandas.DataFrame): water flux_output

water_output (pandas.DataFrame):  water storage in each compartment

growth_outputs (pandas.DataFrame):  crop growth variables

Source code in aquacrop/timestep/outputs_when_model_is_finished.py

def outputs_when_model_is_finished(
    model_is_finished: bool,
    flux_output: "ndarray",
    water_output: "ndarray",
    growth_outputs: "ndarray",
    steps_are_finished: bool,
):
    """
    Function that turns numpy array outputs into pandas dataframes

    Arguments:

        model_is_finished (bool):  is model finished

        flux_output (numpy.array): water flux_output

        water_output (numpy.array):  water storage in each compartment

        growth_outputs (numpy.array):  crop growth variables

        n_seasons (int):  total number of seasons being simulated

        steps_are_finished (bool):  have the simulated num_steps finished

    Returns:

        flux_output (pandas.DataFrame): water flux_output

        water_output (pandas.DataFrame):  water storage in each compartment

        growth_outputs (pandas.DataFrame):  crop growth variables


    """
    if model_is_finished is True or steps_are_finished is True:
        # ClockStruct.step_start_time = ClockStruct.step_end_time
        # ClockStruct.step_end_time = ClockStruct.step_end_time + np.timedelta64(1, "D")
        flux_output_df = pd.DataFrame(
            flux_output,
            columns=[
                "time_step_counter",
                "season_counter",
                "dap",
                "Wr",
                "z_gw",
                "surface_storage",
                "IrrDay",
                "Infl",
                "Runoff",
                "DeepPerc",
                "CR",
                "GwIn",
                "Es",
                "EsPot",
                "Tr",
                "TrPot",
            ],
        )

        water_output_df = pd.DataFrame(
            water_output,
            columns=["time_step_counter", "growing_season", "dap"]
            + ["th" + str(i) for i in range(1, water_output.shape[1] - 2)],
        )

        growth_outputs_df = pd.DataFrame(
            growth_outputs,
            columns=[
                "time_step_counter",
                "season_counter",
                "dap",
                "gdd",
                "gdd_cum",
                "z_root",
                "canopy_cover",
                "canopy_cover_ns",
                "biomass",
                "biomass_ns",
                "harvest_index",
                "harvest_index_adj",
                "DryYield",
                "FreshYield",
                "YieldPot",
            ],
        )

        return flux_output_df, water_output_df, growth_outputs_df

    return False

prepare_lars_weather(file, year, generated=True, order=['year', 'jday', 'minTemp', 'maxTemp', 'precip', 'rad'], wind_speed=3.4)

Uses FAO-PM to calculate reference evapotranspiration for LARS generated and baseline input data.

Source code in aquacrop/utils/lars.py

def prepare_lars_weather(
    file,
    year,
    generated=True,
    order=["year", "jday", "minTemp", "maxTemp", "precip", "rad"],
    wind_speed=3.4,
):
    """
    Uses FAO-PM to calculate reference evapotranspiration for LARS generated and baseline input data.

    """

    def vap_pres(t):
        return 0.6108 * np.exp((17.27 * t) / (t + 237.3))

    df = pd.read_csv(file, delim_whitespace=True, header=None)

    if generated:
        df.columns = order
        df["tdelta"] = pd.to_timedelta(df.jday, unit="D")
        df["date"] = pd.to_datetime(f"{year-1}/12/31") + df["tdelta"]

        psyc = 0.054  # sychometric constant
        tmean = (df.maxTemp + df.minTemp) / 2
        e_s = (vap_pres(df.maxTemp) + vap_pres(df.minTemp)) / 2
        e_a = vap_pres(df.minTemp)
        slope = 4098 * vap_pres(tmean) / (tmean + 237.3) ** 2
        R_ns = (1 - 0.23) * df.rad
        sb_const = 4.903e-9
        R_nl = (
            sb_const
            * 0.5
            * ((df.maxTemp + 273.15) ** 4 + (df.minTemp + 273.15) ** 4)
            * (0.34 - 0.14 * (e_a) ** 0.5)
            * (1.35 * 0.77 - 0.35)
        )
        Rn = R_ns - R_nl
        u2 = wind_speed

        eto = 0.408 * slope * Rn + (psyc * 900 * u2 * (e_s - e_a) / (tmean + 273)) / (
            slope + psyc * (1 + 0.34 * u2)
        )
        df["eto"] = eto

        # df["eto"] = df.rad*(0.0023)*(((df.maxTemp+df.minTemp)/2)+17.8)*(df.maxTemp-df.minTemp)**0.5
        df.eto = df.eto.clip(0.1)
        df = df[["simyear", "minTemp", "maxTemp", "precip", "eto", "date"]]
        df.columns = ["simyear", "MinTemp", "MaxTemp", "Precipitation", "ReferenceET", "Date"]

    else:
        df.columns = order
        df["date"] = pd.to_datetime(df.year, format="%Y") + pd.to_timedelta(df.jday - 1, unit="d")

        psyc = 0.054  # sychometric constant
        tmean = (df.maxTemp + df.minTemp) / 2
        e_s = (vap_pres(df.maxTemp) + vap_pres(df.minTemp)) / 2
        e_a = vap_pres(df.minTemp)
        slope = 4098 * vap_pres(tmean) / (tmean + 237.3) ** 2
        R_ns = (1 - 0.23) * df.rad
        sb_const = 4.903e-9
        R_nl = (
            sb_const
            * 0.5
            * ((df.maxTemp + 273.15) ** 4 + (df.minTemp + 273.15) ** 4)
            * (0.34 - 0.14 * (e_a) ** 0.5)
            * (1.35 * 0.77 - 0.35)
        )
        Rn = R_ns - R_nl
        u2 = wind_speed

        eto = 0.408 * slope * Rn + (psyc * 900 * u2 * (e_s - e_a) / (tmean + 273)) / (
            slope + psyc * (1 + 0.34 * u2)
        )
        df["eto"] = eto

        # df["eto"] = df.rad*(0.0023)*(((df.maxTemp+df.minTemp)/2)+17.8)*(df.maxTemp-df.minTemp)**0.5
        df.eto = df.eto.clip(0.1)
        df = df[["minTemp", "maxTemp", "precip", "eto", "date"]]
        df.columns = ["MinTemp", "MaxTemp", "Precipitation", "ReferenceET", "Date"]

    return df

read_clock_parameters(sim_start_time, sim_end_time, off_season=False)

Function to read in start and end simulation time and return a ClockStruct object

Arguments:

sim_start_time (str): simulation start date

sim_end_time (str): simulation start date

off_season (bool): True, simulate off season
                  False, skip ahead to next season post-harvest

Returns:

clock_struct (ClockStruct): simulation time paramaters

Source code in aquacrop/initialize/read_clocks_parameters.py

def read_clock_parameters(
    sim_start_time: str,
    sim_end_time: str,
    off_season: bool=False) -> ClockStruct:
    """
    Function to read in start and end simulation time and return a ClockStruct object

    Arguments:

        sim_start_time (str): simulation start date

        sim_end_time (str): simulation start date

        off_season (bool): True, simulate off season
                          False, skip ahead to next season post-harvest

    Returns:

        clock_struct (ClockStruct): simulation time paramaters


    """
    check_max_simulation_days(sim_start_time, sim_end_time)

    # Extract data and put into pandas datetime format
    pandas_sim_start_time = pd.to_datetime(sim_start_time)
    pandas_sim_end_time = pd.to_datetime(sim_end_time)

    # create ClockStruct object
    clock_struct = ClockStruct()

    # Add variables
    clock_struct.simulation_start_date = pandas_sim_start_time
    clock_struct.simulation_end_date = pandas_sim_end_time

    clock_struct.n_steps = (pandas_sim_end_time - pandas_sim_start_time).days + 1
    clock_struct.time_span = pd.date_range(
        freq="D", start=pandas_sim_start_time, end=pandas_sim_end_time
    )

    clock_struct.step_start_time = clock_struct.time_span[0]
    clock_struct.step_end_time = clock_struct.time_span[1]

    clock_struct.sim_off_season = off_season

    return clock_struct

read_field_management(ParamStruct, FieldMngt, FallowFieldMngt)

store field management variables as FieldMngtStruct object

Arguments:

ParamStruct (ParamStruct):  Contains model crop and soil paramaters

FieldMngt (FieldMngt):  field mngt params

FallowFieldMngt (FieldMngt): fallow field mngt params

Returns:

ParamStruct (ParamStruct):  updated ParamStruct with field management info

Source code in aquacrop/initialize/read_field_managment.py

def read_field_management(
    ParamStruct: "ParamStruct",
    FieldMngt: "FieldMngt",
    FallowFieldMngt: "FieldMngt") -> "ParamStruct":

    """
    store field management variables as FieldMngtStruct object

    Arguments:

        ParamStruct (ParamStruct):  Contains model crop and soil paramaters

        FieldMngt (FieldMngt):  field mngt params

        FallowFieldMngt (FieldMngt): fallow field mngt params

    Returns:

        ParamStruct (ParamStruct):  updated ParamStruct with field management info


    """

    field_mngt_struct = FieldMngtStruct()
    for a, v in FieldMngt.__dict__.items():
        if hasattr(field_mngt_struct, a):
            field_mngt_struct.__setattr__(a, v)

    fallow_field_mngt_struct = FieldMngtStruct()
    for a, v in FallowFieldMngt.__dict__.items():
        if hasattr(fallow_field_mngt_struct, a):
            fallow_field_mngt_struct.__setattr__(a, v)

    ParamStruct.FieldMngt = field_mngt_struct
    ParamStruct.FallowFieldMngt = fallow_field_mngt_struct

    return ParamStruct

read_groundwater_table(ParamStruct, GwStruct, ClockStruct)

Function to initialise groundwater parameters

Arguments:

ParamStruct (ParamStruct): Contains model paramaters

GwStruct (GroundWater): groundwater params

ClockStruct (ClockStruct): time params

Returns:

ParamStruct (ParamStruct): updated with GW info

Source code in aquacrop/initialize/read_groundwater_table.py

def read_groundwater_table(
    ParamStruct: "ParamStruct",
    GwStruct: "GroundWater",
    ClockStruct: "ClockStruct") -> "ParamStruct":
    """
    Function to initialise groundwater parameters

    Arguments:

        ParamStruct (ParamStruct): Contains model paramaters

        GwStruct (GroundWater): groundwater params

        ClockStruct (ClockStruct): time params

    Returns:

        ParamStruct (ParamStruct): updated with GW info

    """

    # assign water table value and method
    WT = GwStruct.water_table
    WTMethod = GwStruct.method

    # check if water table present
    if WT == "N":
        ParamStruct.water_table = 0
        ParamStruct.z_gw = 999 * np.ones(len(ClockStruct.time_span))
        ParamStruct.zGW_dates = ClockStruct.time_span
        ParamStruct.WTMethod = "None"
    elif WT == "Y":
        ParamStruct.water_table = 1

        df = pd.DataFrame([GwStruct.dates, GwStruct.values]).T
        df.columns = ["Date", "Depth(mm)"]

        # get date in correct format
        df.Date = pd.DatetimeIndex(df.Date)
        # print(f'DF length: {len(df)}')
        # print(f'Index length: {len(df.index)}')

        if len(df) == 1:

            # if only 1 watertable depth then set that value to be constant
            # accross whole simulation            
            z_gw = pd.DataFrame(
                data=df["Depth(mm)"].iloc[0]*np.ones(len(ClockStruct.time_span)),
                index=pd.to_datetime(ClockStruct.time_span),
                columns=['Depth(mm)']
            )['Depth(mm)']

        elif len(df) > 1:
            # check water table method
            if WTMethod == "Constant":

                # No interpolation between dates

                # create daily depths for each simulation day
                z_gw = pd.Series(
                    np.nan * np.ones(len(ClockStruct.time_span)), index=ClockStruct.time_span
                )

                # assign constant depth for all dates in between
                for row in range(len(df)):
                    date = df.Date.iloc[row]
                    depth = df["Depth(mm)"].iloc[row]
                    z_gw.loc[z_gw.index >= date] = depth
                    if row == 0:
                        z_gw.loc[z_gw.index <= date] = depth

            elif WTMethod == "Variable":

                # Linear interpolation between dates

                # create daily depths for each simulation day
                # fill unspecified days with NaN
                z_gw = pd.Series(
                    np.nan * np.ones(len(ClockStruct.time_span)), index=ClockStruct.time_span
                )

                for row in range(len(df)):
                    date = df.Date.iloc[row]
                    depth = df["Depth(mm)"].iloc[row]
                    z_gw.loc[date] = depth

                # Interpolate daily groundwater depths
                z_gw = z_gw.interpolate()

        # assign values to Paramstruct object
        ParamStruct.z_gw = z_gw.values
        ParamStruct.zGW_dates = z_gw.index.values
        ParamStruct.WTMethod = WTMethod

    return ParamStruct

read_irrigation_management(ParamStruct, IrrMngt, ClockStruct)

initilize irrigation management and store as IrrMngtStruct object

Arguments:

ParamStruct (ParamStruct):  Contains model crop and soil paramaters

IrrMngt (IrrigationManagement):  irr mngt params object

ClockStruct (ClockStruct):  time paramaters

Returns:

ParamStruct (ParamStruct):  updated model paramaters

Source code in aquacrop/initialize/read_irrigation_management.py

def read_irrigation_management(
    ParamStruct: "ParamStruct",
    IrrMngt: "IrrigationManagement",
    ClockStruct: "ClockStruct") -> "ParamStruct":
    """
    initilize irrigation management and store as IrrMngtStruct object

    Arguments:

        ParamStruct (ParamStruct):  Contains model crop and soil paramaters

        IrrMngt (IrrigationManagement):  irr mngt params object

        ClockStruct (ClockStruct):  time paramaters


    Returns:

        ParamStruct (ParamStruct):  updated model paramaters



    """
    # If specified, read input irrigation time-series
    if IrrMngt.irrigation_method == 3:

        df = IrrMngt.Schedule.copy()
        # change the index to the date
        df.index = pd.DatetimeIndex(df.Date)

        try:
            # create a dateframe containing the daily irrigation to
            # be applied for every day in the simulation
            df = df.reindex(ClockStruct.time_span, fill_value=0).drop("Date", axis=1)

            IrrMngt.Schedule = np.array(df.values, dtype=float).flatten()

        except TypeError:
            # older version of pandas with not reindex

            # create new dataframe for whole simulation
            # populate new dataframe with old values
            new_df = pd.DataFrame(data=np.zeros(len(ClockStruct.time_span)),
                index=pd.to_datetime(ClockStruct.time_span),
                columns=['Depth']
                )

            # fill in the new dataframe with irrigation schedule
            new_df.loc[df.index]=df.Depth.values

            IrrMngt.Schedule = np.array(new_df.values, dtype=float).flatten()

    else:

        IrrMngt.Schedule = np.zeros(len(ClockStruct.time_span))

    IrrMngt.SMT = np.array(IrrMngt.SMT, dtype=float)

    irr_mngt_struct = IrrMngtStruct(len(ClockStruct.time_span))
    for a, v in IrrMngt.__dict__.items():
        if hasattr(irr_mngt_struct, a):
            irr_mngt_struct.__setattr__(a, v)

    ParamStruct.IrrMngt = irr_mngt_struct
    ParamStruct.FallowIrrMngt = IrrMngtStruct(len(ClockStruct.time_span))

    return ParamStruct

read_model_initial_conditions(ParamStruct, ClockStruct, InitWC, crop)

Function to set up initial model conditions

Arguments:

ParamStruct (ParamStruct):  Contains model paramaters

ClockStruct (ClockStruct):  time paramaters

InitWC (InitialWaterContent):  initial water content

crop (Crop): crop parameters

Returns:

ParamStruct (ParamStruct):  updated ParamStruct object

InitCond (InitialCondition):  containing initial model conditions/counters

Source code in aquacrop/initialize/read_model_initial_conditions.py

def read_model_initial_conditions(
    ParamStruct: "ParamStruct",
    ClockStruct: "ClockStruct",
    InitWC: "InitialWaterContent",
    crop: "Crop") -> Tuple["ParamStruct", "InitialCondition"]:
    """
    Function to set up initial model conditions

    Arguments:

        ParamStruct (ParamStruct):  Contains model paramaters

        ClockStruct (ClockStruct):  time paramaters

        InitWC (InitialWaterContent):  initial water content

        crop (Crop): crop parameters


    Returns:

        ParamStruct (ParamStruct):  updated ParamStruct object

        InitCond (InitialCondition):  containing initial model conditions/counters

    """

    ###################
    # creat initial condition class
    ###################

    InitCond = InitialCondition(len(ParamStruct.Soil.profile))

    # class_args = {key:value for key, value in InitCond_class.__dict__.items() if not key.startswith('__') and not callable(key)}
    # InitCond = InitCondStruct(**class_args)

    if ClockStruct.season_counter == -1:
        InitCond.z_root = 0.
        InitCond.cc0_adj = 0.

    elif ClockStruct.season_counter == 0:
        InitCond.z_root = ParamStruct.Seasonal_Crop_List[0].Zmin
        InitCond.cc0_adj = ParamStruct.Seasonal_Crop_List[0].CC0

    # Set HIfinal to crop's reference harvest index
    InitCond.HIfinal = crop.HI0

    ##################
    # save field management
    ##################

    # Initial surface storage between any soil bunds
    if ClockStruct.season_counter == -1:
        # First day of simulation is in fallow period
        if (ParamStruct.FallowFieldMngt.bunds) and (
            float(ParamStruct.FallowFieldMngt.z_bund) > 0.001
        ):
            # Get initial storage between surface bunds
            InitCond.surface_storage = float(ParamStruct.FallowFieldMngt.bund_water)
            if InitCond.surface_storage > float(ParamStruct.FallowFieldMngt.z_bund):
                InitCond.surface_storage = float(ParamStruct.FallowFieldMngt.z_bund)
        else:
            # No surface bunds
            InitCond.surface_storage = 0

    elif ClockStruct.season_counter == 0:
        # First day of simulation is in first growing season
        # Get relevant field management structure parameters
        FieldMngtTmp = ParamStruct.FieldMngt
        if (FieldMngtTmp.bunds) and (float(FieldMngtTmp.z_bund) > 0.001):
            # Get initial storage between surface bunds
            InitCond.surface_storage = float(FieldMngtTmp.bund_water)
            if InitCond.surface_storage > float(FieldMngtTmp.z_bund):
                InitCond.surface_storage = float(FieldMngtTmp.z_bund)
        else:
            # No surface bunds
            InitCond.surface_storage = 0

    ############
    # watertable
    ############

    profile = ParamStruct.Soil.profile

    # Check for presence of groundwater table
    if ParamStruct.water_table == 0:  # No water table present
        # Set initial groundwater level to dummy value
        InitCond.z_gw = ModelConstants.NO_VALUE
        InitCond.wt_in_soil = False
        # Set adjusted field capacity to default field capacity
        InitCond.th_fc_Adj = profile.th_fc.values
    elif ParamStruct.water_table == 1:  # Water table is present
        # Set initial groundwater level
        InitCond.z_gw = float(ParamStruct.z_gw[ClockStruct.time_step_counter])
        # Find compartment mid-points
        zMid = profile.zMid
        # Check if water table is within modelled soil profile
        if InitCond.z_gw >= 0:
            idx = zMid[zMid >= InitCond.z_gw].index
            if idx.shape[0] == 0:
                InitCond.wt_in_soil = False
            else:
                InitCond.wt_in_soil = True
        else:
            InitCond.wt_in_soil = False

        # Adjust compartment field capacity
        compi = int(len(profile)) - 1
        thfcAdj = np.zeros(compi + 1)
        while compi >= 0:
            # get soil layer of compartment
            compdf = profile.loc[compi]
            if compdf.th_fc <= 0.1:
                Xmax = 1
            else:
                if compdf.th_fc >= 0.3:
                    Xmax = 2
                else:
                    pF = 2 + 0.3 * (compdf.th_fc - 0.1) / 0.2
                    Xmax = (np.exp(pF * np.log(10))) / 100

            if (InitCond.z_gw < 0) or ((InitCond.z_gw - zMid.iloc[compi]) >= Xmax):
                for ii in range(compi+1):
                    compdfii = profile.loc[ii]
                    thfcAdj[ii] = compdfii.th_fc

                compi = -1
            else:
                if compdf.th_fc >= compdf.th_s:
                    thfcAdj[compi] = compdf.th_fc
                else:
                    if zMid.iloc[compi] >= InitCond.z_gw:
                        thfcAdj[compi] = compdf.th_s
                    else:
                        dV = compdf.th_s - compdf.th_fc
                        dFC = (dV / (Xmax ** 2)) * ((zMid.iloc[compi] - (InitCond.z_gw - Xmax)) ** 2)
                        thfcAdj[compi] = compdf.th_fc + dFC

                compi = compi - 1

        # Store adjusted field capacity values
        InitCond.th_fc_Adj = np.round(thfcAdj, 3)

    profile["th_fc_Adj"] = np.round(InitCond.th_fc_Adj, 3)

    # create hydrology df to group by layer instead of compartment
    ParamStruct.Soil.Hydrology = profile.groupby("Layer").mean().drop(["dz", "dzsum"], axis=1)
    ParamStruct.Soil.Hydrology["dz"] = profile.groupby("Layer").sum().dz

    ###################
    # initial water contents
    ###################

    typestr = InitWC.wc_type
    methodstr = InitWC.method

    depth_layer = InitWC.depth_layer
    datapoints = InitWC.value

    values = np.zeros(len(datapoints))

    hydf = ParamStruct.Soil.Hydrology

    # Assign data
    if typestr == "Num":
        # Values are defined as numbers (m3/m3) so no calculation required
        depth_layer = np.array(depth_layer, dtype=float)
        values = np.array(datapoints, dtype=float)

    elif typestr == "Pct":
        # Values are defined as percentage of taw. Extract and assign value for
        # each soil layer based on calculated/input soil hydraulic properties
        depth_layer = np.array(depth_layer, dtype=float)
        datapoints = np.array(datapoints, dtype=float)

        for ii in range(len(values)):
            if methodstr == "Depth":
                depth = depth_layer[ii]
                value = datapoints[ii]

                # Find layer at specified depth
                if depth < profile.dzsum.iloc[-1]:
                    layer = profile.query(f"{depth}<dzsum").Layer.iloc[0]
                else:
                    layer = profile.Layer.iloc[-1]

                compdf = hydf.loc[layer]

                # Calculate moisture content at specified depth
                values[ii] = compdf.th_wp + ((value / 100) * (compdf.th_fc - compdf.th_wp))
            elif methodstr == "Layer":
                # Calculate moisture content at specified layer
                layer = depth_layer[ii]
                value = datapoints[ii]

                compdf = hydf.loc[layer]

                values[ii] = compdf.th_wp + ((value / 100) * (compdf.th_fc - compdf.th_wp))

    elif typestr == "Prop":
        # Values are specified as soil hydraulic properties (SAT, FC, or WP).
        # Extract and assign value for each soil layer
        depth_layer = np.array(depth_layer, dtype=float)
        datapoints = np.array(datapoints, dtype=str)

        for ii in range(len(values)):
            if methodstr == "Depth":
                # Find layer at specified depth
                depth = depth_layer[ii]
                value = datapoints[ii]

                # Find layer at specified depth
                if depth < profile.dzsum.iloc[-1]:
                    layer = profile.query(f"{depth}<dzsum").Layer.iloc[0]
                else:
                    layer = profile.Layer.iloc[-1]

                compdf = hydf.loc[layer]

                # Calculate moisture content at specified depth
                if value == "SAT":
                    values[ii] = compdf.th_s
                if value == "FC":
                    values[ii] = compdf.th_fc
                if value == "WP":
                    values[ii] = compdf.th_wp

            elif methodstr == "Layer":
                # Calculate moisture content at specified layer
                layer = depth_layer[ii]
                value = datapoints[ii]

                compdf = hydf.loc[layer]

                if value == "SAT":
                    values[ii] = compdf.th_s
                if value == "FC":
                    values[ii] = compdf.th_fc
                if value == "WP":
                    values[ii] = compdf.th_wp

    # Interpolate values to all soil compartments

    thini = np.zeros(int(profile.shape[0]))
    if methodstr == "Layer":
        for ii in range(len(values)):
            layer = depth_layer[ii]
            value = values[ii]

            idx = profile.query(f"Layer=={int(layer)}").index

            thini[idx] = value

        InitCond.th = thini

    elif methodstr == "Depth":
        depths = depth_layer

        # Add zero point
        if depths[0] > 0:
            depths = np.append([0], depths)
            values = np.append([values[0]], values)

        # Add end point (bottom of soil profile)
        if depths[-1] < ParamStruct.Soil.zSoil:
            depths = np.append(depths, [ParamStruct.Soil.zSoil])
            values = np.append(values, [values[-1]])

        # Find centroids of compartments
        SoilDepths = profile.dzsum.values
        comp_top = np.append([0], SoilDepths[:-1])
        comp_bot = SoilDepths
        comp_mid = (comp_top + comp_bot) / 2
        # Interpolate initial water contents to each compartment
        thini = np.interp(comp_mid, depths, values)
        InitCond.th = thini

    # If groundwater table is present and calculating water contents based on
    # field capacity, then reset value to account for possible changes in field
    # capacity caused by capillary rise effects
    if ParamStruct.water_table == 1:
        if (typestr == "Prop") and (datapoints[-1] == "FC"):
            InitCond.th = InitCond.th_fc_Adj

    # If groundwater table is present in soil profile then set all water
    # contents below the water table to saturation
    if InitCond.wt_in_soil is True:
        # Find compartment mid-points
        SoilDepths = profile.dzsum.values
        comp_top = np.append([0], SoilDepths[:-1])
        comp_bot = SoilDepths
        comp_mid = (comp_top + comp_bot) / 2
        idx = np.where(comp_mid >= InitCond.z_gw)[0][0]
        for ii in range(idx, len(profile)):
            layeri = profile.loc[ii].Layer
            InitCond.th[ii] = hydf.th_s.loc[layeri]

    InitCond.thini = InitCond.th

    ParamStruct.Soil.profile = profile
    ParamStruct.Soil.Hydrology = hydf

    return ParamStruct, InitCond

read_model_parameters(clock_struct, soil, crop, weather_df)

Finalise soil and crop paramaters including planting and harvest dates save to new object param_struct

Arguments:

clock_struct (ClockStruct):  time params

soil (Soil):  soil object

crop (Crop):  crop object

weather_df (DataFrame): list of datetimes

Returns:

clock_struct (ClockStruct): updated time paramaters

param_struct (ParamStruct):  Contains model crop and soil paramaters

Source code in aquacrop/initialize/read_model_parameters.py

def read_model_parameters(
    clock_struct: "ClockStruct",
    soil: "Soil",
    crop: "Crop",
    weather_df: "DataFrame"):

    """
    Finalise soil and crop paramaters including planting and harvest dates
    save to new object param_struct


    Arguments:

        clock_struct (ClockStruct):  time params

        soil (Soil):  soil object

        crop (Crop):  crop object

        weather_df (DataFrame): list of datetimes

    Returns:

        clock_struct (ClockStruct): updated time paramaters

        param_struct (ParamStruct):  Contains model crop and soil paramaters

    """
    # create param_struct object
    param_struct = ParamStruct()

    soil.fill_nan()

    # Assign soil object to param_struct
    param_struct.Soil = soil

    while soil.zSoil < crop.Zmax + 0.1:
        for i in soil.profile.index[::-1]:
            if soil.profile.loc[i, "dz"] < 0.25:
                soil.profile.loc[i, "dz"] += 0.1
                soil.fill_nan()
                break

    # TODO: Why all these commented lines? The model does not allow rotations now?
    ###########
    # crop
    ###########

    #     if isinstance(crop, Iterable):
    #         cropList=list(crop)
    #     else:
    #         cropList = [crop]

    #     # assign variables to paramstruct
    #     paramStruct.nCrops = len(cropList)
    #     if paramStruct.nCrops > 1:
    #         paramStruct.SpecifiedPlantcalendar = 'yield_'
    #     else:
    #         paramStruct.SpecifiedPlantcalendar = 'N'

    #     # add crop list to paramStruct
    #     paramStruct.cropList = cropList

    ############################
    # plant and harvest times
    ############################

    #     # find planting and harvest dates
    #     # check if there is more than 1 crop or multiple plant dates in sim year
    #     if paramStruct.SpecifiedPlantcalendar == "yield_":
    #         # if here than crop rotation occours during same period

    #         # create variables from dataframe
    #         plantingDates = pd.to_datetime(planting_dates)
    #         harvestDates = pd.to_datetime(harvest_dates)

    #         if (paramStruct.nCrops > 1):

    #             cropChoices = [crop.name for crop in paramStruct.cropList]

    #         assert len(cropChoices) == len(plantingDates) == len(harvestDates)

    # elif paramStruct.nCrops == 1:
    # Only one crop type considered during simulation - i.e. no rotations
    # either within or between years
    crop_list = [crop]
    param_struct.CropList = crop_list
    param_struct.NCrops = 1

    # Get start and end years for full simulation
    sim_start_date = clock_struct.simulation_start_date
    sim_end_date = clock_struct.simulation_end_date

    if crop.harvest_date is None:
        crop = compute_crop_calendar(
            crop,
            clock_struct.planting_dates,
            clock_struct.simulation_start_date,
            clock_struct.simulation_end_date,
            clock_struct.time_span,
            weather_df,
        )
        mature = int(crop.MaturityCD + 30)
        plant = pd.to_datetime("1990/" + crop.planting_date)
        harv = plant + np.timedelta64(mature, "D")
        new_harvest_date = str(harv.month) + "/" + str(harv.day)
        crop.harvest_date = new_harvest_date

    # extract years from simulation start and end date
    start_end_years = [sim_start_date.year, sim_end_date.year]

    # check if crop growing season runs over calander year
    # Planting and harvest dates are in days/months format so just add arbitrary year
    single_year = pd.to_datetime("1990/" + crop.planting_date) < pd.to_datetime(
        "1990/" + crop.harvest_date
    )

    if single_year:
        # if normal year

        # Check if the simulation in the following year does not exceed planting date.
        mock_simulation_end_date = pd.to_datetime("1990/" + f'{sim_end_date.month}' + "/" + f'{sim_end_date.day}')
        mock_simulation_start_date = pd.to_datetime("1990/" + crop.planting_date)
        last_simulation_year_does_not_start = mock_simulation_end_date <= mock_simulation_start_date

        if last_simulation_year_does_not_start:
            start_end_years[1] = start_end_years[1] - 1

        # specify the planting and harvest years as normal
        plant_years = list(range(start_end_years[0], start_end_years[1] + 1))
        harvest_years = plant_years
    else:
        # if it takes over a year then the plant year finishes 1 year before end of sim
        # and harvest year starts 1 year after sim start

        if (
            pd.to_datetime(str(start_end_years[1] + 2) + "/" + crop.harvest_date)
            < sim_end_date
        ):

            # specify shifted planting and harvest years
            plant_years = list(range(start_end_years[0], start_end_years[1] + 1))
            harvest_years = list(range(start_end_years[0] + 1, start_end_years[1] + 2))
        else:

            plant_years = list(range(start_end_years[0], start_end_years[1]))
            harvest_years = list(range(start_end_years[0] + 1, start_end_years[1] + 1))

    # Correct for partial first growing season (may occur when simulating
    # off-season soil water balance)
    if (
        pd.to_datetime(str(plant_years[0]) + "/" + crop.planting_date)
        < clock_struct.simulation_start_date
    ):
        # shift everything by 1 year
        plant_years = plant_years[1:]
        harvest_years = harvest_years[1:]

    # ensure number of planting and harvest years are the same
    assert len(plant_years) == len(harvest_years)

    # create lists to hold variables
    planting_dates = []
    harvest_dates = []
    crop_choices = []

    # save full harvest/planting dates and crop choices to lists
    for i, _ in enumerate(plant_years):
        planting_dates.append(
            str(plant_years[i]) + "/" + param_struct.CropList[0].planting_date
        )
        harvest_dates.append(
            str(harvest_years[i]) + "/" + param_struct.CropList[0].harvest_date
        )
        crop_choices.append(param_struct.CropList[0].Name)

    # save crop choices
    param_struct.CropChoices = list(crop_choices)

    # save clock paramaters
    clock_struct.planting_dates = pd.to_datetime(planting_dates)
    clock_struct.harvest_dates = pd.to_datetime(harvest_dates)
    clock_struct.n_seasons = len(planting_dates)

    # Initialise growing season counter
    if pd.to_datetime(clock_struct.step_start_time) == clock_struct.planting_dates[0]:
        clock_struct.season_counter = 0
    else:
        clock_struct.season_counter = -1

    # return the FileLocations object as i have added some elements
    return clock_struct, param_struct

read_weather_inputs(clock_sctruct, weather_df)

Clip weather to start and end simulation dates

Arguments:

clock_sctruct (ClockStruct): ClockStruct object

weather_df (DataFrame): weather dataframe

Returns:

weather_df (DataFrame): clipped weather dataframe

Source code in aquacrop/initialize/read_weather_inputs.py

def read_weather_inputs(
    clock_sctruct: "ClockStruct",
    weather_df: "DataFrame") -> "DataFrame":
    """
    Clip weather to start and end simulation dates

    Arguments:

        clock_sctruct (ClockStruct): ClockStruct object

        weather_df (DataFrame): weather dataframe

    Returns:

        weather_df (DataFrame): clipped weather dataframe

    """

    # get the start and end dates of simulation
    start_date = clock_sctruct.simulation_start_date
    end_date = clock_sctruct.simulation_end_date

    if weather_df.Date.iloc[0] > start_date:
        raise ValueError(
            "The first date of the climate data cannot be longer than the start date of the model."
        )

    if weather_df.Date.iloc[-1] < end_date:
        raise ValueError(
            "The model end date cannot be longer than the last date of climate data."
        )

    # remove weather data outside of simulation dates
    weather_df = weather_df[weather_df.Date >= start_date]
    weather_df = weather_df[weather_df.Date <= end_date]

    return weather_df

solution_single_time_step(init_cond, param_struct, clock_struct, weather_step, outputs)

Function to perform AquaCrop solution for a single time step

Arguments:

init_cond (InitialCondition):  containing current variables+counters

param_struct (ParamStruct):  contains model paramaters

clock_struct (ClockStruct):  model time paramaters

weather_step (numpy.ndarray):  contains precipitation,ET,temp_max,temp_min for current day

outputs (Output):  object to store outputs

Returns:

NewCond (InitialCondition):  containing updated simulation variables+counters

param_struct (ParamStruct):  contains model paramaters

outputs (Output):  object to store outputs

Source code in aquacrop/timestep/run_single_timestep.py

def solution_single_time_step(
    init_cond: "InitialCondition",
    param_struct: "ParamStruct",
    clock_struct: "ClockStruct",
    weather_step: "ndarray",
    outputs:"Output",
) ->  Tuple["InitialCondition", "ParamStruct","Output"]:
    """
    Function to perform AquaCrop solution for a single time step

    Arguments:

        init_cond (InitialCondition):  containing current variables+counters

        param_struct (ParamStruct):  contains model paramaters

        clock_struct (ClockStruct):  model time paramaters

        weather_step (numpy.ndarray):  contains precipitation,ET,temp_max,temp_min for current day

        outputs (Output):  object to store outputs

    Returns:

        NewCond (InitialCondition):  containing updated simulation variables+counters

        param_struct (ParamStruct):  contains model paramaters

        outputs (Output):  object to store outputs

    """

    # Unpack structures
    Soil = param_struct.Soil
    CO2 = param_struct.CO2
    precipitation = weather_step[2]
    temp_max = weather_step[1]
    temp_min = weather_step[0]
    et0 = weather_step[3]

    # Store initial conditions in structure for updating %%
    NewCond = init_cond
    if param_struct.water_table == 1:
        GroundWater = param_struct.z_gw[clock_struct.time_step_counter]
    else:
        GroundWater = 0

    # Check if growing season is active on current time step %%
    if clock_struct.season_counter >= 0:
        # Check if in growing season
        CurrentDate = clock_struct.step_start_time
        planting_date = clock_struct.planting_dates[clock_struct.season_counter]
        harvest_date = clock_struct.harvest_dates[clock_struct.season_counter]

        if (
            (planting_date <= CurrentDate)
            and (harvest_date >= CurrentDate)
            and (NewCond.crop_mature is False)
            and (NewCond.crop_dead is False)
        ):
            growing_season = True
        else:
            growing_season = False

        # Assign crop, irrigation management, and field management structures
        Crop_ = param_struct.Seasonal_Crop_List[clock_struct.season_counter]
        Crop_Name = param_struct.CropChoices[clock_struct.season_counter]
        IrrMngt = param_struct.IrrMngt

        if growing_season is True:
            FieldMngt = param_struct.FieldMngt
        else:
            FieldMngt = param_struct.FallowFieldMngt

    else:
        # Not yet reached start of first growing season
        growing_season = False
        # Assign crop, irrigation management, and field management structures
        # Assign first crop as filler crop
        Crop_ = param_struct.Fallow_Crop
        Crop_Name = "fallow"

        Crop_.Aer = 5
        Crop_.Zmin = 0.3
        IrrMngt = param_struct.FallowIrrMngt
        FieldMngt = param_struct.FallowFieldMngt

    # Increment time counters %%
    if growing_season is True:
        # Calendar days after planting
        NewCond.dap = NewCond.dap + 1
        # Growing degree days after planting

        gdd = growing_degree_day(
            Crop_.GDDmethod, Crop_.Tupp, Crop_.Tbase, temp_max, temp_min
        )

        # Update cumulative gdd counter
        NewCond.gdd = gdd
        NewCond.gdd_cum = NewCond.gdd_cum + gdd

        NewCond.growing_season = True
    else:
        NewCond.growing_season = False

        # Calendar days after planting
        NewCond.dap = 0
        # Growing degree days after planting
        gdd = 0.3
        NewCond.gdd_cum = 0

    # save current timestep counter
    NewCond.time_step_counter = clock_struct.time_step_counter
    NewCond.precipitation = weather_step[2]
    NewCond.temp_max = weather_step[1]
    NewCond.temp_min = weather_step[0]
    NewCond.et0 = weather_step[3]


    class_args = {
        key: value
        for key, value in Crop_.__dict__.items()
        if not key.startswith("__") and not callable(key)
    }
    Crop = CropStructNT(**class_args)

    # Run simulations %%
    # 1. Check for groundwater table
    NewCond.th_fc_Adj, NewCond.wt_in_soil, NewCond.z_gw = check_groundwater_table(
        Soil.Profile,
        NewCond.z_gw,
        NewCond.th,
        NewCond.th_fc_Adj,
        param_struct.water_table,
        GroundWater
    )

    # 2. Root development
    NewCond.z_root, NewCond.r_cor = root_development(
        Crop,
        Soil.Profile,
        NewCond.dap,
        NewCond.z_root,
        NewCond.delayed_cds,
        NewCond.gdd_cum,
        NewCond.delayed_gdds,
        NewCond.tr_ratio,
        NewCond.th,
        NewCond.canopy_cover,
        NewCond.canopy_cover_ns,
        NewCond.germination,
        NewCond.r_cor,
        NewCond.t_pot,
        NewCond.z_gw,
        gdd,
        growing_season,
        param_struct.water_table,
    )

    # 3. Pre-irrigation
    NewCond, PreIrr = pre_irrigation(
        Soil.Profile, Crop, NewCond, growing_season, IrrMngt
    )

    # 4. Drainage

    NewCond.th, DeepPerc, FluxOut = drainage(
        Soil.Profile,
        NewCond.th,
        NewCond.th_fc_Adj,
    )

    # 5. Surface runoff
    Runoff, Infl, NewCond.day_submerged = rainfall_partition(
        precipitation,
        NewCond.th,
        NewCond.day_submerged,
        FieldMngt.sr_inhb,
        FieldMngt.bunds,
        FieldMngt.z_bund,
        FieldMngt.curve_number_adj_pct,
        Soil.cn,
        Soil.adj_cn,
        Soil.z_cn,
        Soil.nComp,
        Soil.Profile,
    )

    # 6. Irrigation
    NewCond.depletion, NewCond.taw, NewCond.irr_cum, Irr = irrigation(
        IrrMngt.irrigation_method,
        IrrMngt.SMT,
        IrrMngt.AppEff,
        IrrMngt.MaxIrr,
        IrrMngt.IrrInterval,
        IrrMngt.Schedule,
        IrrMngt.depth,
        IrrMngt.MaxIrrSeason,
        NewCond.growth_stage,
        NewCond.irr_cum,
        NewCond.e_pot,
        NewCond.t_pot,
        NewCond.z_root,
        NewCond.th,
        NewCond.dap,
        NewCond.time_step_counter,
        Crop,
        Soil.Profile,
        Soil.z_top,
        growing_season,
        precipitation,
        Runoff,
    )

    # 7. Infiltration
    (
        NewCond.th,
        NewCond.surface_storage,
        DeepPerc,
        Runoff,
        Infl,
        FluxOut,
    ) = infiltration(
        Soil.Profile,
        NewCond.surface_storage,
        NewCond.th_fc_Adj,
        NewCond.th,
        Infl,
        Irr,
        IrrMngt.AppEff,
        FieldMngt.bunds,
        FieldMngt.z_bund,
        FluxOut,
        DeepPerc,
        Runoff,
        growing_season,
    )
    # 8. Capillary Rise
    NewCond, CR = capillary_rise(
        Soil.Profile,
        Soil.nLayer,
        Soil.fshape_cr,
        NewCond,
        FluxOut,
        param_struct.water_table,
    )

    # 9. Check germination
    NewCond = germination(
        NewCond,
        Soil.z_germ,
        Soil.Profile,
        Crop.GermThr,
        Crop.PlantMethod,
        gdd,
        growing_season,
    )

    # 10. Update growth stage
    NewCond = growth_stage(Crop, NewCond, growing_season)

    # 11. Canopy cover development
    NewCond = canopy_cover(
        Crop, Soil.Profile, Soil.z_top, NewCond, gdd, et0, growing_season
    )

    # 12. Soil evaporation
    (
        NewCond.e_pot,
        NewCond.th,
        NewCond.stage2,
        NewCond.w_stage_2,
        NewCond.w_surf,
        NewCond.surface_storage,
        NewCond.evap_z,
        Es,
        EsPot,
    ) = soil_evaporation(
        clock_struct.evap_time_steps,
        clock_struct.sim_off_season,
        clock_struct.time_step_counter,
        Soil.Profile,
        Soil.evap_z_min,
        Soil.evap_z_max,
        Soil.rew,
        Soil.kex,
        Soil.fwcc,
        Soil.f_wrel_exp,
        Soil.f_evap,
        Crop.CalendarType,
        Crop.Senescence,
        IrrMngt.irrigation_method,
        IrrMngt.WetSurf,
        FieldMngt.mulches,
        FieldMngt.f_mulch,
        FieldMngt.mulch_pct,
        NewCond.dap,
        NewCond.w_surf,
        NewCond.evap_z,
        NewCond.stage2,
        NewCond.th,
        NewCond.delayed_cds,
        NewCond.gdd_cum,
        NewCond.delayed_gdds,
        NewCond.ccx_w,
        NewCond.canopy_cover_adj,
        NewCond.ccx_act,
        NewCond.canopy_cover,
        NewCond.premat_senes,
        NewCond.surface_storage,
        NewCond.w_stage_2,
        NewCond.e_pot,
        et0,
        Infl,
        precipitation,
        Irr,
        growing_season,
    )

    # 13. Crop transpiration
    Tr, TrPot_NS, TrPot, NewCond, IrrNet = transpiration(
        Soil.Profile,
        Soil.nComp,
        Soil.z_top,
        Crop,
        IrrMngt.irrigation_method,
        IrrMngt.NetIrrSMT,
        NewCond,
        et0,
        CO2,
        growing_season,
        gdd,
    )

    # 14. Groundwater inflow
    NewCond, GwIn = groundwater_inflow(Soil.Profile, NewCond)

    # 15. Reference harvest index
    (NewCond.hi_ref, NewCond.yield_form, NewCond.pct_lag_phase) = HIref_current_day( # ,NewCond.HIfinal
        NewCond.hi_ref,
        NewCond.HIfinal,
        NewCond.dap,
        NewCond.delayed_cds,
        NewCond.yield_form,
        NewCond.pct_lag_phase,
        NewCond.canopy_cover,
        NewCond.cc_prev,
        NewCond.ccx_w,
        Crop,
        growing_season,
    )

    # 16. Biomass accumulation
    (NewCond.biomass, NewCond.biomass_ns) = biomass_accumulation(
        Crop,
        NewCond.dap,
        NewCond.delayed_cds,
        NewCond.hi_ref,
        NewCond.pct_lag_phase,
        NewCond.biomass,
        NewCond.biomass_ns,
        Tr,
        TrPot_NS,
        et0,
        growing_season,
    )

    # 17. Harvest index
    NewCond = harvest_index(
        Soil.Profile, Soil.z_top, Crop, NewCond, et0, temp_max, temp_min, growing_season
    )

    # 18. Yield potential
    NewCond.YieldPot = (NewCond.biomass_ns / 100) * NewCond.harvest_index

    # 19. Crop yield_ (dry and fresh)
    if growing_season is True:
        # Calculate crop yield_ (tonne/ha)
        NewCond.DryYield = (NewCond.biomass / 100) * NewCond.harvest_index_adj
        NewCond.FreshYield = NewCond.DryYield / (Crop.YldWC / 100)
        # print( clock_struct.time_step_counter,(NewCond.biomass/100),NewCond.harvest_index_adj)
        # Check if crop has reached maturity
        if ((Crop.CalendarType == 1) and (NewCond.dap >= Crop.Maturity)) or (
            (Crop.CalendarType == 2) and (NewCond.gdd_cum >= Crop.Maturity)
        ):
            # Crop has reached maturity
            NewCond.crop_mature = True

    elif growing_season is False:
        # Crop yield_ is zero outside of growing season
        NewCond.DryYield = 0
        NewCond.FreshYield = 0

    # 20. Root zone water
    _TAW = TAW()
    _water_root_depletion = Dr()
    # thRZ = RootZoneWater()

    Wr, _water_root_depletion.Zt, _water_root_depletion.Rz, _TAW.Zt, _TAW.Rz, _, _, _, _, _, _ = root_zone_water(
        Soil.Profile,
        float(NewCond.z_root),
        NewCond.th,
        Soil.z_top,
        float(Crop.Zmin),
        Crop.Aer,
    )

    # 21. Update net irrigation to add any pre irrigation
    IrrNet = IrrNet + PreIrr
    NewCond.irr_net_cum = NewCond.irr_net_cum + PreIrr

    # Update model outputs %%
    row_day = clock_struct.time_step_counter
    row_gs = clock_struct.season_counter

    # Irrigation
    if growing_season is True:
        if IrrMngt.irrigation_method == 4:
            # Net irrigation
            IrrDay = IrrNet
            IrrTot = NewCond.irr_net_cum
        else:
            # Irrigation
            IrrDay = Irr
            IrrTot = NewCond.irr_cum

    else:
        IrrDay = 0
        IrrTot = 0

        NewCond.depletion = _water_root_depletion.Rz
        NewCond.taw = _TAW.Rz

    # Water contents
    outputs.water_storage[row_day, :3] = np.array(
        [clock_struct.time_step_counter, growing_season, NewCond.dap]
    )
    outputs.water_storage[row_day, 3:] = NewCond.th

    # Water fluxes
    # print(f'Saving NewCond.z_gw to outputs: {NewCond.z_gw}')
    outputs.water_flux[row_day, :] = [
        clock_struct.time_step_counter,
        clock_struct.season_counter,
        NewCond.dap,
        Wr,
        NewCond.z_gw,
        NewCond.surface_storage,
        IrrDay,
        Infl,
        Runoff,
        DeepPerc,
        CR,
        GwIn,
        Es,
        EsPot,
        Tr,
        TrPot,
    ]

    # Crop growth
    outputs.crop_growth[row_day, :] = [
        clock_struct.time_step_counter,
        clock_struct.season_counter,
        NewCond.dap,
        gdd,
        NewCond.gdd_cum,
        NewCond.z_root,
        NewCond.canopy_cover,
        NewCond.canopy_cover_ns,
        NewCond.biomass,
        NewCond.biomass_ns,
        NewCond.harvest_index,
        NewCond.harvest_index_adj,
        NewCond.DryYield,
        NewCond.FreshYield,
        NewCond.YieldPot,
    ]

    # Final output (if at end of growing season)
    if clock_struct.season_counter > -1:
        if (
            (NewCond.crop_mature is True)
            or (NewCond.crop_dead is True)
            or (
                clock_struct.harvest_dates[clock_struct.season_counter]
                == clock_struct.step_end_time
            )
        ) and (NewCond.harvest_flag is False):

            # Store final outputs
            outputs.final_stats.loc[row_gs] = [
                clock_struct.season_counter,
                Crop_Name,
                clock_struct.step_end_time,
                clock_struct.time_step_counter,
                NewCond.DryYield,
                NewCond.FreshYield,
                NewCond.YieldPot,
                IrrTot,
            ]

            # Set harvest flag
            NewCond.harvest_flag = True

    return NewCond, param_struct, outputs

update_time(clock_struct, init_cond, param_struct, weather, crop)

Function to update current time in model.

Arguments:

clock_struct (ClockStruct):  model time paramaters

init_cond (InitialCondition):  containing sim variables+counters

param_struct (ParamStruct):  containing model paramaters

weather (numpy.array):  weather data for simulation period

Returns:

clock_struct (ClockStruct):  model time paramaters

init_cond (InitialCondition):  containing reset model paramaters

param_struct (ParamStruct):  containing model paramaters

Source code in aquacrop/timestep/update_time.py

def update_time(
    clock_struct: "ClockStruct",
    init_cond: "InitialCondition",
    param_struct: "ParamStruct",
    weather: "ndarray",
    crop: "Crop",
    ) -> Tuple["ClockStruct","InitialCondition", "ParamStruct"]:
    """
    Function to update current time in model.

    Arguments:

        clock_struct (ClockStruct):  model time paramaters

        init_cond (InitialCondition):  containing sim variables+counters

        param_struct (ParamStruct):  containing model paramaters

        weather (numpy.array):  weather data for simulation period

    Returns:

        clock_struct (ClockStruct):  model time paramaters

        init_cond (InitialCondition):  containing reset model paramaters

        param_struct (ParamStruct):  containing model paramaters
    """
    # Update time
    if clock_struct.model_is_finished is False:
        if (init_cond.harvest_flag is True) and (
            (clock_struct.sim_off_season is False)
        ):
            # TODO: sim_off_season will always be False.

            # End of growing season has been reached and not simulating
            # off-season soil water balance. Advance time to the start of the
            # next growing season.
            # Check if in last growing season
            if clock_struct.season_counter < clock_struct.n_seasons - 1:
                # Update growing season counter
                clock_struct.season_counter = clock_struct.season_counter + 1
                # Update time-step counter

                clock_struct.time_step_counter = clock_struct.time_span.get_loc(
                    clock_struct.planting_dates[clock_struct.season_counter]
                )
                # Update start time of time-step
                clock_struct.step_start_time = clock_struct.time_span[
                    clock_struct.time_step_counter
                ]
                # Update end time of time-step
                clock_struct.step_end_time = clock_struct.time_span[
                    clock_struct.time_step_counter + 1
                ]
                # Reset initial conditions for start of growing season
                init_cond, param_struct = reset_initial_conditions(
                    clock_struct, init_cond, param_struct, weather, crop
                )

        else:
            # Simulation considers off-season, so progress by one time-step
            # (one day)
            # Time-step counter
            clock_struct.time_step_counter = clock_struct.time_step_counter + 1
            # Start of time step (beginning of current day)
            # clock_struct.time_span = pd.Series(clock_struct.time_span)
            clock_struct.step_start_time = clock_struct.time_span[
                clock_struct.time_step_counter
            ]
            # End of time step (beginning of next day)
            clock_struct.step_end_time = clock_struct.time_span[
                clock_struct.time_step_counter + 1
            ]
            # Check if it is not the last growing season
            if clock_struct.season_counter < clock_struct.n_seasons - 1:
                # Check if upcoming day is the start of a new growing season
                if (
                    clock_struct.step_start_time
                    == clock_struct.planting_dates[clock_struct.season_counter + 1]
                ):
                    # Update growing season counter
                    clock_struct.season_counter = clock_struct.season_counter + 1
                    # Reset initial conditions for start of growing season
                    init_cond, param_struct = reset_initial_conditions(
                        clock_struct, init_cond, param_struct, weather, crop
                    )

    return clock_struct, init_cond, param_struct

aquacrop.utils.prepare_weather

prepare_weather(weather_file_path)

function to read in weather data and return a dataframe containing
the weather data

Arguments:

FileLocations): FileLocationsClass: input File Locations

weather_file_path): `str): file location of weather data

Returns:

weather_df (pandas.DataFrame): weather data for simulation period

Source code in aquacrop/utils/prepare_weather.py

def prepare_weather(weather_file_path):
    """
    function to read in weather data and return a dataframe containing
    the weather data

    Arguments:\n

FileLocations): `FileLocationsClass`:  input File Locations

weather_file_path): `str):  file location of weather data



    Returns:

weather_df (pandas.DataFrame):  weather data for simulation period

    """

    weather_df = pd.read_csv(weather_file_path, header=0, delim_whitespace=True)

    assert len(weather_df.columns) == 7

    # rename the columns
    weather_df.columns = str(
        "Day Month Year MinTemp MaxTemp Precipitation ReferenceET").split()

    # put the weather dates into datetime format
    weather_df["Date"] = pd.to_datetime(weather_df[["Year", "Month", "Day"]])

    # drop the day month year columns
    weather_df = weather_df.drop(["Day", "Month", "Year"], axis=1)

    # set limit on ET0 to avoid divide by zero errors
    weather_df.ReferenceET.clip(lower=0.1, inplace=True)

    return weather_df