Spaces:
Sleeping
Sleeping
| import requests | |
| import os | |
| import pandas as pd | |
| import datetime | |
| import numpy as np | |
| from sklearn.preprocessing import OrdinalEncoder | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| ## TODO: write function to display the color coding of the categoies both in the df and as a guide. | |
| #sg like: | |
| def color_aq(val): | |
| color = 'green' if val else 'red' | |
| return f'background-color: {color}' | |
| # but better | |
| def get_air_quality_data(station_name): | |
| AIR_QUALITY_API_KEY = os.getenv('AIR_QUALITY_API_KEY') | |
| request_value = f'https://api.waqi.info/feed/{station_name}/?token={AIR_QUALITY_API_KEY}' | |
| answer = requests.get(request_value).json()["data"] | |
| forecast = answer['forecast']['daily'] | |
| return [ | |
| answer["time"]["s"][:10], # Date | |
| int(forecast['pm25'][0]['avg']), # avg predicted pm25 | |
| int(forecast['pm10'][0]['avg']), # avg predicted pm10 | |
| max(int(forecast['pm25'][0]['avg']), int(forecast['pm10'][0]['avg'])) # avg predicted aqi | |
| ] | |
| def get_air_quality_df(data): | |
| col_names = [ | |
| 'date', | |
| 'pm25', | |
| 'pm10', | |
| 'aqi' | |
| ] | |
| new_data = pd.DataFrame( | |
| data | |
| ).T | |
| new_data.columns = col_names | |
| new_data['pm25'] = pd.to_numeric(new_data['pm25']) | |
| new_data['pm10'] = pd.to_numeric(new_data['pm10']) | |
| new_data['aqi'] = pd.to_numeric(new_data['aqi']) | |
| return new_data | |
| def get_weather_data_daily(city): | |
| WEATHER_API_KEY = os.getenv('WEATHER_API_KEY') | |
| answer = requests.get(f'https://weather.visualcrossing.com/VisualCrossingWebServices/rest/services/timeline/{city}/today?unitGroup=metric&include=days&key={WEATHER_API_KEY}&contentType=json').json() | |
| data = answer['days'][0] | |
| return [ | |
| answer['address'].lower(), | |
| data['datetime'], | |
| data['tempmax'], | |
| data['tempmin'], | |
| data['temp'], | |
| data['feelslikemax'], | |
| data['feelslikemin'], | |
| data['feelslike'], | |
| data['dew'], | |
| data['humidity'], | |
| data['precip'], | |
| data['precipprob'], | |
| data['precipcover'], | |
| data['snow'], | |
| data['snowdepth'], | |
| data['windgust'], | |
| data['windspeed'], | |
| data['winddir'], | |
| data['pressure'], | |
| data['cloudcover'], | |
| data['visibility'], | |
| data['solarradiation'], | |
| data['solarenergy'], | |
| data['uvindex'], | |
| data['conditions'] | |
| ] | |
| def get_weather_data_weekly(city: str, start_date: datetime) -> pd.DataFrame: | |
| WEATHER_API_KEY = os.getenv('WEATHER_API_KEY') | |
| end_date = f"{start_date + datetime.timedelta(days=6):%Y-%m-%d}" | |
| answer = requests.get(f'https://weather.visualcrossing.com/VisualCrossingWebServices/rest/services/timeline/{city}/{start_date}/{end_date}?unitGroup=metric&include=days&key={WEATHER_API_KEY}&contentType=json').json() | |
| weather_data = answer['days'] | |
| final_df = pd.DataFrame() | |
| for i in range(7): | |
| data = weather_data[i] | |
| list_of_data = [ | |
| answer['address'].lower(), | |
| data['datetime'], | |
| data['tempmax'], | |
| data['tempmin'], | |
| data['temp'], | |
| data['feelslikemax'], | |
| data['feelslikemin'], | |
| data['feelslike'], | |
| data['dew'], | |
| data['humidity'], | |
| data['precip'], | |
| data['precipprob'], | |
| data['precipcover'], | |
| data['snow'], | |
| data['snowdepth'], | |
| data['windgust'], | |
| data['windspeed'], | |
| data['winddir'], | |
| data['pressure'], | |
| data['cloudcover'], | |
| data['visibility'], | |
| data['solarradiation'], | |
| data['solarenergy'], | |
| data['uvindex'], | |
| data['conditions'] | |
| ] | |
| weather_df = get_weather_df(list_of_data) | |
| final_df = pd.concat([final_df, weather_df]) | |
| return final_df | |
| def get_weather_df(data): | |
| col_names = [ | |
| 'name', | |
| 'date', | |
| 'tempmax', | |
| 'tempmin', | |
| 'temp', | |
| 'feelslikemax', | |
| 'feelslikemin', | |
| 'feelslike', | |
| 'dew', | |
| 'humidity', | |
| 'precip', | |
| 'precipprob', | |
| 'precipcover', | |
| 'snow', | |
| 'snowdepth', | |
| 'windgust', | |
| 'windspeed', | |
| 'winddir', | |
| 'pressure', | |
| 'cloudcover', | |
| 'visibility', | |
| 'solarradiation', | |
| 'solarenergy', | |
| 'uvindex', | |
| 'conditions' | |
| ] | |
| new_data = pd.DataFrame( | |
| data | |
| ).T | |
| new_data.columns = col_names | |
| for col in col_names: | |
| if col not in ['name', 'date', 'conditions']: | |
| new_data[col] = pd.to_numeric(new_data[col]) | |
| return new_data | |
| def data_encoder(X): | |
| X.drop(columns=['date', 'name'], inplace=True) | |
| X['conditions'] = OrdinalEncoder().fit_transform(X[['conditions']]) | |
| return X | |
| def get_aplevel(temps:np.ndarray, table:list): | |
| boundary_list = np.array([0, 50, 100, 150, 200, 300]) # assert temps.shape == [x, 1] | |
| redf = np.logical_not(temps<=boundary_list) # temps.shape[0] x boundary_list.shape[0] ndarray | |
| hift = np.concatenate((np.roll(redf, -1)[:, :-1], np.full((temps.shape[0], 1), False)), axis = 1) | |
| cat = np.nonzero(np.not_equal(redf,hift)) | |
| level = [table[el] for el in cat[1]] | |
| return level | |
| def get_color(level:list): | |
| air_pollution_level = ['Good', 'Moderate', 'Unhealthy for sensitive Groups','Unhealthy' ,'Very Unhealthy', 'Hazardous'] | |
| color_list = ["Green", "Yellow", "DarkOrange", "Red", "Purple", "DarkRed"] | |
| ind = [air_pollution_level.index(lel) for lel in level] | |
| text = [f"color:{color_list[idex]};" for idex in ind] | |
| return text |