-
Notifications
You must be signed in to change notification settings - Fork 903
/
ESKOM.py
130 lines (105 loc) · 4.88 KB
/
ESKOM.py
1
2
3
4
5
6
7
8
9
10
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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
import csv
from datetime import datetime, timedelta, timezone
from logging import Logger, getLogger
from pprint import PrettyPrinter
from zoneinfo import ZoneInfo
from numpy import nan
from requests import Response, Session
from electricitymap.contrib.lib.models.event_lists import ProductionBreakdownList
from electricitymap.contrib.lib.models.events import ProductionMix, StorageMix
from electricitymap.contrib.lib.types import ZoneKey
from parsers.lib.exceptions import ParserException
pp = PrettyPrinter(indent=4)
TIMEZONE = ZoneInfo("Africa/Johannesburg")
SOURCE = "eskom.co.za"
# Mapping columns to keys
# Helpful: https://www.eskom.co.za/dataportal/glossary/
COLUMN_MAPPING = {
0: "coal", # Thermal_Gen_Excl_Pumping_and_SCO
1: "ignored", # Eskom_OCGT_SCO_Pumping changed to ignored since negative oil is not possible, usually [-6, 0]
2: "ignored", # Eskom_Gas_SCO_Pumping changed to ignored since negative gas is not possible, usually -1 or 0
3: "ignored", # Hydro_Water_SCO_Pumping Probably electricity consumed by the plant itself (even) when not generating power. Can be ignored.
4: "hydro", # Pumped_Water_SCO_Pumping
5: "ignored", # Thermal_Generation sum of 0, 1, 2, 3, 4. Can be ignored.
6: "nuclear", # Nuclear_Generation
7: "ignored", # International_Imports
8: "oil", # Eskom_OCGT_Generation
9: "gas", # Eskom_Gas_Generation
10: "oil", # Dispatchable_IPP_OCGT
11: "hydro", # Hydro_Water_Generation
12: "hydro", # Pumped_Water_Generation
13: "ignored", # IOS_Excl_ILS_and_MLR Interruption of Supply. Can be ignored.
14: "ignored", # ILS_Usage Interruptible Load Shed = companies paid not to consume electricity. Can be ignored.
15: "ignored", # Manual_Load_Reduction_MLR MLS = forced load shedding. Can be ignored.
16: "wind", # Wind
17: "solar", # PV
18: "solar", # CSP
19: "biomass", # Other_RE - looking at capacity data and the IEA annual balances, other RE is likely to be biomass
}
# Ignored values
# 1, 2, 3, 5, 7, 13, 14, 15
# TODO:
# - 7 (international imports) can be further implemented in exchange function.
STORAGE_IDS = [4, 12]
PRODUCTION_IDS = [0, 6, 8, 9, 10, 11, 16, 17, 18, 19]
def get_url() -> str:
"""Returns the formatted URL"""
date = datetime.now(timezone.utc)
return f"https://www.eskom.co.za/dataportal/wp-content/uploads/{date.strftime('%Y')}/{date.strftime('%m')}/Station_Build_Up.csv"
def fetch_production(
zone_key: ZoneKey = ZoneKey("ZA"),
session: Session = Session(),
target_datetime: datetime | None = None,
logger: Logger = getLogger(__name__),
) -> list[dict]:
if target_datetime is not None:
local_target_datetime = target_datetime.astimezone(TIMEZONE)
local_one_week_ago = datetime.now(TIMEZONE) - timedelta(days=7)
if local_target_datetime < local_one_week_ago:
raise NotImplementedError(
f"No production data is available for {local_target_datetime}."
)
res: Response = session.get(get_url())
if not res.ok:
raise ParserException(
"ESKOM.py",
f"Exception when fetching production for {zone_key}: error when calling url={get_url()}",
zone_key=zone_key,
)
csv_data = csv.reader(res.text.splitlines())
return_list = ProductionBreakdownList(logger)
for row in csv_data:
if row[0] == "Date_Time_Hour_Beginning":
continue
returned_datetime = datetime.fromisoformat(row[0]).replace(tzinfo=TIMEZONE)
returned_production = row[1:] # First column is datetime
production = ProductionMix()
storage = StorageMix()
if all(value == "" for value in returned_production):
logger.warning(
f"Empty data for {returned_datetime} in {zone_key}. Skipping."
)
continue
else:
for index, prod_data_value in enumerate(returned_production):
prod_data_value = float(prod_data_value) if prod_data_value else nan
if index in PRODUCTION_IDS:
production.add_value(
COLUMN_MAPPING[index],
prod_data_value,
correct_negative_with_zero=True,
)
elif index in STORAGE_IDS:
storage.add_value(COLUMN_MAPPING[index], prod_data_value * -1)
return_list.append(
zoneKey=zone_key,
datetime=returned_datetime,
production=production,
storage=storage,
source=SOURCE,
)
return return_list.to_list()
if __name__ == "__main__":
"""Main method, never used by the Electricity Maps backend, but handy for testing."""
print("fetch_production() ->")
pp.pprint(fetch_production())