|
| 1 | +# SPDX-FileCopyrightText: 2022 Contributors to the Power Grid Model project <dynamic.grid.calculation@alliander.com> |
| 2 | +# |
| 3 | +# SPDX-License-Identifier: MPL-2.0 |
| 4 | +""" |
| 5 | +Sim Bench Converter: Download, extract and convert a sim bench dataset to PGM |
| 6 | +""" |
| 7 | + |
| 8 | +from pathlib import Path |
| 9 | +from tempfile import gettempdir |
| 10 | +from typing import Optional |
| 11 | + |
| 12 | +from power_grid_model_io.converters.tabular_converter import TabularConverter |
| 13 | +from power_grid_model_io.data_stores.csv_dir_store import CsvDirStore |
| 14 | +from power_grid_model_io.data_types import TabularData |
| 15 | +from power_grid_model_io.utils.download import download_and_extract |
| 16 | + |
| 17 | +DEFAULT_MAPPING_FILE = Path(__file__).parent.parent / "config" / "csv" / "sim_bench.yaml" |
| 18 | +DEFAULT_DOWNLOAD_URL = "http://141.51.193.167/simbench/gui/usecase/download/?simbench_code={simbench_code:s}&format=csv" |
| 19 | + |
| 20 | + |
| 21 | +class SimBenchConverter(TabularConverter): |
| 22 | + """ |
| 23 | + Sim Bench Converter: Download, extract and convert a sim bench dataset to PGM |
| 24 | + """ |
| 25 | + |
| 26 | + __slots__ = ("_simbench_code", "_download_url", "_download_dir") |
| 27 | + |
| 28 | + def __init__(self, simbench_code: Optional[str] = None, download_dir: Optional[Path] = None): |
| 29 | + super().__init__(mapping_file=DEFAULT_MAPPING_FILE) |
| 30 | + self.simbench_code: Optional[str] = simbench_code |
| 31 | + self._download_url: Optional[str] = None |
| 32 | + self._download_dir: Path = download_dir or Path(gettempdir()) |
| 33 | + if simbench_code is not None: |
| 34 | + self.set_download_url(DEFAULT_DOWNLOAD_URL.format(simbench_code=simbench_code)) |
| 35 | + |
| 36 | + def set_download_url(self, url: str): |
| 37 | + """ |
| 38 | + Use a custom sim bench URL |
| 39 | + """ |
| 40 | + self._download_url = url |
| 41 | + |
| 42 | + def _load_data(self, data: Optional[TabularData]) -> TabularData: |
| 43 | + if data is None and self._source is None and self._download_url is not None: |
| 44 | + try: |
| 45 | + csv_dir = download_and_extract(self._download_url, dir_path=self._download_dir) |
| 46 | + except ValueError as ex: |
| 47 | + if str(ex).endswith(".download"): |
| 48 | + raise ValueError(f"Invalid SimBench dataset URL: {self._download_url}") from None |
| 49 | + raise ex |
| 50 | + self._source = CsvDirStore(csv_dir, delimiter=";") |
| 51 | + return super()._load_data(data=data) |
0 commit comments