Skip to content

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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
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 
701
702
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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
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 
309
310
311
312
313
314
315
316
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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
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 
597
598
599
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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
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 
164
165
166
167
168
169
170
171
172
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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
68
69
70
71
72
73
@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 
61
62
63
64
65
66
@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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
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 
467
468
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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
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 
661
662
663
664
665
666
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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
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 
499
500
501
502
503
504
505
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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
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 
435
436
437
438
439
440
441
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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
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 
262
263
264
265
266
267
268
269
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})"