"""The :mod:`~virtual_ecosystem.models.abiotic.abiotic_model` module creates a
:class:`~virtual_ecosystem.models.abiotic.abiotic_model.AbioticModel`
class as a child of the :class:`~virtual_ecosystem.core.base_model.BaseModel` class.
This implements the full complexity abiotic model.
""" # noqa: D205
from __future__ import annotations
from typing import Any
import numpy as np
from pyrealm.constants import CoreConst as PyrealmCoreConst
from virtual_ecosystem.core.base_model import BaseModel
from virtual_ecosystem.core.configuration import CompiledConfiguration
from virtual_ecosystem.core.core_components import CoreComponents
from virtual_ecosystem.core.data import Data
from virtual_ecosystem.core.logger import LOGGER
from virtual_ecosystem.core.model_config import CoreConfiguration
from virtual_ecosystem.models.abiotic.energy_balance import (
initialise_canopy_and_soil_fluxes,
)
from virtual_ecosystem.models.abiotic.microclimate import run_microclimate
from virtual_ecosystem.models.abiotic.model_config import (
AbioticConfiguration,
AbioticConstants,
)
from virtual_ecosystem.models.abiotic_simple.microclimate_simple import (
calculate_vapour_pressure_deficit,
run_simple_microclimate,
)
from virtual_ecosystem.models.abiotic_simple.model_config import AbioticSimpleBounds
[docs]
class AbioticModel(
BaseModel,
model_name="abiotic",
model_update_bounds=("1 hour", "1 month"),
vars_required_for_init=(
"air_temperature_ref",
"atmospheric_co2_ref",
"atmospheric_pressure_ref",
"layer_heights",
"leaf_area_index",
"mean_annual_temperature",
"relative_humidity_ref",
"shortwave_absorption",
"wind_speed_ref",
"downward_longwave_radiation",
# These four aren't actually required but they _are_ populated by
# HydrologyModel.__init__ and the current logic for static model update checking
# objects when the data provides _some_ of the variables that a model updates
# and that do not appear in this list.
"aerodynamic_resistance_canopy",
"specific_heat_air",
"latent_heat_vapourisation",
"density_air",
"condensation",
),
vars_updated=(
"air_temperature",
"canopy_temperature",
"soil_temperature",
"vapour_pressure",
"vapour_pressure_deficit",
"relative_humidity",
"wind_speed",
"sensible_heat_flux",
"latent_heat_flux",
"ground_heat_flux",
"density_air",
"specific_heat_air",
"latent_heat_vapourisation",
"aerodynamic_resistance_canopy",
"net_radiation",
"longwave_emission",
"diurnal_temperature_range",
"condensation",
),
vars_required_for_update=(
"air_temperature_ref",
"relative_humidity_ref",
"vapour_pressure_deficit_ref",
"atmospheric_pressure_ref",
"atmospheric_co2_ref",
"wind_speed_ref",
"leaf_area_index",
"layer_heights",
"downward_shortwave_radiation",
"stomatal_conductance",
"shortwave_absorption",
"downward_longwave_radiation",
"aerodynamic_resistance_soil",
"soil_evaporation",
"canopy_evaporation",
"transpiration",
"condensation",
),
vars_populated_by_init=(
"soil_temperature",
"vapour_pressure_ref",
"vapour_pressure_deficit_ref",
"air_temperature",
"relative_humidity",
"vapour_pressure_deficit",
"wind_speed",
"atmospheric_pressure",
"atmospheric_co2",
"canopy_temperature",
"sensible_heat_flux",
"latent_heat_flux",
"ground_heat_flux",
"net_radiation",
"longwave_emission",
"vapour_pressure",
"diurnal_temperature_range",
),
vars_populated_by_first_update=(),
):
"""A class describing the abiotic model.
Args:
data: The data object to be used in the model.
latitude: Mean latitude of the study grid
core_components: The core components used across models.
model_constants: Set of constants for the abiotic model.
pyrealm_core_constants: Additional configuration options to the pyrealm
package.
bounds: A set of bounds to be applied to abiotic variables.
static: Boolean flag indicating if the model should run in static mode.
"""
def __init__(
self,
data: Data,
latitude: float,
core_components: CoreComponents,
model_constants: AbioticConstants = AbioticConstants(),
pyrealm_core_constants: PyrealmCoreConst = PyrealmCoreConst(),
bounds: AbioticSimpleBounds = AbioticSimpleBounds(),
static: bool = False,
):
"""Abiotic init function.
The init function is used only to define class attributes. Any logic should be
handled in :fun:`~virtual_ecosystem.abiotic.abiotic_model._setup`.
"""
super().__init__(data, core_components, static)
self.model_constants: AbioticConstants
"""Set of constants for the abiotic model."""
self.bounds: AbioticSimpleBounds
"""A set of bounds on microclimates variables, used with both the simple model
of the initial state and the full energy balance calculations."""
self.pyrealm_core_constants: PyrealmCoreConst
"""Pyrealm core constants."""
self.latitude: float
"""Latitude in degrees. This is required to generate the diurnal cycle."""
# Run the setup if the model is not in deep static mode
if self._run_setup:
self._setup(
latitude=latitude,
model_constants=model_constants,
pyrealm_core_constants=pyrealm_core_constants,
bounds=bounds,
)
def _setup(
self,
latitude: float,
model_constants: AbioticConstants = AbioticConstants(),
pyrealm_core_constants: PyrealmCoreConst = PyrealmCoreConst(),
bounds: AbioticSimpleBounds = AbioticSimpleBounds(),
) -> None:
"""Function to set up the abiotic model.
This function initializes soil temperature and canopy temperature for all
corresponding layers and calculates the reference vapour pressure deficit for
all time steps of the simulation. All variables are added directly to the
self.data object.
See __init__ for argument descriptions.
"""
self.latitude = latitude
self.model_constants = model_constants
self.pyrealm_core_constants = pyrealm_core_constants
self.bounds = bounds
# Calculate vapour pressure deficit at reference height for all time steps
vapour_pressure_and_deficit = calculate_vapour_pressure_deficit(
temperature=self.data["air_temperature_ref"],
relative_humidity=self.data["relative_humidity_ref"],
pyrealm_core_constants=self.pyrealm_core_constants,
)
self.data["vapour_pressure_deficit_ref"] = (
vapour_pressure_and_deficit["vapour_pressure_deficit"]
).rename("vapour_pressure_deficit_ref")
self.data["vapour_pressure_ref"] = (
vapour_pressure_and_deficit["vapour_pressure"]
).rename("vapour_pressure_ref")
# Generate initial profiles of air temperature [C], relative humidity [-],
# vapour pressure deficit [kPa], soil temperature [C], atmospheric pressure
# [kPa], and atmospheric :math:`\ce{CO2}` [ppm]
initial_microclimate = run_simple_microclimate(
data=self.data,
layer_structure=self.layer_structure,
time_index=0,
constants=self.model_constants,
core_constants=self.core_constants,
pyrealm_core_constants=pyrealm_core_constants,
bounds=self.bounds,
)
# Initialise diurnal temperature range, [C]
diurnal_temperature_range = initial_microclimate["air_temperature"].copy()
self.data["diurnal_temperature_range"] = diurnal_temperature_range.where(
diurnal_temperature_range.isnull(),
other=5.0, # TODO add data when available
)
# Generate initial profiles of canopy temperature and heat fluxes from soil and
# canopy
initial_canopy_and_soil = initialise_canopy_and_soil_fluxes(
air_temperature=initial_microclimate["air_temperature"],
layer_structure=self.layer_structure,
initial_flux_value=self.model_constants.initial_flux_value,
)
# Update data object
for output_dict in (
initial_microclimate,
initial_canopy_and_soil,
):
self.data.add_from_dict(output_dict=output_dict)
[docs]
@classmethod
def from_config(
cls,
data: Data,
configuration: CompiledConfiguration,
core_components: CoreComponents,
) -> AbioticModel:
"""Factory function to initialise the abiotic model from configuration.
This function unpacks the relevant information from the configuration file, and
then uses it to initialise the model. If any information from the config is
invalid rather than returning an initialised model instance an error is raised.
Args:
data: A :class:`~virtual_ecosystem.core.data.Data` instance.
configuration: A validated Virtual Ecosystem model configuration object.
core_components: The core components used across models.
"""
# Extract the validated model configuration from the complete compiled
# configuration. This syntax is odd but required to support static typing
model_configuration: AbioticConfiguration = configuration.get_subconfiguration(
"abiotic", AbioticConfiguration
)
core_configuration: CoreConfiguration = configuration.get_subconfiguration(
"core", CoreConfiguration
)
LOGGER.info(
"Information required to initialise the abiotic model successfully "
"extracted."
)
return cls(
data=data,
core_components=core_components,
static=model_configuration.static,
latitude=model_configuration.latitude,
model_constants=model_configuration.constants,
pyrealm_core_constants=core_configuration.pyrealm.core,
bounds=model_configuration.bounds,
)
[docs]
def spinup(self) -> None:
"""Placeholder function to spin up the abiotic model."""
def _update(self, time_index: int, **kwargs: Any) -> None:
"""Function to update the abiotic model.
Args:
time_index: The index of the current time step in the data object.
**kwargs: Further arguments to the update method.
"""
month = (
self.model_timing.update_datestamps[time_index]
.astype("datetime64[M]")
.astype(int)
% 12
+ 1
)
# Determine number of days
days_float: float = (
self.model_timing.update_interval_seconds
/ self.core_constants.seconds_to_day
)
days: int = int(days_float // 1)
# Check if the number of days is exact and warn if not
if not np.allclose(days_float % 1, 0):
LOGGER.warning(
f"Update interval is not a whole number of days ({days_float}),"
f" partitioning inputs among {days} days."
)
# Run microclimate model
update_dict = run_microclimate(
data=self.data,
vars_updated=self.vars_updated,
time_index=time_index,
time_dim=self.core_constants.hours_per_day,
time_interval=self.model_timing.update_interval_seconds,
month=month,
days=days,
latitude=self.latitude,
layer_structure=self.layer_structure,
abiotic_constants=self.model_constants,
core_constants=self.core_constants,
pyrealm_core_constants=self.pyrealm_core_constants,
abiotic_bounds=self.bounds,
)
self.data.add_from_dict(output_dict=update_dict)
[docs]
def cleanup(self) -> None:
"""Placeholder function for abiotic model cleanup."""
