Skip to content

Instantly share code, notes, and snippets.

@andy-pi
Created December 30, 2024 18:06
Show Gist options
  • Select an option

  • Save andy-pi/9351e08d18add2e9a3c4e433dc5cc68e to your computer and use it in GitHub Desktop.

Select an option

Save andy-pi/9351e08d18add2e9a3c4e433dc5cc68e to your computer and use it in GitHub Desktop.
Personal Strava Dashboard - Github Actions & Python Static Site Generator
import os
import pickle
import time
from datetime import datetime, timedelta
from dotenv import load_dotenv
from jinja2 import Environment, FileSystemLoader
from stravalib.client import Client as StravaClient
load_dotenv()
strava = StravaClient()
CLIENT_ID = os.getenv('CLIENT_ID')
CLIENT_SECRET = os.getenv('CLIENT_SECRET')
AUTH_CODE = os.getenv('AUTH_CODE')
RUN_DISTANCES = {'5k': 5000, '10k': 10000, 'Half Marathon': 21100}
MEDALS = {1: "#ffd700", 2: "#C0C0C0", 3: "#CD7F32"}
CACHE_FILE = "detailed_activities_cache.pickle"
def render_template(template_name, **kwargs):
env = Environment(loader=FileSystemLoader('templates'))
env.filters['floatformat'] = lambda x, precision=0: f"{x:.{precision}f}"
template = env.get_template(template_name)
return template.render(**kwargs)
def save_activity_to_cache(activity_id, activity_data, cache_dir='activities_cache'):
"""Save an individual activity to a separate file."""
os.makedirs(cache_dir, exist_ok=True)
cache_file = os.path.join(cache_dir, f'{activity_id}.pickle')
with open(cache_file, 'wb') as f:
pickle.dump(activity_data, f)
print(f"Cached activity ID {activity_id} to {cache_file}")
def load_activity_from_cache(activity_id, cache_dir='activities_cache'):
"""Load a single activity from its cache file."""
cache_file = os.path.join(cache_dir, f'{activity_id}.pickle')
if os.path.exists(cache_file):
with open(cache_file, 'rb') as f:
print(f"Loading activity ID {activity_id} from cache")
return pickle.load(f)
print(f"Activity ID {activity_id} not found in cache")
return None
def load_all_cached_activities(cache_dir='activities_cache'):
"""Load all activities from the cache directory."""
if not os.path.exists(cache_dir):
print("Cache directory not found.")
return {}
activities = {}
for file_name in os.listdir(cache_dir):
if file_name.endswith('.pickle'):
activity_id = os.path.splitext(file_name)[0]
activity = load_activity_from_cache(activity_id, cache_dir)
if activity:
activities[activity_id] = activity
return activities
def fetch_and_cache_detailed_activities(activities, cache_dir='activities_cache'):
"""
Fetch detailed activities and cache them individually by ID.
Skip fetching if an activity is already cached.
"""
detailed_activities = load_all_cached_activities(cache_dir)
for activity in activities:
activity_id = str(activity.id) # Ensure the ID is a string for consistency
if activity_id in detailed_activities:
print(f"Activity ID {activity_id} already cached. Skipping fetch.")
continue
print(f"Fetching detailed data for activity ID: {activity_id}")
try:
detailed_activity = get_detailed_activity(activity.id)
detailed_activities[activity_id] = detailed_activity
save_activity_to_cache(activity_id, detailed_activity, cache_dir)
except Exception as e:
print(f"Error fetching activity ID {activity_id}: {e}")
return detailed_activities
def get_first_access_token():
access_token = strava.exchange_code_for_token(client_id=CLIENT_ID, client_secret=CLIENT_SECRET, code=AUTH_CODE)
with open('../access_token.pickle', 'wb') as f:
pickle.dump(access_token, f)
def get_access_token():
"""
Manage and refresh access tokens. Save the token file only if running locally.
Strava tokens only valid for 6 hours
"""
token_file = 'access_token.pickle'
is_github_actions = os.getenv('GITHUB_ACTIONS') == 'true'
# Check if token file exists locally and only use it if not running in GitHub Actions
if not is_github_actions and os.path.exists(token_file):
print("Access token file found locally. Loading token...")
with open(token_file, 'rb') as f:
access_token = pickle.load(f)
else:
print("Access token file not found, loading from GH secrets")
# WIP BROKEN
# These will become really old, but it doesn't matter, we just need one to refresh
access_token = {}
access_token['refresh_token'] = os.getenv('REFRESH_TOKEN')
access_token['expires_at'] = os.getenv('EXPIRES_AT')
# Check if the token has expired and refresh it if needed
if time.time() > float(access_token['expires_at']):
print("Token has expired, refreshing...")
refresh_response = strava.refresh_access_token(
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
refresh_token=access_token['refresh_token']
)
access_token = refresh_response
# Save refreshed token locally only if not running in GitHub Actions
if not is_github_actions:
with open(token_file, 'wb') as f:
pickle.dump(refresh_response, f)
print("Refreshed token saved locally.")
# Set the access token for the Strava client
strava.access_token = access_token['access_token']
strava.refresh_token = access_token['refresh_token']
strava.token_expires_at = access_token['expires_at']
print("Access token is valid.")
def get_last_x_days_activities(days=365):
'''
https://stravalib.readthedocs.io/en/latest/get-started/activities.html
'''
current_date = datetime.now()
x_days_ago = current_date - timedelta(days=days)
formatted_x_days_ago = x_days_ago.strftime("%Y-%m-%d")
activities = strava.get_activities(after=formatted_x_days_ago, limit=1000)
return activities
def get_detailed_activity(id):
return strava.get_activity(id, include_all_efforts=True)
def calc_best_efforts_from_cache(detailed_activities):
"""
Calculate the top 3 best efforts using cached detailed activities with extensive debugging.
Returns a structured dictionary with best efforts for each target distance.
"""
best_efforts = {distance: [] for distance in RUN_DISTANCES.keys()}
for activity_id, detailed_activity in detailed_activities.items():
# print(f"Processing cached detailed activity ID: {activity_id}")
# Check if `best_efforts` exists and is not None
if not hasattr(detailed_activity, 'best_efforts') or detailed_activity.best_efforts is None:
# print(f"No best efforts available for activity ID {activity_id}. Skipping...")
continue
# Debugging `best_efforts` field
# print(f"Activity has {len(detailed_activity.best_efforts)} best efforts")
# for effort in detailed_activity.best_efforts:
# try:
# print(f"Best effort: Distance = {effort.distance}, Elapsed Time = {effort.elapsed_time}, Date = {effort.start_date}, Title = {detailed_activity.name}")
# except AttributeError as e:
# print(f"Error accessing effort details: {e}, Effort: {effort}")
# Process pre-calculated best efforts
for effort in detailed_activity.best_efforts:
for target_distance, target_distance_meters in RUN_DISTANCES.items():
if abs(effort.distance - target_distance_meters) < 100:
# Use elapsed_time directly as it's likely a numeric value
duration = effort.elapsed_time
title = detailed_activity.name
date = effort.start_date.strftime("%A %B %d, %Y")
# Convert seconds to hh:mm:ss or mm:ss if no hours
hours, remainder = divmod(duration, 3600)
minutes, seconds = divmod(remainder, 60)
if hours > 0:
formatted_time = f"{hours}:{minutes:02}:{seconds:02}"
else:
formatted_time = f"{minutes}:{seconds:02}"
# print(f"Matching best effort: Target = {target_distance}, Distance = {effort.distance}, Duration = {formatted_time}, Date = {date}, Title = {title}")
# Append to best efforts for sorting later
best_efforts[target_distance].append({
'time': formatted_time,
'duration': duration, # Keep raw duration for sorting
'date': date,
'title': title
})
# Sort and trim the best efforts for each distance
final_best_efforts = {distance: [] for distance in RUN_DISTANCES.keys()}
for distance, efforts in best_efforts.items():
sorted_efforts = sorted(efforts, key=lambda x: x['duration']) # Sort by duration (ascending)
for rank, effort in enumerate(sorted_efforts[:3], start=1): # Take top 3
effort_with_rank_and_color = {
'time': effort['time'],
'date': effort['date'],
'title': effort['title'],
'rank': rank,
'color': MEDALS.get(rank, "#ffffff") # Default color for ranks beyond 3
}
final_best_efforts[distance].append(effort_with_rank_and_color)
# print("Final Running Best Effortss:", final_best_efforts)
return final_best_efforts
def get_longest_three_bike_rides(detailed_activities):
"""
Get the top three longest bike rides based on distance.
Returns a structured dictionary with the longest rides ranked.
"""
longest_rides = []
# print("Processing bike rides for top distances...")
for activity_id, detailed_activity in detailed_activities.items():
# Ensure the activity is a bike ride
activity_type = getattr(detailed_activity.type, 'root', '').lower()
if activity_type not in ['ride', 'cycling']:
# print(f"Skipping non-bike activity: {activity_type}")
continue
# Extract required details
try:
distance = detailed_activity.distance # Distance in meters
duration = detailed_activity.moving_time # Duration in seconds
title = detailed_activity.name
date = detailed_activity.start_date.strftime("%A %B %d, %Y")
# Format time to hh:mm:ss or mm:ss
hours, remainder = divmod(duration, 3600)
minutes, seconds = divmod(remainder, 60)
if hours > 0:
formatted_time = f"{hours}:{minutes:02}:{seconds:02}"
else:
formatted_time = f"{minutes}:{seconds:02}"
# Append ride details to the list
longest_rides.append({
'distance': distance / 1000, # Keep numeric for sorting
'time': formatted_time,
'date': date,
'title': title
})
# print(f"Processed ride: {distance / 1000:.1f}k, {formatted_time}, {date}, {title}")
except AttributeError as e:
pass
# print(f"Error processing activity ID {activity_id}: {e}")
# Sort by distance (descending) and take the top 3
longest_rides = sorted(longest_rides, key=lambda x: x['distance'], reverse=True)[:3]
# Format the result into a ranked dictionary with colors
ranked_rides = {rank + 1: {**ride, 'distance': f"{ride['distance']:.1f}k", 'color': MEDALS.get(rank + 1, "")}
for rank, ride in enumerate(longest_rides)}
# print("Top three longest rides calculated:")
# for rank, ride in ranked_rides.items():
# print(f"Rank {rank}: {ride}")
return ranked_rides
def get_fastest_three_surprise_loops(detailed_activities):
"""
Get the fastest 3 overall times for any bike ride titled 'Surprise Loop'.
Returns a list of dictionaries containing the top 3 rides ranked by fastest time.
"""
surprise_loops = []
for activity_id, detailed_activity in detailed_activities.items():
# Ensure the activity is a bike ride
activity_type = getattr(detailed_activity.type, 'root', '').lower()
if activity_type not in ['ride', 'cycling']:
# print(f"Skipping non-bike activity: {activity_type}")
continue
# Filter by title 'Surprise Loop'
if detailed_activity.name not in ['Surprise Loop', 'Suprise Loop']:
# print(f"Skipping non-Surprise Loop activity: {detailed_activity.name}")
continue
# Extract required details
try:
duration = detailed_activity.moving_time # Duration in seconds
distance = detailed_activity.distance # Distance in meters
date = detailed_activity.start_date.strftime("%A %B %d, %Y")
# Format time to hh:mm:ss or mm:ss
hours, remainder = divmod(duration, 3600)
minutes, seconds = divmod(remainder, 60)
formatted_time = f"{hours}:{minutes:02}:{seconds:02}" if hours > 0 else f"{minutes}:{seconds:02}"
# Append ride details to the list
surprise_loops.append({
'time': formatted_time,
'duration': duration, # Keep raw duration for sorting
'distance': f"{distance / 1000:.1f}k", # Convert meters to kilometers
'date': date,
'title': detailed_activity.name
})
except AttributeError as e:
# print(f"Error processing activity ID {activity_id}: {e}")
pass
# Sort by time (ascending) and take the top 3
surprise_loops = sorted(surprise_loops, key=lambda x: x['duration'])[:3]
# Add rank and color for each loop
for rank, sloop in enumerate(surprise_loops, start=1):
sloop['rank'] = rank
sloop['color'] = {1: '#ffd700', 2: '#C0C0C0', 3: '#CD7F32'}.get(rank, '#ffffff')
return surprise_loops
def total_distance_for_activity(detailed_activities, activity_type=None, sport_type=None, days=365):
"""
Calculate the total distance for a single activity type or sport type in the past given days.
Args:
detailed_activities (dict): A dictionary of activity data where keys are activity IDs and values are detailed activity objects.
activity_type (str, optional): The main activity type (e.g., 'ride', 'run'). Defaults to None.
sport_type (str, optional): Specific sport type (e.g., 'MountainBikeRide'). Defaults to None.
days (int): Number of days in the past to consider (default is 365).
Returns:
float: Total distance in kilometers for the specified activity or sport type.
"""
# Define the date range
start_date = datetime.now() - timedelta(days=days)
total_distance = 0.0
for activity_id, detailed_activity in detailed_activities.items():
# Check if the main activity type matches
if activity_type:
detailed_type = getattr(detailed_activity.type, 'root', '').lower()
if detailed_type != activity_type.lower():
continue
# Check if the specific sport type matches
if sport_type:
detailed_sport_type = getattr(detailed_activity.sport_type, 'root', '').lower()
if detailed_sport_type != sport_type.lower():
continue
# Check if the activity is within the given date range
if detailed_activity.start_date_local >= start_date:
# Add the distance (convert to kilometers if in meters)
total_distance += detailed_activity.distance / 1000 # Assuming distance is in meters
return total_distance
if __name__ == '__main__':
get_access_token()
activities = get_last_x_days_activities()
detailed_activities = fetch_and_cache_detailed_activities(activities)
# TOTALS
total_run_distance = total_distance_for_activity(detailed_activities, 'run')
total_mtb_distance = total_distance_for_activity(detailed_activities, sport_type="MountainBikeRide")
total_road_bike_distance = total_distance_for_activity(detailed_activities, 'ride') - total_mtb_distance
total_swim_distance = total_distance_for_activity(detailed_activities, 'swim')
# BIKING
longest_bike_rides = get_longest_three_bike_rides(detailed_activities)
fastest_surprise_loops = get_fastest_three_surprise_loops(detailed_activities)
# RUNNING: best efforts
best_efforts = calc_best_efforts_from_cache(detailed_activities)
output_path = "output"
os.makedirs(output_path, exist_ok=True)
output_html = render_template(
'index.html',
total_run_distance=total_run_distance,
total_mtb_distance=total_mtb_distance,
total_swim_distance=total_swim_distance,
longest_bike_rides=longest_bike_rides,
total_road_bike_distance=total_road_bike_distance,
fastest_surprise_loops=fastest_surprise_loops,
best_efforts=best_efforts,
last_update=datetime.now().strftime("%A, %d %B %Y")
)
with open(os.path.join(output_path, 'index.html'), 'w') as f:
f.write(output_html)
print("Static site generated at ./output/index.html")
name: Daily Update of Strava Dashboard
on:
schedule:
- cron: "0 12 * * *" # Runs daily at midday UTC
workflow_dispatch: # Allows manual trigger
jobs:
daily-update:
runs-on: ubuntu-latest
steps:
# Step 1: Checkout repository
- name: Checkout repository
uses: actions/checkout@v3
# Step 2: Set up Python (3.11.8)
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: 3.11.8
# Step 3: Install dependencies
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
# Step 4: Run the script
- name: Run the Python script
run: python dashboard.py
env:
CLIENT_ID: ${{ secrets.CLIENT_ID }}
CLIENT_SECRET: ${{ secrets.CLIENT_SECRET }}
AUTH_CODE: ${{ secrets.AUTH_CODE }}
REFRESH_TOKEN: ${{ secrets.REFRESH_TOKEN }}
EXPIRES_AT: ${{ secrets.EXPIRES_AT }}
# Step 5: Commit and push changes
- name: Commit and push changes
run: |
echo "Checking for changes..."
git status --porcelain
git diff
git config --local user.name "GitHub Actions"
git config --local user.email "actions@github.com"
if [ -n "$(git status --porcelain)" ]; then
git add output/index.html
git commit -m "Daily update: $(date +'%Y-%m-%d %H:%M:%S')"
git push
else
echo "No changes to commit"
fi
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment