Skip to content

gcages.completeness#

General tools for checking completeness

Classes:

Name Description
NotCompleteError

Raised when a pd.DataFrame is not complete

Functions:

Name Description
assert_all_groups_are_complete

Assert all groups have 'complete' data
get_missing_levels

Get missing levels in an index

NotCompleteError #

Bases: ValueError

Raised when a pd.DataFrame is not complete

Methods:

Name Description
__init__

Initialise the error
Source code in src/gcages/completeness.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class NotCompleteError(ValueError):
    """
    Raised when a [pd.DataFrame][pandas.DataFrame] is not complete
    """

    def __init__(
        self,
        missing: pd.DataFrame,
        complete_index: pd.MultiIndex,
    ) -> None:
        """
        Initialise the error

        Parameters
        ----------
        missing
            Index levels that don't have a complete index

        complete_index
            Definition of a complete index
        """
        error_msg = (
            "The DataFrame is not complete. "
            f"The following expected levels are missing:\n{missing}\n"
            f"The complete index expected for each level is:\n"
            f"{complete_index.to_frame(index=False)}"
        )
        super().__init__(error_msg)

__init__ #

__init__(
    missing: DataFrame, complete_index: MultiIndex
) -> None

Initialise the error

Parameters:

Name Type Description Default
missing DataFrame

Index levels that don't have a complete index

 required
complete_index MultiIndex

Definition of a complete index

 required
Source code in src/gcages/completeness.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(
    self,
    missing: pd.DataFrame,
    complete_index: pd.MultiIndex,
) -> None:
    """
    Initialise the error

    Parameters
    ----------
    missing
        Index levels that don't have a complete index

    complete_index
        Definition of a complete index
    """
    error_msg = (
        "The DataFrame is not complete. "
        f"The following expected levels are missing:\n{missing}\n"
        f"The complete index expected for each level is:\n"
        f"{complete_index.to_frame(index=False)}"
    )
    super().__init__(error_msg)

assert_all_groups_are_complete #

assert_all_groups_are_complete(
    to_check: DataFrame,
    complete_index: MultiIndex,
    group_keys: list[str] | None = None,
    unit_col: str = "unit",
) -> None

Assert all groups have 'complete' data

Here, complete is defined by complete_index, which specifies the metadata that should be included for each group.

Parameters:

Name Type Description Default
to_check DataFrame

Data to check

 required
complete_index MultiIndex

Index which defines the meaning of 'complete'

 required
group_keys list[str] | None

Keys to use to group to_check into groups when checking for completeness.

If not supplied, we use all the index levels in to_check except those that appear in complete_index and the unit_col, specifically to_check.index.names.difference([*complete_index.names, unit_col]).

 None
unit_col str

Unit column (differences here do not indicate incompleteness)

 'unit'

Raises:

Type Description
ValueError

to_check is empty
NotCompleteError

to_check is not complete

Examples:

>>> to_check = pd.DataFrame(
...     [
...         [1.0, 2.0],
...         [3.0, 2.0],
...         [1.0, 2.0],
...         [3.0, 2.0],
...     ],
...     columns=[2015, 2100],
...     index=pd.MultiIndex.from_tuples(
...         [
...             ("sa", "va", "W"),
...             ("sa", "vb", "W"),
...             ("sb", "va", "W"),
...             ("sb", "vb", "W"),
...         ],
...         names=["scenario", "variable", "unit"],
...     ),
... )
>>> to_check
                        2015  2100
scenario variable unit
sa       va       W      1.0   2.0
         vb       W      3.0   2.0
sb       va       W      1.0   2.0
         vb       W      3.0   2.0

>>> # A checker, by which `to_check` is complete
>>> checker_a = pd.MultiIndex.from_tuples(
...     [
...         ("va",),
...         ("vb",),
...     ],
...     names=["variable"],
... )
>>> assert_all_groups_are_complete(to_check, complete_index=checker_a)
>>> # No error raised, all happy
>>>
>>> # A checker which includes variables that aren't present in `to_check`
>>> checker_b = pd.MultiIndex.from_tuples(
...     [
...         ("va",),
...         ("vb",),
...         ("vc",),
...     ],
...     names=["variable"],
... )
>>> assert_all_groups_are_complete(to_check, complete_index=checker_b)
Traceback (most recent call last):
    ...
gcages.completeness.NotCompleteError: The DataFrame is not complete. The following expected levels are missing:
  variable scenario
0       vc       sa
0       vc       sb
The complete index expected for each level is:
  variable
0       va
1       vb
2       vc
Source code in src/gcages/completeness.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
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
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def assert_all_groups_are_complete(
    to_check: pd.DataFrame,
    complete_index: pd.MultiIndex,
    group_keys: list[str] | None = None,
    unit_col: str = "unit",
) -> None:
    """
    Assert all groups have 'complete' data

    Here, complete is defined by `complete_index`,
    which specifies the metadata that should be included for each group.

    Parameters
    ----------
    to_check
        Data to check

    complete_index
        Index which defines the meaning of 'complete'

    group_keys
        Keys to use to group `to_check` into groups when checking for completeness.

        If not supplied, we use all the index levels in `to_check`
        except those that appear in `complete_index` and the `unit_col`,
        specifically
        `to_check.index.names.difference([*complete_index.names, unit_col])`.

    unit_col
        Unit column (differences here do not indicate incompleteness)

    Raises
    ------
    ValueError
        `to_check` is empty

    NotCompleteError
        `to_check` is not complete

    Examples
    --------
    >>> to_check = pd.DataFrame(
    ...     [
    ...         [1.0, 2.0],
    ...         [3.0, 2.0],
    ...         [1.0, 2.0],
    ...         [3.0, 2.0],
    ...     ],
    ...     columns=[2015, 2100],
    ...     index=pd.MultiIndex.from_tuples(
    ...         [
    ...             ("sa", "va", "W"),
    ...             ("sa", "vb", "W"),
    ...             ("sb", "va", "W"),
    ...             ("sb", "vb", "W"),
    ...         ],
    ...         names=["scenario", "variable", "unit"],
    ...     ),
    ... )
    >>> to_check  # doctest: +NORMALIZE_WHITESPACE
                            2015  2100
    scenario variable unit
    sa       va       W      1.0   2.0
             vb       W      3.0   2.0
    sb       va       W      1.0   2.0
             vb       W      3.0   2.0

    >>> # A checker, by which `to_check` is complete
    >>> checker_a = pd.MultiIndex.from_tuples(
    ...     [
    ...         ("va",),
    ...         ("vb",),
    ...     ],
    ...     names=["variable"],
    ... )
    >>> assert_all_groups_are_complete(to_check, complete_index=checker_a)
    >>> # No error raised, all happy
    >>>
    >>> # A checker which includes variables that aren't present in `to_check`
    >>> checker_b = pd.MultiIndex.from_tuples(
    ...     [
    ...         ("va",),
    ...         ("vb",),
    ...         ("vc",),
    ...     ],
    ...     names=["variable"],
    ... )
    >>> assert_all_groups_are_complete(to_check, complete_index=checker_b)
    Traceback (most recent call last):
        ...
    gcages.completeness.NotCompleteError: The DataFrame is not complete. The following expected levels are missing:
      variable scenario
    0       vc       sa
    0       vc       sb
    The complete index expected for each level is:
      variable
    0       va
    1       vb
    2       vc
    """  # noqa: E501
    if to_check.empty:
        msg = "`to_check` is empty"
        raise ValueError(msg)

    # Probably a smarter way to do this rather than looping, I can't see it now
    if unit_col not in to_check.index.names:
        msg = f"{unit_col=} is not in {to_check.index.names=}"
        raise KeyError(msg)

    if group_keys is None:
        group_keys = to_check.index.names.difference([*complete_index.names, unit_col])  # type: ignore # pandas-stubs confused

    missing_l = []
    # Check against the levels that are in `complete_index`
    # (also ignoring units if they are there)
    idx_to_check_drop_levels = list(
        {*to_check.index.names.difference(complete_index.names), unit_col}  # type: ignore # pandas-stubs confused
    )
    for group_values, gdf in to_check.groupby(group_keys):
        missing_levels = get_missing_levels(
            gdf.index,  # type: ignore # fix when moving to pandas-openscm
            complete_index,
            levels_to_drop=idx_to_check_drop_levels,
        )

        if not missing_levels.empty:
            tmp = missing_levels.to_frame(index=False)
            # Could probably do this better too if we need speed
            for key, value in zip(group_keys, group_values):
                tmp[key] = value

            missing_l.append(tmp)

    if missing_l:
        raise NotCompleteError(pd.concat(missing_l), complete_index=complete_index)

get_missing_levels #

get_missing_levels(
    index: MultiIndex,
    complete_index: MultiIndex,
    levels_to_drop: list[str] | None = None,
    unit_col: str | None = None,
) -> MultiIndex

Get missing levels in an index

Here, complete is defined by complete_index, which specifies the levels that should be in index.

Parameters:

Name Type Description Default
index MultiIndex

Index to check

 required
complete_index MultiIndex

Index which defines the meaning of 'complete'

 required
levels_to_drop list[str] | None

Levels to drop from index before checking for completeness

If not supplied, we use all the index levels in index except those that appear in complete_index and the unit_col, specifically index.names.difference([*complete_index.names, unit_col]).

 None
unit_col str | None

Unit column (differences here do not indicate missing levels)

Only needed if levels_to_drop is None.

 None

Examples:

>>> to_check = pd.MultiIndex.from_tuples(
...     [
...         ("sa", "va", "W"),
...         ("sa", "vb", "W"),
...         ("sb", "va", "W"),
...         ("sb", "vb", "W"),
...     ],
...     names=["scenario", "variable", "unit"],
... )
>>> to_check
MultiIndex([('sa', 'va', 'W'),
        ('sa', 'vb', 'W'),
        ('sb', 'va', 'W'),
        ('sb', 'vb', 'W')],
       names=['scenario', 'variable', 'unit'])
>>> # A checker, by which `to_check` is complete
>>> checker_a = pd.MultiIndex.from_tuples(
...     [
...         ("va",),
...         ("vb",),
...     ],
...     names=["variable"],
... )
>>> get_missing_levels(to_check, complete_index=checker_a, unit_col="unit")
MultiIndex([], names=['variable'])
>>> # Empty index i.e. no missing levels
>>>
>>> # A checker which includes variables that aren't present in `to_check`
>>> checker_b = pd.MultiIndex.from_tuples(
...     [
...         ("va",),
...         ("vb",),
...         ("vc",),
...     ],
...     names=["variable"],
... )
>>> get_missing_levels(
...     to_check, complete_index=checker_b, unit_col="unit"
... )
MultiIndex([('vc',)],
       names=['variable'])
Source code in src/gcages/completeness.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def get_missing_levels(
    index: pd.MultiIndex,
    complete_index: pd.MultiIndex,
    levels_to_drop: list[str] | None = None,
    unit_col: str | None = None,
) -> pd.MultiIndex:
    """
    Get missing levels in an index

    Here, complete is defined by `complete_index`,
    which specifies the levels that should be in `index`.

    Parameters
    ----------
    index
        Index to check

    complete_index
        Index which defines the meaning of 'complete'

    levels_to_drop
        Levels to drop from `index` before checking for completeness

        If not supplied, we use all the index levels in `index`
        except those that appear in `complete_index` and the `unit_col`,
        specifically
        `index.names.difference([*complete_index.names, unit_col])`.

    unit_col
        Unit column (differences here do not indicate missing levels)

        Only needed if `levels_to_drop` is `None`.

    Examples
    --------
    >>> to_check = pd.MultiIndex.from_tuples(
    ...     [
    ...         ("sa", "va", "W"),
    ...         ("sa", "vb", "W"),
    ...         ("sb", "va", "W"),
    ...         ("sb", "vb", "W"),
    ...     ],
    ...     names=["scenario", "variable", "unit"],
    ... )
    >>> to_check  # doctest: +NORMALIZE_WHITESPACE
    MultiIndex([('sa', 'va', 'W'),
            ('sa', 'vb', 'W'),
            ('sb', 'va', 'W'),
            ('sb', 'vb', 'W')],
           names=['scenario', 'variable', 'unit'])
    >>> # A checker, by which `to_check` is complete
    >>> checker_a = pd.MultiIndex.from_tuples(
    ...     [
    ...         ("va",),
    ...         ("vb",),
    ...     ],
    ...     names=["variable"],
    ... )
    >>> get_missing_levels(to_check, complete_index=checker_a, unit_col="unit")
    MultiIndex([], names=['variable'])
    >>> # Empty index i.e. no missing levels
    >>>
    >>> # A checker which includes variables that aren't present in `to_check`
    >>> checker_b = pd.MultiIndex.from_tuples(
    ...     [
    ...         ("va",),
    ...         ("vb",),
    ...         ("vc",),
    ...     ],
    ...     names=["variable"],
    ... )
    >>> get_missing_levels(
    ...     to_check, complete_index=checker_b, unit_col="unit"
    ... )  # doctest: +NORMALIZE_WHITESPACE
    MultiIndex([('vc',)],
           names=['variable'])
    """
    if levels_to_drop is None:
        if unit_col is None:
            msg = "If levels_to_drop is `None` then `unit_col` must be supplied"
            raise AssertionError(msg)

        # Check against the levels that are in `complete_index`
        # (also ignoring units if they are there)
        levels_to_drop = list(
            {*index.names.difference(complete_index.names), unit_col}  # type: ignore # pandas-stubs confused
        )

    index_to_check = index.droplevel(levels_to_drop)  # type: ignore # something wrong with pandas-stubs

    if not isinstance(index_to_check, pd.MultiIndex):
        index_to_check = pd.MultiIndex.from_arrays(
            [index_to_check.values], names=[index_to_check.name]
        )

    missing_levels: pd.MultiIndex = complete_index.difference(  # type: ignore # pandas-stubs out of date
        index_to_check.reorder_levels(complete_index.names)
    )

    return missing_levels