Spaces:
Runtime error
Runtime error
| import streamlit as st | |
| import pandas as pd | |
| import geopandas as gpd | |
| import requests | |
| import json | |
| import time | |
| class DataModel: | |
| def __init__(self,): | |
| self.base_url = "https://infrasense-dashboard-aggregation-service-prod.delta.k8s-wdy.de" | |
| self.crs_dist = 32630 | |
| self.crs_ll = 4326 | |
| self.sleep = 5 | |
| def get_data(_self, osm_id, radius=25): | |
| """ | |
| Method for getting and extracting data from backend | |
| :param osm_id: id of the osm shape | |
| :param radius: radius around shape for data selection | |
| """ | |
| # get data from backend | |
| response = requests.get(f"{_self.base_url}/estimateShapeStatsAsync/{osm_id}/{radius}", headers={'X-API-KEY':st.secrets['api_key']}) | |
| loading_finished = False | |
| # data is ready for processing | |
| if response.status_code == 200: | |
| loading_finished = True | |
| # data are calculated, wait until finished | |
| elif response.status_code == 202: | |
| response, loading_finished = _self.__waiting_for_calculation(osm_id, radius) | |
| # get error meassage if data loading failed | |
| if not loading_finished: | |
| st.error('Error while data loading') | |
| st.stop() | |
| # extract data from backend | |
| return _self.extract_backend_data(response) | |
| def __waiting_for_calculation(self, osm_id, radius): | |
| """ | |
| Method for asyncron requests for Biqe monitor endpoint | |
| :param node_id: node id for calculation | |
| :param radius: radius around node for more possible data points | |
| :return data_json: json with response data | |
| :return status: status of finished data loading | |
| """ | |
| # proof if data is loaded | |
| time.sleep(self.sleep) | |
| response = self.__check_request(osm_id, radius) | |
| while response.status_code == 202: | |
| time.sleep(self.sleep) | |
| response = self.__check_request(osm_id, radius) | |
| # if finished | |
| if response.status_code == 200: | |
| status = True | |
| else: | |
| status = False | |
| return response, status | |
| def __check_request(self, osm_id, radius): | |
| """ | |
| Method for looking if async calculation of node data is finished | |
| :param node_id: node id for calculation | |
| :param extension_radius: radius around node for more possible data points | |
| :return req: response data from endpoint | |
| """ | |
| url = f'{self.base_url}/task_status/{osm_id}/{radius}/ShapeStats' | |
| return requests.get(url, headers={'X-API-KEY':st.secrets['api_key']}) | |
| def extract_backend_data(self, response): | |
| """ | |
| Method for extracting and procrocess the data from backend json | |
| :param response: request response from backend with data | |
| :return df: dataframe with extracted quality data | |
| :return geom: geometry of the shape | |
| """ | |
| # extraced data from response and parse to json | |
| json_data = response.json() | |
| json_data = json.loads(json_data['state']) | |
| # extract geometry element | |
| geom = json.loads(json_data['feature']) | |
| geom = gpd.GeoDataFrame.from_features(geom["features"]) | |
| # extract data and set feature | |
| df = pd.DataFrame(json_data['data']) | |
| df['date'] = pd.to_datetime(df['int_date'].astype(str), format='%Y%m%d') | |
| df['weekday'] = df['date'].dt.weekday | |
| df.sort_values(by='date', ignore_index=True, inplace=True) | |
| return df, geom |