-
Notifications
You must be signed in to change notification settings - Fork 903
/
create_aggregated_zone_config.py
90 lines (73 loc) · 2.99 KB
/
create_aggregated_zone_config.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
#!/usr/bin/env python3
"""
Create an aggregation zone config file by looking up sub zones based on the
zone name. Sums capacities, all contributors.
Example usage:
poetry run python scripts/create_aggregated_zone_config US US-Central
"""
import argparse
from copy import deepcopy
from pathlib import Path
import yaml
from electricitymap.contrib.config.model import Capacity, Zone
NAME_MAPPING = {"hydro storage": "hydro_storage", "battery storage": "battery_storage"}
def create_aggregated_config(zoneKey: str, timezone: str):
zone = Zone(
key=zoneKey,
sub_zone_names=[],
timezone=timezone,
contributors=[],
capacity=Capacity(),
)
zone.parsers = None
for path in Path("config/zones").glob(f"{zoneKey}-*.yaml"):
if path.is_file():
with open(path) as file:
subzone = yaml.safe_load(file.read())
zone.sub_zone_names.append(path.stem)
if "contributors" in subzone:
zone.contributors = list(
set(zone.contributors + subzone["contributors"])
)
if "capacity" in subzone:
capacities: dict = subzone["capacity"]
for key, capacity in capacities.items():
mapped_key = NAME_MAPPING.get(key, key)
if capacity is not None:
current_capacity = zone.capacity.__getattribute__(
mapped_key
)
if current_capacity is None:
current_capacity = 0
zone.capacity.__setattr__(
mapped_key, current_capacity + capacity
)
zoneDict = deepcopy(zone.__dict__)
for key, item in zone.__dict__.items():
if item is None:
del zoneDict[key]
del zoneDict["key"]
zoneDict["subZoneNames"] = zoneDict.pop("sub_zone_names")
zoneDict["subZoneNames"] = sorted(zoneDict["subZoneNames"])
zoneDict["capacity"] = deepcopy(zone.capacity.__dict__)
zoneDict["capacity"]["hydro storage"] = zoneDict["capacity"].pop("hydro_storage")
zoneDict["capacity"]["battery storage"] = zoneDict["capacity"].pop(
"battery_storage"
)
for key, value in zoneDict["capacity"].items():
if value is not None:
zoneDict["capacity"][key] = round(value, 1)
with open(f"config/zones/{zoneKey}.yaml", "w") as file:
yaml.safe_dump(zoneDict, file)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("zone", help="The parent zone name (e.g. US)")
parser.add_argument("timezone", help="The timezone of the parentzone")
args = parser.parse_args()
zone = args.zone
timezone = args.timezone
print(f"Creating {zone}")
create_aggregated_config(zone, timezone)
print(f"Created {zone}.yaml in config/zones.")
if __name__ == "__main__":
main()