302 lines
15 KiB
Python
302 lines
15 KiB
Python
# app.py
|
||
#
|
||
# author: deng
|
||
# date: 20251127
|
||
|
||
from datetime import datetime
|
||
|
||
import streamlit as st
|
||
import yaml
|
||
from elevation import ElevationRenderer
|
||
from gpx import GPXProcessor
|
||
from map import MapRenderer
|
||
from streamlit_folium import st_folium
|
||
from weather import WeatherFetcher
|
||
|
||
|
||
class HikingAssistant:
|
||
def __init__(self):
|
||
self._config = self._load_config()
|
||
|
||
def _load_config(self):
|
||
with open('assets/config.yaml', 'r') as f:
|
||
return yaml.safe_load(f)
|
||
|
||
def run(self):
|
||
"""Hiking Assistant application"""
|
||
# Page configuration
|
||
st.set_page_config(
|
||
page_title=self._config['app']['page_title'],
|
||
page_icon=self._config['app']['page_favicon_path'],
|
||
layout='wide',
|
||
initial_sidebar_state='collapsed',
|
||
)
|
||
|
||
# [Block1]
|
||
# Title and description
|
||
st.title('⛰️ ' + self._config['app']['page_title'])
|
||
|
||
# File uploader
|
||
uploaded_file = st.file_uploader(
|
||
'上傳您的 GPX 檔案開始分析路線！',
|
||
type=['gpx'],
|
||
help='可以從[健行筆記](https://hiking.biji.co/index.php?q=trail&act=gpx_list)、'
|
||
'[Hikingbook](https://zh-tw.hikingbook.net/explore/trails?regions=Taiwan)等網站下載 GPX 檔案',
|
||
)
|
||
|
||
# [Block2]
|
||
# Analysis
|
||
if uploaded_file is not None:
|
||
try:
|
||
# Create cache key based on file name and size
|
||
file_key = f'{uploaded_file.name}_{uploaded_file.size}'
|
||
|
||
# Check if file is already processed
|
||
if 'gpx_file_key' not in st.session_state or st.session_state.gpx_file_key != file_key:
|
||
# Process GPX file (only when it's a new file)
|
||
with st.spinner('正在解析 GPX 檔案...'):
|
||
gpx_processor = GPXProcessor(uploaded_file)
|
||
|
||
if not gpx_processor.validate_and_parse():
|
||
st.error('❌ GPX 檔案格式錯誤或不包含有效的軌跡點')
|
||
return
|
||
|
||
# Calculate statistics
|
||
total_distance = gpx_processor.calculate_distance()
|
||
elevation_gain, elevation_loss = gpx_processor.calculate_elevation_gain_loss()
|
||
min_elevation, max_elevation = gpx_processor.get_min_max_elevation()
|
||
start_point, end_point = gpx_processor.get_start_end_points()
|
||
all_points = gpx_processor.get_all_points()
|
||
distances, elevations = gpx_processor.get_elevation_profile_data()
|
||
gradients = gpx_processor.get_gradients()
|
||
estimated_time = gpx_processor.calculate_naismith_time()
|
||
|
||
# Cache all processed data
|
||
st.session_state.gpx_file_key = file_key
|
||
st.session_state.total_distance = total_distance
|
||
st.session_state.elevation_gain = elevation_gain
|
||
st.session_state.elevation_loss = elevation_loss
|
||
st.session_state.min_elevation = min_elevation
|
||
st.session_state.max_elevation = max_elevation
|
||
st.session_state.start_point = start_point
|
||
st.session_state.end_point = end_point
|
||
st.session_state.all_points = all_points
|
||
st.session_state.distances = distances
|
||
st.session_state.elevations = elevations
|
||
st.session_state.gradients = gradients
|
||
st.session_state.estimated_time = estimated_time
|
||
|
||
# Show altitude sickness warning for first gpx loading
|
||
if max_elevation >= self._config['app']['altitude_sickness']['elevation_threshold']:
|
||
st.toast(self._config['app']['altitude_sickness']['warning_text'], icon=self._config['app']['altitude_sickness']['emoji'])
|
||
|
||
# Use cached data
|
||
total_distance = st.session_state.total_distance
|
||
elevation_gain = st.session_state.elevation_gain
|
||
elevation_loss = st.session_state.elevation_loss
|
||
min_elevation = st.session_state.min_elevation
|
||
max_elevation = st.session_state.max_elevation
|
||
start_point = st.session_state.start_point
|
||
end_point = st.session_state.end_point
|
||
all_points = st.session_state.all_points
|
||
distances = st.session_state.distances
|
||
elevations = st.session_state.elevations
|
||
gradients = st.session_state.gradients
|
||
estimated_time = st.session_state.estimated_time
|
||
|
||
# Display statistics section
|
||
st.header('📊 路線五四三')
|
||
col1, col2, col3, col4, col5, col6 = st.columns(6)
|
||
|
||
with col1:
|
||
st.metric(label='總距離', value=f'{total_distance:.1f}km')
|
||
|
||
with col2:
|
||
st.metric(label='總爬升', value=f'{elevation_gain:.0f}m')
|
||
|
||
with col3:
|
||
st.metric(label='總下降', value=f'{elevation_loss:.0f}m')
|
||
|
||
with col4:
|
||
st.metric(label='最高海拔', value=f'{max_elevation:.0f}m')
|
||
|
||
with col5:
|
||
st.metric(label='最低海拔', value=f'{min_elevation:.0f}m')
|
||
|
||
with col6:
|
||
st.metric(
|
||
label='預估行進時間',
|
||
value=f'{estimated_time // 60}h {estimated_time % 60}m',
|
||
help='依據[Naismith’s Rule](https://en.wikipedia.org/wiki/Naismith%27s_rule)計算',
|
||
)
|
||
|
||
# Map section
|
||
with st.spinner('正在渲染地圖...'):
|
||
map_renderer = MapRenderer()
|
||
route_map = map_renderer.create_route_map(all_points, start_point, end_point)
|
||
|
||
if route_map:
|
||
st_folium(route_map, width=None, height=500, key='route_map', returned_objects=[])
|
||
else:
|
||
st.error('無法渲染地圖')
|
||
|
||
# Create two columns: elevation profile on left, pie chart on right
|
||
profile_col, pie_col = st.columns([2, 1])
|
||
|
||
with profile_col:
|
||
with st.spinner('正在繪製海拔剖面圖...'):
|
||
profile_renderer = ElevationRenderer()
|
||
elevation_fig = profile_renderer.create_elevation_profile(distances, elevations, gradients)
|
||
|
||
if elevation_fig:
|
||
# elevation_fig = profile_renderer.add_gradient_legend(elevation_fig)
|
||
st.plotly_chart(elevation_fig, width='stretch')
|
||
else:
|
||
st.error('無法繪製海拔剖面圖')
|
||
|
||
with pie_col:
|
||
with st.spinner('正在分析坡度分布...'):
|
||
pie_fig = profile_renderer.create_gradient_pie_chart(distances, gradients)
|
||
|
||
if pie_fig:
|
||
st.plotly_chart(pie_fig, width='stretch')
|
||
else:
|
||
st.error('無法繪製坡度分布圖')
|
||
|
||
st.divider()
|
||
|
||
# Weather section
|
||
st.header('☀️ 天氣安抓')
|
||
|
||
if start_point:
|
||
weather_fetcher = WeatherFetcher(forecast_days=self._config['app']['weather']['forecast_days'])
|
||
weather_key = f'weather_{start_point[0]:.6f}_{start_point[1]:.6f}'
|
||
|
||
# Check if weather data is already cached
|
||
if 'weather_cache_key' not in st.session_state or st.session_state.weather_cache_key != weather_key:
|
||
# Fetch weather data (only when coordinates change)
|
||
with st.spinner('正在獲取天氣資訊...'):
|
||
weather_data = weather_fetcher.get_weather(start_point[0], start_point[1])
|
||
|
||
# Cache weather data
|
||
st.session_state.weather_cache_key = weather_key
|
||
st.session_state.weather_data = weather_data
|
||
else:
|
||
# Use cached weather data
|
||
weather_data = st.session_state.weather_data
|
||
|
||
if weather_data:
|
||
daily_times = weather_data.get('daily_time', [])
|
||
if daily_times:
|
||
# Date selector
|
||
date_labels = [datetime.fromisoformat(d).strftime('%Y/%m/%d') for d in daily_times]
|
||
selected_date_label = st.selectbox(
|
||
'出發日期',
|
||
options=date_labels,
|
||
index=0,
|
||
key='weather_date_selector',
|
||
)
|
||
selected_index = date_labels.index(selected_date_label)
|
||
|
||
st.subheader('起點天氣')
|
||
|
||
# Get weather data
|
||
# Current
|
||
current_humidity = weather_data.get('current_humidity')
|
||
current_apparent_temperature = weather_data.get('current_apparent_temperature')
|
||
current_precipitation_probability = weather_data.get('current_precipitation_probability')
|
||
current_precipitation = weather_data.get('current_precipitation')
|
||
current_wind_speed = weather_data.get('current_wind_speed')
|
||
current_wind_direction = weather_data.get('current_wind_direction')
|
||
# Daily
|
||
daily_temp_max = weather_data.get('daily_temp_max', [])
|
||
daily_temp_min = weather_data.get('daily_temp_min', [])
|
||
daily_apparent_max = weather_data.get('daily_apparent_temp_max', [])
|
||
daily_apparent_min = weather_data.get('daily_apparent_temp_min', [])
|
||
daily_mean_relative_humidity = weather_data.get('daily_mean_relative_humidity', [])
|
||
daily_precip_sum = weather_data.get('daily_precipitation_sum', [])
|
||
daily_precip_prob = weather_data.get('daily_precipitation_prob', [])
|
||
daily_wind_speed_max = weather_data.get('daily_wind_speed_max', [])
|
||
daily_wind_direction_dominant = weather_data.get('daily_wind_direction_dominant', [])
|
||
daily_sunrise = weather_data.get('daily_sunrise', [])
|
||
daily_sunset = weather_data.get('daily_sunset', [])
|
||
daily_uv_index_max = weather_data.get('daily_uv_index_max', [])
|
||
|
||
# Row 1: Temperature (3 columns)
|
||
col1, col2, col3 = st.columns(3)
|
||
with col1:
|
||
st.metric('🔥 最高溫度', f'{daily_temp_max[selected_index]:.1f}°C')
|
||
with col2:
|
||
st.metric('❄️ 最低溫度', f'{daily_temp_min[selected_index]:.1f}°C')
|
||
with col3:
|
||
if selected_index == 0:
|
||
apparent_temp = current_apparent_temperature
|
||
else:
|
||
apparent_temp = (daily_apparent_max[selected_index] + daily_apparent_min[selected_index]) / 2
|
||
st.metric('🌡️ 體感溫度', f'{apparent_temp:.1f}°C')
|
||
|
||
# Row 2: Humidity, Precipitation, Wind (3 columns)
|
||
col1, col2, col3 = st.columns(3)
|
||
with col1:
|
||
if selected_index == 0:
|
||
relative_humidity = current_humidity
|
||
else:
|
||
relative_humidity = daily_mean_relative_humidity[selected_index]
|
||
st.metric('💧 相對濕度', f'{relative_humidity}%')
|
||
|
||
with col2:
|
||
if selected_index == 0:
|
||
precipitation_probability = current_precipitation_probability
|
||
precipitation = current_precipitation
|
||
else:
|
||
precipitation_probability = daily_precip_prob[selected_index]
|
||
precipitation = daily_precip_sum[selected_index]
|
||
st.metric('🌧️ 降雨機率 / 雨量', f'{precipitation_probability}% / {precipitation:.1f}mm')
|
||
|
||
with col3:
|
||
wind_dir_text = weather_fetcher.convert_wind_degrees_to_flow_direction(
|
||
degrees=current_wind_direction if selected_index == 0 else daily_wind_direction_dominant[selected_index]
|
||
)
|
||
wind_level = weather_fetcher.get_wind_speed_indicator(
|
||
wind_speed=current_wind_speed if selected_index == 0 else daily_wind_speed_max[selected_index]
|
||
)
|
||
st.metric('💨 風速 / 流向', f'{daily_wind_speed_max[selected_index]:.1f} km/h {wind_dir_text} {wind_level}')
|
||
|
||
# Row 3: Sunrise, Sunset
|
||
col1, col2, col3 = st.columns(3)
|
||
with col1:
|
||
sunrise_time = datetime.fromisoformat(daily_sunrise[selected_index]).strftime('%H:%M')
|
||
st.metric('🌅 日出時間', sunrise_time)
|
||
|
||
with col2:
|
||
sunset_time = datetime.fromisoformat(daily_sunset[selected_index]).strftime('%H:%M')
|
||
st.metric('🌇 日落時間', sunset_time)
|
||
|
||
with col3:
|
||
uv_index = daily_uv_index_max[selected_index]
|
||
uv_level = weather_fetcher.get_uv_index_indicator(uv_index)
|
||
st.metric('☀️ UV指數', f'{uv_index} {uv_level}')
|
||
|
||
if selected_index >= 3:
|
||
st.info('❗ 三天後的天氣資訊誤差可能較大')
|
||
else:
|
||
st.warning('⚠️ 無法獲取足夠天數的天氣資料')
|
||
else:
|
||
st.warning('⚠️ 無法獲取天氣資訊，請稍後再試')
|
||
else:
|
||
st.warning('⚠️ 無法確定起點位置')
|
||
|
||
except Exception as e:
|
||
st.error(f'❌ 處理檔案時發生錯誤: {str(e)}')
|
||
|
||
# [Block3]
|
||
# Footer
|
||
st.divider()
|
||
footer_text = f'<div style="text-align:center; font-size: 0.8em"><p>{self._config["app"]["page_footer_text"]}</p></div>'
|
||
st.markdown(footer_text, unsafe_allow_html=True)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
app = HikingAssistant()
|
||
app.run()
|