-
Notifications
You must be signed in to change notification settings - Fork 902
/
GT.py
131 lines (116 loc) · 4.52 KB
/
GT.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
#!/usr/bin/env python3
"""Parse Guatemalan electricity data from the Administrador del Mercado
Mayorista (AMM) API.
"""
# Standard library imports
import collections
from datetime import datetime
from logging import Logger, getLogger
from typing import Optional
# Third-party library imports
import arrow
from requests import Session
# Local library imports
from parsers.lib import validation
DEFAULT_ZONE_KEY = "GT"
PRODUCTION_THRESHOLD = 10 # MW
TIMEZONE = "America/Guatemala"
DOMAIN = "wl12.amm.org.gt"
URL = f"https://{DOMAIN}/GraficaPW/graficaCombustible"
def fetch_consumption(
zone_key: str = DEFAULT_ZONE_KEY,
session: Optional[Session] = None,
target_datetime: Optional[datetime] = None,
logger: Logger = getLogger(__name__),
):
"""Fetch a list of hourly consumption data, in MW, for the day of the
requested date-time.
"""
date_time = arrow.get(target_datetime).to(TIMEZONE).floor("hour")
results = [
{
"consumption": row["DEM SNI"],
"datetime": date_time.replace(hour=hour).datetime,
"source": DOMAIN,
"zoneKey": zone_key,
}
for hour, row in enumerate(
index_api_data_by_hour(get_api_data(session, URL, date_time))
)
]
# An hour's data isn't updated until the hour has passed, so the current
# (and future) hour(s) is/are not included in the results.
return results if target_datetime else results[: date_time.hour]
def fetch_production(
zone_key: str = DEFAULT_ZONE_KEY,
session: Optional[Session] = None,
target_datetime: Optional[datetime] = None,
logger: Logger = getLogger(__name__),
):
"""Fetch a list of hourly production data, in MW, for the day of the
requested date-time.
"""
date_time = arrow.get(target_datetime).to(TIMEZONE).floor("hour")
results = [
{
"datetime": date_time.replace(hour=hour).datetime,
"production": {
"biomass": row["BIOGAS"] + row["BIOMASA"],
"coal": row["CARBÓN"],
"gas": row["GAS NATURAL"],
"geothermal": row["VAPOR"],
"hydro": row["AGUA"],
"oil": row["BUNKER"] + row["DIESEL"],
"solar": row["IRRADIACIÓN"],
"unknown": row["BIOMASA/CARBÓN"]
+ row["CARBÓN/PETCOKE"]
+ row["SYNGAN"],
"wind": row["VIENTO"],
},
"source": DOMAIN,
"zoneKey": zone_key,
}
for hour, row in enumerate(
index_api_data_by_hour(get_api_data(session, URL, date_time))
)
]
# If the current day is selected, the API will return zero-filled future
# data until the end of the day. Truncate the list to avoid returning any
# future data.
if not target_datetime:
results = results[: date_time.hour + 1]
return [
validation.validate(
result, logger, floor=PRODUCTION_THRESHOLD, remove_negative=True
)
for result in results
]
def index_api_data_by_hour(json):
"""The JSON data returned by the API is a list of objects, each
representing one technology type. Collect this information into a list,
with the list index representing the hour of day.
"""
results = [collections.defaultdict(float) for _ in range(24)]
for row in json:
# The API returns hours in the range [1, 24], so each one refers to the
# past hour (e.g., 1 is the time period [00:00, 01:00)). Shift the hour
# so each index represents the hour ahead and is in the range [0, 24),
# e.g., hour 0 represents the period [00:00, 01:00).
results[int(row["hora"]) - 1][row["tipo"]] = row["potencia"]
return results
def get_api_data(session: Session, url, date_time):
"""Get the JSON-formatted response from the AMM API for the desired
date-time.
"""
session = session or Session()
return session.get(url, params={"dt": date_time.format("DD/MM/YYYY")}).json()
if __name__ == "__main__":
# Never used by the electricityMap back-end, but handy for testing.
print("fetch_production():")
print(fetch_production(), end="\n\n")
print("fetch_production(target_datetime='2022-01-01T:12:00:00Z'):")
print(fetch_production(target_datetime="2022-01-01T12:00:00Z"), end="\n\n")
print("fetch_consumption():")
print(fetch_consumption(), end="\n\n")
print("fetch_consumption(target_datetime='2022-01-01T:12:00:00Z'):")
print(fetch_consumption(target_datetime="2022-01-01T12:00:00Z"), end="\n\n")