deng/hiking_assistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

1) add uv to weather, 2) add caching to accelerate loading speed

Browse Source
This commit is contained in:
deng
2025-11-27 22:20:54 +08:00
parent 5e8e51ee82
commit 1c0e72f0e1
3 changed files with 206 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

219
hiking_assistant/app.py
View File

@ -3,6 +3,8 @@
# author: deng
# date: 20251127
from datetime import datetime
import streamlit as st
import yaml
from elevation_profile import ElevationProfileRenderer
@ -30,6 +32,7 @@ class HikingAssistant:
initial_sidebar_state='collapsed',
)
# [Block1]
# Title and description
st.title('⛰️ ' + self._config['app']['page_title'])
@ -41,25 +44,61 @@ class HikingAssistant:
'[Hikingbook](https://zh-tw.hikingbook.net/explore/trails?regions=Taiwan)等網站下載 GPX 檔案',
)
# [Block2]
# Analysis
if uploaded_file is not None:
try:
# Process GPX file
with st.spinner('正在解析 GPX 檔案...'):
gpx_processor = GPXProcessor(uploaded_file)
# Create cache key based on file name and size
file_key = f'{uploaded_file.name}_{uploaded_file.size}'
if not gpx_processor.validate_and_parse():
st.error('❌ GPX 檔案格式錯誤或不包含有效的軌跡點')
return
# 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)
# 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()
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
# 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('📊 路線五四三')
@ -126,70 +165,128 @@ class HikingAssistant:
st.header('☀️ 天氣安抓')
if start_point:
with st.spinner('正在獲取天氣資訊...'):
weather_fetcher = WeatherFetcher()
weather_data = weather_fetcher.get_weather(start_point[0], start_point[1])
weather_fetcher = WeatherFetcher(forecast_days=self._config['app']['weather']['forecast_days'])
weather_key = f'weather_{start_point[0]:.6f}_{start_point[1]:.6f}'
if weather_data:
weather_info = weather_fetcher.format_weather_display(weather_data)
# 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])
if weather_info:
# Current weather
st.subheader('此刻起點天氣')
col1, col2, col3, col4, col5 = st.columns(5)
# 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
with col1:
st.metric(label='🌡️ 溫度', value=weather_info['temperature'])
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)
with col2:
st.metric(label='💧 濕度', value=weather_info['humidity'])
st.subheader('起點天氣')
with col3:
# Get precipitation probability from weather_data
precip_prob = weather_data.get('current_precipitation_prob')
if precip_prob is not None:
st.metric(label='☔ 降雨機率', value=f'{precip_prob}%')
else:
st.metric(label='☔ 降雨機率', value='N/A')
# 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', [])
with col4:
st.metric(label='🌧️ 降雨量', value=weather_info['precipitation'])
# 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')
with col5:
# Get wind direction and convert to flow direction
wind_dir_degrees = weather_data.get('current_wind_direction')
wind_dir_text = weather_fetcher.convert_wind_degrees_to_flow_direction(wind_dir_degrees)
# 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}%')
# Get wind speed and level indicator
wind_speed_value = weather_data.get('current_wind_speed')
wind_level = weather_fetcher.get_wind_speed_indicator(wind_speed_value)
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')
st.metric(
label='💨 風速與流向',
value=f'{weather_info["wind_speed"]} {wind_dir_text} {wind_level}',
)
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}')
# Forecast
if weather_info['forecast']:
st.subheader('未來三天預報')
forecast_cols = st.columns(3)
# 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)
for idx, forecast in enumerate(weather_info['forecast']):
with forecast_cols[idx]:
st.markdown(f'**{forecast["date"]}**')
st.write(f'🌡️ 高溫: {forecast["temp_max"]}°C')
st.write(f'🌡️ 低溫: {forecast["temp_min"]}°C')
st.write(f'☔ 降雨機率: {forecast["precip_prob"]}%')
else:
st.warning('⚠️ 無法格式化天氣資料')
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('⚠️ 無法獲取天氣資訊，請稍後再試')
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)

2
hiking_assistant/assets/config.yaml
View File

@ -2,3 +2,5 @@ app:
page_title: 臺灣登山小幫手
page_favicon_path: ./assets/favicon_compressed.jpg
page_footer_text: ⚠️ 本服務提供之資訊僅供規劃參考，山區氣候瞬息萬變，請務必依據現場狀況與自身能力進行風險評估<br>Made with ❤️ by <a href="https://gitea.guineapig.love/deng">deng</a>
weather:
forecast_days: 7

91
hiking_assistant/weather.py
View File

@ -9,10 +9,11 @@ import requests
class WeatherFetcher:
"""Fetch weather data for hiking locations."""
def __init__(self):
def __init__(self, api_url='https://api.open-meteo.com/v1/forecast', request_timeout=8, forecast_days=7):
"""Initialize weather fetcher."""
self.api_url = 'https://api.open-meteo.com/v1/forecast'
self.request_timeout = 8 # unit: second
self.api_url = api_url
self.request_timeout = request_timeout
self.forecast_days = max(forecast_days, 1)
def convert_wind_degrees_to_flow_direction(self, degrees):
"""Convert wind direction from degrees to flow direction emoji.
@ -50,6 +51,29 @@ class WeatherFetcher:
else:
return '🔴' # Dangerous
def get_uv_index_indicator(self, uv_index):
"""Get UV index level indicator based on index.
Args:
uv_index: UV index
Returns:
str: Emoji indicator (🟢/🟡/🟠/🔴/🟣)
"""
if uv_index is None:
return ''
if uv_index <= 2:
return '🟢' # Safe
elif uv_index <= 5:
return '🟡' # Caution
elif uv_index <= 7:
return '🟠' # Alert
elif uv_index <= 10:
return '🔴' # Dangerous
else:
return '🟣' # Extreme
def get_weather(self, latitude, longitude):
"""Fetch weather data for given coordinates.
@ -65,10 +89,14 @@ class WeatherFetcher:
params = {
'latitude': latitude,
'longitude': longitude,
'current': 'temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m,wind_direction_10m',
'daily': 'temperature_2m_max,temperature_2m_min,precipitation_probability_max',
'current': ('relative_humidity_2m,apparent_temperature,precipitation_probability,precipitation,wind_speed_10m,wind_direction_10m',),
'daily': (
'temperature_2m_max,temperature_2m_min,apparent_temperature_max,apparent_temperature_min,'
'relative_humidity_2m_mean,precipitation_sum,precipitation_probability_max,sunrise,sunset,'
'wind_speed_10m_max,wind_direction_10m_dominant,uv_index_max'
),
'timezone': 'auto',
'forecast_days': 3,
'forecast_days': self.forecast_days,
}
response = requests.get(self.api_url, params=params, timeout=self.request_timeout)
@ -81,15 +109,24 @@ class WeatherFetcher:
daily = data.get('daily', {})
weather_info = {
'current_temperature': current.get('temperature_2m'),
'current_humidity': current.get('relative_humidity_2m'),
'current_apparent_temperature': current.get('apparent_temperature'),
'current_precipitation_probability': current.get('precipitation_probability'),
'current_precipitation': current.get('precipitation'),
'current_precipitation_prob': daily.get('precipitation_probability_max', [None])[0], # Today's rain probability
'current_wind_speed': current.get('wind_speed_10m'),
'current_wind_direction': current.get('wind_direction_10m'),
'daily_temp_max': daily.get('temperature_2m_max', []),
'daily_temp_min': daily.get('temperature_2m_min', []),
'daily_apparent_temp_max': daily.get('apparent_temperature_max', []),
'daily_apparent_temp_min': daily.get('apparent_temperature_min', []),
'daily_mean_relative_humidity': daily.get('relative_humidity_2m_mean', []),
'daily_precipitation_sum': daily.get('precipitation_sum', []),
'daily_precipitation_prob': daily.get('precipitation_probability_max', []),
'daily_wind_speed_max': daily.get('wind_speed_10m_max', []),
'daily_wind_direction_dominant': daily.get('wind_direction_10m_dominant', []),
'daily_sunrise': daily.get('sunrise', []),
'daily_sunset': daily.get('sunset', []),
'daily_uv_index_max': daily.get('uv_index_max', []),
'daily_time': daily.get('time', []),
}
@ -101,41 +138,3 @@ class WeatherFetcher:
except Exception as e:
print(f'天氣資料處理失敗: {str(e)}')
return None
def format_weather_display(self, weather_info):
"""Format weather data for display.
Args:
weather_info: Weather data dictionary from get_weather()
Returns:
dict: Formatted weather data for UI display
"""
if not weather_info:
return None
formatted = {
'temperature': f'{weather_info.get("current_temperature", "N/A")}°C',
'humidity': f'{weather_info.get("current_humidity", "N/A")}%',
'precipitation': f'{weather_info.get("current_precipitation", 0)} mm',
'wind_speed': f'{weather_info.get("current_wind_speed", "N/A")} km/h',
'forecast': [],
}
# Add 3-day forecast
times = weather_info.get('daily_time', [])
temp_max = weather_info.get('daily_temp_max', [])
temp_min = weather_info.get('daily_temp_min', [])
precip_prob = weather_info.get('daily_precipitation_prob', [])
for i in range(min(3, len(times))):
formatted['forecast'].append(
{
'date': times[i],
'temp_max': temp_max[i] if i < len(temp_max) else 'N/A',
'temp_min': temp_min[i] if i < len(temp_min) else 'N/A',
'precip_prob': precip_prob[i] if i < len(precip_prob) else 'N/A',
}
)
return formatted