-
Notifications
You must be signed in to change notification settings - Fork 902
/
BO.py
153 lines (120 loc) · 4.15 KB
/
BO.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
#!/usr/bin/env python3
import json
import re
from datetime import datetime
from logging import Logger, getLogger
from typing import List, NamedTuple, Optional
import arrow
from requests import Session
tz_bo = "America/La_Paz"
SOURCE = "cndc.bo"
class HourlyProduction(NamedTuple):
datetime: datetime
forecast: Optional[float]
total: Optional[float]
thermo: Optional[float]
hydro: Optional[float]
wind: Optional[float]
solar: Optional[float]
bagasse: Optional[float]
def extract_xsrf_token(html):
"""Extracts XSRF token from the source code of the generation graph page."""
return re.search(r'var ttoken = "([a-f0-9]+)";', html).group(1)
def fetch_data(
session: Optional[Session] = None,
target_datetime: Optional[datetime] = None,
logger: Logger = getLogger(__name__),
) -> List[HourlyProduction]:
if target_datetime is not None:
dt = arrow.get(target_datetime, tz_bo)
else:
dt = arrow.now(tz=tz_bo)
print("current dt", dt)
r = session or Session()
# Define actual and previous day (for midnight data).
formatted_dt = dt.format("YYYY-MM-DD")
# initial path for url to request
url_init = "https://www.cndc.bo/gene/dat/gene.php?fechag={0}"
# XSRF token for the initial request
xsrf_token = extract_xsrf_token(r.get("https://www.cndc.bo/gene/index.php").text)
resp = r.get(url_init.format(formatted_dt), headers={"x-csrf-token": xsrf_token})
hour_rows = json.loads(resp.text.replace("", ""))["data"]
result: List[HourlyProduction] = []
for hour_row in hour_rows:
[hour, forecast, total, thermo, hydro, wind, solar, bagasse] = hour_row
# isn't this Bolivia time and not UTC?
timestamp = dt.replace(
# "hour" is one-indexed
hour=hour - 1,
minute=0,
second=0,
microsecond=0,
)
result.append(
HourlyProduction(
datetime=timestamp.datetime,
forecast=forecast,
total=total,
thermo=thermo,
hydro=hydro,
wind=wind,
solar=solar,
bagasse=bagasse,
)
)
return result
def fetch_production(
zone_key: str = "BO",
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."""
payload = []
for row in fetch_data(
session=session, target_datetime=target_datetime, logger=logger
):
# NOTE: thermo includes gas + oil mixed, so we set these as unknown for now
# The modes here should match the ones we extract in the production payload
modes_extracted = [row.hydro, row.solar, row.wind, row.bagasse]
if row.total is None or None in modes_extracted:
continue
payload.append(
{
"zoneKey": zone_key,
"datetime": row.datetime,
"production": {
"biomass": row.bagasse,
"hydro": row.hydro,
"solar": row.solar,
"unknown": row.total - sum(modes_extracted),
"wind": row.wind,
},
"storage": {},
"source": SOURCE,
}
)
return payload
def fetch_generation_forecast(
zone_key: str = "BO",
session: Optional[Session] = None,
target_datetime: Optional[datetime] = None,
logger: Logger = getLogger(__name__),
) -> list:
return [
{
"zoneKey": zone_key,
"datetime": row.datetime,
"value": row.forecast,
"source": SOURCE,
}
for row in fetch_data(
session=session, target_datetime=target_datetime, logger=logger
)
]
if __name__ == "__main__":
"""Main method, never used by the Electricity Map backend, but handy for testing."""
print("fetch_production() ->")
print(fetch_production())
print("fetch_generation_forecast() ->")
print(fetch_generation_forecast())