-
Notifications
You must be signed in to change notification settings - Fork 902
/
US_MISO.py
147 lines (111 loc) · 4.21 KB
/
US_MISO.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
#!/usr/bin/env python3
"""Parser for the MISO area of the United States."""
from datetime import datetime
from logging import Logger, getLogger
from typing import Optional
from dateutil import parser, tz
from requests import Session
mix_url = (
"https://api.misoenergy.org/MISORTWDDataBroker/DataBrokerServices.asmx?messageType"
"=getfuelmix&returnType=json"
)
mapping = {
"Coal": "coal",
"Natural Gas": "gas",
"Nuclear": "nuclear",
"Wind": "wind",
"Solar": "solar",
"Other": "unknown",
}
wind_forecast_url = "https://api.misoenergy.org/MISORTWDDataBroker/DataBrokerServices.asmx?messageType=getWindForecast&returnType=json"
# To quote the MISO data source;
# "The category listed as “Other” is the combination of Hydro, Pumped Storage Hydro, Diesel, Demand Response Resources,
# External Asynchronous Resources and a varied assortment of solid waste, garbage and wood pulp burners".
# Timestamp reported by data source is in format 23-Jan-2018 - Interval 11:45 EST
# Unsure exactly why EST is used, possibly due to operational connections with PJM.
def get_json_data(logger: Logger, session: Optional[Session] = None):
"""Returns 5 minute generation data in json format."""
s = session or Session()
json_data = s.get(mix_url).json()
return json_data
def data_processer(json_data, logger: Logger):
"""
Identifies any unknown fuel types and logs a warning.
Returns a tuple containing datetime object and production dictionary.
"""
generation = json_data["Fuel"]["Type"]
production = {}
for fuel in generation:
try:
k = mapping[fuel["CATEGORY"]]
except KeyError as e:
logger.warning(
"Key '{}' is missing from the MISO fuel mapping.".format(
fuel["CATEGORY"]
)
)
k = "unknown"
v = float(fuel["ACT"])
production[k] = production.get(k, 0.0) + v
# Remove unneeded parts of timestamp to allow datetime parsing.
timestamp = json_data["RefId"]
split_time = timestamp.split(" ")
time_junk = {1, 2} # set literal
useful_time_parts = [v for i, v in enumerate(split_time) if i not in time_junk]
if useful_time_parts[-1] != "EST":
raise ValueError("Timezone reported for US-MISO has changed.")
time_data = " ".join(useful_time_parts)
tzinfos = {"EST": tz.gettz("America/New_York")}
dt = parser.parse(time_data, tzinfos=tzinfos)
return dt, production
def fetch_production(
zone_key: str = "US-MISO",
session: Optional[Session] = None,
target_datetime: Optional[datetime] = None,
logger: Logger = getLogger(__name__),
) -> dict:
"""Requests the last known production mix (in MW) of a given country."""
if target_datetime:
raise NotImplementedError("This parser is not yet able to parse past dates")
json_data = get_json_data(logger, session=session)
processed_data = data_processer(json_data, logger)
data = {
"zoneKey": zone_key,
"datetime": processed_data[0],
"production": processed_data[1],
"storage": {},
"source": "misoenergy.org",
}
return data
def fetch_wind_forecast(
zone_key: str = "US-MISO",
session: Optional[Session] = None,
target_datetime: Optional[datetime] = None,
logger: Logger = getLogger(__name__),
) -> list:
"""Requests the day ahead wind forecast (in MW) of a given zone."""
if target_datetime:
raise NotImplementedError("This parser is not yet able to parse past dates")
s = session or Session()
req = s.get(wind_forecast_url)
raw_json = req.json()
raw_data = raw_json["Forecast"]
data = []
for item in raw_data:
dt = parser.parse(item["DateTimeEST"]).replace(
tzinfo=tz.gettz("America/New_York")
)
value = float(item["Value"])
datapoint = {
"datetime": dt,
"production": {"wind": value},
"source": "misoenergy.org",
"zoneKey": zone_key,
}
data.append(datapoint)
return data
if __name__ == "__main__":
print("fetch_production() ->")
print(fetch_production())
print("fetch_wind_forecast() ->")
print(fetch_wind_forecast())