-
Notifications
You must be signed in to change notification settings - Fork 903
/
update_data_sources.py
135 lines (109 loc) · 4.43 KB
/
update_data_sources.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
""" This script aims at automatically updating the data sources file with sources listed in the zones config. """
import logging
from collections import namedtuple
from copy import copy
from pathlib import Path
from typing import Dict, List, Set
import arrow
import pandas as pd
import yaml
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
CONFIG_DIR = Path(__file__).parent.parent.joinpath("config").resolve()
EMISSION_FACTORS_SOURCES_FILENAME = (
Path(__file__).parent.parent.joinpath("EMISSION_FACTORS_SOURCES.md").resolve()
)
MD_CONTENT_HEADER = """
# Emission factors sources
This file describes data sources used for generating the emission factors for all zones.
It only describes zone specific emission factors. Our default emission factors come from the [IPCC (2014) Fith Assessment Report](https://www.ipcc.ch/site/assets/uploads/2018/02/ipcc_wg3_ar5_annex-iii.pdf#page=7) report, and are fully described in our [wiki](https://github.com/electricitymaps/electricitymaps-contrib/wiki/Default-emission-factors).
## Zone specific emission factors
<details><summary>Click to see the full list of sources</summary>
"""
def _find_emission_factor_sources(
zone_config: dict,
) -> Dict[str, Dict[str, Dict[str, str]]]:
zone_sources = zone_config.get("sources", {})
def _get_sources_for_type(_type: str) -> Dict[str, Dict[str, str]]:
sources = {}
for mode, ef in zone_config.get("emissionFactors", {}).get(_type, {}).items():
sources_per_mode = {}
if isinstance(ef, list):
for _ef in ef:
for s in zone_sources:
if s in _ef.get("source"):
sources_per_mode[s] = zone_sources[s].get("link")
else:
for s in zone_sources:
if s in ef.get("source"):
sources_per_mode[s] = zone_sources[s].get("link")
if sources_per_mode != {}:
sources[mode] = sources_per_mode
return sources
sources = {
"direct": _get_sources_for_type("direct"),
"lifecycle": _get_sources_for_type("lifecycle"),
}
if sources["direct"] == {}:
del sources["direct"]
if sources["lifecycle"] == {}:
del sources["lifecycle"]
return sources
def read_zone_config(zone_key: str) -> dict:
with open(CONFIG_DIR.joinpath(f"zones/{zone_key}.yaml")) as f:
return yaml.safe_load(f)
def update_data_sources() -> None:
all_emission_factor_sources = {}
for zone_key in sorted(CONFIG_DIR.joinpath("zones").glob("*.yaml")):
zone_key = zone_key.stem
zone_config = read_zone_config(zone_key)
all_emission_factor_sources[zone_key] = _find_emission_factor_sources(
zone_config
)
# Filter out empty sources
all_emission_factor_sources = {
k: v
for k, v in all_emission_factor_sources.items()
if v.get("direct", {}) != {} or v.get("lifecycle", {}) != {}
}
md_content = copy(MD_CONTENT_HEADER)
for zone_key, sources in all_emission_factor_sources.items():
zone_content = f"""
* {zone_key}
"""
if "direct" in sources:
zone_content += """
* Direct emission factors
"""
for mode, mode_sources in sources["direct"].items():
zone_content += f"""
* {mode}
"""
for source, link in mode_sources.items():
# We must be careful to not add ";" in the sources
for i, _s in enumerate(source.split("; ")):
zone_content += f"""
* [{_s}]({link.split(', ')[i]})
"""
if "lifecycle" in sources:
zone_content += """
* Lifecycle emission factors
"""
for mode, mode_sources in sources["lifecycle"].items():
zone_content += f"""
* {mode}
"""
for source, link in mode_sources.items():
# We must be careful to not add ";" in the sources
for i, _s in enumerate(source.split("; ")):
zone_content += f"""
* [{_s}]({link.split(', ')[i]})
"""
md_content += zone_content
md_content += """
</details>
"""
with open(EMISSION_FACTORS_SOURCES_FILENAME, "w") as f:
f.write(md_content)
if __name__ == "__main__":
update_data_sources()