# -*- coding: utf-8 -*-
"""Prepare parts of the energy balance of Germany and its federal states.
SPDX-FileCopyrightText: 2016-2021 Uwe Krien <krien@uni-bremen.de>
SPDX-License-Identifier: MIT
"""
__copyright__ = "Uwe Krien <krien@uni-bremen.de>"
__license__ = "MIT"
# Python libraries
import os
import logging
# Internal modules
from reegis import config as cfg
from reegis import inhabitants
from reegis import geometries
# External packages
import pandas as pd
import requests
[docs]def get_eb_index_translation_dict():
dic = cfg.get_dict("EB_INDEX_TRANSLATION")
dic_keys = list(dic.keys())
for key in dic_keys:
for keyword in ["Umw-Einsatz", "Umw-Ausstoß", "Umw-Verbrauch"]:
if keyword in key:
value = dic.pop(key)
key = key.replace(keyword, keyword + ":")
dic[key] = value
if dic[key] == "":
value = key
value = value.replace("Umw-Einsatz", "transformation input")
value = value.replace("Umw-Ausstoß", "transformation output")
value = value.replace("Umw-Verbrauch", "transformation demand")
dic[key] = value
return dic
[docs]def get_de_balance(year):
"""Download and return energy balance of germany for a given year."""
url = cfg.get("energy_balance", "url_energy_balance_germany")
req = requests.get(url.format(year=str(year)[-2:], suffix="xls"))
if int(req.headers["Content-length"]) > 0:
fn_de = os.path.join(
cfg.get("paths", "energy_balance"),
cfg.get("energy_balance", "energy_balance_de_original"),
).format(year=year, suffix="xls")
with open(fn_de, "wb") as fout:
fout.write(req.content)
else:
req = requests.get(url.format(year=str(year)[-2:], suffix="xlsx"))
if int(req.headers["Content-length"]) > 0:
fn_de = os.path.join(
cfg.get("paths", "energy_balance"),
cfg.get("energy_balance", "energy_balance_de_original"),
).format(year=year, suffix="xlsx")
with open(fn_de, "wb") as fout:
fout.write(req.content)
else:
raise ValueError("No file received. Check url.")
fn_h = os.path.join(
cfg.get("paths", "static_sources"), "energy_balance_header_germany.csv"
)
head = pd.read_csv(fn_h, header=[0]).columns
df = pd.read_excel(fn_de, "tj", index_col=[0], skiprows=6, usecols="A:AI",
nrows=68)
df.columns = head[1:]
return df
[docs]def get_de_usage_balance(year, grouped=False):
"""
Parameters
----------
year
grouped
Returns
-------
Examples
--------
>>> df=get_de_usage_balance(2015, True)
>>> df.loc['total', 'total']
8898093
"""
df = get_de_balance(year)
df["Braunkohle (sonstige)"] += df["Hartbraunkohle"]
df.drop(
["Hartbraunkohle", "primär (gesamt)", "sekundär (gesamt)", "Row"],
axis=1,
inplace=True,
)
df = df.rename(columns=cfg.get_dict("COLUMN_TRANSLATION"))
df = df.rename(cfg.get_dict("SECTOR"))
df = df.loc[set(cfg.get_dict("SECTOR_OLD").values())]
if grouped:
df = df.groupby(by=cfg.get_dict("FUEL_GROUPS"), axis=1).sum()
return df
[docs]def get_domestic_retail_share(year, grouped=False):
"""
Parameters
----------
year
grouped
Returns
-------
Examples
--------
>>> df=get_domestic_retail_share(2014, True)
>>> df.loc['district heating', 'domestic']
0.73
"""
deb = get_de_usage_balance(year=year, grouped=grouped)
deb.sort_index(1, inplace=True)
# deb=deb.groupby(level=[1]).sum()
share = pd.DataFrame()
share["domestic"] = (
deb.loc["domestic"] / deb.loc["domestic and retail"]
).round(2)
share["retail"] = (
(deb.loc["retail"] / deb.loc["domestic and retail"])
.round(2)
.transpose()
)
return share
[docs]def get_states_energy_balance(year=None):
"""
Get the energy balance for a given year. The input file is the csv-file
downloaded from:
https://www.lak-energiebilanzen.de/eingabe-dynamisch/?a=e900
Parameters
----------
year : int or None
If year is None all possible years will be returned.
Returns
-------
pandas.DataFrame
Notes
-----
Translation of the index is incomplete.
Examples
--------
>>> eb=get_states_energy_balance(2012)
>>> eb.loc[(['BB', 'NW'], 'extraction'), 'lignite (raw)'].round(1)
BB extraction 316931.2
NW extraction 927025.0
Name: lignite (raw), dtype: float64
>>> eb=get_states_energy_balance()
>>> eb.loc[([2012, 2013], ['BB', 'NW'], 'extraction'), 'lignite (raw)'
... ].round(1).sort_index()
2012 BB extraction 316931.2
NW extraction 927025.0
2013 BB extraction 318703.2
NW extraction 894546.0
Name: lignite (raw), dtype: float64
"""
header_fn = os.path.join(
cfg.get("paths", "static_sources"),
cfg.get("energy_balance", "energy_balance_header"),
)
header = pd.read_csv(header_fn)
if os.path.sep not in cfg.get("energy_balance", "energy_balance_states"):
fn = os.path.join(
cfg.get("paths", "static_sources"),
cfg.get("energy_balance", "energy_balance_states"),
)
else:
fn = cfg.get("energy_balance", "energy_balance_states")
eb = pd.read_csv(
fn,
sep=";",
skiprows=4,
index_col=[0, 1, 2],
skipfooter=10,
engine="python",
encoding="utf-8",
)
eb.columns = header.columns
codes = {
"Baden-Württemberg": "BW",
"Bayern": "BY",
"Berlin": "BE",
"Brandenburg": "BB",
"Bremen": "HB",
"Hamburg": "HH",
"Hessen": "HE",
"Mecklenburg-Vorpommern": "MV",
"Niedersachsen": "NI",
"Nordrhein-Westfalen": "NW",
"Rheinland-Pfalz": "RP",
"Saarland": "SL",
"Sachsen": "SN",
"Sachsen-Anhalt": "ST",
"Schleswig-Holstein": "SH",
"Thüringen": "TH",
}
fs_list = [codes[x] for x in eb.index.get_level_values(0).unique()]
eb.index.set_levels(fs_list, level=0, inplace=True)
eb = eb.fillna(0)
eb.drop(["Anmerkung", "Stand"], axis=1, inplace=True)
eb = eb.swaplevel(0, 1)
eb.index = eb.index.rename(["", "", ""])
eb = eb.rename(columns=cfg.get_dict("COLUMN_TRANSLATION"))
eb = eb.rename(index=get_eb_index_translation_dict(), level=2)
if year is not None:
eb = eb.loc[year]
return eb
[docs]def get_usage_balance(year, grouped=False):
"""
Get the usage part of the energy balance.
Parameters
----------
year : int
Year of the energy balance.
grouped : bool
If set to True the fuels will be grouped to main groups like hard coal
or lignite.
Returns
-------
pandas.DataFrame
Examples
--------
>>> year=2013
>>> cb=get_usage_balance(year)
>>> total=cb.pop('total')
>>> int((cb.loc['BE'].sum(axis=1) - total.loc['BE']).sum())
0
>>> int((cb.loc['ST'].sum(axis=1) - total.loc['ST']).sum())
-8952
>>> int((cb.loc['BY'].sum(axis=1) - total.loc['BY']).sum())
-17731
>>> cb=get_usage_balance(year)
>>> cb=fix_usage_balance(cb, year)
>>> total=cb.pop('total')
>>> int((cb.loc['BE'].sum(axis=1) - total.loc['BE']).sum())
0
>>> int((cb.loc['ST'].sum(axis=1) - total.loc['ST']).sum())
0
>>> int((cb.loc['BY'].sum(axis=1) - total.loc['BY']).sum())
0
"""
eb = get_states_energy_balance(year)
eb = eb.loc[
(slice(None), list(cfg.get_dict("SECTOR").keys())), slice(None)
]
eb = eb.rename(index=cfg.get_dict("SECTOR"), level=1)
if grouped:
eb = eb.groupby(by=cfg.get_dict("FUEL_GROUPS"), axis=1).sum()
return eb
[docs]def fix_usage_balance(eb, year):
"""
Fixes the energy balances after analysing them. This is done manually.
"""
if year not in [2012, 2013, 2014]:
raise ValueError(
"You cannot edit the balance for year {0}.".format(year)
)
# ******************************************************************
# Bavaria (Bayern) - Missing coal values
# Difference between fuel sum and LAK table
missing = {2012: 10529, 2013: 8995, 2014: 9398}
fix = missing[year]
# the missing value is added to 'hard coal raw' even though it is not
# specified which hard coal product is missing.
eb.loc[("BY", "total"), "hard coal (raw)"] = fix
# There is a small amount specified in the 'domestic and retail'
# sector.
dom_retail = eb.loc[("BY", "domestic and retail"), "hard coal (raw)"]
# The rest of the total hard coal consumption comes from the industrial
# sector.
eb.loc[("BY", "industrial"), "hard coal (raw)"] = fix - dom_retail
# ******************************************************************
# Berlin (Berlin) - corrected values for domestic gas and electricity
# In new publications (e.g. LAK table) these values have changed. The newer
# values will be used.
if year == 2013 or year == 2012:
electricity = {2012: 9150, 2013: 7095}
gas = {2012: -27883, 2013: -13317}
total = {2012: -18733, 2013: -6222}
for row in ["total", "domestic and retail", "retail"]:
eb.loc[("BE", row), "electricity"] += electricity[year]
eb.loc[("BE", row), "natural gas"] += gas[year]
eb.loc[("BE", row), "total"] += total[year]
# ******************************************************************
# Saxony-Anhalt (Sachsen Anhalt) - missing values for hard coal, oil and
# other depending on the year. Due to a lack of information the
# difference
# will be halved between the sectors.
missing = {2012: 5233, 2013: 4396, 2014: 3048}
if year == 2012:
fix = missing[year]
# the missing value is added to 'hard coal raw' even though it is not
# specified which hard coal product is missing.
eb.loc[("ST", "industrial"), "other"] += fix / 2
eb.loc[("ST", "industrial"), "hard coal (raw)"] += fix / 2
# There is a small amount specified in the 'domestic and retail'
# sector.
dom_retail_hc = eb.loc[
("ST", "domestic and retail"), "hard coal (raw)"
]
# The rest of the total hard coal consumption comes from the industrial
# sector.
eb.loc[("ST", "total"), "other"] += fix / 2
eb.loc[("ST", "total"), "hard coal (raw)"] += fix / 2 + dom_retail_hc
if year == 2013:
fix = missing[year]
# the missing value is added to 'hard coal raw' even though it is not
# specified which hard coal product is missing.
eb.loc[("ST", "industrial"), "mineral oil products"] += fix / 2
eb.loc[("ST", "industrial"), "hard coal (raw)"] += fix / 2
# There is a small amount specified in the 'domestic and retail'
# sector.
dom_retail_hc = eb.loc[
("ST", "domestic and retail"), "hard coal (raw)"
]
dom_retail_oil = eb.loc[
("ST", "domestic and retail"), "mineral oil products"
]
# The rest of the total hard coal consumption comes from the industrial
# sector.
eb.loc[("ST", "total"), "mineral oil products"] += fix / 2 + (
dom_retail_oil
)
eb.loc[("ST", "total"), "hard coal (raw)"] += fix / 2 + dom_retail_hc
if year == 2014:
fix = missing[year]
# the missing value is added to 'hard coal raw' even though it is not
# specified which hard coal product is missing.
eb.loc[("ST", "industrial"), "mineral oil products"] += fix / 2
eb.loc[("ST", "industrial"), "hard coal (coke)"] += fix / 2
# There is a small amount specified in the 'domestic and retail'
# sector.
dom_retail = eb.loc[
("ST", "domestic and retail"), "mineral oil products"
]
# The rest of the total hard coal consumption comes from the industrial
# sector.
eb.loc[("ST", "total"), "mineral oil products"] += fix / 2 + dom_retail
eb.loc[("ST", "total"), "hard coal (coke)"] += fix / 2
return eb
if __name__ == "__main__":
pass