calliope.backend.helper_functions

Functions that can be used to process data in math where and expression strings.

NAME is the function name to use in the math strings.

DefaultIfEmpty(return_type, *, equation_name, input_data, backend_interface=None, **kwargs)

Bases: ParsingHelperFunction

Abstract helper function class, which all helper functions must subclass.

The abstract properties and methods defined here must be defined by all helper functions.

Source code in src/calliope/backend/helper_functions.py

def __init__(
    self,
    return_type: Literal["array", "math_string"],
    *,
    equation_name: str,
    input_data: xr.Dataset,
    backend_interface: Optional[type[BackendModel]] = None,
    **kwargs,
) -> None:
    """Abstract helper function class, which all helper functions must subclass.

    The abstract properties and methods defined here must be defined by all helper functions.

    """
    self._equation_name = equation_name
    self._input_data = input_data
    self._backend_interface = backend_interface
    self._return_type = return_type

ALLOWED_IN = ['expression'] class-attribute instance-attribute

NAME = 'default_if_empty' class-attribute instance-attribute

ignore_where: bool property

If True, where arrays will not be applied to the incoming data variables (valid for expression helpers)

as_array(var, default)

Get an array with filled NaNs if present in the model, or a single default value if not.

This function is useful for avoiding excessive sub expressions where it is a choice between an expression or a single numeric value.

Parameters:

Name	Type	Description	Default
var	DataArray	array of backend expression objects or an un-indexed "string" object array with the var name (if not present in the model).	required
default	float | int	Numeric value with which to fill / replace var.	required

Returns:

Type	Description
DataArray	xr.DataArray: If var is an array of backend expression objects, NaNs will be filled with default. If var is an unindexed array with a single string value, an unindexed array with the default value.

Examples:

>>> default_if_empty(flow_cap, 0)
[out] <xarray.DataArray (techs: 2)>
      array(['ac_transmission:node1', 'ac_transmission:node2'], dtype=object)
>>> default_if_empty(flow_export, 0)
[out] <xarray.DataArray ()>
      array(0, dtype=np.int)

Source code in src/calliope/backend/helper_functions.py

def as_array(self, var: xr.DataArray, default: float | int) -> xr.DataArray:
    """Get an array with filled NaNs if present in the model, or a single default value if not.

    This function is useful for avoiding excessive sub expressions where it is a choice between an expression or a single numeric value.

    Args:
        var (xr.DataArray): array of backend expression objects or an un-indexed "string" object array with the var name (if not present in the model).
        default (float | int): Numeric value with which to fill / replace `var`.

    Returns:
        xr.DataArray:
            If var is an array of backend expression objects, NaNs will be filled with `default`.
            If var is an unindexed array with a single string value, an unindexed array with the default value.

    Examples:
        ```
        >>> default_if_empty(flow_cap, 0)
        [out] <xarray.DataArray (techs: 2)>
              array(['ac_transmission:node1', 'ac_transmission:node2'], dtype=object)
        >>> default_if_empty(flow_export, 0)
        [out] <xarray.DataArray ()>
              array(0, dtype=np.int)
        ```
    """
    if var.attrs.get("obj_type", "") == "string":
        return xr.DataArray(default)
    else:
        return var.fillna(default)

as_math_string(var, default)

Source code in src/calliope/backend/helper_functions.py

def as_math_string(self, var: str, default: float | int) -> str:
    return rf"({var}\vee{{}}{default})"

Defined(return_type, *, equation_name, input_data, backend_interface=None, **kwargs)

Bases: ParsingHelperFunction

Abstract helper function class, which all helper functions must subclass.

The abstract properties and methods defined here must be defined by all helper functions.

Source code in src/calliope/backend/helper_functions.py

def __init__(
    self,
    return_type: Literal["array", "math_string"],
    *,
    equation_name: str,
    input_data: xr.Dataset,
    backend_interface: Optional[type[BackendModel]] = None,
    **kwargs,
) -> None:
    """Abstract helper function class, which all helper functions must subclass.

    The abstract properties and methods defined here must be defined by all helper functions.

    """
    self._equation_name = equation_name
    self._input_data = input_data
    self._backend_interface = backend_interface
    self._return_type = return_type

ALLOWED_IN = ['where'] class-attribute instance-attribute

NAME = 'defined' class-attribute instance-attribute

ignore_where: bool property

If True, where arrays will not be applied to the incoming data variables (valid for expression helpers)

as_array(*, within, how, **dims)

Find whether members of a model dimension are defined inside another.

For instance, whether a node defines a specific tech (or group of techs). Or, whether a tech defines a specific carrier.

Parameters:

Name	Type	Description	Default
within	str	the model dimension to check.	required
how	Literal[all, any]	Whether to return True for any match of nested members or for all nested members.	required

Kwargs

dims (dict[str, str]): key: dimension whose members will be searched for as being defined under the primary dimension (within). Must be one of the core model dimensions: [nodes, techs, carriers] value: subset of the dimension members to find. Transmission techs can be called using the base tech name (e.g., ac_transmission) and all link techs will be collected (e.g., [ac_transmission:region1, ac_transmission:region2]).

Returns:

Type	Description
DataArray	xr.DataArray: For each member of within, True if any/all member(s) in dims is nested within that member.

Examples:

Check for any of a list of techs being defined at nodes. Assuming a YAML definition of:

nodes:
  node1:
    techs:
      tech1:
      tech3:
  node2:
    techs:
      tech2:
      tech3:

Then:

    >>> defined(techs=[tech1, tech2], within=nodes, how=any)
    [out] <xarray.DataArray (nodes: 2)>
          array([ True, False])
          Coordinates:
          * nodes    (nodes) <U5 'node1' 'node2'

    >>> defined(techs=[tech1, tech2], within=nodes, how=all)
    [out] <xarray.DataArray (nodes: 2)>
          array([ False, False])
          Coordinates:
          * nodes    (nodes) <U5 'node1' 'node2'

Source code in src/calliope/backend/helper_functions.py

def as_array(
    self, *, within: str, how: Literal["all", "any"], **dims: str
) -> xr.DataArray:
    """Find whether members of a model dimension are defined inside another.

    For instance, whether a node defines a specific tech (or group of techs).
    Or, whether a tech defines a specific carrier.

    Args:
        within (str): the model dimension to check.
        how (Literal[all, any]): Whether to return True for `any` match of nested members or for `all` nested members.

    Kwargs:
        dims (dict[str, str]):
            **key**: dimension whose members will be searched for as being defined under the primary dimension (`within`).
            Must be one of the core model dimensions: [nodes, techs, carriers]
            **value**: subset of the dimension members to find.
            Transmission techs can be called using the base tech name (e.g., `ac_transmission`) and all link techs will be collected (e.g., [`ac_transmission:region1`, `ac_transmission:region2`]).


    Returns:
        xr.DataArray:
            For each member of `within`, True if any/all member(s) in `dims` is nested within that member.

    Examples:
        Check for any of a list of techs being defined at nodes.
        Assuming a YAML definition of:

        ```yaml
        nodes:
          node1:
            techs:
              tech1:
              tech3:
          node2:
            techs:
              tech2:
              tech3:
        ```
        Then:
        ```
            >>> defined(techs=[tech1, tech2], within=nodes, how=any)
            [out] <xarray.DataArray (nodes: 2)>
                  array([ True, False])
                  Coordinates:
                  * nodes    (nodes) <U5 'node1' 'node2'

            >>> defined(techs=[tech1, tech2], within=nodes, how=all)
            [out] <xarray.DataArray (nodes: 2)>
                  array([ False, False])
                  Coordinates:
                  * nodes    (nodes) <U5 'node1' 'node2'
        ```
    """
    dim_names = list(dims.keys())
    dims_with_list_vals = {
        dim: self._listify(vals) if dim == "techs" else self._listify(vals)
        for dim, vals in dims.items()
    }
    definition_matrix = self._input_data.definition_matrix
    dim_within_da = definition_matrix.any(self._dims_to_remove(dim_names, within))
    within_da = getattr(dim_within_da.sel(**dims_with_list_vals), how)(dim_names)

    return within_da

as_math_string(*, within, how, **dims)

Source code in src/calliope/backend/helper_functions.py

def as_math_string(self, *, within: str, how: Literal["all", "any"], **dims) -> str:
    substrings = []
    for name, vals in dims.items():
        substrings.append(self._latex_substring(how, name, vals, within))
    if len(substrings) == 1:
        return substrings[0]
    else:
        return rf"\bigwedge({', '.join(substrings)})"

GetValAtIndex(return_type, *, equation_name, input_data, backend_interface=None, **kwargs)

Bases: ParsingHelperFunction

Abstract helper function class, which all helper functions must subclass.

The abstract properties and methods defined here must be defined by all helper functions.

Source code in src/calliope/backend/helper_functions.py

def __init__(
    self,
    return_type: Literal["array", "math_string"],
    *,
    equation_name: str,
    input_data: xr.Dataset,
    backend_interface: Optional[type[BackendModel]] = None,
    **kwargs,
) -> None:
    """Abstract helper function class, which all helper functions must subclass.

    The abstract properties and methods defined here must be defined by all helper functions.

    """
    self._equation_name = equation_name
    self._input_data = input_data
    self._backend_interface = backend_interface
    self._return_type = return_type

ALLOWED_IN = ['expression', 'where'] class-attribute instance-attribute

NAME = 'get_val_at_index' class-attribute instance-attribute

ignore_where: bool property

If True, where arrays will not be applied to the incoming data variables (valid for expression helpers)

as_array(**dim_idx_mapping)

Get value of a model dimension at a given integer index.

This function is primarily useful for timeseries data.

Other Parameters:

Name	Type	Description
key	str	Model dimension in which to extract value.
value	int	Integer index of the value to extract (assuming zero-indexing).

Raises:

Type	Description
ValueError	Exactly one dimension:index mapping is expected.

Returns:

Type	Description
DataArray	xr.DataArray: Dimensionless array containing one value.

Examples:

>>> coords = {"timesteps": ["2000-01-01 00:00", "2000-01-01 01:00", "2000-01-01 02:00"]}
>>> model_data = xr.Dataset(coords=coords)
>>> get_val_at_index = GetValAtIndex(model_data=model_data)
>>> get_val_at_index(model_data)(timesteps=0)
<xarray.DataArray 'timesteps' ()>
array('2000-01-01 00:00', dtype='<U16')
Coordinates:
    timesteps  <U16 '2000-01-01 00:00'
>>> get_val_at_index(model_data)(timesteps=-1)
<xarray.DataArray 'timesteps' ()>
array('2000-01-01 00:00', dtype='<U16')
Coordinates:
    timesteps  <U16 '2000-01-01 02:00'

Source code in src/calliope/backend/helper_functions.py

def as_array(self, **dim_idx_mapping: int) -> xr.DataArray:
    """Get value of a model dimension at a given integer index.

    This function is primarily useful for timeseries data.

    Keyword Args:
        key (str): Model dimension in which to extract value.
        value (int): Integer index of the value to extract (assuming zero-indexing).

    Raises:
        ValueError: Exactly one dimension:index mapping is expected.

    Returns:
        xr.DataArray: Dimensionless array containing one value.

    Examples:
        >>> coords = {"timesteps": ["2000-01-01 00:00", "2000-01-01 01:00", "2000-01-01 02:00"]}
        >>> model_data = xr.Dataset(coords=coords)
        >>> get_val_at_index = GetValAtIndex(model_data=model_data)
        >>> get_val_at_index(model_data)(timesteps=0)
        <xarray.DataArray 'timesteps' ()>
        array('2000-01-01 00:00', dtype='<U16')
        Coordinates:
            timesteps  <U16 '2000-01-01 00:00'
        >>> get_val_at_index(model_data)(timesteps=-1)
        <xarray.DataArray 'timesteps' ()>
        array('2000-01-01 00:00', dtype='<U16')
        Coordinates:
            timesteps  <U16 '2000-01-01 02:00'
    """
    dim, idx = self._mapping_to_dim_idx(**dim_idx_mapping)
    return self._input_data.coords[dim][int(idx)]

as_math_string(**dim_idx_mapping)

Source code in src/calliope/backend/helper_functions.py

def as_math_string(self, **dim_idx_mapping: str) -> str:
    dim, idx = self._mapping_to_dim_idx(**dim_idx_mapping)
    return f"{dim}[{idx}]"

Inheritance(return_type, *, equation_name, input_data, backend_interface=None, **kwargs)

Bases: ParsingHelperFunction

Abstract helper function class, which all helper functions must subclass.

The abstract properties and methods defined here must be defined by all helper functions.

Source code in src/calliope/backend/helper_functions.py

def __init__(
    self,
    return_type: Literal["array", "math_string"],
    *,
    equation_name: str,
    input_data: xr.Dataset,
    backend_interface: Optional[type[BackendModel]] = None,
    **kwargs,
) -> None:
    """Abstract helper function class, which all helper functions must subclass.

    The abstract properties and methods defined here must be defined by all helper functions.

    """
    self._equation_name = equation_name
    self._input_data = input_data
    self._backend_interface = backend_interface
    self._return_type = return_type

ALLOWED_IN = ['where'] class-attribute instance-attribute

NAME = 'inheritance' class-attribute instance-attribute

ignore_where: bool property

If True, where arrays will not be applied to the incoming data variables (valid for expression helpers)

as_array(*, nodes=None, techs=None)

Find all technologies and/or nodes which inherit from a particular technology or node group.

The group items being referenced must be defined by the user in node_groups/tech_groups.

Parameters:

Name	Type	Description	Default
nodes	Optional[str]	group name to search for inheritance of on the nodes dimension. Default is None.	None
techs	Optional[str]	group name to search for inheritance of on the techs dimension. Default is None.	None

Returns:

Type	Description
DataArray	xr.Dataset: Boolean array where values are True where the group is inherited, False otherwise. Array dimensions will equal the number of non-None inputs.

Examples:

With:

node_groups:
  foo:
    available_area: 1
tech_groups:
  bar:
    flow_cap_max: 1
  baz:
    inherits: bar
    flow_out_eff: 0.5
nodes:
  node_1:
    inherits: foo
    techs: {tech_1, tech_2}
  node_2:
    techs: {tech_1, tech_2}
techs:
  tech_1:
    ...
    inherits: bar
  tech_2:
    ...
    inherits: baz

>>> inheritance(nodes=foo)
<xarray.DataArray (nodes: 2)>
array([True, False])
Coordinates:
* nodes      (nodes) <U1 'node_1' 'node_2'

>>> inheritance(techs=bar)  # tech_2 inherits `bar` via `baz`.
<xarray.DataArray (techs: 2)>
array([True, True])
Coordinates:
* techs      (techs) <U1 'tech_1' 'tech_2'

>>> inheritance(techs=baz)
<xarray.DataArray (techs: 2)>
array([False, True])
Coordinates:
* techs      (techs) <U1 'tech_1' 'tech_2'

>>> inheritance(nodes=foo, techs=baz)
<xarray.DataArray (nodes: 2, techs: 2)>
array([[False, False],
       [True, False]])
Coordinates:
* nodes      (nodes) <U1 'node_1' 'node_2'
* techs      (techs) <U1 'tech_1' 'tech_2'

Source code in src/calliope/backend/helper_functions.py

def as_array(
    self, *, nodes: Optional[str] = None, techs: Optional[str] = None
) -> xr.DataArray:
    """Find all technologies and/or nodes which inherit from a particular technology or node group.

    The group items being referenced must be defined by the user in `node_groups`/`tech_groups`.

    Args:
        nodes (Optional[str], optional): group name to search for inheritance of on the `nodes` dimension. Default is None.
        techs (Optional[str], optional): group name to search for inheritance of on the `techs` dimension. Default is None.

    Returns:
        xr.Dataset: Boolean array where values are True where the group is inherited, False otherwise. Array dimensions will equal the number of non-None inputs.

    Examples:
        With:
        ```yaml
        node_groups:
          foo:
            available_area: 1
        tech_groups:
          bar:
            flow_cap_max: 1
          baz:
            inherits: bar
            flow_out_eff: 0.5
        nodes:
          node_1:
            inherits: foo
            techs: {tech_1, tech_2}
          node_2:
            techs: {tech_1, tech_2}
        techs:
          tech_1:
            ...
            inherits: bar
          tech_2:
            ...
            inherits: baz
        ```

        >>> inheritance(nodes=foo)
        <xarray.DataArray (nodes: 2)>
        array([True, False])
        Coordinates:
        * nodes      (nodes) <U1 'node_1' 'node_2'

        >>> inheritance(techs=bar)  # tech_2 inherits `bar` via `baz`.
        <xarray.DataArray (techs: 2)>
        array([True, True])
        Coordinates:
        * techs      (techs) <U1 'tech_1' 'tech_2'

        >>> inheritance(techs=baz)
        <xarray.DataArray (techs: 2)>
        array([False, True])
        Coordinates:
        * techs      (techs) <U1 'tech_1' 'tech_2'

        >>> inheritance(nodes=foo, techs=baz)
        <xarray.DataArray (nodes: 2, techs: 2)>
        array([[False, False],
               [True, False]])
        Coordinates:
        * nodes      (nodes) <U1 'node_1' 'node_2'
        * techs      (techs) <U1 'tech_1' 'tech_2'

    """
    inherits_nodes = xr.DataArray(True)
    inherits_techs = xr.DataArray(True)
    if nodes is not None:
        inherits_nodes = self._input_data.get(
            "nodes_inheritance", xr.DataArray("")
        ).str.contains(f"{nodes}(?:,|$)")
    if techs is not None:
        inherits_techs = self._input_data.get(
            "techs_inheritance", xr.DataArray("")
        ).str.contains(f"{techs}(?:,|$)")
    return inherits_nodes & inherits_techs

as_math_string(nodes=None, techs=None)

Source code in src/calliope/backend/helper_functions.py

def as_math_string(
    self, nodes: Optional[str] = None, techs: Optional[str] = None
) -> str:
    strings = []
    if nodes is not None:
        strings.append(f"nodes={nodes}")
    if techs is not None:
        strings.append(f"techs={techs}")
    return rf"\text{{inherits({','.join(strings)})}}"

ParsingHelperFunction(return_type, *, equation_name, input_data, backend_interface=None, **kwargs)

Bases: ABC

Abstract helper function class, which all helper functions must subclass.

The abstract properties and methods defined here must be defined by all helper functions.

Source code in src/calliope/backend/helper_functions.py

def __init__(
    self,
    return_type: Literal["array", "math_string"],
    *,
    equation_name: str,
    input_data: xr.Dataset,
    backend_interface: Optional[type[BackendModel]] = None,
    **kwargs,
) -> None:
    """Abstract helper function class, which all helper functions must subclass.

    The abstract properties and methods defined here must be defined by all helper functions.

    """
    self._equation_name = equation_name
    self._input_data = input_data
    self._backend_interface = backend_interface
    self._return_type = return_type

ALLOWED_IN: list[Literal['where', 'expression']] abstractmethod property

List of parseable math strings that this function can be accessed from.

NAME: str abstractmethod property

Helper function name that is used in the math expression/where string.

ignore_where: bool property

If True, where arrays will not be applied to the incoming data variables (valid for expression helpers)

as_array(*args, **kwargs) abstractmethod

Method to apply the helper function to provide an n-dimensional array output.

This method is called when the class is initialised with return_type=array.

Source code in src/calliope/backend/helper_functions.py

@abstractmethod
def as_array(self, *args, **kwargs) -> xr.DataArray:
    """Method to apply the helper function to provide an n-dimensional array output.

    This method is called when the class is initialised with ``return_type=array``.
    """

as_math_string(*args, **kwargs) abstractmethod

Method to update LaTeX math strings to include the action applied by the helper function.

This method is called when the class is initialised with return_type=math_string.

Source code in src/calliope/backend/helper_functions.py

@abstractmethod
def as_math_string(self, *args, **kwargs) -> str:
    """Method to update LaTeX math strings to include the action applied by the helper function.

    This method is called when the class is initialised with ``return_type=math_string``.
    """

ReduceCarrierDim(return_type, *, equation_name, input_data, backend_interface=None, **kwargs)

Bases: ParsingHelperFunction

Abstract helper function class, which all helper functions must subclass.

The abstract properties and methods defined here must be defined by all helper functions.

Source code in src/calliope/backend/helper_functions.py

def __init__(
    self,
    return_type: Literal["array", "math_string"],
    *,
    equation_name: str,
    input_data: xr.Dataset,
    backend_interface: Optional[type[BackendModel]] = None,
    **kwargs,
) -> None:
    """Abstract helper function class, which all helper functions must subclass.

    The abstract properties and methods defined here must be defined by all helper functions.

    """
    self._equation_name = equation_name
    self._input_data = input_data
    self._backend_interface = backend_interface
    self._return_type = return_type

ALLOWED_IN = ['expression'] class-attribute instance-attribute

NAME = 'reduce_carrier_dim' class-attribute instance-attribute

ignore_where: bool property

If True, where arrays will not be applied to the incoming data variables (valid for expression helpers)

as_array(array, flow_direction)

Reduce expression array data by selecting the carrier that corresponds to the given carrier tier and then dropping the carriers dimension.

Parameters:

Name	Type	Description	Default
array	DataArray	Expression array.	required
flow_direction	Literal['in', 'out']	Flow direction in which to check for the existence of carrier(s) for technologies defined in array.	required

Returns:

Type	Description
DataArray	xr.DataArray: array reduced by the carriers dimension.

Source code in src/calliope/backend/helper_functions.py

def as_array(
    self, array: xr.DataArray, flow_direction: Literal["in", "out"]
) -> xr.DataArray:
    """Reduce expression array data by selecting the carrier that corresponds to the given carrier tier and then dropping the `carriers` dimension.

    Args:
        array (xr.DataArray): Expression array.
        flow_direction (Literal["in", "out"]): Flow direction in which to check for the existence of carrier(s) for technologies defined in `array`.

    Returns:
        xr.DataArray: `array` reduced by the `carriers` dimension.
    """
    sum_helper = Sum(
        return_type=self._return_type,
        equation_name=self._equation_name,
        input_data=self._input_data,
    )

    return sum_helper(
        array.where(self._input_data[f"carrier_{flow_direction}"]), over="carriers"
    )

as_math_string(array, flow_direction)

Source code in src/calliope/backend/helper_functions.py

def as_math_string(self, array: str, flow_direction: Literal["in", "out"]) -> str:
    return rf"\sum\limits_{{\text{{carrier}} \in \text{{carrier_{flow_direction}}}}} ({array})"

Roll(return_type, *, equation_name, input_data, backend_interface=None, **kwargs)

Bases: ParsingHelperFunction

Abstract helper function class, which all helper functions must subclass.

The abstract properties and methods defined here must be defined by all helper functions.

Source code in src/calliope/backend/helper_functions.py

def __init__(
    self,
    return_type: Literal["array", "math_string"],
    *,
    equation_name: str,
    input_data: xr.Dataset,
    backend_interface: Optional[type[BackendModel]] = None,
    **kwargs,
) -> None:
    """Abstract helper function class, which all helper functions must subclass.

    The abstract properties and methods defined here must be defined by all helper functions.

    """
    self._equation_name = equation_name
    self._input_data = input_data
    self._backend_interface = backend_interface
    self._return_type = return_type

ALLOWED_IN = ['expression'] class-attribute instance-attribute

NAME = 'roll' class-attribute instance-attribute

ignore_where: bool property

as_array(array, **roll_kwargs)

Roll (a.k.a., shift) the array along the given dimension(s) by the given number of places. Rolling keeps the array index labels in the same position, but moves the data by the given number of places.

Parameters:

Name	Type	Description	Default
array	DataArray	Array on which to roll data.	required

Keyword Args: key (str): name of dimension on which to roll. value (int): number of places to roll data.

Returns:

Type	Description
DataArray	xr.DataArray: array with rolled data.

Examples:

>>> array = xr.DataArray([1, 2, 3], coords={"foo": ["A", "B", "C"]})
>>> model_data = xr.Dataset({"bar": array})
>>> roll = Roll()
>>> roll("bar", foo=1)
<xarray.DataArray 'bar' (foo: 3)>
array([3, 1, 2])
Coordinates:
* foo      (foo) <U1 'A' 'B' 'C'

Source code in src/calliope/backend/helper_functions.py

def as_array(self, array: xr.DataArray, **roll_kwargs: int) -> xr.DataArray:
    """Roll (a.k.a., shift) the array along the given dimension(s) by the given number of places.
    Rolling keeps the array index labels in the same position, but moves the data by the given number of places.

    Args:
        array (xr.DataArray): Array on which to roll data.
    Keyword Args:
        key (str): name of dimension on which to roll.
        value (int): number of places to roll data.

    Returns:
        xr.DataArray: `array` with rolled data.

    Examples:
        >>> array = xr.DataArray([1, 2, 3], coords={"foo": ["A", "B", "C"]})
        >>> model_data = xr.Dataset({"bar": array})
        >>> roll = Roll()
        >>> roll("bar", foo=1)
        <xarray.DataArray 'bar' (foo: 3)>
        array([3, 1, 2])
        Coordinates:
        * foo      (foo) <U1 'A' 'B' 'C'
    """
    roll_kwargs_int: Mapping = {k: int(v) for k, v in roll_kwargs.items()}
    return array.roll(roll_kwargs_int)

as_math_string(array, **roll_kwargs)

Source code in src/calliope/backend/helper_functions.py

def as_math_string(self, array: str, **roll_kwargs: str) -> str:
    new_strings = {
        k.removesuffix("s"): f"{-1 * int(v):+d}" for k, v in roll_kwargs.items()
    }
    component = self._add_to_iterator(array, new_strings)
    return component

SelectFromLookupArrays(return_type, *, equation_name, input_data, backend_interface=None, **kwargs)

Bases: ParsingHelperFunction

Abstract helper function class, which all helper functions must subclass.

The abstract properties and methods defined here must be defined by all helper functions.

Source code in src/calliope/backend/helper_functions.py

def __init__(
    self,
    return_type: Literal["array", "math_string"],
    *,
    equation_name: str,
    input_data: xr.Dataset,
    backend_interface: Optional[type[BackendModel]] = None,
    **kwargs,
) -> None:
    """Abstract helper function class, which all helper functions must subclass.

    The abstract properties and methods defined here must be defined by all helper functions.

    """
    self._equation_name = equation_name
    self._input_data = input_data
    self._backend_interface = backend_interface
    self._return_type = return_type

ALLOWED_IN = ['expression'] class-attribute instance-attribute

NAME = 'select_from_lookup_arrays' class-attribute instance-attribute

ignore_where: bool property

If True, where arrays will not be applied to the incoming data variables (valid for expression helpers)

as_array(array, **lookup_arrays)

Apply vectorised indexing on an arbitrary number of an input array's dimensions.

Parameters:

Name	Type	Description	Default
array	DataArray	Array on which to apply vectorised indexing.	required

Kwargs

lookup_arrays (dict[str, xr.DataArray]): key: dimension on which to apply vectorised indexing value: array whose values are either NaN or values from the dimension given in the key.

Raises: BackendError: array must be indexed over the dimensions given in the lookup_arrays dict keys. BackendError: All lookup_arrays must be indexed over all the dimensions given in the lookup_arrays dict keys.

Returns:

Type	Description
DataArray	xr.DataArray: array with rearranged values (coordinates remain unchanged). Any NaN index coordinates in the lookup arrays will be NaN in the returned array.

Examples:

>>> coords = {"foo": ["A", "B", "C"]}
>>> array = xr.DataArray([1, 2, 3], coords=coords)
>>> lookup_array = xr.DataArray(np.array(["B", "A", np.nan], dtype="O"), coords=coords, name="bar")
>>> model_data = xr.Dataset({"bar": lookup_array})
>>> select_from_lookup_arrays = SelectFromLookupArrays(model_data=model_data)
>>> select_from_lookup_arrays(array, foo=lookup_array)
<xarray.DataArray 'bar' (foo: 3)>
array([ 2.,  1., nan])
Coordinates:
* foo      (foo) object 'A' 'B' 'C'

The lookup array assigns the value at "B" to "A" and vice versa. "C" is masked since the lookup array value is NaN.

Source code in src/calliope/backend/helper_functions.py

def as_array(
    self, array: xr.DataArray, **lookup_arrays: xr.DataArray
) -> xr.DataArray:
    """Apply vectorised indexing on an arbitrary number of an input array's dimensions.

    Args:
        array (xr.DataArray): Array on which to apply vectorised indexing.

    Kwargs:
        lookup_arrays (dict[str, xr.DataArray]):
            key: dimension on which to apply vectorised indexing
            value: array whose values are either NaN or values from the dimension given in the key.
    Raises:
        BackendError: `array` must be indexed over the dimensions given in the `lookup_arrays` dict keys.
        BackendError: All `lookup_arrays` must be indexed over all the dimensions given in the `lookup_arrays` dict keys.

    Returns:
        xr.DataArray:
            `array` with rearranged values (coordinates remain unchanged).
            Any NaN index coordinates in the lookup arrays will be NaN in the returned array.

    Examples:
        >>> coords = {"foo": ["A", "B", "C"]}
        >>> array = xr.DataArray([1, 2, 3], coords=coords)
        >>> lookup_array = xr.DataArray(np.array(["B", "A", np.nan], dtype="O"), coords=coords, name="bar")
        >>> model_data = xr.Dataset({"bar": lookup_array})
        >>> select_from_lookup_arrays = SelectFromLookupArrays(model_data=model_data)
        >>> select_from_lookup_arrays(array, foo=lookup_array)
        <xarray.DataArray 'bar' (foo: 3)>
        array([ 2.,  1., nan])
        Coordinates:
        * foo      (foo) object 'A' 'B' 'C'

    The lookup array assigns the value at "B" to "A" and vice versa.
    "C" is masked since the lookup array value is NaN.
    """
    # Inspired by https://github.com/pydata/xarray/issues/1553#issuecomment-748491929
    # Reindex does not presently support vectorized lookups: https://github.com/pydata/xarray/issues/1553
    # Sel does (e.g. https://github.com/pydata/xarray/issues/4630) but can't handle missing keys

    dims = set(lookup_arrays.keys())
    missing_dims_in_component = dims.difference(array.dims)
    missing_dims_in_lookup_tables = any(
        dim not in lookup.dims for dim in dims for lookup in lookup_arrays.values()
    )
    if missing_dims_in_component:
        raise BackendError(
            f"Cannot select items from `{array.name}` on the dimensions {dims} since the array is not indexed over the dimensions {missing_dims_in_component}"
        )
    if missing_dims_in_lookup_tables:
        raise BackendError(
            f"All lookup arrays used to select items from `{array.name}` must be indexed over the dimensions {dims}"
        )

    dim = "dim_0"
    ixs = {}
    masks = []

    # Turn string lookup values to numeric ones.
    # We stack the dimensions to handle multidimensional lookups
    for index_dim, index in lookup_arrays.items():
        stacked_lookup = self._input_data[index.name].stack({dim: dims})
        ix = array.indexes[index_dim].get_indexer(stacked_lookup)
        if (ix == -1).all():
            received_lookup = self._input_data[index.name].to_series().dropna()
            raise IndexError(
                f"Trying to select items on the dimension {index_dim} from the {index.name} lookup array, but no matches found. Received: {received_lookup}"
            )
        ixs[index_dim] = xr.DataArray(
            np.fmax(0, ix), coords={dim: stacked_lookup[dim]}
        )
        masks.append(ix >= 0)

    # Create a mask to nullify any lookup values that are not given (i.e., are np.nan in the lookup array)
    mask = functools.reduce(lambda x, y: x & y, masks)

    result = array[ixs]

    if not mask.all():
        result[{dim: ~mask}] = np.nan
    unstacked_result = result.drop_vars(dims).unstack(dim)
    return unstacked_result

as_math_string(array, **lookup_arrays)

Source code in src/calliope/backend/helper_functions.py

def as_math_string(self, array: str, **lookup_arrays: str) -> str:
    new_strings = {
        (iterator := dim.removesuffix("s")): rf"={array}[{iterator}]"
        for dim, array in lookup_arrays.items()
    }
    array = self._add_to_iterator(array, new_strings)
    return array

Sum(return_type, *, equation_name, input_data, backend_interface=None, **kwargs)

Bases: ParsingHelperFunction

Abstract helper function class, which all helper functions must subclass.

The abstract properties and methods defined here must be defined by all helper functions.

Source code in src/calliope/backend/helper_functions.py

def __init__(
    self,
    return_type: Literal["array", "math_string"],
    *,
    equation_name: str,
    input_data: xr.Dataset,
    backend_interface: Optional[type[BackendModel]] = None,
    **kwargs,
) -> None:
    """Abstract helper function class, which all helper functions must subclass.

    The abstract properties and methods defined here must be defined by all helper functions.

    """
    self._equation_name = equation_name
    self._input_data = input_data
    self._backend_interface = backend_interface
    self._return_type = return_type

ALLOWED_IN = ['expression'] class-attribute instance-attribute

NAME = 'sum' class-attribute instance-attribute

ignore_where: bool property

If True, where arrays will not be applied to the incoming data variables (valid for expression helpers)

as_array(array, *, over)

Sum an expression array over the given dimension(s).

Parameters:

Name	Type	Description	Default
array	DataArray	expression array	required
over	Union[str, list[str]]	dimension(s) over which to apply sum.	required

Returns:

Type	Description
DataArray	xr.DataArray: Array with dimensions reduced by applying a summation over the dimensions given in over. NaNs are ignored (xarray.DataArray.sum arg: skipna: True) and if all values along the dimension(s) are NaN, the summation will lead to a NaN (xarray.DataArray.sum arg: min_count=1).

Source code in src/calliope/backend/helper_functions.py

def as_array(
    self, array: xr.DataArray, *, over: Union[str, list[str]]
) -> xr.DataArray:
    """Sum an expression array over the given dimension(s).

    Args:
        array (xr.DataArray): expression array
        over (Union[str, list[str]]): dimension(s) over which to apply `sum`.

    Returns:
        xr.DataArray:
            Array with dimensions reduced by applying a summation over the dimensions given in `over`.
            NaNs are ignored (xarray.DataArray.sum arg: `skipna: True`) and if all values along the dimension(s) are NaN,
            the summation will lead to a NaN (xarray.DataArray.sum arg: `min_count=1`).
    """
    return array.sum(over, min_count=1, skipna=True)

as_math_string(array, *, over)

Source code in src/calliope/backend/helper_functions.py

def as_math_string(self, array: str, *, over: Union[str, list[str]]) -> str:
    if isinstance(over, str):
        overstring = self._instr(over)
    else:
        foreach_string = r" \\ ".join(self._instr(i) for i in over)
        overstring = rf"\substack{{{foreach_string}}}"
    return rf"\sum\limits_{{{overstring}}} ({array})"

WhereAny(return_type, *, equation_name, input_data, backend_interface=None, **kwargs)

Bases: ParsingHelperFunction

Abstract helper function class, which all helper functions must subclass.

The abstract properties and methods defined here must be defined by all helper functions.

Source code in src/calliope/backend/helper_functions.py

def __init__(
    self,
    return_type: Literal["array", "math_string"],
    *,
    equation_name: str,
    input_data: xr.Dataset,
    backend_interface: Optional[type[BackendModel]] = None,
    **kwargs,
) -> None:
    """Abstract helper function class, which all helper functions must subclass.

    The abstract properties and methods defined here must be defined by all helper functions.

    """
    self._equation_name = equation_name
    self._input_data = input_data
    self._backend_interface = backend_interface
    self._return_type = return_type

ALLOWED_IN = ['where'] class-attribute instance-attribute

NAME = 'any' class-attribute instance-attribute

ignore_where: bool property

If True, where arrays will not be applied to the incoming data variables (valid for expression helpers)

as_array(parameter, *, over)

Reduce the boolean where array of a model parameter by applying any over some dimension(s).

Parameters:

Name	Type	Description	Default
parameter	str	Reference to a model input parameter	required
over	Union[str, list[str]]	dimension(s) over which to apply any.	required

Returns:

Type	Description
DataArray	xr.DataArray: If the parameter exists in the model, returns a boolean array with dimensions reduced by applying a boolean OR operation along the dimensions given in over. If the parameter does not exist, returns a dimensionless False array.

Source code in src/calliope/backend/helper_functions.py

def as_array(self, parameter: str, *, over: Union[str, list[str]]) -> xr.DataArray:
    """Reduce the boolean where array of a model parameter by applying `any` over some dimension(s).

    Args:
        parameter (str): Reference to a model input parameter
        over (Union[str, list[str]]): dimension(s) over which to apply `any`.

    Returns:
        xr.DataArray:
            If the parameter exists in the model, returns a boolean array with dimensions reduced by applying a boolean OR operation along the dimensions given in `over`.
            If the parameter does not exist, returns a dimensionless False array.
    """
    if parameter in self._input_data.data_vars:
        parameter_da = self._input_data[parameter]
        bool_parameter_da = (
            parameter_da.notnull()
            & (parameter_da != np.inf)
            & (parameter_da != -np.inf)
        )
    elif (
        self._backend_interface is not None
        and parameter in self._backend_interface._dataset
    ):
        bool_parameter_da = self._backend_interface._dataset[parameter].notnull()
    else:
        bool_parameter_da = xr.DataArray(False)
    over = self._listify(over)
    available_dims = set(bool_parameter_da.dims).intersection(over)

    return bool_parameter_da.any(dim=available_dims, keep_attrs=True)

as_math_string(array, *, over)

Source code in src/calliope/backend/helper_functions.py

def as_math_string(self, array: str, *, over: Union[str, list[str]]) -> str:
    if isinstance(over, str):
        overstring = self._instr(over)
    else:
        foreach_string = r" \\ ".join(self._instr(i) for i in over)
        overstring = rf"\substack{{{foreach_string}}}"
    # Using bigvee for "collective-or"
    return rf"\bigvee\limits_{{{overstring}}} ({array})"