Passed
Pull Request — dev (#943)
by
unknown
02:17
created

unzip_file()   A

Complexity

Conditions 2

Size

Total Lines 3
Code Lines 3

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 3
dl 0
loc 3
rs 10
c 0
b 0
f 0
cc 2
nop 2
1
"""
2
Motorized Individual Travel (MIT) Charging Infrastructure
3
4
Main module for preparation of static model data for cahrging infrastructure for
5
motorized individual travel.
6
7
**Contents of this module**
8
* Creation of DB tables
9
* Download and preprocessing of vehicle registration data from zenodo
10
* Determination of all potential charging locations for the four charging use cases
11
  home, work, public and hpc per mv grid district
12
* Write results to DB
13
14
**Configuration**
15
16
The config of this dataset can be found in *datasets.yml* in section
17
*charging_infrastructure*.
18
19
**Charging Infrastructure**
20
21
The charging infrastructure allocation is based on [TracBEV[(
22
https://github.com/rl-institut/tracbev). TracBEV is a tool for the regional allocation
23
of charging infrastructure. In practice this allows users to use results generated via
24
[SimBEV](https://github.com/rl-institut/simbev) and place the corresponding charging
25
points on a map. These are split into the four use cases home, work, public and hpc.
26
"""
27
from __future__ import annotations
28
29
from pathlib import Path
30
import zipfile
31
32
from loguru import logger
33
import requests
34
35
from egon.data import config, db
36
from egon.data.datasets import Dataset
37
from egon.data.datasets.emobility.motorized_individual_travel_charging_infrastructure.db_classes import (  # noqa: E501
38
    EgonEmobChargingInfrastructure,
39
)
40
from egon.data.datasets.emobility.motorized_individual_travel_charging_infrastructure.infrastructure_allocation import (  # noqa: E501
41
    run_tracbev,
42
)
43
44
WORKING_DIR = Path(".", "charging_infrastructure").resolve()
45
DATASET_CFG = config.datasets()["charging_infrastructure"]
46
47
48
def create_tables() -> None:
49
    engine = db.engine()
50
    EgonEmobChargingInfrastructure.__table__.drop(bind=engine, checkfirst=True)
51
    EgonEmobChargingInfrastructure.__table__.create(
52
        bind=engine, checkfirst=True
53
    )
54
55
    logger.debug("Created tables.")
56
57
58
def download_zip(url: str, target: Path, chunk_size: int = 128) -> None:
59
    r = requests.get(url, stream=True)
60
61
    target.parent.mkdir(parents=True, exist_ok=True)
62
63
    with open(target, "wb") as fd:
64
        for chunk in r.iter_content(chunk_size=chunk_size):
65
            fd.write(chunk)
66
67
68
def unzip_file(source: Path, target: Path) -> None:
69
    with zipfile.ZipFile(source, "r") as zip_ref:
70
        zip_ref.extractall(target)
71
72
73
def get_tracbev_data() -> None:
74
    tracbev_cfg = DATASET_CFG["original_data"]["sources"]["tracbev"]
75
    file = WORKING_DIR / tracbev_cfg["file"]
76
77
    download_zip(url=tracbev_cfg["url"], target=file)
78
79
    unzip_file(source=file, target=WORKING_DIR)
80
81
82
class MITChargingInfrastructure(Dataset):
83
    def __init__(self, dependencies):
84
        super().__init__(
85
            name="MITChargingInfrastructure",
86
            version="0.0.1.dev",
87
            dependencies=dependencies,
88
            tasks=(
89
                {
90
                    create_tables,
91
                    get_tracbev_data,
92
                },
93
                run_tracbev,
94
            ),
95
        )
96