-
Notifications
You must be signed in to change notification settings - Fork 903
/
BG.py
67 lines (55 loc) · 2.01 KB
/
BG.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
from datetime import datetime, timezone
from logging import Logger, getLogger
from requests import Session
from electricitymap.contrib.lib.models.event_lists import ProductionBreakdownList
from electricitymap.contrib.lib.models.events import ProductionMix
from electricitymap.contrib.lib.types import ZoneKey
from parsers.lib.exceptions import ParserException
PARSER = "BG.py"
SOURCE = "eso.bg"
SOURCE_API_URL = "http://www.eso.bg/api/rabota_na_EEC_json.php?en"
PRODUCTION_TYPE_TO_PRODUCTION_MODE = {
"NPP": "nuclear",
"CHP": "coal",
"Heating TPPs": "gas",
"Factory TPPs": "gas",
"HPP": "hydro",
"Small HPPs": "hydro",
"Wind": "wind",
"PV": "solar",
"Bio": "biomass",
}
def fetch_production(
zone_key: ZoneKey = ZoneKey("BG"),
session: Session | None = None,
target_datetime: datetime | None = None,
logger: Logger = getLogger(__name__),
) -> list[dict]:
session = session or Session()
if target_datetime is not None:
raise ParserException(
PARSER, "This parser is not yet able to parse historical data", zone_key
)
time = datetime.now(timezone.utc)
response = session.get(SOURCE_API_URL)
if not response.ok:
raise ParserException(
PARSER,
f"Exception when fetching production error code: {response.status_code}: {response.text}",
zone_key,
)
response_payload = response.json()
logger.debug(f"Raw generation breakdown: {response_payload}")
production_mix = ProductionMix()
for header, value in response_payload:
production_type, _delimiter, _percentage = header.rpartition(" ")
production_mode = PRODUCTION_TYPE_TO_PRODUCTION_MODE[production_type]
production_mix.add_value(production_mode, value)
production_breakdown_list = ProductionBreakdownList(logger)
production_breakdown_list.append(
zoneKey=zone_key,
datetime=time,
source=SOURCE,
production=production_mix,
)
return production_breakdown_list.to_list()