-
Notifications
You must be signed in to change notification settings - Fork 902
/
PE.py
117 lines (94 loc) · 3.14 KB
/
PE.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
#!/usr/bin/env python3
# coding=utf-8
from datetime import datetime
from logging import Logger, getLogger
from typing import Optional
import arrow
import dateutil
from requests import Session
from .lib.validation import validate
tz = "America/Lima"
MAP_GENERATION = {
"DIESEL": "oil",
"RESIDUAL": "biomass",
"CARBÓN": "coal",
"GAS": "gas",
"HÍDRICO": "hydro",
"BIOGÁS": "unknown",
"BAGAZO": "biomass",
"SOLAR": "solar",
"EÓLICA": "wind",
}
def parse_date(item):
return arrow.get(item["Nombre"], "YYYY/MM/DD hh:mm:ss").replace(
tzinfo=dateutil.tz.gettz(tz)
)
def fetch_production(
zone_key: str = "PE",
session: Optional[Session] = None,
target_datetime: Optional[datetime] = None,
logger: Logger = getLogger(__name__),
) -> list:
"""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")
r = session or Session()
url = "https://www.coes.org.pe/Portal/portalinformacion/generacion"
current_date = arrow.now(tz=tz)
today = current_date.format("DD/MM/YYYY")
yesterday = current_date.shift(days=-1).format("DD/MM/YYYY")
end_date = current_date.shift(days=+1).format("DD/MM/YYYY")
# To guarantee a full 24 hours of data we must make 2 requests.
response_today = r.post(
url, data={"fechaInicial": today, "fechaFinal": end_date, "indicador": 0}
)
response_yesterday = r.post(
url, data={"fechaInicial": yesterday, "fechaFinal": today, "indicador": 0}
)
data_today = response_today.json()["GraficoTipoCombustible"]["Series"]
data_yesterday = response_yesterday.json()["GraficoTipoCombustible"]["Series"]
raw_data = data_today + data_yesterday
# Note: We receive MWh values between two intervals!
interval_hours = (
parse_date(raw_data[0]["Data"][1]) - parse_date(raw_data[0]["Data"][0])
).total_seconds() / 3600
data = []
datetimes = []
for series in raw_data:
k = series["Name"]
if k not in MAP_GENERATION:
logger.warning(f'Unknown production type "{k}" for Peru')
continue
for v in series["Data"]:
dt = parse_date(v)
try:
i = datetimes.index(dt)
except ValueError:
i = len(datetimes)
datetimes.append(dt)
data.append(
{
"zoneKey": zone_key,
"datetime": dt.datetime,
"production": {},
"source": "coes.org.pe",
}
)
data[i]["production"][MAP_GENERATION[k]] = (
data[i]["production"].get(MAP_GENERATION[k], 0)
+ v["Valor"] / interval_hours
)
return list(
filter(
lambda x: validate(
x,
logger,
required=["gas"],
floor=0.0,
)
is not None,
data,
)
)
if __name__ == "__main__":
print(fetch_production())