scenario_builder.demand.get_heat_profiles_deflex() - Code Metrics - Inspection of "Add deflex scenario builder" - reegis/scenario_builder - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#2)

by Uwe

created 2020-10-20 09:08 UTC

scenario_builder.demand.get_heat_profiles_deflex() C

↳ Parent: scenario_builder.demand

Complexity

Conditions

Size

Total Lines	84
Code Lines	43

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	10
eloc	43
nop	5
dl	0
loc	84
rs	5.9999
c	0
b	0
f	0

How to fix Long Method Complexity

"""Create a basic scenario from the internal data structure.

SPDX-FileCopyrightText: 2016-2019 Uwe Krien <[email protected]>

SPDX-License-Identifier: MIT
"""


import calendar
import logging
import os

import pandas as pd
from reegis import demand_elec
from reegis import demand_heat

from scenario_builder import config as cfg


def get_heat_profiles_deflex(
    deflex_geo, year, time_index=None, weather_year=None, keep_unit=False
):
    """

    Parameters
    ----------
    year
    deflex_geo
    time_index
    weather_year
    keep_unit

    Returns
    -------

    """
    # separate_regions=keep all demand connected to the region
    separate_regions = cfg.get_list("demand_heat", "separate_heat_regions")
    # Add lower and upper cases to be not case sensitive
    separate_regions = ([x.upper() for x in separate_regions] +
                        [x.lower() for x in separate_regions])

    # add second fuel to first
    combine_fuels = cfg.get_dict("combine_heat_fuels")

    # fuels to be dissolved per region
    region_fuels = cfg.get_list("demand_heat", "local_fuels")

    fn = os.path.join(
        cfg.get("paths", "demand"),
        "heat_profiles_{year}_{map}".format(year=year, map=deflex_geo.name),
    )

    demand_region = (
        demand_heat.get_heat_profiles_by_region(
            deflex_geo, year, to_csv=fn, weather_year=weather_year
        )
        .groupby(level=[0, 1], axis=1)
        .sum()
    )

    # Decentralised demand is combined to a nation-wide demand if not part
    # of region_fuels.
    regions = list(
        set(demand_region.columns.get_level_values(0).unique())
        - set(separate_regions)
    )

    # If region_fuels is 'all' fetch all fuels to be local.
    if "all" in region_fuels:
        region_fuels = demand_region.columns.get_level_values(1).unique()

    for fuel in demand_region.columns.get_level_values(1).unique():
        demand_region["DE_demand", fuel] = 0

    for region in regions:
        for f1, f2 in combine_fuels.items():
            demand_region[region, f1] += demand_region[region, f2]
            demand_region.drop((region, f2), axis=1, inplace=True)
        cols = list(set(demand_region[region].columns) - set(region_fuels))
        for col in cols:
            demand_region["DE_demand", col] += demand_region[region, col]
            demand_region.drop((region, col), axis=1, inplace=True)

    if time_index is not None:
        demand_region.index = time_index

    if not keep_unit:
        msg = (
            "The unit of the source is 'TJ'. "
            "Will be divided by {0} to get 'MWh'."
        )
        converter = 0.0036
        demand_region = demand_region.div(converter)
        logging.debug(msg.format(converter))

    demand_region.sort_index(1, inplace=True)

    for c in demand_region.columns:
        if demand_region[c].sum() == 0:
            demand_region.drop(c, axis=1, inplace=True)

    return demand_region


def scenario_demand(regions, year, name, weather_year=None):
    """

    Parameters
    ----------
    regions
    year
    name
    weather_year

    Returns
    -------

    Examples
    --------
    >>> regions=geometries.deflex_regions(rmap="de21")  # doctest: +SKIP
    >>> my_demand=scenario_demand(regions, 2014, "de21")  # doctest: +SKIP
    >>> int(my_demand["DE01", "district heating"].sum())  # doctest: +SKIP
    18639262
    >>> int(my_demand["DE05", "electrical_load"].sum())  # doctest: +SKIP
    10069304

    """
    demand_series = scenario_elec_demand(
        pd.DataFrame(), regions, year, name, weather_year=weather_year
    )
    if cfg.get("basic", "heat"):
        demand_series = scenario_heat_demand(
            demand_series, regions, year, weather_year=weather_year
        )
    return demand_series


def scenario_heat_demand(table, regions, year, weather_year=None):
    """

    Parameters
    ----------
    table
    regions
    year
    weather_year

    Returns
    -------

    """
    idx = table.index  # Use the index of the existing time series
    table = pd.concat(
        [
            table,
            get_heat_profiles_deflex(
                regions, year, idx, weather_year=weather_year
            ),
        ],
        axis=1,
    )
    return table.sort_index(1)


def scenario_elec_demand(table, regions, year, name, weather_year=None):
    """

    Parameters
    ----------
    table
    regions
    year
    name
    weather_year

    Returns
    -------

    """
    if weather_year is None:
        demand_year = year
    else:
        demand_year = weather_year

    df = demand_elec.get_entsoe_profile_by_region(
        regions, demand_year, name, annual_demand="bmwi"
    )
    df = pd.concat([df], axis=1, keys=["electrical_load"]).swaplevel(0, 1, 1)
    df = df.reset_index(drop=True)
    if not calendar.isleap(year) and len(df) > 8760:
        df = df.iloc[:8760]
    return pd.concat([table, df], axis=1).sort_index(1)


if __name__ == "__main__":
    pass


1			"""Create a basic scenario from the internal data structure.
2
3			SPDX-FileCopyrightText: 2016-2019 Uwe Krien <[email protected]>
4
5			SPDX-License-Identifier: MIT
6			"""
7
8
9			import calendar
10			import logging
11			import os
12
13			import pandas as pd
14			from reegis import demand_elec
15			from reegis import demand_heat
16
17			from scenario_builder import config as cfg
18
19
20			def get_heat_profiles_deflex(
21			deflex_geo, year, time_index=None, weather_year=None, keep_unit=False
22			):
23			"""
24
25			Parameters
26			----------
27			year
28			deflex_geo
29			time_index
30			weather_year
31			keep_unit
32
33			Returns
34			-------
35
36			"""
37			# separate_regions=keep all demand connected to the region
38			separate_regions = cfg.get_list("demand_heat", "separate_heat_regions")
39			# Add lower and upper cases to be not case sensitive
40			separate_regions = ([x.upper() for x in separate_regions] +
41			[x.lower() for x in separate_regions])
42
43			# add second fuel to first
44			combine_fuels = cfg.get_dict("combine_heat_fuels")
45
46			# fuels to be dissolved per region
47			region_fuels = cfg.get_list("demand_heat", "local_fuels")
48
49			fn = os.path.join(
50			cfg.get("paths", "demand"),
51			"heat_profiles_{year}_{map}".format(year=year, map=deflex_geo.name),
52			)
53
54			demand_region = (
55			demand_heat.get_heat_profiles_by_region(
56			deflex_geo, year, to_csv=fn, weather_year=weather_year
57			)
58			.groupby(level=[0, 1], axis=1)
59			.sum()
60			)
61
62			# Decentralised demand is combined to a nation-wide demand if not part
63			# of region_fuels.
64			regions = list(
65			set(demand_region.columns.get_level_values(0).unique())
66			- set(separate_regions)
67			)
68
69			# If region_fuels is 'all' fetch all fuels to be local.
70			if "all" in region_fuels:
71			region_fuels = demand_region.columns.get_level_values(1).unique()
72
73			for fuel in demand_region.columns.get_level_values(1).unique():
74			demand_region["DE_demand", fuel] = 0
75
76			for region in regions:
77			for f1, f2 in combine_fuels.items():
78			demand_region[region, f1] += demand_region[region, f2]
79			demand_region.drop((region, f2), axis=1, inplace=True)
80			cols = list(set(demand_region[region].columns) - set(region_fuels))
81			for col in cols:
82			demand_region["DE_demand", col] += demand_region[region, col]
83			demand_region.drop((region, col), axis=1, inplace=True)
84
85			if time_index is not None:
86			demand_region.index = time_index
87
88			if not keep_unit:
89			msg = (
90			"The unit of the source is 'TJ'. "
91			"Will be divided by {0} to get 'MWh'."
92			)
93			converter = 0.0036
94			demand_region = demand_region.div(converter)
95			logging.debug(msg.format(converter))
96
97			demand_region.sort_index(1, inplace=True)
98
99			for c in demand_region.columns:
100			if demand_region[c].sum() == 0:
101			demand_region.drop(c, axis=1, inplace=True)
102
103			return demand_region
104
105
106			def scenario_demand(regions, year, name, weather_year=None):
107			"""
108
109			Parameters
110			----------
111			regions
112			year
113			name
114			weather_year
115
116			Returns
117			-------
118
119			Examples
120			--------
121			>>> regions=geometries.deflex_regions(rmap="de21") # doctest: +SKIP
122			>>> my_demand=scenario_demand(regions, 2014, "de21") # doctest: +SKIP
123			>>> int(my_demand["DE01", "district heating"].sum()) # doctest: +SKIP
124			18639262
125			>>> int(my_demand["DE05", "electrical_load"].sum()) # doctest: +SKIP
126			10069304
127
128			"""
129			demand_series = scenario_elec_demand(
130			pd.DataFrame(), regions, year, name, weather_year=weather_year
131			)
132			if cfg.get("basic", "heat"):
133			demand_series = scenario_heat_demand(
134			demand_series, regions, year, weather_year=weather_year
135			)
136			return demand_series
137
138
139			def scenario_heat_demand(table, regions, year, weather_year=None):
140			"""
141
142			Parameters
143			----------
144			table
145			regions
146			year
147			weather_year
148
149			Returns
150			-------
151
152			"""
153			idx = table.index # Use the index of the existing time series
154			table = pd.concat(
155			[
156			table,
157			get_heat_profiles_deflex(
158			regions, year, idx, weather_year=weather_year
159			),
160			],
161			axis=1,
162			)
163			return table.sort_index(1)
164
165
166			def scenario_elec_demand(table, regions, year, name, weather_year=None):
167			"""
168
169			Parameters
170			----------
171			table
172			regions
173			year
174			name
175			weather_year
176
177			Returns
178			-------
179
180			"""
181			if weather_year is None:
182			demand_year = year
183			else:
184			demand_year = weather_year
185
186			df = demand_elec.get_entsoe_profile_by_region(
187			regions, demand_year, name, annual_demand="bmwi"
188			)
189			df = pd.concat([df], axis=1, keys=["electrical_load"]).swaplevel(0, 1, 1)
190			df = df.reset_index(drop=True)
191			if not calendar.isleap(year) and len(df) > 8760:
192			df = df.iloc[:8760]
193			return pd.concat([table, df], axis=1).sort_index(1)
194
195
196			if __name__ == "__main__":
197			pass
198

reegis / scenario_builder

Pull Request — master (#2)

scenario_builder.demand.get_heat_profiles_deflex() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like