-
Notifications
You must be signed in to change notification settings - Fork 910
/
PrinceEdwardIsland.py
151 lines (127 loc) · 5.26 KB
/
PrinceEdwardIsland.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
#!/usr/bin/env python3
import json
from datetime import datetime
from logging import Logger, getLogger
from typing import Any
from zoneinfo import ZoneInfo
from requests import Session
from electricitymap.contrib.lib.models.event_lists import (
ExchangeList,
ProductionBreakdownList,
)
from electricitymap.contrib.lib.models.events import ProductionMix
from electricitymap.contrib.lib.types import ZoneKey
from parsers.lib.exceptions import ParserException
SOURCE = "PrinceEdwardIsland.ca"
TIMEZONE = ZoneInfo("Canada/Atlantic")
# The dashboard associated with this API can be found at
# https://www.PrinceEdwardIsland.ca/en/feature/pei-wind-energy.
URL = "https://wdf.PrinceEdwardIsland.ca/api/workflow" # API
ZONE_KEY = ZoneKey("CA-PE")
def _parse_power(text: str) -> float:
return float(text[: text.index(" MW")])
def _get_event(session: Session):
# Request an event from the source.
event = session.post(
URL,
data=json.dumps({"featureName": "WindEnergy", "queryName": "WindEnergy"}),
headers={"Content-Type": "application/json"},
).json()["data"]
# Collect the production modes and power values into a lookup table.
modes = dict(
object_["data"]["header"].split(": ", maxsplit=1)
for object_ in event
if object_["type"] == "GaugeChart"
)
# Extract the timestamp.
timestamp = datetime.strptime(
event[-1]["data"]["text"][len("Last updated ") :], "%B %d, %Y %I:%M %p"
).replace(tzinfo=TIMEZONE)
# Return a new, more convenient lookup table.
return {
"datetime": timestamp,
"fossil": _parse_power(modes["Total On-Island Fossil Fuel Generation"]),
"load": _parse_power(modes["Total On-Island Load"]),
"wind": _parse_power(modes["Total On-Island Wind Generation"]),
}
def fetch_production(
zone_key: ZoneKey = ZONE_KEY,
session: Session | None = None,
target_datetime: datetime | None = None,
logger: Logger = getLogger(__name__),
) -> list[dict[str, Any]]:
"""Requests the last known production mix (in MW) of a given country."""
if zone_key != ZONE_KEY:
raise ParserException(
"PrinceEdwardIsland.py", "Cannot parse zone '{zone_key}'", zone_key
)
if target_datetime:
raise ParserException(
"PrinceEdwardIsland.py", "Unable to fetch historical data", ZONE_KEY
)
event = _get_event(session or Session())
production_mix = ProductionMix()
# These are oil-fueled ("heavy fuel oil" and "diesel") generators used as
# peakers and back-up.
production_mix.add_value("oil", event["fossil"])
production_mix.add_value("wind", event["wind"])
production_breakdowns = ProductionBreakdownList(logger)
production_breakdowns.append(
datetime=event["datetime"],
production=production_mix,
source=SOURCE,
zoneKey=ZONE_KEY,
)
return production_breakdowns.to_list()
def fetch_exchange(
zone_key1: ZoneKey,
zone_key2: ZoneKey,
session: Session | None = None,
target_datetime: datetime | None = None,
logger: Logger = getLogger(__name__),
) -> list[dict[str, Any]]:
"""Requests the last known power exchange (in MW) between two regions."""
sorted_zone_keys = ZoneKey("->".join(sorted((zone_key1, zone_key2))))
if target_datetime:
raise ParserException(
"PrinceEdwardIsland.py", "Unable to fetch historical data", sorted_zone_keys
)
if sorted_zone_keys != ZoneKey("CA-NB->CA-PE"):
raise ParserException(
"PrinceEdwardIsland.py",
f"The exchange pair '{sorted_zone_keys}' is not implemented",
sorted_zone_keys,
)
# PEI imports most of its electricity. Everything not generated on the
# island is imported from New Brunswick.
#
# In the case of wind, some is "paper-exported" even if there is a net
# import, and the "Wind Power Exported Off Island" gauge on the dashboard
# indicates their accounting of these exports. The dashboard says:
#
# Wind Power Exported Off-Island is that portion of wind generation that
# is supplying contracts elsewhere. The actual electricity from this
# portion of wind generation may stay within PEI but is satisfying a
# contractual arrangement in another jurisdiction.
#
# We are ignoring these paper exports, as they are an accounting/legal
# detail that doesn't actually reflect what happens on the wires. Given
# that NB is the only interconnection with PEI, physically exporting wind
# power to NB while simultaneously importing the balance seems unlikely.
event = _get_event(session or Session())
exchanges = ExchangeList(logger)
exchanges.append(
datetime=event["datetime"],
# The sorted zones are CA-NB -> CA-PE, so a positive value represents
# export from NB / import into PE.
netFlow=event["load"] - event["fossil"] - event["wind"],
source=SOURCE,
zoneKey=sorted_zone_keys,
)
return exchanges.to_list()
if __name__ == "__main__":
# Never used by the Electricity Map backend, but handy for testing.
print("fetch_production() ->")
print(fetch_production())
print('fetch_exchange("CA-PE", "CA-NB") ->')
print(fetch_exchange(ZoneKey("CA-PE"), ZoneKey("CA-NB")))