scenario_builder.demand.get_heat_profiles_deflex() - Code Metrics - Inspection of "Fix index" - reegis/scenario_builder - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 3b218c...39186e )

by Uwe

created 2021-03-25 12:26 UTC

scenario_builder.demand.get_heat_profiles_deflex() C

↳ Parent: scenario_builder.demand

Complexity

Conditions

Size

Total Lines	86
Code Lines	43

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	10
eloc	43
nop	5
dl	0
loc	86
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-2021 Uwe Krien <[email protected]>

SPDX-License-Identifier: MIT
"""


import calendar
import logging
import os

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


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("creator", "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")
    combine_fuels = {"natural gas": "gas"}

    # fuels to be dissolved per region
    region_fuels = cfg.get_list("creator", "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", 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", 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, opsd_version=None, weather_year=None):
    """

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

    Returns
    -------

    Examples
    --------
    >>> from reegis import geometries  # doctest: +SKIP
    >>> fs=geometries.get_federal_states_polygon()  # 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", "all"].sum())  # doctest: +SKIP
    10069304

    """
    demand_series = {
        "electricity demand series": scenario_elec_demand(
            pd.DataFrame(),
            regions,
            year,
            name,
            weather_year=weather_year,
            version=opsd_version,
        )
    }
    if cfg.get("creator", "heat"):
        demand_series["heat demand series"] = scenario_heat_demand(
            regions, year, weather_year=weather_year
        ).reset_index(drop=True)
    return demand_series


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

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

    Returns
    -------

    """
    return get_heat_profiles_deflex(
        regions, year, weather_year=weather_year
    ).sort_index(1)


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

reegis / scenario_builder

Push — master ( 3b218c...39186e )

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