Spaces:
Running
Running
import traceback | |
import os | |
import json | |
import pathlib | |
import gpxpy | |
import numpy as np | |
import pandas as pd | |
import gradio as gr | |
from datetime import datetime | |
import pytz | |
from sunrisesunset import SunriseSunset | |
from timezonefinder import TimezoneFinder | |
tf = TimezoneFinder() | |
from beaufort_scale.beaufort_scale import beaufort_scale_kmh | |
import srtm | |
elevation_data = srtm.get_data() | |
import openmeteo_requests | |
import requests_cache | |
from retry_requests import retry | |
from geopy import distance | |
from geopy.geocoders import Nominatim | |
geolocator = Nominatim(user_agent='FreeLetzWeather') | |
from apscheduler.schedulers.background import BackgroundScheduler | |
### Default variables ### | |
# Setup the Open-Meteo API client with cache and retry on error | |
cache_session = requests_cache.CachedSession('.cache', expire_after = 3600) | |
retry_session = retry(cache_session, retries = 5, backoff_factor = 0.2) | |
openmeteo = openmeteo_requests.Client(session = retry_session) | |
# Open Meteo weather forecast API | |
url = 'https://api.open-meteo.com/v1/forecast' | |
params = { | |
'timezone': 'auto', | |
'hourly': ['temperature_2m', 'rain', 'wind_speed_10m', 'weather_code', 'is_day'] | |
} | |
# Weather icons URL | |
icon_url = 'https://raw.githubusercontent.com/basmilius/weather-icons/refs/heads/dev/production/fill/svg/' | |
# Custom CSS | |
css = ''' | |
#button {background: DarkGoldenrod;} | |
.buttons {color: white;} | |
#table {height: 1080px;} | |
.tables {height: 1080px;} | |
.required-dropdown input:focus { | |
color: white; | |
background-color: DarkGoldenrod; | |
box-shadow: 0 0 0 12px DarkGoldenrod; | |
} | |
''' | |
# Default GPX if none is uploaded | |
gpx_file = os.path.join(os.getcwd(), 'default_gpx.gpx') | |
gpx_path = pathlib.Path(gpx_file) | |
### Functions ### | |
with open('weather_icons_custom.json', 'r') as file: | |
icons = json.load(file) | |
def add_ele(x): | |
return elevation_data.get_elevation(x['latitude'], x['longitude'], 0) | |
def map_icons(df): | |
code = df['weather_code'] | |
if df['is_day'] == 1: | |
icon = icons[str(code)]['day']['icon'] | |
description = icons[str(code)]['day']['description'] | |
elif df['is_day'] == 0: | |
icon = icons[str(code)]['night']['icon'] | |
description = icons[str(code)]['night']['description'] | |
df['Weather'] = '<img style="float: left; padding: 0; margin: -6px; display: block;" width=32px; src=' + icon_url + icon + '>' | |
df['Weather outline'] = description | |
return df | |
# Pluviometry to natural language | |
def rain_intensity(precipt): | |
if precipt >= 50: | |
rain = 'Extreme rain' | |
elif 50 < precipt <= 16: | |
rain = 'Very heavy rain' | |
elif 4 <= precipt < 16: | |
rain = 'Heavy rain' | |
elif 1 <= precipt < 4: | |
rain = 'Moderate rain' | |
elif 0.25 <= precipt < 1: | |
rain = 'Light rain' | |
elif 0 < precipt < 0.25: | |
rain = 'Light drizzle' | |
else: | |
rain = '' | |
return rain | |
def gen_dates_list(): | |
global day_print | |
global dates_filt | |
global dates_dict | |
global dates_list | |
global day_read | |
global today | |
today = datetime.today() | |
day_read = today.strftime('%A %-d %B') | |
day_print = '<h2>' + day_read + '</h2>' | |
dates_aval = pd.date_range(datetime.today(), periods=7).to_pydatetime().tolist() | |
dates_read = [x.strftime('%A %-d %B %Y') for x in dates_aval] | |
dates_filt = [x.strftime('%Y-%m-%d') for x in dates_aval] | |
dates_dict = dict(zip(dates_read, dates_filt)) | |
dates_list = list(dates_dict.keys()) | |
return dates_list | |
def sunrise_sunset(lat, lon, day): | |
tz = tf.timezone_at(lng=lon, lat=lat) | |
zone = pytz.timezone(tz) | |
dt = day.astimezone(zone) | |
rs = SunriseSunset(dt, lat=lat, lon=lon, zenith='official') | |
rise_time, set_time = rs.sun_rise_set | |
sunrise = rise_time.strftime('%H:%M') | |
sunset = set_time.strftime('%H:%M') | |
sunrise_icon = '<img style="float: left;" width="32px" src=' + icon_url + 'sunrise.svg>' | |
sunset_icon = '<img style="float: left;" width="32px" src=' + icon_url + 'sunset.svg>' | |
sunrise = '<h6>' + sunrise_icon + ' Sunrise ' + sunrise + '</h6>' | |
sunset = '<h6>' + sunset_icon + ' Sunset ' + sunset + '</h6>' | |
return sunrise, sunset | |
# Download the JSON and filter it per date | |
def json_parser(date): | |
global dfs | |
responses = openmeteo.weather_api(url, params=params) | |
# Process first location. Add a for-loop for multiple locations or weather models | |
response = responses[0] | |
# Process hourly data. The order of variables needs to be the same as requested. | |
hourly = response.Hourly() | |
hourly_temperature_2m = hourly.Variables(0).ValuesAsNumpy() | |
rain = hourly.Variables(1).ValuesAsNumpy() | |
hourly_wind_speed_10m = hourly.Variables(2).ValuesAsNumpy() | |
weather_code = hourly.Variables(3).ValuesAsNumpy() | |
is_day = hourly.Variables(4).ValuesAsNumpy() | |
hourly_data = {'date': pd.date_range( | |
start = pd.to_datetime(hourly.Time(), unit = 's', utc = True), | |
end = pd.to_datetime(hourly.TimeEnd(), unit = 's', utc = True), | |
freq = pd.Timedelta(seconds = hourly.Interval()), | |
inclusive = 'left' | |
)} | |
hourly_data['Temp (°C)'] = hourly_temperature_2m.round(0).astype(int) | |
hourly_data['weather_code'] = weather_code.astype(int) | |
hourly_data['is_day'] = is_day.astype(int) | |
v_rain_intensity = np.vectorize(rain_intensity) | |
hourly_data['Rain level'] = v_rain_intensity(rain) | |
v_beaufort_scale_kmh = np.vectorize(beaufort_scale_kmh) | |
hourly_data['Wind level'] = v_beaufort_scale_kmh(hourly_wind_speed_10m, language='en') | |
hourly_data['Rain (mm/h)'] = rain.round(1) | |
hourly_data['Wind (km/h)'] = hourly_wind_speed_10m.round(1) | |
hourly_dataframe = pd.DataFrame(data = hourly_data) | |
hourly_dataframe['Temp (°C)'] = hourly_dataframe['Temp (°C)'].astype(str) + '°' | |
hourly_dataframe['Wind (km/h)'] = hourly_dataframe['Wind (km/h)'].astype(str).replace('0.0', '') | |
hourly_dataframe['Rain (mm/h)'] = hourly_dataframe['Rain (mm/h)'].astype(str).replace('0.0', '') | |
hourly_dataframe['Time'] = hourly_dataframe['date'].dt.hour.astype(str).str.zfill(2) | |
hourly_dataframe = hourly_dataframe.apply(map_icons, axis=1) | |
dfs = hourly_dataframe[hourly_dataframe['date'].dt.strftime('%Y-%m-%d') == date] | |
dfs = dfs[['Time', 'Weather', 'Weather outline', 'Temp (°C)', 'Rain (mm/h)', 'Rain level', 'Wind (km/h)', 'Wind level']] | |
dfs = dfs.style.set_properties(**{'border': '0px'}) | |
return dfs | |
# Extract coordinates and location from GPX file | |
def coor_gpx(gpx): | |
def parse_gpx(gpx): | |
global gpx_name | |
global params | |
global lat | |
global lon | |
global altitude | |
global location | |
global dates_dict | |
global dates_list | |
global day_read | |
global dates | |
global sunrise | |
global sunset | |
with open(gpx) as f: | |
gpx_parsed = gpxpy.parse(f) | |
# Convert to a dataframe one point at a time. | |
points = [] | |
for track in gpx_parsed.tracks: | |
for segment in track.segments: | |
for p in segment.points: | |
points.append({ | |
'latitude': p.latitude, | |
'longitude': p.longitude, | |
'elevation': p.elevation, | |
}) | |
df_gpx = pd.DataFrame.from_records(points) | |
#gpx_dict = df_gpx.iloc[-1].to_dict() | |
df_gpx['srtm'] = df_gpx.apply(lambda x: add_ele(x), axis=1) | |
# Distance estimation function | |
def eukarney(lat1, lon1, alt1, lat2, lon2, alt2): | |
p1 = (lat1, lon1) | |
p2 = (lat2, lon2) | |
karney = distance.distance(p1, p2).m | |
return np.sqrt(karney**2 + (alt2 - alt1)**2) | |
# Create shifted columns in order to facilitate distance calculation | |
df_gpx['lat_shift'] = df_gpx['latitude'].shift(periods=-1).fillna(df_gpx['latitude']) | |
df_gpx['lon_shift'] = df_gpx['longitude'].shift(periods=-1).fillna(df_gpx['longitude']) | |
df_gpx['alt_shift'] = df_gpx['srtm'].shift(periods=-1).fillna(df_gpx['srtm']) | |
# Apply the distance function to the dataframe | |
df_gpx['distances'] = df_gpx.apply(lambda x: eukarney(x['latitude'], x['longitude'], x['srtm'], x['lat_shift'], x['lon_shift'], x['alt_shift']), axis=1).fillna(0) | |
df_gpx['distance'] = df_gpx['distances'].cumsum().round(decimals = 0).astype(int) | |
gpx_dict = df_gpx.iloc[(df_gpx.distance - df_gpx.distance.median()).abs().argsort()[:1]].to_dict('records')[0] | |
params['latitude'] = gpx_dict['latitude'] | |
params['longitude'] = gpx_dict['longitude'] | |
params['elevation'] = gpx_dict['elevation'] | |
lat = params['latitude'] | |
lon = params['longitude'] | |
if params['elevation'] == None: | |
params['elevation'] = int(round(gpx_dict['srtm'], 0)) | |
else: | |
params['elevation'] = int(round(params['elevation'], 0)) | |
altitude = params['elevation'] | |
location = geolocator.reverse('{}, {}'.format(lat, lon), zoom=14) | |
gpx_name = 'You have uploaded <b style="color: #004170;">' + os.path.basename(gpx.name) + '</b>' | |
location = '<p style="color: #004170">' + str(location) + '</p>' | |
dates_list = gen_dates_list() | |
day_read = dates_list[0] | |
date_filt = datetime.strptime(day_read, '%A %d %B %Y') | |
date_filt = date_filt.strftime('%Y-%m-%d') | |
day_print = '<h2>' + day_read + '</h2>' | |
sunrise, sunset = sunrise_sunset(lat, lon, datetime.strptime(day_read, '%A %d %B %Y')) | |
dates = gr.Dropdown(choices=dates_list, label='2. Next, pick up the date of your hike', value=dates_list[0], interactive=True, elem_classes='required-dropdown') | |
dfs = json_parser(date_filt) | |
try: | |
parse_gpx(gpx) | |
except Exception as error: | |
traceback.print_exc() | |
parse_gpx(gpx_path) | |
global gpx_name | |
gpx_name = '<b style="color: firebrick;">ERROR: Not a valid GPX file. Upload another file.</b>' | |
return gpx_name, location, dates, day_print, sunrise, sunset, dfs | |
# Choose a date from the dropdown menu | |
def date_chooser(day): | |
global day_read | |
global sunrise | |
global sunset | |
global sunrise_icon | |
global sunset_icon | |
global dates_dict | |
global dates_list | |
dates_list = gen_dates_list() | |
day_read = day | |
day_print = '<h2>' + day_read + '</h2>' | |
date = datetime.strptime(day, '%A %d %B %Y') | |
index = dates_list.index(day) | |
sunrise, sunset = sunrise_sunset(lat, lon, date) | |
date_filt = date.strftime('%Y-%m-%d') | |
dfs = json_parser(date_filt) | |
dates = gr.Dropdown(choices=dates_list, label='2. Next, pick up the date of your hike', value=dates_list[index], interactive=True, elem_classes='required-dropdown') | |
return day_print, sunrise, sunset, dfs, dates | |
# Call functions with default values | |
coor_gpx(gpx_path) | |
sunrise, sunset = sunrise_sunset(lat, lon, today) | |
dfs = json_parser(dates_filt[0]) | |
### Gradio app ### | |
with gr.Blocks(theme='ParityError/Interstellar', css=css, fill_height=True) as demo: | |
with gr.Column(): | |
with gr.Row(): | |
gr.HTML('<h1 style="color: DarkGoldenrod">Freedom Luxembourg<br><h3 style="color: #004170">The Weather for Hikers</h3></h1>') | |
with gr.Column(): | |
upload_gpx = gr.UploadButton(label='1. Upload your GPX track', file_count='single', size='lg', file_types=['.gpx', '.GPX'], elem_id='button', elem_classes='buttons', interactive=True) | |
file_name = gr.HTML('<h6>' + gpx_name + '</h6>') | |
dates = gr.Dropdown(choices=gen_dates_list(), label='2. Pick up the date of your hike', value=dates_list[0], interactive=True, elem_classes='required-dropdown') | |
gr.HTML('<h1><br></h1>') | |
with gr.Row(): | |
choosen_date = gr.HTML(day_print) | |
loc = gr.HTML('<p style="color: #004170">' + str(location) + '</p>') | |
sunrise = gr.HTML(sunrise) | |
sunset = gr.HTML(sunset) | |
table = gr.DataFrame(dfs, max_height=1000, type='pandas', headers=None, line_breaks=False, interactive=False, wrap=True, visible=True, render=True, | |
elem_id='table', elem_classes='tables', | |
datatype=['str', 'html', 'str', 'str', 'str', 'str', 'str', 'str'], | |
) | |
gr.HTML('<center>Freedom Luxembourg<br><a style="color: DarkGoldenrod; font-style: italic; text-decoration: none" href="https://www.freeletz.lu/freeletz/" target="_blank">freeletz.lu</a></center>') | |
gr.HTML('<center>Powered by <a style="color: #004170; text-decoration: none" href="https://open-meteo.com/" target="_blank">Open Meteo</a></center>') | |
upload_gpx.upload(fn=coor_gpx, inputs=upload_gpx, outputs=[file_name, loc, dates, choosen_date, sunrise, sunset, table]) | |
dates.input(fn=date_chooser, inputs=dates, outputs=[choosen_date, sunrise, sunset, table, dates]) | |
def restart_app(): | |
demo.close() | |
port = int(os.environ.get('PORT', 7860)) | |
demo.launch(server_name="0.0.0.0", server_port=port) | |
scheduler = BackgroundScheduler({'apscheduler.timezone': 'Europe/Luxembourg'}) | |
scheduler.add_job(func=restart_app, trigger='cron', hour='05', minute='55') | |
scheduler.start() | |
port = int(os.environ.get('PORT', 7860)) | |
demo.launch(server_name="0.0.0.0", server_port=port) | |