dac_workflow_manager

Classes:

• DACWorkflowManager –

DACWorkflowManager

DACWorkflowManager(placements)

Bases: WorkflowManager

__init_(self, placements)

Initialization of an instance of the generic DACWorkflowManager class.

Parameters:

• placements

  (pandas Dataframe) –

  The locations that the simulation should be run for. Columns must include "lon", "lat" (CRS: 4326) and "capacity" -The capacity is the nominal capacity of the DAC plant in tCO2/h

Returns:

• DACWorkflowManager –

Methods:

• adjust_variable_to_long_run_average –

  Adjusts the average mean of the specified variable to a known long-run-average

• apply_loss_factor –

  Applies a loss factor onto a specified variable

• extract_raster_values_at_placements –

  Extracts pixel values at each of the configured placements from the specified raster file

• get_scalar_values_from_raster –

  Auxiliary function to extract raster values with NaN fallback options.

• load_lt_dac_model_data –

  Function to load the DAC model data of a given model. The model data maps temperature and relative humidity to energy demand, relative productivity and water desorption.

• read –

  Reads the specified variables from the NetCDF4-style weather dataset, and then extracts

• register_workflow_parameter –

  Add a parameter to the WorkflowManager which will be included in the output XArray dataset

• set_time_index –

  Sets the time index of the WorkflowManager

• simulate_ht_dac_model –

  Simulate the high-temperature (HT), liquid-solvent DAC (Direct Air Capture) model for a given model type.

• simulate_lt_dac_model –

  Simulate the LT DAC (Direct Air Capture) model for the specified plant locations.

• spatial_disaggregation –

  [summary]

• to_netcdf –

  Saves an XArray dataset to netCDF4 format

• to_xarray –

  Generates an XArray dataset from the data currently contained in the WorkflowManager

Source code in reskit/dac/workflows/dac_workflow_manager.py

def __init__(self, placements):
    """

    __init_(self, placements)

    Initialization of an instance of the generic DACWorkflowManager class.

    Parameters
    ----------
    placements : pandas Dataframe
        The locations that the simulation should be run for.
        Columns must include "lon", "lat" (CRS: 4326) and "capacity"
        -The capacity is the nominal capacity of the DAC plant in tCO2/h

    Returns
    -------
    DACWorkflowManager

    """
    # Do basic workflow construction
    assert all([a in placements.columns for a in ["lon", "lat", "capacity"]]), (
        "Placements must contain the columns lon,lat and capacity"
    )
    super().__init__(placements)

    units = {
        "capacity": "t_CO2/h",
        "capacity_factor": "-",
        "conversion_factor_electricity": "MWh_el/t_CO2",
        "conversion_factor_heat": "MWh_heat/t_CO2",
        "conversion_factor_water": "t_H2O/t_CO2",
        "CO2_output": "t_CO2",
        "H2O_output": "t_H2O",
        "electricity_input": "MWh_el",
        "heat_input": "MWh_heat",
    }
    self.units = OrderedDict(units)

adjust_variable_to_long_run_average

adjust_variable_to_long_run_average(
    variable: str,
    source_long_run_average: Union[str, float, ndarray],
    real_long_run_average: Union[str, float, ndarray],
    real_lra_scaling: float = 1,
    spatial_interpolation: str = "linear-spline",
    nodata_fallback: str = "nan",
    nodata_fallback_scaling: float = 1,
    allow_nans: bool = True,
)

Adjusts the average mean of the specified variable to a known long-run-average

Note:

uses the equation: variable[t] = variable[t] * real_long_run_average / source_long_run_average

Parameters:

• variable

  (str) –

  The variable to be adjusted

• source_long_run_average

  (Union[str, float, ndarray]) –

  The variable's native long run average (the average in the weather file) - If a string is given, it is expected to be a path to a raster file which can be used to look up the average values from using the coordinates in .placements - If a numpy ndarray (or derivative) is given, the shape must be one of (time, placements) or at least (placements)

• real_long_run_average

  (Union[str, float, ndarray]) –

  The variables 'true' long run average - If a string is given, it is expected to be a path to a raster file which can be used to look up the average values from using the coordinates in .placements - If a numpy ndarray (or derivative) is given, the shape must be one of (time, placements) or at least (placements)

• real_lra_scaling

  (float, default: 1 ) –

  An optional scaling factor to apply to the values derived from real_long_run_average. - This is primarily useful when real_long_run_average is a path to a raster file - By default 1

• spatial_interpolation

  (str, default: 'linear-spline' ) –

  When either source_long_run_average or real_long_run_average are a path to a raster file, this input specifies which interpolation algorithm should be used - Options are: "near", "linear-spline", "cubic-spline", "average" - By default "linear-spline" - See for more info: geokit.raster.interpolateValues

• nodata_fallback

  (str, default: 'nan' ) –

  When real_long_run_average has no data, one can decide between different fallback options, by default np.nan: - np.nan or None : return np.nan for missing values in real_long_run_average - float : Apply this float value as a scaling factor for all no-data locations only: source_long_run_average * nodata_fallback. NOTE: A value of 1.0 will return the source lra value in case of missing real lra values (no additional nodata_fallback_scaling applied). - str : Will be interpreted as a filepath to a raster with alternative real_long_run_average values, scaled by nodata_fallback_scaling. - callable : any callable method taking the arguments (all iterables): 'locs' and 'source_long_run_average_value' (the locations as gk.geom.point objects and original value from source data). The output values will be considered as the new real_long_run_average for missing locations only (absolute data, no additional nodata_fallback_scaling applied).

  NOTE: np.nan will also be returned in case that the nodata fallback does not yield values either.

• nodata_fallback_scaling

  (float, default: 1 ) –

  An optional scaling factor to apply to the values derived from nodata_fallback. - This is primarily useful when nodata_fallback is a path to a raster file - By default 1

• allow_nans

  (boolean, default: True ) –

  If True, NaN values may remain after scaling, else an error will raised. By default True.

Returns:

• WorkflowManager –

  Returns the invoking WorkflowManager (for chaining)

Source code in reskit/workflow_manager.py

def adjust_variable_to_long_run_average(
    self,
    variable: str,
    source_long_run_average: Union[str, float, np.ndarray],
    real_long_run_average: Union[str, float, np.ndarray],
    real_lra_scaling: float = 1,
    spatial_interpolation: str = "linear-spline",
    nodata_fallback: str = "nan",
    nodata_fallback_scaling: float = 1,
    allow_nans: bool = True,
):
    """Adjusts the average mean of the specified variable to a known long-run-average

    Note:
    -----
    uses the equation: variable[t] = variable[t] * real_long_run_average / source_long_run_average

    Parameters
    ----------
    variable : str
        The variable to be adjusted

    source_long_run_average : Union[str, float, np.ndarray]
        The variable's native long run average (the average in the weather file)
        - If a string is given, it is expected to be a path to a raster file which can be
        used to look up the average values from using the coordinates in `.placements`
        - If a numpy ndarray (or derivative) is given, the shape must be one of (time, placements)
        or at least (placements)

    real_long_run_average : Union[str, float, np.ndarray]
        The variables 'true' long run average
        - If a string is given, it is expected to be a path to a raster file which can be
        used to look up the average values from using the coordinates in `.placements`
        - If a numpy ndarray (or derivative) is given, the shape must be one of (time, placements)
        or at least (placements)

    real_lra_scaling : float, optional
        An optional scaling factor to apply to the values derived from `real_long_run_average`.
        - This is primarily useful when `real_long_run_average` is a path to a raster file
        - By default 1

    spatial_interpolation : str, optional
        When either `source_long_run_average` or `real_long_run_average` are a path to a raster
        file, this input specifies which interpolation algorithm should be used
        - Options are: "near", "linear-spline", "cubic-spline", "average"
        - By default "linear-spline"
        - See for more info: geokit.raster.interpolateValues

    nodata_fallback: float, str, callable, optional
        When real_long_run_average has no data, one can decide between different fallback options, by default np.nan:
        - np.nan or None : return np.nan for missing values in real_long_run_average
        - float : Apply this float value as a scaling factor for all no-data locations only: source_long_run_average * nodata_fallback.
        NOTE: A value of 1.0 will return the source lra value in case of missing real lra values (no additional nodata_fallback_scaling applied).
        - str : Will be interpreted as a filepath to a raster with alternative real_long_run_average values, scaled by nodata_fallback_scaling.
        - callable : any callable method taking the arguments (all iterables): 'locs' and 'source_long_run_average_value'
        (the locations as gk.geom.point objects and original value from source data). The output values will be considered as
        the new real_long_run_average for missing locations only (absolute data, no additional nodata_fallback_scaling applied).

        NOTE: np.nan will also be returned in case that the nodata fallback does not yield values either.

    nodata_fallback_scaling: float
        An optional scaling factor to apply to the values derived from `nodata_fallback`.
        - This is primarily useful when `nodata_fallback` is a path to a raster file
        - By default 1

    allow_nans : boolean, optional
        If True, NaN values may remain after scaling, else an error will raised. By default True.

    Returns
    -------
    WorkflowManager
        Returns the invoking WorkflowManager (for chaining)
    """
    if not (nodata_fallback is None or callable(nodata_fallback) or isinstance(nodata_fallback, (float, int, str))):
        raise TypeError(f"'nodata_fallback' must be a float or a Callable.")

    # first get source values
    if isinstance(source_long_run_average, str):
        # assume raster fp
        source_lra = self.get_scalar_values_from_raster(
            fp=source_long_run_average, spatial_interpolation="linear-spline"
        )
    else:
        source_lra = source_long_run_average

    # then get lng-run average values for scaling
    if isinstance(real_long_run_average, str):
        # assume a raster path
        real_lra = self.get_scalar_values_from_raster(
            fp=real_long_run_average, spatial_interpolation=spatial_interpolation
        )
    else:
        real_lra = real_long_run_average

    # replace missing values with no-data fallback if needed
    if isinstance(nodata_fallback, str) and nodata_fallback.lower() == "source":
        warnings.warn(
            "'source' value for 'nodata_fallback' is deprecated and will be removed soon. Use 1.0 instead.",
            DeprecationWarning,
        )
        nodata_fallback = 1.0
    if isinstance(nodata_fallback, str) and nodata_fallback.lower() == "nan":
        warnings.warn(
            "'nan' value for 'nodata_fallback' is deprecated and will be removed soon. Use np.nan instead.",
            DeprecationWarning,
        )
        nodata_fallback = np.nan
    if any(np.isnan(real_lra)):  # TODO currently all real_lra are replaced by fallback, is this intentional?
        # we are lacking long-run average values
        if nodata_fallback is None or (isinstance(nodata_fallback, float) and np.isnan(nodata_fallback)):
            # nans will be returned for missing lra values
            fallback_lra = np.array([np.nan] * len(real_lra))
        elif isinstance(nodata_fallback, (int, float)):
            # apply factor to source_lra to scale missing values
            fallback_lra = nodata_fallback * source_lra  # no additional scaling
        elif callable(nodata_fallback):
            # apply function to calculate missing values
            fallback_lra = nodata_fallback(
                locs=self.locs, source_long_run_average_value=source_lra
            )  # no additional scaling
        elif isinstance(nodata_fallback, str):
            # assume this is yet another raster path as fallback and extract missing values
            fallback_lra = (
                self.get_scalar_values_from_raster(fp=nodata_fallback, spatial_interpolation=spatial_interpolation)
                * nodata_fallback_scaling
            )

        # divide by real_lra_scaling once to compensate later multiplication below for
        # scaling factor, nodata_fallback should not be multiplied by real_lra_scaling
        fallback_lra = fallback_lra / real_lra_scaling
        # set fallback values where real_lra is nan
        real_lra[np.isnan(real_lra)] = fallback_lra[np.isnan(real_lra)]

    # save LRA as attribute
    self.real_lra = real_lra

    # calculate scaling factor:
    # nan result will stay nan results, as these placements cannot be calculated any more
    factors = real_lra * real_lra_scaling / source_lra
    if any(np.isnan(real_lra)):
        if allow_nans:
            warnings.warn(f"NaN values remaining in real lra after application of nodata_fallback.")
        else:
            raise ValueError(f"Missing values for variable '{variable}' and NaNs not allowed.")

    # write info with missing values to sim_data:
    self.placements[f"missing_values_{basename(real_long_run_average)}_nodata_fallback{nodata_fallback}"] = (
        np.isnan(factors)
    )

    self.sim_data[variable] = factors * self.sim_data[variable]
    self.placements[f"LRA_factor_{variable}"] = factors
    return self

apply_loss_factor

apply_loss_factor(
    loss: Union[float, ndarray, FunctionType],
    variables: Union[str, List[str]] = ["capacity_factor"],
)

Applies a loss factor onto a specified variable

Parameters:

• loss

  (Union[float, ndarray, FunctionType]) –

  The loss factor(s) to be applied - If a float or a numpy ndarray is given, then the following operation is performed:

  variable = variable * (1 - loss) - If a function is given, then the following operation is performed: variable = variable * (1 - loss(variable) ) - If a numpy ndarray is given, it must be broadcastable to the variable's shape in .sim_data

• variables

  (Union[str, List[str]], default: ['capacity_factor'] ) –

  The variable or variables to apply the loss factor to - By default ["capacity_factor"]

Returns:

• WorkflowManager –

  Returns the invoking WorkflowManager (for chaining)

Source code in reskit/workflow_manager.py

def apply_loss_factor(
    self,
    loss: Union[float, np.ndarray, FunctionType],
    variables: Union[str, List[str]] = ["capacity_factor"],
):
    """Applies a loss factor onto a specified variable

    Parameters
    ----------
    loss : Union[float, np.ndarray, FunctionType]
        The loss factor(s) to be applied
        - If a float or a numpy ndarray is given, then the following operation is performed:
        > variable = variable * (1 - loss)
        - If a function is given, then  the following operation is performed:
        > variable = variable * (1 - loss(variable) )
        - If a numpy ndarray is given, it must be broadcastable to the variable's shape in
        `.sim_data`

    variables : Union[str, List[str]], optional
        The variable or variables to apply the loss factor to
        - By default ["capacity_factor"]

    Returns
    -------
    WorkflowManager
        Returns the invoking WorkflowManager (for chaining)
    """
    # filter only existing variables
    _variables = [_var for _var in variables if _var in self.sim_data.keys()]
    if len(_variables) < len(variables):
        warnings.warn(
            f"Loss factor could not be applied to the following requested variables because variables are not in sim_data: {', '.join(sorted(set(variables) - set(_variables)))}"
        )

    for var in _variables:
        if isinstance(loss, FunctionType):
            self.sim_data[var] *= 1 - loss(self.sim_data[var])
        else:
            self.sim_data[var] *= 1 - loss

    return self

extract_raster_values_at_placements

extract_raster_values_at_placements(raster, **kwargs)

Extracts pixel values at each of the configured placements from the specified raster file

Source code in reskit/workflow_manager.py

def extract_raster_values_at_placements(self, raster, **kwargs):
    """Extracts pixel values at each of the configured placements from the specified raster file"""
    return gk.raster.interpolateValues(raster, points=self.locs, **kwargs)

get_scalar_values_from_raster

get_scalar_values_from_raster(
    fp, spatial_interpolation, points=None
)

Auxiliary function to extract raster values with NaN fallback options.

Source code in reskit/workflow_manager.py

def get_scalar_values_from_raster(self, fp, spatial_interpolation, points=None):
    """
    Auxiliary function to extract raster values with NaN fallback options.
    """
    assert isfile(fp), f"File '{fp}' in adjust_variable_to_long_run_average() does not exist."
    # execute with warnings filter since values outside of source data would trigger geokit UserWarning every time
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")

        if points is None:
            points = [(loc.lon, loc.lat) for loc in self.locs._locations]
        else:
            assert isinstance(points, list) and all([isinstance(x, tuple) and len(x) == 2 for x in points]), (
                "points must be a list of (lon, lat) tuples."
            )

        # interpolateValues returns a scalar for a single location; ensure a 1-D array
        # so the nan handling below (and callers) work regardless of the number of points
        _lra = np.atleast_1d(gk.raster.interpolateValues(fp, points, mode=spatial_interpolation))
        # if getting values fails, it could be because of interpolation method.
        # these values will be replaced with the nearest interpolation method
        if np.isnan(_lra).any():
            _lra_near = np.atleast_1d(gk.raster.interpolateValues(fp, self.locs, mode="near"))
            _lra[np.isnan(_lra)] = _lra_near[np.isnan(_lra)]
        # still nans, i.e. the cell itself is nan, but maybe its neighbors are not
        # try the (nan)median of the surrounding cells
        if np.isnan(_lra).any():

            def _nanmedian(vals, xOff, yOff):
                """Aux function to mimic the 3 expected inputs in interpolateValues()"""
                return np.nanmedian(vals)

            points = [(loc.lon, loc.lat) for loc in self.locs._locations]
            _lra_near = np.atleast_1d(gk.raster.interpolateValues(fp, points, mode="func", func=_nanmedian))
            _lra[np.isnan(_lra)] = _lra_near[np.isnan(_lra)]
    return _lra

load_lt_dac_model_data

load_lt_dac_model_data(model: str)

Function to load the DAC model data of a given model. The model data maps temperature and relative humidity to energy demand, relative productivity and water desorption. Description: The DAC model data needs columns "T" and "RH" where the temperature (°C) and relative humidity are included. Its a csv file. Additionally, the data needs columns: -"totalElectricity" where the needed electricity input in MWh/tCO2 is stated at the specified ambient conditions -"totalThermal" where the needed heat input in MWh/tCO2 is stated at the specified ambient conditions -"relProd" where the relative productivity (i.e. the cpaacity factor) is stated at the specified ambient conditions -"waterDesorption" where the produced water is stated in tH2O/tCO2. It might also be possible that this is negative (if water is consumed rather than produced) The currently available models: -LT_jajjawi: Data from the developed low temperature (solid sorbent) DAC model by Jajjawi et al. [1]. Here, the heat is needed at 90 °C. -LT_sendi: Data from the developed low temperature (solid sorbent) DAC model by Sendi et al. [2]. Here, the heat is needed at 110 °C (for steam generation). The original Sendi data has been adapted as described in Wenzel 2025 [3].

Parameters:

• model

  (str) –

  type of DAC model to use. Valid inputs are: "LT_sendi", "LT_jajjawi" or a path to a csv with DAC model data in the same format as in ./data/

References

[1] http://dx.doi.org/10.2139/ssrn.5230783 [2] 10.1016/j.oneear.2022.09.003 [3] 10.1016/j.adapen.2025.100229

Source code in reskit/dac/workflows/dac_workflow_manager.py

def load_lt_dac_model_data(self, model: str):
    """
    Function to load the DAC model data of a given model. The model data maps temperature and relative humidity to energy demand, relative productivity and water desorption.
    Description:
    The DAC model data needs columns "T" and "RH" where the temperature (°C) and relative humidity are included. Its a csv file.
    Additionally, the data needs columns:
        -"totalElectricity" where the needed electricity input in MWh/tCO2 is stated at the specified ambient conditions
        -"totalThermal" where the needed heat input in MWh/tCO2 is stated at the specified ambient conditions
        -"relProd" where the relative productivity (i.e. the cpaacity factor) is stated at the specified ambient conditions
        -"waterDesorption" where the produced water is stated in tH2O/tCO2. It might also be possible that this is negative (if water is consumed rather than produced)
    The currently available models:
        -LT_jajjawi: Data from the developed low temperature (solid sorbent) DAC model by Jajjawi et al. [1]. Here, the heat is needed at 90 °C.
        -LT_sendi: Data from the developed low temperature (solid sorbent) DAC model by Sendi et al. [2]. Here, the heat is needed at 110 °C (for steam generation). The original Sendi data has been adapted as described in Wenzel 2025 [3].

    Parameters
    ----------
    model: str
        type of DAC model to use. Valid inputs are: "LT_sendi", "LT_jajjawi" or a path to a csv with DAC model data in the same format as in ./data/

    References
    ----------
    [1] http://dx.doi.org/10.2139/ssrn.5230783
    [2] 10.1016/j.oneear.2022.09.003
    [3] 10.1016/j.adapen.2025.100229

    """
    model_path_dict = {"LT_sendi": "LT_sendi.csv", "LT_jajjawi": "LT_jajjawi.csv"}

    assert model in model_path_dict.keys() or (
        isinstance(model, str) and model.endswith(".csv") and os.path.isfile(model)
    ), (
        f"Invalid model: {model}. Not one of the base models (LT_jajjawi or LT_sendi) and no valid path to an existing csv with custom data."
    )

    if model in model_path_dict.keys():
        path = os.path.join(DATAFOLDER, model_path_dict[model])
    else:
        path = model
    self.dac_data = pd.read_csv(path, index_col=0)

    required_cols = [
        "totalElectricity",
        "totalThermal",
        "relProd",
        "waterDesorption",
    ]
    assert all(col in self.dac_data.columns for col in required_cols), (
        f"Missing columns: {set(required_cols) - set(self.dac_data.columns)}"
    )

read

read(
    variables: Union[str, List[str]],
    source_type: str,
    source: str,
    set_time_index: bool = False,
    spatial_interpolation_mode: str = "bilinear",
    temporal_reindex_method: str = "nearest",
    time_index_from=None,
    **kwargs,
)

Reads the specified variables from the NetCDF4-style weather dataset, and then extracts those variables for each of the coordinates configured in .placements. The resulting data is then available in .sim_data.

Parameters:

• variables

  (str or list of strings) –

  The variables (or variables) to be read from the specified source - If a path to a weather source is given, then only the 'standard' variables configured for that source type are available (see the doc string for the weather source you are interested in) - If either 'elevated_wind_speed' or 'surface_wind_speed' is included in the variable list, then the members .elevated_wind_speed_height and .surface_wind_speed_height, respectfully, are also added. These are constants which specify what the 'native' wind speed height is, which depends on the source - A pre-loaded NCSource can also be given, thus allowing for any variable in the source to be specified in the variables list. But the user needs to take care of initializing the NCSource and loading the data they want

• source_type

  (str) –

  The type of weather datasource which is to be loaded. Can be one of: "ERA5", "SARAH", "MERRA", or 'user' - If a pre-loaded NCSource is given for the source object, then the source_type should be "user"

• source

  (str or NCSource) –

  The source to read weather variables from

• set_time_index

  (bool, default: False ) –

  If True, instructs the workflow manager to set the time index to that which is read from the weather source - By default False

• spatial_interpolation_mode

  (str, default: 'bilinear' ) –

  The spatial interpolation mode to use while reading data from the weather source at each of the placement coordinates - By default "bilinear"

• temporal_reindex_method

  (str, default: 'nearest' ) –

  In the event of missing data, this algorithm is used to fill in the missing data. - Can be, for example, "nearest", "ffill", "bfill", "interpolate" - By default "nearest"

Returns:

• WorkflowManager –

  Returns the invoking WorkflowManager (for chaining)

Raises:

• RuntimeError –

  If set_time_index is False but no .time_index exists

• RuntimeError –

  If source_type is unknown

Source code in reskit/workflow_manager.py

def read(
    self,
    variables: Union[str, List[str]],
    source_type: str,
    source: str,
    set_time_index: bool = False,
    spatial_interpolation_mode: str = "bilinear",
    temporal_reindex_method: str = "nearest",
    time_index_from=None,
    **kwargs,
):
    """Reads the specified variables from the NetCDF4-style weather dataset, and then extracts
    those variables for each of the coordinates configured in `.placements`. The resulting
    data is then available in `.sim_data`.

    Parameters
    ----------
    variables : str or list of strings
        The variables (or variables) to be read from the specified source
        - If a path to a weather source is given, then only the 'standard' variables
        configured for that source type are available (see the doc string for the
        weather source you are interested in)
        - If either 'elevated_wind_speed' or 'surface_wind_speed' is included in the
        variable list, then the members `.elevated_wind_speed_height` and
        `.surface_wind_speed_height`, respectfully, are also added. These are constants
        which specify what the 'native' wind speed height is, which depends on the source
        - A pre-loaded NCSource can also be given, thus allowing for any variable in the
        source to be specified in the `variables` list. But the user needs to take care
        of initializing the NCSource and loading the data they want

    source_type : str
        The type of weather datasource which is to be loaded. Can be one of:
        "ERA5", "SARAH", "MERRA", or 'user'
        - If a pre-loaded NCSource is given for the `source` object, then the `source_type`
        should be "user"

    source : str or rk.weather.NCSource
        The source to read weather variables from

    set_time_index : bool, optional
        If True, instructs the workflow manager to set the time index to that which is read
        from the weather source
        - By default False

    spatial_interpolation_mode : str, optional
        The spatial interpolation mode to use while reading data from the weather source at
        each of the placement coordinates
        - By default "bilinear"

    temporal_reindex_method : str, optional
        In the event of missing data, this algorithm is used to fill in the missing data.
        - Can be, for example, "nearest", "ffill", "bfill", "interpolate"
        - By default "nearest"

    Returns
    -------
    WorkflowManager
        Returns the invoking WorkflowManager (for chaining)

    Raises
    ------
    RuntimeError
        If set_time_index is False but no `.time_index` exists
    RuntimeError
        If source_type is unknown
    """
    if not set_time_index and self.time_index is None:
        raise RuntimeError("Time index is not available")

    if not isinstance(variables, list):
        variables = [
            variables,
        ]

    if isinstance(source, str) and source_type != "user":
        storage_format = kwargs.pop("storage_format", None)
        is_zarr = storage_format == "zarr" or source.endswith(".zarr") or source.startswith("gs://")
        if source_type == "ERA5":
            source_constructor = rk_weather.Era5ZarrSource if is_zarr else rk_weather.Era5Source
        elif source_type == "SARAH":
            source_constructor = rk_weather.SarahSource
        elif source_type == "MERRA":
            source_constructor = rk_weather.MerraSource
        elif source_type == "ICON-LAM":
            source_constructor = rk_weather.IconlamSource
        else:
            raise RuntimeError("Unknown source_type")

        if source_type == "ERA5":
            time_slice = kwargs.pop("time_slice", None)
            era5_kwargs = dict(kwargs)
            if time_slice is not None:
                if not is_zarr:
                    raise RuntimeError(
                        "'time_slice' is only supported for Zarr-backed ERA5 sources; support for "
                        "netCDF4-backed ERA5 sources is planned. Until then, restrict the time span "
                        "of netCDF4 ERA5 data by selecting the corresponding files instead."
                    )
                era5_kwargs["time_slice"] = time_slice
            source = source_constructor(source, bounds=self.ext, time_index_from=time_index_from, **era5_kwargs)
        else:
            source = source_constructor(source, bounds=self.ext, **kwargs)

        # Load the requested variables
        source.sload(*variables)

    else:  # Assume source is already an initialized NCSource-like object
        missing_variables = [var for var in variables if var not in source.data]
        if missing_variables:
            if hasattr(source, "sload"):
                source.sload(*missing_variables)
            else:
                raise AssertionError(
                    "The given source has no '.sload()' method and is missing the variable(s): "
                    + ", ".join(missing_variables)
                )

    if set_time_index:
        self.set_time_index(source.time_index)

    # read variables
    for var in variables:
        self.sim_data[var] = source.get(
            var,
            self.locs,  # Manipulate locs here
            interpolation=spatial_interpolation_mode,
            force_as_data_frame=True,
        )

        if not set_time_index:
            self.sim_data[var] = self.sim_data[var].reindex(self.time_index, method=temporal_reindex_method)

        self.sim_data[var] = self.sim_data[var].values

        # Special check for wind speed height
        if var == "elevated_wind_speed":
            self.elevated_wind_speed_height = source.ELEVATED_WIND_SPEED_HEIGHT

        if var == "surface_wind_speed":
            self.surface_wind_speed_height = source.SURFACE_WIND_SPEED_HEIGHT

    return self

register_workflow_parameter

register_workflow_parameter(
    key: str, value: Union[str, float]
)

Add a parameter to the WorkflowManager which will be included in the output XArray dataset

Parameters:

• key

  (str) –

  The workflow parameter's access key

• value

  (Union[str, float]) –

  The workflow parameter's value. Only strings and floats are allowed

Source code in reskit/workflow_manager.py

def register_workflow_parameter(self, key: str, value: Union[str, float]):
    """Add a parameter to the WorkflowManager which will be included in the output XArray dataset

    Parameters
    ----------
    key : str
        The workflow parameter's access key

    value : Union[str,float]
        The workflow parameter's value. Only strings and floats are allowed
    """
    self.workflow_parameters[key] = value

set_time_index

set_time_index(times: DatetimeIndex)

Sets the time index of the WorkflowManager

Source code in reskit/workflow_manager.py

def set_time_index(self, times: pd.DatetimeIndex):
    """Sets the time index of the WorkflowManager

    Parameters
    ----------
        times : pd.DatetimeIndex
            The timesteps to use throughout the WorkflowManager's life cycle. The
            length of this dataset must match the shape of data which is loaded into
            the WorkflorManager.sim_data member.
    """
    self.time_index = times

    self._time_sel_ = None
    self._time_index_ = self.time_index.copy()
    self._set_sim_shape()

simulate_ht_dac_model

simulate_ht_dac_model(model: str = 'HT_okosun')

Simulate the high-temperature (HT), liquid-solvent DAC (Direct Air Capture) model for a given model type.

This function maps ambient temperature and relative humidity to energy demand, relative productivity, and water desorption for the specified DAC model. Currently, only the electrified HT DAC model "HT_okosun" is available.

Parameters:

• model

  (str, default: 'HT_okosun' ) –

  Type of DAC model to use. Currently, only "HT_okosun" is implemented. Default is "HT_okosun".

Raises:

• NotImplementedError –

  If a DAC model type other than "HT_okosun" is requested.

Notes

The "HT_okosun" model is based on a natural gas-fired HT-DAC system [1], adapted to an electrified version as described in [2,3]. The electrified DAC model only consumes electricity.

References

[1] 10.1016/j.apenergy.2022.119895 [2] 10.3389/fclim.2020.618644 [3] 10.1016/j.adapen.2025.100229

Source code in reskit/dac/workflows/dac_workflow_manager.py

def simulate_ht_dac_model(self, model: str = "HT_okosun"):
    """
    Simulate the high-temperature (HT), liquid-solvent DAC (Direct Air Capture) model for a given model type.

    This function maps ambient temperature and relative humidity to energy demand,
    relative productivity, and water desorption for the specified DAC model.
    Currently, only the electrified HT DAC model "HT_okosun" is available.

    Parameters
    ----------
    model : str, optional
        Type of DAC model to use. Currently, only "HT_okosun" is implemented.
        Default is "HT_okosun".

    Raises
    ------
    NotImplementedError
        If a DAC model type other than "HT_okosun" is requested.

    Notes
    -----
    The "HT_okosun" model is based on a natural gas-fired HT-DAC system [1],
    adapted to an electrified version as described in [2,3]. The electrified
    DAC model only consumes electricity.

    References
    ----------
    [1] 10.1016/j.apenergy.2022.119895
    [2] 10.3389/fclim.2020.618644
    [3] 10.1016/j.adapen.2025.100229
    """
    # Calculate capture rate, relative productivity and energy (w/o compression)
    if model == "HT_okosun":
        capture_rate = (
            48.8371759783294
            + 0.141875496 * self.sim_data["relative_humidity"]
            + 0.961897256 * self.sim_data["surface_air_temperature"]
            - 0.000550616476 * self.sim_data["relative_humidity"] ** 2
            + 0.00266221049 * self.sim_data["surface_air_temperature"] * self.sim_data["relative_humidity"]
            - 0.00588467947 * self.sim_data["surface_air_temperature"] ** 2
        )  # equation fitted by k.okosun as described in [3]. Describes the share [%] of co2 captured from the incoming air dependent on the ambient conditions. See also [1,2].

        ElecDemand = 7.2082 * capture_rate ** (
            -0.317
        )  # equation fitted by k.okosun as described in [3]. Relates the capture rate to the energy demand.
        relative_productivity = (
            capture_rate / 40 * 527702.4 / 1000000
        )  # equation fitted by k.okosun as described in [3]. Relates the capture rate to the relative productivity.
    else:
        raise NotImplementedError(f"HT-DAC Model of type {model} not implemented.")

    self.sim_data["capacity_factor"] = (
        relative_productivity  # the relative productivity for DAC plants equals to the capacity factor for other renewable energy plants, i.e. wind turbines
    )
    self.sim_data["conversion_factor_electricity"] = -ElecDemand  # MWh_el/t_CO2
    self.sim_data["conversion_factor_heat"] = np.nan  # MWh_th/t_CO2
    self.sim_data["conversion_factor_water"] = np.nan  # tH2O/tCO2

    # Now, besides the conversion factors which are relative to the produced CO2 mass, also simulate the specified plant with the specified capacity:
    self.sim_data["CO2_output"] = self.sim_data["capacity_factor"] * np.array(
        self.placements["capacity"]
    )  # t_CO2/h
    self.sim_data["H2O_output"] = np.nan  # t_H2O/h
    self.sim_data["electricity_input"] = (
        self.sim_data["CO2_output"] * -self.sim_data["conversion_factor_electricity"]
    )  # MWh_el/h
    self.sim_data["heat_input"] = np.nan  # MWh_th/h

simulate_lt_dac_model

simulate_lt_dac_model(fillMethod: str = 'nearest')

Simulate the LT DAC (Direct Air Capture) model for the specified plant locations.

This function interpolates DAC model data to the simulation grid and calculates electricity, heat, and water requirements as well as CO2 output for the plants. It also handles points outside the convex hull of the DAC data using a specified fill method.

Parameters:

• fillMethod

  (str, default: 'nearest' ) –

  Method to fill values for weather conditions outside the convex hull of the DAC model data. Options are: - "nearest" : use the nearest available datapoint (default) - "offTmin" : cut off for temperatures below the DAC data range, use nearest for relative humidity

Raises:

• NotImplementedError –

  If a filling method other than "nearest" or "offTmin" is requested.

Source code in reskit/dac/workflows/dac_workflow_manager.py

def simulate_lt_dac_model(self, fillMethod: str = "nearest"):
    """
    Simulate the LT DAC (Direct Air Capture) model for the specified plant locations.

    This function interpolates DAC model data to the simulation grid and calculates
    electricity, heat, and water requirements as well as CO2 output for the plants.
    It also handles points outside the convex hull of the DAC data using a specified
    fill method.

    Parameters
    ----------
    fillMethod : str, optional
        Method to fill values for weather conditions outside the convex hull of
        the DAC model data. Options are:
        - "nearest" : use the nearest available datapoint (default)
        - "offTmin" : cut off for temperatures below the DAC data range, use nearest
        for relative humidity

    Raises
    ------
    NotImplementedError
        If a filling method other than "nearest" or "offTmin" is requested.
    """
    assert fillMethod in [
        "offTmin",
        "nearest",
    ], f"Filling Method: {fillMethod} not implemented."

    # create unique grid as well as interpolators and evaluate for each property:
    properties = ["totalElectricity", "totalThermal", "waterDesorption", "relProd"]
    interpolated_outputs = {}

    for prop in properties:
        # Pivot table for this property
        pivot = self.dac_data.pivot_table(values=prop, index="T", columns="RH")

        # Create interpolator
        interpolator = RegularGridInterpolator(
            points=(pivot.index.values, pivot.columns.values.astype("float64")),
            values=pivot.values,
            bounds_error=False,
        )

        # Evaluate interpolator at simulation data and store into dict
        interpolated_outputs[prop] = interpolator(
            (self.sim_data["surface_air_temperature"], self.sim_data["relative_humidity"])
        )

    if (fillMethod == "offTmin") or (fillMethod == "nearest"):
        # fill points outside the convex hull with "nearest" :
        fill_outputs = {}
        for prop in properties:
            # Create interpolator using nearest-neighbor
            pivot = self.dac_data.pivot_table(values=prop, index="T", columns="RH")
            interpolator = RegularGridInterpolator(
                points=(pivot.index.values, pivot.columns.values.astype("float64")),
                values=pivot.values,
                method="nearest",
                bounds_error=False,
                fill_value=None,
            )
            # Evaluate at simulation points
            fill_outputs[prop] = interpolator(
                (self.sim_data["surface_air_temperature"], self.sim_data["relative_humidity"])
            )
    if fillMethod == "offTmin":
        # fill RH values outside range by nearest and force no operation below/above T bounds by setting relProd=0
        Tmin = self.dac_data["T"].min()
        fill_outputs.loc[self.sim_data["surface_air_temperature"] < Tmin, "relProd"] = 0

    # combine:
    for prop in properties:
        interpolated_outputs[prop] = np.where(
            np.isnan(interpolated_outputs[prop]), fill_outputs[prop], interpolated_outputs[prop]
        )

    self.sim_data["capacity_factor"] = interpolated_outputs[
        "relProd"
    ]  # the relative productivity for DAC plants equals to the capacity factor for other renewable energy plants, i.e. wind turbines
    self.sim_data["conversion_factor_electricity"] = interpolated_outputs["totalElectricity"]  # MWh_el/t_CO2
    self.sim_data["conversion_factor_heat"] = interpolated_outputs["totalThermal"]  # MWh_th/t_CO2
    self.sim_data["conversion_factor_water"] = interpolated_outputs["waterDesorption"]  # t_H2O/t_CO2

    # Now, besides the conversion factors which are relative to the produced CO2 mass, also simulate the specified plant with the specified capacity:
    self.sim_data["CO2_output"] = self.sim_data["capacity_factor"] * np.array(
        self.placements["capacity"]
    )  # t_CO2/h
    self.sim_data["H2O_output"] = self.sim_data["CO2_output"] * self.sim_data["conversion_factor_water"]  # t_H2O/h
    self.sim_data["electricity_input"] = (
        self.sim_data["CO2_output"] * -self.sim_data["conversion_factor_electricity"]
    )  # MWh_el/h
    self.sim_data["heat_input"] = self.sim_data["CO2_output"] * -self.sim_data["conversion_factor_heat"]  # MWh_th/h

spatial_disaggregation

spatial_disaggregation(
    variable: str,
    source_high_resolution: Union[str, float, ndarray],
    source_low_resolution: Union[str, float, ndarray],
    real_lra_scaling: float = 1,
    spatial_interpolation: str = "linear-spline",
)

[summary]

Parameters:

• variable

  (str) –

  [description]

• source_long_run_average

  (Union[str, float, ndarray]) –

  [description]

• real_long_run_average

  (Union[str, float, ndarray]) –

  [description]

• real_lra_scaling

  (float, default: 1 ) –

  [description], by default 1

• spatial_interpolation

  (str, default: 'linear-spline' ) –

  [description], by default "linear-spline"

Source code in reskit/workflow_manager.py

def spatial_disaggregation(
    self,
    variable: str,
    source_high_resolution: Union[str, float, np.ndarray],
    source_low_resolution: Union[str, float, np.ndarray],
    real_lra_scaling: float = 1,
    spatial_interpolation: str = "linear-spline",
):
    """[summary]

    Parameters
    ----------
    variable : str
        [description]
    source_long_run_average : Union[str, float, np.ndarray]
        [description]
    real_long_run_average : Union[str, float, np.ndarray]
        [description]
    real_lra_scaling : float, optional
        [description], by default 1
    spatial_interpolation : str, optional
        [description], by default "linear-spline"
    """
    # Get values from high resolution tiff file
    if isinstance(source_high_resolution, str):
        points = [(loc.lon, loc.lat) for loc in self.locs._locations]
        correction_values_high_res = gk.raster.interpolateValues(  # TODO change here
            source_high_resolution, points, mode=spatial_interpolation
        )
        # assert not np.isnan(correction_values_high_res).any() and (correction_values_high_res > 0).all()
    else:
        correction_values_high_res = source_high_resolution

    # Get values from low resolution tiff file (meant over eg. ERA5)
    if isinstance(source_low_resolution, str):
        points = [(loc.lon, loc.lat) for loc in self.locs._locations]
        correction_values_low_res = gk.raster.interpolateValues(  # TODO change here
            source_low_resolution, points, mode=spatial_interpolation
        )
        # assert not np.isnan(correction_values_low_res).any() and (correction_values_low_res > 0).all()
    else:
        correction_values_low_res = source_low_resolution

    # correction factors:
    factors = correction_values_high_res / correction_values_low_res
    factors = np.nan_to_num(factors, nan=1 / real_lra_scaling)
    assert (factors > 0).all()

    # update values
    self.sim_data[variable] = self.sim_data[variable] * factors * real_lra_scaling
    return self

to_netcdf

to_netcdf(
    xds: Dataset,
    output_netcdf_path: str = None,
    output_variables: List[str] = None,
    custom_attributes: dict = None,
    _intermediate_dict=False,
) -> str

Saves an XArray dataset to netCDF4 format

Note: - The .placements data is automatically added to the XArray dataset along the 'locations' dimension - The workflow_parameters data is automatically added as dimensionless variables - The .sim_data is automatically added along the dimensions (time, locations) - The .time_index is automatically added along the dimension 'time'

Parameters:

• xds

  (Dataset) –

  The XArray dataset to save

• output_netcdf_path

  (str, default: None ) –

  If given, the XArray dataset will be written to disc at the specified path - By default None

• output_variables

  (List[str], default: None ) –

  If given, specifies the variables which should be included in the resulting dataset. Otherwise all suitable variables found in .placements, .workflow_parameters, .sim_data, and .time_index will be included - Only variables of numeric or string type are suitable due to NetCDF4 limitations - By default None

• custom_attributes

  (dict, default: None ) –

  If given, adds the key-value pairs as attributes to the XArray dataset before saving - These will be added in addition to existing attributes - By default None

Returns:

• output_netcdf_path –

  The resulting output_netcdf_path

Source code in reskit/workflow_manager.py

def to_netcdf(
    self,
    xds: xarray.Dataset,
    output_netcdf_path: str = None,
    output_variables: List[str] = None,
    custom_attributes: dict = None,
    _intermediate_dict=False,
) -> str:
    """Saves an XArray dataset to netCDF4 format

    Note:
    - The `.placements` data is automatically added to the XArray dataset along the 'locations' dimension
    - The `workflow_parameters` data is automatically added as dimensionless variables
    - The `.sim_data` is automatically added along the dimensions (time, locations)
    - The `.time_index` is automatically added along the dimension 'time'

    Parameters
    ----------
    xds : xarray.Dataset
        The XArray dataset to save

    output_netcdf_path : str
        If given, the XArray dataset will be written to disc at the specified path
        - By default None

    output_variables : List[str], optional
        If given, specifies the variables which should be included in the resulting
        dataset. Otherwise all suitable variables found in `.placements`, `.workflow_parameters`,
        `.sim_data`, and `.time_index` will be included
        - Only variables of numeric or string type are suitable due to NetCDF4 limitations
        - By default None

    custom_attributes : dict, optional
        If given, adds the key-value pairs as attributes to the XArray dataset before saving
        - These will be added in addition to existing attributes
        - By default None

    Returns
    -------
    output_netcdf_path
        The resulting output_netcdf_path
    """
    encoding = dict()

    # write sim_data
    for key in self.sim_data.keys():
        # check if key in requestet output_variables
        if output_variables is not None:
            if key not in output_variables:
                continue
        encoding[key] = dict(zlib=True)

    # FIXME: WHAT IS THIS FOR?
    # #write sim_data_daily, only if exists
    # if hasattr(self, 'sim_data_daily'):
    #     for key in self.sim_data_daily.keys():
    #         #check if key in requestet output_variables
    #         if output_variables is not None:
    #             if key not in output_variables:
    #                 continue
    #         encoding[key] = dict(zlib=True)

    # Add custom attributes if provided
    if custom_attributes is not None:
        for k, v in custom_attributes.items():
            xds.attrs[k] = v

    if output_netcdf_path is not None:
        xds.to_netcdf(output_netcdf_path, encoding=encoding)
        return output_netcdf_path
    else:
        return xds

to_xarray

to_xarray(
    output_netcdf_path: str = None,
    output_variables: List[str] = None,
    custom_attributes: dict = None,
    _intermediate_dict=False,
) -> Dataset

Generates an XArray dataset from the data currently contained in the WorkflowManager

Note: - The .placements data is automatically added to the XArray dataset along the 'locations' dimension - The workflow_parameters data is automatically added as dimensionless variables - The .sim_data is automatically added along the dimensions (time, locations) - The .time_index is automatically added along the dimension 'time'

Parameters:

• output_netcdf_path

  (str, default: None ) –

  If given, the XArray dataset will be written to disc at the specified path - By default None

• output_variables

  (List[str], default: None ) –

  If given, specifies the variables which should be included in the resulting dataset. Otherwise all suitable variables found in .placements, .workflow_parameters, .sim_data, and .time_index will be included - Only variables of numeric or string type are suitable due to NetCDF4 limitations - By default None

• custom_attributes

  (dict, default: None ) –

  If given, adds the key-value pairs as attributes to the XArray dataset - These will be added in addition to the workflow_parameters - By default None

Returns:

• Dataset –

  The resulting XArray dataset

Source code in reskit/workflow_manager.py

def to_xarray(
    self,
    output_netcdf_path: str = None,
    output_variables: List[str] = None,
    custom_attributes: dict = None,
    _intermediate_dict=False,
) -> xarray.Dataset:
    """Generates an XArray dataset from the data currently contained in the WorkflowManager

    Note:
    - The `.placements` data is automatically added to the XArray dataset along the 'locations' dimension
    - The `workflow_parameters` data is automatically added as dimensionless variables
    - The `.sim_data` is automatically added along the dimensions (time, locations)
    - The `.time_index` is automatically added along the dimension 'time'

    Parameters
    ----------
    output_netcdf_path : str, optional
        If given, the XArray dataset will be written to disc at the specified path
        - By default None

    output_variables : List[str], optional
        If given, specifies the variables which should be included in the resulting
        dataset. Otherwise all suitable variables found in `.placements`, `.workflow_parameters`,
        `.sim_data`, and `.time_index` will be included
        - Only variables of numeric or string type are suitable due to NetCDF4 limitations
        - By default None

    custom_attributes : dict, optional
        If given, adds the key-value pairs as attributes to the XArray dataset
        - These will be added in addition to the workflow_parameters
        - By default None

    Returns
    -------
    xarray.Dataset
        The resulting XArray dataset
    """
    if isinstance(output_variables, str):
        output_variables = [output_variables]
    if isinstance(output_variables, list) and not "RESKit_sim_order" in output_variables:
        output_variables.append("RESKit_sim_order")

    times = self.time_index
    if times[0].tz is not None:
        times = [np.datetime64(dt.tz_convert("UTC").tz_convert(None)) for dt in times]
    times_days = np.unique(pd.DatetimeIndex(times).date).astype("datetime64")
    if times_days[0].astype("datetime64[Y]") != times_days[-1].astype("datetime64[Y]"):
        # old tiles where shifted by 1 hour, so the last day of the previous year also appears. catch this problem whti this if clause
        times_days = times_days[1:]
    xds = OrderedDict()
    encoding = dict()

    if "location_id" in self.placements.columns:
        location_coords = self.placements["location_id"].copy()
        del self.placements["location_id"]
    else:
        location_coords = np.arange(self.placements.shape[0])

    # write placements
    for c in self.placements.columns:
        # check if c in requestet output_variables
        if output_variables is not None:
            if c not in output_variables:
                continue
        if np.issubdtype(self.placements[c].dtype, np.number):
            write = True
        else:
            write = True
            for element in self.placements[c]:
                if not isinstance(element, (str, bytearray)):
                    write = False
                    break
        if write:
            xds[c] = xarray.DataArray(
                self.placements[c],
                dims=["location"],
                coords=dict(location=location_coords),
            )

    # write sim_data
    for key in self.sim_data.keys():
        # check if key in requestet output_variables
        if output_variables is not None:
            if key not in output_variables:
                continue

        tmp = np.full((len(self.time_index), self.locs.count), 0.0, dtype=float)
        tmp[self._time_sel_, :] = self.sim_data[key]

        xds[key] = xarray.DataArray(
            tmp,
            dims=["time", "location"],
            coords=dict(time=times, location=location_coords),
        )
        encoding[key] = dict(zlib=True)

    # write sim_data_daily, only if exists
    if hasattr(self, "sim_data_daily"):
        for key in self.sim_data_daily.keys():
            # check if key in requestet output_variables
            if output_variables is not None:
                if key not in output_variables:
                    continue

            tmp = np.full((len(times_days), self.locs.count), np.nan)
            tmp[:, :] = self.sim_data_daily[key]

            xds[key] = xarray.DataArray(
                tmp,
                dims=["time_days", "location"],
                coords=dict(time_days=times_days, location=location_coords),
            )
            encoding[key] = dict(zlib=True)

    if _intermediate_dict:
        return xds

    xds = xarray.Dataset(xds)

    for k, v in self.workflow_parameters.items():
        xds.attrs[k] = v

    # Add custom attributes if provided
    if custom_attributes is not None:
        for k, v in custom_attributes.items():
            xds.attrs[k] = v

    if output_netcdf_path is not None:
        xds.to_netcdf(output_netcdf_path, encoding=encoding)
        return output_netcdf_path
    else:
        return xds