Explore
Groups
No topics yet. Start the conversation.
Summary
User supplied summary for the plot
Electricity generation (GW) for Germany last 7 days.
Description
The below description is supplied in free-text by the user
import sys
import requests, pandas as pd
from lxml import etree
from novem import Plot
import inspect
import os
import dotenv
dotenv.load_dotenv()
TOKEN = os.getenv("ENTSO_TOKEN")
REGIONS = {
"Norway": {
"documentType": "A75",
"domains": [
"10YNO-1--------2", # NO1
"10YNO-2--------T", # NO2
"10YNO-3--------J", # NO3
"10YNO-4--------9", # NO4
"10Y1001A1001A48H", # NO5
],
},
"Spain": {
"documentType": "A75",
"domains": [
"10YES-REE------0", # ES mainland
],
},
"Sweden": {
"documentType": "A75",
"domains": [
"10Y1001A1001A44P", # SE1
"10Y1001A1001A45N", # SE2
"10Y1001A1001A46L", # SE3
"10Y1001A1001A47J", # SE4
],
},
"Germany": {
"documentType": "A75",
"domains": [
"10Y1001A1001A82H", # Germany
],
}
}
PSR_TO_NAME = {
'B01': 'Biomass',
'B02': 'Fossil Brown coal/Lignite',
'B03': 'Fossil Coal-derived gas',
'B04': 'Fossil Gas',
'B05': 'Fossil Hard coal',
'B06': 'Fossil Oil',
'B07': 'Fossil Oil shale',
'B08': 'Fossil Peat',
'B09': 'Geothermal',
'B10': 'Hydro Pumped Storage',
'B11': 'Hydro Run-of-river and poundage',
'B12': 'Hydro Water Reservoir',
'B13': 'Marine',
'B14': 'Nuclear',
'B15': 'Other renewable',
'B16': 'Solar',
'B17': 'Waste',
'B18': 'Wind Offshore',
'B19': 'Wind Onshore',
'B20': 'Other',
'B25': 'Energy storage'
}
AGGREGATION = {
'Biomass': 'Organic',
'Energy storage': 'Battery',
'Fossil Brown coal/Lignite': 'Fossil',
'Fossil Coal-derived gas': 'Fossil',
'Fossil Gas': 'Fossil',
'Fossil Hard coal': 'Fossil',
'Fossil Oil': 'Fossil',
'Fossil Oil shale': 'Fossil',
'Fossil Peat': 'Fossil',
'Geothermal': 'Other',
'Hydro Pumped Storage': 'Hydro',
'Hydro Run-of-river and poundage': 'Hydro',
'Hydro Water Reservoir': 'Hydro',
'Marine': 'Other',
'Nuclear': 'Nuclear',
'Other': 'Other',
'Other renewable': 'Other',
'Solar': 'Solar',
'Waste': 'Organic',
'Wind Offshore': 'Wind',
'Wind Onshore': 'Wind'
}
T_END = pd.Timestamp.utcnow().floor('60min')
T_START = T_END - pd.Timedelta('7d')
T_INTERVAL = f"{T_START:%Y-%m-%dT%H:%MZ}/{T_END:%Y-%m-%dT%H:%MZ}"
PARAMETERS_BASE = {
"securityToken": TOKEN,
"documentType": "A65",
"processType": "A16",
"TimeInterval": T_INTERVAL,
}
def fetch_data(eics, document_type):
dfs = []
for eic in eics:
p = PARAMETERS_BASE.copy()
if document_type == "A16":
p["area"] = eic
else:
p["in_Domain"] = eic
p["documentType"] = document_type
if document_type == "A65":
p["outBiddingZone_Domain"] = eic
if document_type == "A16":
p["processType"] = "A01"
print(f"Fetching data for {eic}...")
r = requests.get("https://web-api.tp.entsoe.eu/api", params=p)
try:
r.raise_for_status()
except requests.HTTPError as e:
print("\n❌ HTTP error for:", p)
print("Response:", r.text)
raise
dfs.append(parse_gl(r.content))
df = pd.concat(dfs).groupby(["ts_utc","psrType"], as_index=False)["MW"].sum()
return df
# Minimal XML → tidy dataframe (timestamp, psrType, MW)
def parse_gl(xml_bytes):
ns = {"gl":"urn:iec62325.351:tc57wg16:451-6:generationloaddocument:3:0"}
root = etree.fromstring(xml_bytes)
rows = []
for ts in root.findall(".//gl:TimeSeries", ns):
psr = ts.findtext(".//gl:MktPSRType/gl:psrType", namespaces=ns)
period = ts.find(".//gl:Period", ns)
start = pd.to_datetime(period.findtext("./gl:timeInterval/gl:start", namespaces=ns))
res = pd.Timedelta(period.findtext("./gl:resolution", namespaces=ns).replace("PT","").lower())
for pt in period.findall("./gl:Point", ns):
pos = int(pt.findtext("./gl:position", namespaces=ns))
qty = float(pt.findtext("./gl:quantity", namespaces=ns))
ts_utc = start + (pos-1)*res
rows.append((ts_utc, psr, qty))
return pd.DataFrame(rows, columns=["ts_utc","psrType","MW"])
def clean_data(df):
df['ts_utc'] = pd.to_datetime(df['ts_utc'])
all_ts = df['ts_utc'].sort_values().unique()
full_index = pd.MultiIndex.from_product(
[all_ts, df['psrType'].unique()],
names=['ts_utc', 'psrType']
)
df = (
df.set_index(['ts_utc', 'psrType'])
.reindex(full_index)
.groupby(level='psrType')
.ffill()
.reset_index()
)
# Map names
df['Production Type'] = df['psrType'].map(PSR_TO_NAME)
df['prod_type_agg'] = df['Production Type'].map(AGGREGATION)
# Rename and convert
df = df.rename(columns={'ts_utc': 'start', 'MW': 'Generation (MW)'})
df['Generation (MW)'] = pd.to_numeric(df['Generation (MW)'], errors='coerce')
df['Generation (GW)'] = df['Generation (MW)'] / 1000 # Convert to GW
df.index.name = 'Timestamp'
# Drop unused columns early
df = df.drop(columns=['psrType', 'Production Type'])
# Aggregate
df = df.groupby(['start', 'prod_type_agg'])['Generation (GW)'].sum()
# ---- KEY PART: make a complete grid ----
# All timestamps
all_times = df.index.get_level_values('start').unique()
# All categories
all_cats = df.index.get_level_values('prod_type_agg').unique()
# Full multi-index
full_index = pd.MultiIndex.from_product(
[all_times, all_cats],
names=["start", "prod_type_agg"]
)
# Reindex and fill missing values with 0
df = df.reindex(full_index, fill_value=0).reset_index()
return df
def novem_plot(df_hourly, region):
plot_id = f'entso-e-generation-{region.lower()}'
plt = Plot(plot_id)
plt.type = 'custom'
plt.data = df_hourly
plt.name = f'Electricity generation (GW) for {region} last 7 days. '
plt.title = f'Generation (GW) {region}'
plt.caption = 'Source: ENTSO-E, calculations by novem.'
plt.shared = 'public'
with open('./custom/custom-entso-e.js', 'r') as f:
plt.api_write('/config/custom/custom.js', f.read())
plt.description = '```python' + ' ' + inspect.getsource(sys.modules[__name__]) + '```' # Include this python script in Description
plt.summary = plt.name
# %%
for region in REGIONS.keys():
print(f"Fetching {region} data...")
eics = REGIONS[region]['domains']
document_type = REGIONS[region]['documentType']
df = fetch_data(eics,document_type)
df = clean_data(df)
#if region == "Spain":
# df = fix_spain_nuclear(df)
novem_plot(df, region)
# %%