-
Notifications
You must be signed in to change notification settings - Fork 902
/
SG.py
340 lines (259 loc) · 11.9 KB
/
SG.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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
#!/usr/bin/env python3
import re
from collections import defaultdict
from datetime import datetime
from logging import Logger, getLogger
from typing import Optional
import arrow
from PIL import Image, ImageOps
from pytesseract import image_to_string
from requests import Session
TIMEZONE = "Asia/Singapore"
TICKER_URL = "https://www.emcsg.com/ChartServer/blue/ticker"
SOLAR_URL = "https://www.ema.gov.sg/cmsmedia/irradiance/plot.png"
"""
Around 95% of Singapore's generation is done with combined-cycle gas turbines.
Resources:
- https://www.ema.gov.sg/Statistics.aspx particularly filtering by "Generation"
The most recent document at time of writing is
https://www.ema.gov.sg/cmsmedia/Publications_and_Statistics/Statistics/27RSU.pdf
It says that between January 2016 and March 2017, non-gas monthly generation breakdown has been 0.7% to 1.4% coal,
0.1% to 1.0% "petroleum products" (excluding an exceptional value of 3.9% in Nov 2016),
and 2.7% to 3.4% "other", which includes among others waste-to-energy and solar.
- Write-up of energy statistics in 2015 and first half of 2016:
https://www.ema.gov.sg/cmsmedia/Publications_and_Statistics/Publications/SES/2016/Singapore%20Energy%20Statistics%202016.pdf
(referred to as "Singapore Energy Statistics 2016" below)
It states among others "In 2015, [natural gas] accounted for about 95% of fuel mix,
comparable with that recorded in 2014. Petroleum products, mainly in the form of diesel and fuel oil,
made up 0.7% of the fuel mix. Other energy products (e.g. municipal waste, coal and biomass)
accounted for 2.9% while the remaining 1.2% was from coal", and that steam turbines in Singapore
"typically run on fuel oil and diesel'
The real-time information on EMCSG website includes data for three categories: "CCGT/COGEN/TRIGEN", "GT", and "ST".
I take these to mean combined-cycle gas turbines/co-generation/tri-generation,
single-cycle gas turbines (following the Energy Statistics 2016 pg 24), and steam turbines respectively.
There is no real-time information on fuel for steam turbines.
For Electricity Map, we map CCGT and GT to gas, and ST to "unknown".
The Energy Market Authority estimates current solar production and publishes it at
https://www.ema.gov.sg/solarmap.aspx
There exists an interconnection to Malaysia, it is implemented in MY_WM.py.
"""
TYPE_MAPPINGS = {"CCGT/COGEN/TRIGEN": "gas", "GT": "gas", "ST": "unknown"}
def get_solar(session: Session, logger: Logger) -> Optional[float]:
"""
Fetches a graphic showing estimated solar production data.
Uses OCR (tesseract) to extract MW value.
"""
url = SOLAR_URL
solar_image = Image.open(session.get(url, stream=True).raw)
solar_mw = __detect_output_from_solar_image(solar_image, logger)
solar_dt = __detect_datetime_from_solar_image(solar_image, logger)
singapore_dt = arrow.now("Asia/Singapore")
diff = singapore_dt - solar_dt
# Need to be sure we don't get old data if image stops updating.
if diff.total_seconds() > 3600:
msg = (
"Singapore solar data is too old to use, " "parsed data timestamp was {}."
).format(solar_dt)
logger.warning(msg, extra={"key": "SG"})
return None
return solar_mw
def parse_megawatt_value(val) -> int:
"""Turns values like "5,156MW" and "26MW" into 5156 and 26 respectively."""
return int(val.replace(",", "").replace("MW", ""))
def parse_percent(val) -> float:
"""Turns values like "97.92%" into 0.9792."""
return float(val.replace("%", "")) / 100
def parse_price(price_str) -> float:
"""Turns values like "$70.57/MWh" into 70.57"""
return float(price_str.replace("$", "").replace("/MWh", ""))
def find_first_list_item_by_key_value(l, filter_key, filter_value, sought_key):
"""
Parses a common pattern in Singapore JSON response format. Examples:
[d['Value'] for d in energy_section if d['Label'] == 'Demand'][0]
=> find_first_list_item_by_key_value(energy_section, 'Label', 'Demand', 'Value')
[section['SectionData'] for section in sections if section['Name'] == 'Energy'][0]
=> find_first_list_item_by_key_value(sections, 'Name', 'Energy', 'SectionData')
[d['Value'] for d in energy_section if d['Label'] == 'USEP'][0]
=> find_first_list_item_by_key_value(energy_section, 'Label', 'USEP', 'Value')
"""
return [
list_item[sought_key]
for list_item in l
if list_item[filter_key] == filter_value
][0]
def sg_period_to_hour(period_str) -> float:
"""
Singapore electricity markets are split into 48 periods.
Period 1 starts at 00:00 Singapore time, Period 9 starts at 04:00.
Returns hours since midnight, possibly with 0.5 to indicate 30 minutes.
"""
return (float(period_str) - 1) / 2.0
def sg_data_to_datetime(data):
data_date = data["Date"]
data_time = sg_period_to_hour(data["Period"])
date_arrow = arrow.get(data_date, "DD MMM YYYY")
datetime_arrow = date_arrow.shift(hours=data_time)
data_datetime = arrow.get(datetime_arrow.datetime, TIMEZONE).datetime
return data_datetime
def fetch_production(
zone_key: str = "SG",
session: Optional[Session] = None,
target_datetime: Optional[datetime] = None,
logger: Logger = getLogger(__name__),
) -> dict:
"""Requests the last known production mix (in MW) of Singapore."""
if target_datetime:
raise NotImplementedError("This parser is not yet able to parse past dates")
requests_obj = session or Session()
response = requests_obj.get(TICKER_URL)
data = response.json()
sections = data["Sections"]
energy_section = find_first_list_item_by_key_value(
sections, "Name", "Energy", "SectionData"
)
demand_str = find_first_list_item_by_key_value(
energy_section, "Label", "Demand", "Value"
)
demand = parse_megawatt_value(demand_str)
system_loss_str = find_first_list_item_by_key_value(
energy_section, "Label", "System Loss", "Value"
)
system_loss = parse_megawatt_value(system_loss_str)
generation = demand + system_loss
mix_section = find_first_list_item_by_key_value(
sections, "Name", "Generator Type Share", "SectionData"
)
gen_types = {
gen_type["Label"]: parse_percent(gen_type["Value"]) for gen_type in mix_section
}
generation_by_type = defaultdict(float) # this dictionary will default keys to 0.0
for gen_type, gen_percent in gen_types.items():
gen_mw = gen_percent * generation
mapped_type = TYPE_MAPPINGS.get(gen_type, None)
if mapped_type:
generation_by_type[TYPE_MAPPINGS[gen_type]] += gen_mw
else:
# unrecognized - log it, then add into unknown
msg = (
'Singapore has unrecognized generation type "{}" '
"with production share {}%"
).format(gen_type, gen_percent)
logger.warning(msg)
generation_by_type["unknown"] += gen_mw
generation_by_type["solar"] = get_solar(requests_obj, logger)
# some generation methods that are not used in Singapore
generation_by_type.update({"nuclear": 0, "wind": 0, "hydro": 0})
return {
"datetime": sg_data_to_datetime(data),
"zoneKey": zone_key,
"production": generation_by_type,
"storage": {}, # there is no known electricity storage in Singapore
"source": "emcsg.com, ema.gov.sg",
}
def fetch_price(
zone_key: str = "SG",
session: Optional[Session] = None,
target_datetime: Optional[datetime] = None,
logger: Logger = getLogger(__name__),
) -> dict:
"""
Requests the most recent known power prices in Singapore (USEP).
See https://www.emcsg.com/marketdata/guidetoprices for details of what different prices in the data source mean.
We use USEP here: "The Uniform Singapore Energy Price (USEP) is the uniform price of energy
that applies for settlement purposes for all energy injections or withdrawals that are deemed to occur
at the Singapore hub. It is the weighted-average of the nodal prices at all off-take nodes in each half hour."
There are also price forecasts for future prices at https://www.emcsg.com/marketdata/priceinformation
that appears to extend to end of day in Singapore, so up to 24 hours into the future,
however we don't currently use this.
"""
if target_datetime:
raise NotImplementedError("This parser is not yet able to parse past dates")
requests_obj = session or Session()
response = requests_obj.get(TICKER_URL)
data = response.json()
sections = data["Sections"]
energy_section = find_first_list_item_by_key_value(
sections, "Name", "Energy", "SectionData"
)
price_str = find_first_list_item_by_key_value(
energy_section, "Label", "USEP", "Value"
)
price = parse_price(price_str)
return {
"zoneKey": zone_key,
"datetime": sg_data_to_datetime(data),
"currency": "SGD",
"price": price,
"source": "emcsg.com",
}
def __detect_datetime_from_solar_image(solar_image, logger: Logger):
w, h = solar_image.size
crop_left = int(w * 0.75)
crop_top = int(h * 0.87)
crop_right = int(w * 0.93)
crop_bottom = int(h * 0.92)
time_img = solar_image.crop((crop_left, crop_top, crop_right, crop_bottom))
processed_img = __preprocess_image_for_ocr(time_img)
text = image_to_string(
processed_img,
lang="eng",
config='--psm 7 -c tessedit_char_whitelist="0123456789:- "',
)
try:
time_pattern = r"\d+-\d+-\d+\s+\d+:\d+"
time_string = re.search(time_pattern, text, re.MULTILINE).group(0)
except AttributeError:
msg = "Unable to get values for SG solar from OCR text: {}".format(text)
logger.warning(msg, extra={"key": "SG"})
return None
solar_dt = arrow.get(time_string).replace(tzinfo="Asia/Singapore")
return solar_dt
def __detect_output_from_solar_image(solar_image, logger: Logger):
w, h = solar_image.size
crop_left = int(w * 0.65)
crop_top = int(h * 0.74)
crop_right = int(w * 0.93)
crop_bottom = int(h * 0.80)
output_img = solar_image.crop((crop_left, crop_top, crop_right, crop_bottom))
processed_img = __preprocess_image_for_ocr(output_img)
text = image_to_string(processed_img, lang="eng", config="--psm 7")
try:
pattern = r"Est. PV Output: (.*)MWac"
val = re.search(pattern, text, re.MULTILINE).group(1)
except AttributeError:
msg = "Unable to get values for SG solar from OCR text: {}".format(text)
logger.warning(msg, extra={"key": "SG"})
return None
# At night format changes from 0.00 to 0
# tesseract cannot distinguish singular 0 and O in font provided by image.
# This try/except will make sure no invalid data is returned.
try:
solar_mw = float(val)
except ValueError:
if val == "O":
solar_mw = 0.0
else:
msg = "Singapore solar data is unreadable - got {}.".format(val)
logger.warning(msg, extra={"key": "SG"})
return None
return solar_mw
def __preprocess_image_for_ocr(img):
"""
Perform a number of image pre-processing recommendations to improve success of character recognition.
:param img: the image to be processed
:return: pre-processed image, optimized for optical character recognition (OCR)
"""
# https://tesseract-ocr.github.io/tessdoc/ImproveQuality#inverting-images
inverted_img = ImageOps.invert(
img
) # assumes black background of Singapore solar output image
dark_text_on_light_bg = inverted_img.convert("L")
# https://tesseract-ocr.github.io/tessdoc/ImproveQuality#missing-borders
img_with_border = ImageOps.expand(dark_text_on_light_bg, border=2)
return img_with_border
if __name__ == "__main__":
"""Main method, never used by the Electricity Map backend, but handy for testing."""
print("fetch_production() ->")
print(fetch_production())
print('fetch_price("SG") ->')
print(fetch_price("SG"))