Spaces:
Runtime error
Runtime error
| import requests | |
| import pandas as pd | |
| import datetime | |
| import time | |
| import gradio as gr | |
| import os | |
| ########### | |
| # other API's of interest: https://medium.com/@imdipto/best-free-alternatives-to-the-wunderground-weather-api-21acb22450e6 | |
| ########## | |
| OPENWEATHER_API_KEY = os.environ.get('OPENWEATHER_API_KEY') | |
| WEATHERAPI_KEY = os.environ.get('WEATHERAPI_KEY') | |
| def openweather_to_result(lat, lon, gmt_time): | |
| """ | |
| API docs: https://openweathermap.org/api/one-call-api#current | |
| Parameters | |
| ------------ | |
| lat [float]: decimal valued latitude | |
| lon [float]: decimal valued longitude | |
| gmt_time [datetime object]: time of desired forecast, in gmt and as python datetime object | |
| Returns | |
| -------- | |
| cloud_pct Tuple(List, List): list of cloud percent and corresponding time for times within 1.5 hours of input GMT time | |
| """ | |
| exclude_parts = 'current,minutely,daily,alerts' | |
| request_url = f'https://api.openweathermap.org/data/2.5/onecall?lat={lat}&lon={lon}&exclude={exclude_parts}&appid={OPENWEATHER_API_KEY}' | |
| response = requests.get(request_url) | |
| data = response.json() | |
| cloud_pct = [] | |
| forecast_times = [] | |
| # timeframe around input time to check cloud % for | |
| timeframe = datetime.timedelta(hours=1, minutes=30) | |
| for hour in data['hourly']: | |
| # dt property is unix utc time of forecasted data - convert this to python datetime object | |
| forecast_time = datetime.datetime.fromtimestamp( | |
| hour['dt'], tz=datetime.timezone.utc) | |
| if abs(forecast_time - gmt_time) <= timeframe: | |
| # cloud pct is stored in each hour at top level | |
| cloud_pct.append(hour['clouds']) | |
| forecast_times.append(forecast_time) | |
| return cloud_pct, forecast_times | |
| def weatherapi_to_result(lat, lon, gmt_time): | |
| """ | |
| API docs: https://www.weatherapi.com/docs/ | |
| TODO: implement wrapper instead https://github.com/weatherapicom/weatherapi-Python | |
| Parameters | |
| ------------ | |
| lat [float]: decimal valued latitude | |
| lon [float]: decimal values longitude | |
| gmt_time [datetime object]: time of desired forecast, in gmt and as python datetime object | |
| Returns | |
| -------- | |
| cloud_pct Tuple(List, List): list of cloud percent and corresponding time for times within 1.5 hours of input GMT time | |
| """ | |
| request_url = f'http://api.weatherapi.com/v1/forecast.json?key={WEATHERAPI_KEY}&q={lat},{lon}&days=2&alerts=no' | |
| response = requests.get(request_url) | |
| data = response.json() | |
| timezone = data['location']['tz_id'] | |
| cloud_pct = [] | |
| forecast_times = [] | |
| # quick error handling to make sure input time python object has "timezone" property attached | |
| try: | |
| gmt_time = gmt_time.astimezone(datetime.timezone.utc) | |
| except: | |
| gmt_time = gmt_time.tz_localize('utc') | |
| # timeframe around input time to check cloud % for | |
| timeframe = datetime.timedelta(hours=1, minutes=30) | |
| # this api is first divided into days, then hours | |
| for day in data['forecast']['forecastday']: | |
| for hour in day['hour']: | |
| # time_epoch contains unix epoch time in GMT/UTC | |
| #forecast_time = datetime.datetime.fromtimestamp(hour['time_epoch'], ZoneInfo(timezone)) | |
| forecast_time = datetime.datetime.fromtimestamp( | |
| hour['time_epoch'], datetime.timezone.utc) | |
| if abs(forecast_time - gmt_time) <= timeframe: | |
| cloud_pct.append(hour['cloud']) | |
| forecast_times.append( | |
| forecast_time.astimezone(datetime.timezone.utc)) | |
| return cloud_pct, forecast_times | |
| def met_to_result(lat, lon, gmt_time): | |
| """ | |
| API doc: https://api.met.no/weatherapi/locationforecast/2.0/documentation | |
| How to: https://api.met.no/doc/locationforecast/HowTO | |
| Parameters | |
| ------------ | |
| lat [float]: decimal valued latitude | |
| lon [float]: decimal values longitude | |
| gmt_time [datetime object]: time of desired forecast, in gmt and as python datetime object | |
| Returns | |
| -------- | |
| cloud_pct Tuple(List, List): list of cloud percent and corresponding time for times within 1.5 hours of input GMT time | |
| """ | |
| # set user agent https://stackoverflow.com/questions/10606133/sending-user-agent-using-requests-library-in-python | |
| # must be unique per API Terms of Service https://api.met.no/doc/TermsOfService | |
| headers = { | |
| 'User-Agent': 'NASAEarthScienceRemoteSensingUnit alex.h.stoken@nasa.gov'} | |
| request_url = f'https://api.met.no/weatherapi/locationforecast/2.0/compact?lat={lat}&lon={lon}' | |
| response = requests.get(request_url, headers=headers) | |
| data = response.json() | |
| cloud_pct = [] | |
| forecast_times = [] | |
| # timeframe around input time to check cloud % for | |
| timeframe = datetime.timedelta(hours=1, minutes=30) | |
| # walk through json return | |
| for hour in data['properties']['timeseries']: | |
| # time is utc formatted time https://api.met.no/doc/locationforecast/FAQ | |
| forecast_time = datetime.datetime.strptime( | |
| hour['time'], '%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=datetime.timezone.utc) | |
| # check if time of forecast is withing "timeframe" of desired time | |
| if abs(forecast_time - gmt_time) <= timeframe: | |
| # grab cloud pct from location within the nested json, add to list | |
| cloud_pct.append(hour['data']['instant'] | |
| ['details']['cloud_area_fraction']) | |
| # add time of forecast to list. Should be an "on the hour" time | |
| forecast_times.append(forecast_time) | |
| return cloud_pct, forecast_times | |
| ################ | |
| # generate text | |
| ################ | |
| def file_to_cloud_listing(input_file, services): | |
| """ | |
| Args: | |
| input_file (Union[str, gradio FileType]): input csv file with LAT, LON, SITE, GMT cols | |
| services (List): list of weather api servies to check | |
| Returns: | |
| str: formatted string with weather predictions for locations | |
| """ | |
| # this works if the input is from gradio. Then the file has an name property | |
| try: | |
| sites = pd.read_csv(input_file.name, parse_dates=['GMT']) | |
| using_gradio = True | |
| except: | |
| # this is for input from a script or command line | |
| sites = pd.read_csv(input_file, parse_dates=['GMT']) | |
| using_gradio = False | |
| start = time.perf_counter() | |
| date_format = "%H:%M" | |
| text = '' | |
| # each row is a site. Get weather data and then print it for each service for each site. | |
| for row_idx, row in sites.iterrows(): | |
| #time_of_interest = datetime.datetime.strptime(row.GMT, '%m/%d/%y %H:%M') | |
| text += check_row(row, services, date_format) | |
| text += f'{"="*60}\n' | |
| return text | |
| def check_row(row, services, date_format="%H:%M"): | |
| """Check a row of data (a pd.Series with LAT, LON, GMT, SITE cols) | |
| Args: | |
| row (pd.Series): pd.Series with LAT, LON, GMT, SITE cols) | |
| services (List): List of weather services (['OpenWeather', 'MET (Norwegian)', 'WeatherAPI'] or subset) | |
| date_format (str, optional): Format for printing time of site pass over. Defaults to "%H:%M". | |
| Returns: | |
| str: formatted str of text for weather vals | |
| """ | |
| text = "" | |
| text += f'{"Location":13}:\t\t{row.SITE} @ {row["GMT"].strftime(date_format)} GMT\n' | |
| if not isinstance(row.GMT, datetime.datetime): | |
| GMT = row["GMT"].to_pydatetime() | |
| else: | |
| GMT = row["GMT"] | |
| GMT = GMT.replace(tzinfo=datetime.timezone.utc) | |
| if 'OpenWeather' in services: | |
| try: | |
| cldp, times = openweather_to_result(row.LAT, row.LON, GMT) | |
| text += format_cldp_and_time("OpenWeather", cldp=cldp, times=times) | |
| except Exception as e: | |
| text += f'OpenWeather:\t\tError {e} in API processing\n' | |
| if 'MET (Norwegian)' in services: | |
| try: | |
| cldp, times = met_to_result(row.LAT, row.LON, GMT) | |
| text += format_cldp_and_time("Norwegian", cldp=cldp) | |
| except Exception as e: | |
| text += f'Norwegian:\t\tError {e} in API processing\n' | |
| if 'WeatherAPI' in services: | |
| try: | |
| cldp, times = weatherapi_to_result(row.LAT, row.LON, GMT) | |
| text += format_cldp_and_time("WeatherAPI", cldp=cldp) | |
| except Exception as e: | |
| text += f'WeatherAPI:\t\tError {e} in API processing\n' | |
| return text | |
| def format_cldp_and_time(api_name, cldp, times=None): | |
| """Formats output text for lists of cloud percents and forecast times | |
| Args: | |
| api_name ([type]): Name of weather source. | |
| cldp (List): List of floating point cloud percentage values. | |
| times (List, optional): List of forecast times, as datetime objects. Defaults to None. | |
| Returns: | |
| str: formatted text for printing | |
| """ | |
| text = '' | |
| date_format = "%H:%M" | |
| if times is not None: | |
| text += f'{"Forecast Time:":13}\t\t' + ' '.join(time.strftime(date_format) | |
| for time in times) + "\n" | |
| text += f'{api_name:13}:\t\t{" ".join(f"{p:<6.0f}" for p in cldp)}\n' | |
| return text | |
| inputs = [gr.inputs.File(label='Site File with Lat/Lon and GMT Time'), gr.inputs.CheckboxGroup(label='Weather Services', | |
| choices=['OpenWeather', 'MET (Norwegian)', 'WeatherAPI'], default=['OpenWeather', 'MET (Norwegian)'])] | |
| outputs = gr.outputs.Textbox(label ='Cloud % for hour before, hour of, hour after') | |
| css = """* {font-family: "Lucida Console", "Courier New", monospace !important;/* <-- fonts */ | |
| }""" | |
| gr.Interface(fn=file_to_cloud_listing, inputs=inputs, css=css, outputs=outputs, | |
| allow_screenshot=False).launch() | |