Skip to content

Disagg

Disaggregation tools for:

  • Spatial disaggregation
  • Temporal disaggregation

SpatialDisaggregator

Source code in src/rpycpl/disagg.py 
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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class SpatialDisaggregator:

    def __init__(self, targets=None):
        self._target_nodes = targets

    def validate_reference_data(self, reference_data: pd.Series):
        """Check reference data has expected format

        Args:
            reference_data (pd.Series): The reference data for disaggregation.
        Raises:
            TypeError: If reference data is not a pandas Series.
            ValueError: If reference data index does not match target nodes.
            ValueError: If reference data is not normalised to 1.
        """

        if not isinstance(reference_data, pd.Series):
            raise TypeError("Reference data must be a pandas Series")
        if self._target_nodes:
            if not reference_data.index.isin(self._target_nodes).all():
                raise ValueError(
                    f"Reference data index {reference_data.index} does not match target nodes {self._target_nodes}"
                )
        if not reference_data.sum() == 1:
            raise ValueError("Reference data is not normalised to 1")

    def use_static_reference(self, data: pd.Series, reference_data: pd.Series):
        """
        Use a reference year to disaggregate the quantity spatially

        Args:
            data (pd.Series): The data to be disaggregated. Dims: (year,).
            reference_data (pd.Series): The reference data for disaggregation.
                E.g the distribution for a reference year. Dims: (space,).
        Returns:
            pd.DataFrame: The disaggregated data. Dims: (space, year).
        """

        self.validate_reference_data(reference_data)
        # outer/cartersian product to get (years, region) matrix
        return pd.DataFrame(
            np.outer(data, reference_data),
            index=data.index,
            columns=reference_data.index,
        ).T

use_static_reference(data, reference_data)

Use a reference year to disaggregate the quantity spatially

Parameters:

Name Type Description Default
data Series

The data to be disaggregated. Dims: (year,).

 required
reference_data Series

The reference data for disaggregation. E.g the distribution for a reference year. Dims: (space,).

 required

Returns: pd.DataFrame: The disaggregated data. Dims: (space, year).

Source code in src/rpycpl/disagg.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def use_static_reference(self, data: pd.Series, reference_data: pd.Series):
    """
    Use a reference year to disaggregate the quantity spatially

    Args:
        data (pd.Series): The data to be disaggregated. Dims: (year,).
        reference_data (pd.Series): The reference data for disaggregation.
            E.g the distribution for a reference year. Dims: (space,).
    Returns:
        pd.DataFrame: The disaggregated data. Dims: (space, year).
    """

    self.validate_reference_data(reference_data)
    # outer/cartersian product to get (years, region) matrix
    return pd.DataFrame(
        np.outer(data, reference_data),
        index=data.index,
        columns=reference_data.index,
    ).T

validate_reference_data(reference_data)

Check reference data has expected format

Parameters:

Name Type Description Default
reference_data Series

The reference data for disaggregation.

 required

Raises: TypeError: If reference data is not a pandas Series. ValueError: If reference data index does not match target nodes. ValueError: If reference data is not normalised to 1.

Source code in src/rpycpl/disagg.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def validate_reference_data(self, reference_data: pd.Series):
    """Check reference data has expected format

    Args:
        reference_data (pd.Series): The reference data for disaggregation.
    Raises:
        TypeError: If reference data is not a pandas Series.
        ValueError: If reference data index does not match target nodes.
        ValueError: If reference data is not normalised to 1.
    """

    if not isinstance(reference_data, pd.Series):
        raise TypeError("Reference data must be a pandas Series")
    if self._target_nodes:
        if not reference_data.index.isin(self._target_nodes).all():
            raise ValueError(
                f"Reference data index {reference_data.index} does not match target nodes {self._target_nodes}"
            )
    if not reference_data.sum() == 1:
        raise ValueError("Reference data is not normalised to 1")