Source code for dave_core.datapool.oep_request

# Copyright (c) 2022-2024 by Fraunhofer Institute for Energy Economics and Energy System Technology (IEE)
# Kassel and individual contributors (see AUTHORS file for details). All rights reserved.
# Use of this source code is governed by a BSD-style license that can be found in the LICENSE file.

from geopandas import GeoDataFrame
from pandas import DataFrame
from requests import get
from shapely.geometry import Point
from shapely.wkb import loads

from dave_core.settings import dave_settings

oep_url = "https://openenergy-platform.org"


[docs] def request_to_df(request): """ This function converts requested data into a DataFrame """ if request.status_code == 200: # 200 is the code of a successful request # if request is empty their will be an JSONDecodeError try: request_data = DataFrame(request.json()) except Exception: request_data = DataFrame() else: request_data = DataFrame() return request_data
[docs] def oep_request(table, schema=None, where=None, geometry=None, db_update=False): """ This function is to requesting data from the open energy platform.\ The available data is to find on https://openenergy-platform.org/dataedit/schemas INPUT: **table** (string) - table name of the searched data OPTIONAL: **schema** (string, default None) - schema name of the searched data. By default DAVE \ search for the schema in the settings file via table name example: 'postcode=34225' **where** (string, default None) - filter the table of the searched data example: 'postcode=34225' **geometry** (string, default None) - name of the geometry parameter in the OEP dataset to \ transform it from WKB to WKT **db_update** (boolean, default False) - If True in every case the data will be related \ from the oep OUTPUT: **requested_data** (DataFrame) - table of the requested data """ if schema is None: schema = dave_settings["oep_tables"][table][0] # request data directly from oep if where: request = get( "".join( [ oep_url, "/api/v0/schema/", schema, "/tables/", table, "/rows/?where=", where, ] ), timeout=30, ) elif dave_settings["oep_tables"][table][2] is not None: request = get( "".join( [ oep_url, "/api/v0/schema/", schema, "/tables/", table, "/rows/?where=", dave_settings["oep_tables"][table][2], ] ), timeout=30, ) else: request = get( "".join( [ oep_url, "/api/v0/schema/", schema, "/tables/", table, "/rows/", ] ), timeout=30, ) # convert data to dataframe request_data = request_to_df(request) # check for geometry parameter if geometry is None: geometry = dave_settings["oep_tables"][table][1] if geometry is not None: # --- convert into geopandas DataFrame with right crs # transform WKB to WKT / Geometry request_data["geometry"] = request_data[geometry].apply(lambda x: loads(x, hex=True)) # create geoDataFrame request_data = GeoDataFrame( request_data, crs=dave_settings["crs_main"], geometry=request_data.geometry, ) # fix some mistakes in the oep data if table == "ego_pf_hv_transformer": # change geometry to point because in original data the geometry was lines with length 0 request_data["geometry"] = request_data.geometry.apply( lambda x: Point(x.geoms[0].coords[:][0][0], x.geoms[0].coords[:][0][1]) ) if table == "ego_dp_mvlv_substation": # change wrong crs from oep request_data.crs = dave_settings["crs_meter"] request_data = request_data.to_crs(dave_settings["crs_main"]) # --- request meta informations for a dataset # !!! Todo: seperate option for getting data from DB. When there are no meta data in DB then check OEP Url request = get( "".join([oep_url, "/api/v0/schema/", schema, "/tables/", table, "/meta/"]), timeout=30, ) # convert data to meta dict # !!! When getting data from database the meta informations should also came from db if request.status_code == 200: # 200 is the code of a successful request request_meta = request.json() # create dict meta_data = { "Main": DataFrame( { "Titel": request_meta["title"], "Description": request_meta["description"], "Spatial": [request_meta["spatial"]], "Licenses": request_meta["licenses"], "metadata_version": request_meta["metaMetadata"]["metadataVersion"], }, index=[0], ), "Sources": DataFrame(request_meta["sources"]), "Data": DataFrame(request_meta["resources"][0]["schema"]["fields"]), } else: meta_data = {} return request_data, meta_data