app.py

import json
import pytz
import numpy as np
import pandas as pd
from geopy import distance
import plotly.graph_objects as go
from gpx_converter import Converter
from sunrisesunset import SunriseSunset
from datetime import datetime, timedelta
from beaufort_scale.beaufort_scale import beaufort_scale_kmh
from timezonefinder import TimezoneFinder
tf = TimezoneFinder()

from dash import Dash, dcc, html, dash_table, Input, Output, no_update, callback
import dash_bootstrap_components as dbc
from dash_extensions import Purify

import srtm
elevation_data = srtm.get_data()

import requests_cache
import openmeteo_requests
from retry_requests import retry

### VARIABLES ###

# Variables to become widgets
igpx = 'default_gpx.gpx'
date = '2024-12-29'
time = '10:55'
speed = 4.0
granularity = 2000

# 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',
	'minutely_15': ['temperature_2m', 'rain', 'wind_speed_10m', 'weather_code', 'is_day'],
	'hourly': ['rain'],
}

# Load the JSON files mapping weather codes to descriptions and icons
with open('weather_icons_custom.json', 'r') as file:
	icons = json.load(file)

# Weather icons URL
icon_url = 'https://raw.githubusercontent.com/basmilius/weather-icons/refs/heads/dev/production/fill/svg/'
sunrise_icon = icon_url + 'sunrise.svg'
sunset_icon = icon_url + 'sunset.svg'

### FUNCTIONS ###

# Sunrise sunset
def sunrise_sunset(lat_start, lon_start, lat_end, lon_end, date):

    tz = tf.timezone_at(lng=lon_start, lat=lat_start)
    zone = pytz.timezone(tz)

    day = datetime.strptime(date, '%Y-%m-%d')

    dt = day.astimezone(zone)

    rs_start = SunriseSunset(dt, lat=lat_start, lon=lon_start, zenith='official')
    rise_time = rs_start.sun_rise_set[0]

    rs_end = SunriseSunset(dt, lat=lat_end, lon=lon_end, zenith='official')
    set_time = rs_end.sun_rise_set[1]

    sunrise = rise_time.strftime('%H:%M')
    sunset = set_time.strftime('%H:%M')

    return sunrise, sunset

# Map weather codes to descriptions and icons
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'] = icon_url + icon
	df['Weather outline'] = description

	return df

# Quantitative 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 = 'No rain / No info'
    return rain

# Function to add elevation
def add_ele(row):
    if pd.isnull(row['altitude']):
        row['altitude'] = elevation_data.get_elevation(row['latitude'], row['longitude'], 0)
    else:
        row['altitude'] = row['altitude']
    return row

# Compute distances using the Karney algorith with Euclidian altitude correction
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)

# Obtain the weather forecast for each waypoint at each specific time
def get_weather(df_wp):

    params['latitude'] = df_wp['latitude']
    params['longitude'] = df_wp['longitude']
    params['elevation'] = df_wp['longitude']

    start_dt = datetime.strptime(date + 'T' + time, '%Y-%m-%dT%H:%M')

    delta_dt = start_dt + timedelta(seconds=df_wp['seconds'])
    delta_read = delta_dt.strftime('%Y-%m-%dT%H:%M')

    start_period = (delta_dt - timedelta(seconds=1800)).strftime('%Y-%m-%dT%H:%M')
    end_period = (delta_dt + timedelta(seconds=1800)).strftime('%Y-%m-%dT%H:%M')

    time_read = delta_dt.strftime('%H:%M')

    df_wp['Time'] = time_read

    params['start_minutely_15'] = delta_read
    params['end_minutely_15'] = delta_read
    params['start_hour'] = delta_read
    params['end_hour'] = delta_read

    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.
    minutely = response.Minutely15()
    hourly = response.Hourly()

    minutely_temperature_2m = minutely.Variables(0).ValuesAsNumpy()[0]
    rain = hourly.Variables(0).ValuesAsNumpy()[0]
    minutely_wind_speed_10m = minutely.Variables(2).ValuesAsNumpy()[0]
    weather_code = minutely.Variables(3).ValuesAsNumpy()[0]
    is_day = minutely.Variables(4).ValuesAsNumpy()[0]

    df_wp['Temp (°C)'] = minutely_temperature_2m
    df_wp['weather_code'] = weather_code
    df_wp['is_day'] = is_day

    v_rain_intensity = np.vectorize(rain_intensity)
    df_wp['Rain level'] = v_rain_intensity(rain)

    v_beaufort_scale_kmh = np.vectorize(beaufort_scale_kmh)

    df_wp['Wind level'] = v_beaufort_scale_kmh(minutely_wind_speed_10m, language='en')

    df_wp['Rain (mm/h)'] = rain.round(1)
    df_wp['Wind (km/h)'] = minutely_wind_speed_10m.round(1)

    return df_wp

# Parse the GPX track
def parse_gpx(igpx):

    global centre_lat
    global centre_lon
    global sunrise
    global sunset

    df_gpx = Converter(input_file = igpx).gpx_to_dataframe()

    # Sunrise sunset

    lat_start, lon_start = df_gpx[['latitude', 'longitude']].head(1).values.flatten().tolist()
    lat_end, lon_end = df_gpx[['latitude', 'longitude']].tail(1).values.flatten().tolist()

    sunrise, sunset = sunrise_sunset(lat_start, lon_start, lat_end, lon_end, date)

    df_gpx = df_gpx.apply(lambda x: add_ele(x), axis=1)

    centre_lat = (df_gpx['latitude'].max() + df_gpx['latitude'].min()) / 2
    centre_lon = (df_gpx['longitude'].max() + df_gpx['longitude'].min()) / 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['altitude'].shift(periods=-1).fillna(df_gpx['altitude'])

    # Apply the distance function to the dataframe

    df_gpx['distances'] = df_gpx.apply(lambda x: eukarney(x['latitude'], x['longitude'], x['altitude'], 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)

    df_gpx = df_gpx.drop(columns=['lat_shift', 'lon_shift', 'alt_shift', 'distances'])

    start = df_gpx['distance'].min()
    finish = df_gpx['distance'].max()

    dist_rang = list(range(start, finish, granularity))
    dist_rang.append(finish)

    way_list = []
    for waypoint in dist_rang:
        gpx_dict = df_gpx.iloc[(df_gpx.distance - waypoint).abs().argsort()[:1]].to_dict('records')[0]
        way_list.append(gpx_dict)

    df_wp = pd.DataFrame(way_list)

    df_wp['seconds'] = df_wp['distance'].apply(lambda x: int(round(x / (speed * (5/18)), 0)))

    df_wp = df_wp.apply(lambda x: get_weather(x), axis=1)

    df_wp['Temp (°C)'] = df_wp['Temp (°C)'].round(0).astype(int).astype(str) + '°C'
    df_wp['is_day'] = df_wp['is_day'].astype(int)
    df_wp['weather_code'] = df_wp['weather_code'].astype(int)
    df_wp = df_wp.apply(map_icons, axis=1)

    df_wp['Rain level'] = df_wp['Rain level'].astype(str)
    df_wp['Wind level'] = df_wp['Wind level'].astype(str)

    df_wp['dist_read'] = ('<p style="font-family:sans; font-size:14px;"><b>' +
                        df_wp['Weather outline'] + '</b><br><br>' +
                        df_wp['Temp (°C)'] + '<br><br>' +
                        df_wp['Rain level'] + '<br>' +
                        df_wp['Wind level'] + '<br><br>' +
                        df_wp['Time'] + '<br><br>' +
                        df_wp['distance'].apply(lambda x: str(int(round(x / 1000, 0)))).astype(str) + ' km | ' + df_wp['altitude'].round(0).astype(int).astype(str) + ' m</p>')

    df_wp = df_wp.reset_index(drop=True)

    df_wp['Waypoint'] = df_wp.index

    dfs = df_wp[['Waypoint', 'Time', 'Weather', 'Weather outline', 'Temp (°C)', 'Rain (mm/h)', 'Rain level', 'Wind (km/h)', 'Wind level']].copy()

    dfs['Wind (km/h)'] = dfs['Wind (km/h)'].round(1).astype(str).replace('0.0', '')
    dfs['Rain (mm/h)'] = dfs['Rain (mm/h)'].round(1).astype(str).replace('0.0', '')
    dfs['Temp (°C)'] = dfs['Temp (°C)'].str.replace('C', '')

    dfs['Weather'] = '<img style="float: right; padding: 0; margin: -6px; display: block;" width=48px; src=' + dfs['Weather'] + '>'

    return [df_gpx, df_wp, dfs]

df_gpx, df_wp, dfs = parse_gpx(igpx)

### PLOTS ###

# Plot map

fig = go.Figure()

fig.add_trace(go.Scattermap(lon=df_gpx['longitude'],
                 lat=df_gpx['latitude'],
                 mode='lines', line=dict(width=4, color='firebrick'),
                 name='gpx_trace'))

fig.add_trace(go.Scattermap(lon=df_wp['longitude'],
                 lat=df_wp['latitude'],
                 mode='markers+text', marker=dict(size=24, color='firebrick', opacity=0.8, symbol='circle'),
                 textfont=dict(color='white', weight='bold'),
                 text=df_wp.index.astype(str),
                 name='wp_trace'))

fig.update_layout(map_style='open-street-map',
    map=dict(center=dict(lat=centre_lat, lon=centre_lon), zoom=12))

fig.update_traces(showlegend=False, hoverinfo='none', hovertemplate=None, selector=({'name': 'wp_trace'}))
fig.update_traces(showlegend=False, hoverinfo='skip', hovertemplate=None, selector=({'name': 'gpx_trace'}))


### DASH APP ###

external_stylesheets = [dbc.themes.BOOTSTRAP]

app = Dash(__name__, external_stylesheets=external_stylesheets)

server = app.server

# Callbacks

@callback(Output('graph-tooltip', 'show'),
    Output('graph-tooltip', 'bbox'),
    Output('graph-tooltip', 'children'),
    Input('graph-basic-2', 'hoverData'))
def display_hover(hoverData):

    if hoverData is None:
        return False, no_update, no_update

    pt = hoverData['points'][0]
    bbox = pt['bbox']
    num = pt['pointNumber']

    df_row = df_wp.iloc[num]
    img_src = df_row['Weather']
    txt_src = df_row['dist_read']

    children = [html.Div([html.Img(src=img_src, style={'width': '100%'}), Purify(txt_src),],
                         style={'width': '128px', 'white-space': 'normal'})]

    return True, bbox, children

# Layout

def serve_layout():

    layout = html.Div([
        html.Div([dcc.Link('The Weather for Hikers', href='.',
                style={'color': 'darkslategray', 'font-size': 20, 'font-family': 'sans', 'font-weight': 'bold', 'text-decoration': 'none'}),
        ]),
        html.Div([dcc.Link('Freedom Luxembourg', href='https://www.freeletz.lu/freeletz/',
                target='_blank', style={'color': 'goldenrod', 'font-size': 15, 'font-family': 'sans', 'text-decoration': 'none'}),
        ]),
        html.Div([html.Br(),
        dbc.Row([dbc.Col(html.Div('Sunrise '), width={'size': 'auto', 'offset': 4}),
                dbc.Col(html.Img(src=sunrise_icon, style={'height':'42px'}), width={'size': 'auto'}),
                dbc.Col(html.Div(sunrise), width={'size': 'auto'}),
                dbc.Col(html.Div('Sunset '), width={'size': 'auto', 'offset': 1}),
                dbc.Col(html.Img(src=sunset_icon, style={'height':'42px'}), width={'size': 'auto'}),
                dbc.Col(html.Div(sunset), width={'size': 'auto'})]),
        ], style={'font-size': 13, 'font-family': 'sans'}),
        html.Div([dash_table.DataTable(
            id='datatable-interactivity', markdown_options = {'html': True},
            columns=[{'name': i, 'id': i, 'deletable': False, 'selectable': False, 'presentation': 'markdown'} for i in dfs.columns],
            data=dfs.to_dict('records'),
            editable=False,
            row_deletable=False,
            style_as_list_view=True,
            style_cell={'fontSize': '12px', 'text-align': 'center', 'margin-bottom':'0'},
            css=[dict(selector= 'p', rule= 'margin: 0; text-align: center')],
            style_header={'backgroundColor': 'goldenrod', 'color': 'white', 'fontWeight': 'bold'}),
        dcc.Graph(id='graph-basic-2', figure=fig, clear_on_unhover=True, style={'height': '90vh'}),
        dcc.Tooltip(id='graph-tooltip'),
        ]),
        html.Div([dcc.Link('Freedom Luxembourg', href='https://www.freeletz.lu/freeletz/',
                target='_blank', style={'color': 'goldenrod', 'font-size': 15, 'font-family': 'sans', 'text-decoration': 'none'}),
        ], style={'text-align': 'center'},),
        html.Div([dcc.Link('Powered by Open Meteo', href='https://open-meteo.com/',
                target='_blank', style={'color': 'black', 'font-size': 13, 'font-family': 'sans', 'text-decoration': 'none'}),
        ], style={'text-align': 'center'}),
        dcc.Interval(
                    id='interval-component',
                    interval=6 * 60 * 60 * 1000,
                    n_intervals=0),
    ], id='layout-content')

    return layout

app.layout = serve_layout

@callback(Output('layout-content', 'children'),
             [Input('interval-component', 'n_intervals')])
def refresh_layout(n):
    layout = serve_layout()
    return layout

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=7860)