-
Notifications
You must be signed in to change notification settings - Fork 902
/
LK.py
109 lines (86 loc) · 3.66 KB
/
LK.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
#!/usr/bin/env python3
"""Quarter-hourly data parser for Sri Lanka
Fetches data for the previous day in 15-minute increments
Data is from the backend for the load curve graph on https://cebcare.ceb.lk/gensum/details
"""
import json
from datetime import datetime
from logging import Logger, getLogger
from typing import Optional
# The arrow library is used to handle datetimes
import arrow
# The request library is used to fetch content through HTTP
from requests import Session
from parsers.lib.exceptions import ParserException
TIMEZONE_NAME = "Asia/Colombo"
GENERATION_BREAKDOWN_URL = "https://cebcare.ceb.lk/GenSum/GetLoadCurveData"
SOURCE_NAME = "ceb.lk"
def fetch_production(
zone_key: str = "LK",
session: Optional[Session] = None,
target_datetime: Optional[datetime] = None,
logger: Logger = getLogger(__name__),
):
"""Requests the previous day's production mix (in MW) for Sri Lanka, per quarter-hour"""
if target_datetime is not None and target_datetime < arrow.utcnow().shift(days=-2):
raise NotImplementedError(
"The datasource currently only has data for yesterday"
) # so 0-24 hours ago just after local midnight to approx. 24-48 hours ago just before midnight
r = session or Session()
response = r.get(GENERATION_BREAKDOWN_URL)
assert response.status_code == 200, (
"Exception when fetching production for "
"{}: error when calling url={}".format(zone_key, GENERATION_BREAKDOWN_URL)
)
source_data = json.loads(
response.json()
) # Response is double encoded; a JSON array encoded as a JSON string
logger.debug(f"Raw generation breakdown: {source_data}", extra={"key": zone_key})
output = []
for quarter_hourly_source_data in source_data:
output_for_timestamp = {
"zoneKey": zone_key,
"datetime": arrow.get(
quarter_hourly_source_data["DateTime"],
"YYYY-MM-DD HH:mm:ss",
tzinfo=TIMEZONE_NAME,
).datetime,
"production": {
"biomass": 0.0,
"coal": 0.0,
"gas": 0.0,
"hydro": 0.0,
"nuclear": 0.0,
"oil": 0.0,
"solar": 0.0,
"wind": 0.0,
"geothermal": 0.0,
"unknown": 0.0,
},
"source": SOURCE_NAME,
}
for generation_type, outputInMW in quarter_hourly_source_data.items():
if generation_type == "DateTime":
continue
if generation_type == "Coal":
output_for_timestamp["production"]["coal"] += outputInMW
elif generation_type == "Major Hydro" or generation_type == "SPP Minihydro":
output_for_timestamp["production"]["hydro"] += outputInMW
elif generation_type == "SPP Biomass":
output_for_timestamp["production"]["biomass"] += outputInMW
elif generation_type == "Solar":
output_for_timestamp["production"]["solar"] += outputInMW
elif generation_type == "Thermal-Oil":
output_for_timestamp["production"]["oil"] += outputInMW
elif generation_type == "Wind":
output_for_timestamp["production"]["wind"] += outputInMW
else:
raise ParserException(
zone_key, "Unknown production type: " + generation_type
)
output.append(output_for_timestamp)
return output
if __name__ == "__main__":
"""Main method, never used by the Electricity Map backend, but handy for testing."""
print("fetch_production() ->")
print(fetch_production())