first commit

.gitignore

@@ -0,0 +1,2 @@
+.env
+.idea

main.py

@@ -0,0 +1,106 @@
+import altair as alt
+import streamlit as st
+import pandas as pd
+
+import strava
+from utils import find_default_publish_start_end_date
+
+st.set_page_config(
+    page_title="Streamlit Activities analysis for Strava",
+    page_icon=":safety_pin:",
+)
+
+strava_header = strava.header()
+
+st.header(":safety_pin: Streamlit Strava activities analysis")
+
+strava_auth = strava.authenticate(header=strava_header, stop_if_unauthenticated=False)
+
+if strava_auth is None:
+    st.markdown("Click the \"Connect with Strava\" button at the top to login with your Strava account and get started.")
+    st.stop()
+
+# analysis on shoes
+st.divider()
+st.header("Display shoes analysis")
+if 'athlete' not in st.session_state:
+    athlete = strava.get_athlete_detail(strava_auth)
+    st.session_state.athlete = athlete
+
+if 'dict_shoes' not in st.session_state:
+    shoes = strava.get_shoes(st.session_state.athlete)
+    dict_shoes = {shoe["name"]: shoe["converted_distance"] for shoe in shoes}
+    st.session_state.dict_shoes = dict_shoes
+
+all_shoes_names = st.session_state.dict_shoes.keys()
+
+selected_shoes = st.multiselect(
+    label="Select columns to plot",
+    options=all_shoes_names,
+)
+
+distances = [st.session_state.dict_shoes[shoe_name] for shoe_name in selected_shoes]
+if selected_shoes:
+    chart_data = pd.DataFrame({
+        'index':selected_shoes,
+        'kilometers':distances
+    })
+    st.bar_chart(chart_data)
+else:
+    st.write("No column(s) selected")
+
+# get activities on a period
+st.divider()
+st.header("Display zones on a period")
+
+default_start_date, default_end_date = find_default_publish_start_end_date()
+
+col_start_date, col_end_date = st.columns(2)
+with col_start_date:
+    st.date_input("Start date", value=default_start_date, key="start_date")
+with col_end_date:
+    st.date_input("End date", value=default_end_date, key="end_date")
+if st.session_state.start_date > st.session_state.end_date:
+    st.error("Error: End date must fall after start date.")
+else:
+    st.session_state.activities = strava.get_activities_on_period(strava_auth, [], st.session_state.start_date, st.session_state.end_date, 1)
+
+st.write(f"You got {len(st.session_state.activities)} activities")
+
+activities_zones = {}
+for activity in st.session_state.activities:
+    if not activity["has_heartrate"]:
+        continue
+    try:
+        st.session_state.activity_zones = strava.get_activity_zones(strava_auth, activity["id"])[0]["distribution_buckets"]
+    except Exception as e:
+        st.write(e)
+    if not activities_zones:
+        activities_zones = {idx: zone["time"] // 60 for idx, zone in enumerate(st.session_state.activity_zones)}
+    else:
+        for idx, zone in enumerate(st.session_state.activity_zones):
+            activities_zones[idx] += (zone["time"] // 60)
+
+zones_label = ["zone 1", "zone 2", "zone 3", "zone 4", "zone 5"]
+zones_df = pd.DataFrame({
+    'zones': zones_label,
+    'minutes': activities_zones.values()
+})
+
+scale = alt.Scale(
+    domain=zones_label,
+    range=["#008000", "#ffcf3e", "#f67200", "#ee1010", "#3f2204"],
+)
+color = alt.Color("zones:N", scale=scale)
+
+bars = (
+    alt.Chart(zones_df)
+    .mark_bar()
+    .encode(
+        x="zones",
+        y="minutes",
+        color=color,
+    )
+)
+
+st.altair_chart(bars, theme="streamlit", use_container_width=True)

requirements.txt

@@ -0,0 +1,4 @@
+streamlit==1.26.0
+httpx==0.24.1
+pandas==2.1.0
+python-dotenv==1.0.0

strava.py

@@ -0,0 +1,240 @@
+import base64
+import os
+
+import httpx
+import streamlit as st
+import webbrowser
+from dotenv import load_dotenv
+from datetime import datetime
+load_dotenv()
+
+APP_URL = os.environ["APP_URL"]
+STRAVA_CLIENT_ID = os.environ["STRAVA_CLIENT_ID"]
+STRAVA_CLIENT_SECRET = os.environ["STRAVA_CLIENT_SECRET"]
+STRAVA_AUTHORIZATION_URL = "https://www.strava.com/oauth/authorize"
+STRAVA_API_BASE_URL = "https://www.strava.com/api/v3"
+SCOPE = "activity:read_all,profile:read_all,activity:write"
+DEFAULT_ACTIVITY_LABEL = "NO_ACTIVITY_SELECTED"
+STRAVA_ORANGE = "#fc4c02"
+
+
+
+@st.cache_data
+def load_image_as_base64(image_path):
+    with open(image_path, "rb") as f:
+        contents = f.read()
+    return base64.b64encode(contents).decode("utf-8")
+
+
+def powered_by_strava_logo():
+    base64_image = load_image_as_base64("static/api_logo_pwrd_by_strava.png")
+    st.markdown(
+        f'<img src="data:image/png;base64,{base64_image}" width="100%" alt="powered by strava">',
+        unsafe_allow_html=True,
+    )
+
+@st.cache_data
+def authorization_url():
+    request = httpx.Request(
+        method="GET",
+        url=STRAVA_AUTHORIZATION_URL,
+        params={
+            "client_id": STRAVA_CLIENT_ID,
+            "redirect_uri": APP_URL,
+            "response_type": "code",
+            "approval_prompt": "auto",
+            "scope": SCOPE
+        }
+    )
+
+    return request.url
+
+
+def login_header(header=None):
+    strava_authorization_url = authorization_url()
+
+    if header is None:
+        base = st
+    else:
+        col1, _, _, button = header
+        base = button
+
+    with col1:
+        powered_by_strava_logo()
+
+    base64_image = load_image_as_base64("./static/btn_strava_connect.png")
+    base.markdown(
+        (
+            f"<a href=\"{strava_authorization_url}\">"
+            f"  <img alt=\"strava login\" src=\"data:image/png;base64,{base64_image}\" width=\"100%\">"
+            f"</a>"
+        ),
+        unsafe_allow_html=True,
+    )
+
+
+def logout_header(header=None):
+    if header is None:
+        base = st
+    else:
+        _, col2, _, button = header
+        base = button
+
+
+    with col2:
+        powered_by_strava_logo()
+
+    if base.button("Log out"):
+        webbrowser.open(APP_URL, new=0)
+
+
+def logged_in_title(strava_auth, header=None):
+    if header is None:
+        base = st
+    else:
+        col, _, _, _ = header
+        base = col
+
+    first_name = strava_auth["athlete"]["firstname"]
+    last_name = strava_auth["athlete"]["lastname"]
+    col.markdown(f"*Welcome, {first_name} {last_name}!*")
+
+
+@st.cache_data
+def exchange_authorization_code(authorization_code):
+    response = httpx.post(
+        url="https://www.strava.com/oauth/token",
+        json={
+            "client_id": STRAVA_CLIENT_ID,
+            "client_secret": STRAVA_CLIENT_SECRET,
+            "code": authorization_code,
+            "grant_type": "authorization_code",
+        }
+    )
+    try:
+        response.raise_for_status()
+    except httpx.HTTPStatusError:
+        st.error("Something went wrong while authenticating with Strava. Please reload and try again")
+        st.experimental_set_query_params()
+        st.stop()
+        return
+
+    strava_auth = response.json()
+
+    return strava_auth
+
+def authenticate(header=None, stop_if_unauthenticated=True):
+    query_params = st.experimental_get_query_params()
+    authorization_code = query_params.get("code", [None])[0]
+
+    if authorization_code is None:
+        authorization_code = query_params.get("session", [None])[0]
+
+    if authorization_code is None:
+        login_header(header=header)
+        if stop_if_unauthenticated:
+            st.stop()
+        return
+    else:
+        logout_header(header=header)
+        strava_auth = exchange_authorization_code(authorization_code)
+        logged_in_title(strava_auth, header)
+        st.experimental_set_query_params(session=authorization_code)
+
+        return strava_auth
+
+
+def header():
+    col1, col2, col3 = st.columns(3)
+
+    with col3:
+        strava_button = st.empty()
+
+    return col1, col2, col3, strava_button
+
+def catch_strava_api_error(response):
+    if response.status_code == 200:
+        return
+    st.write(response.status_code)
+    if response.status_code == 401:
+        st.error("You are not authorized to access this resource. Please relog yourself.")
+        st.stop()
+        return
+    else:
+        st.error(response)
+        if response["errors"]:
+            st.error(f"Something went wrong while fetching data from Strava. Please reload and try again (error message: {response['message']})")
+            st.stop()
+            return
+
+def strava_call(auth, uri_params, params = None):
+    access_token = auth["access_token"]
+    response = httpx.get(
+        url=f"{STRAVA_API_BASE_URL}/{uri_params}",
+        params=params,
+        headers={
+            "Authorization": f"Bearer {access_token}",
+        },
+    )
+    catch_strava_api_error(response)
+    return response.json()
+
+@st.cache_data
+def get_athlete_detail(auth, page=1):
+    return strava_call(auth, "athlete")
+
+@st.cache_data
+def get_shoes(athlete):
+    return athlete["shoes"]
+
+@st.cache_data
+def get_activity(activity_id, auth):
+    return strava_call(auth, f"activities/{activity_id}")
+
+@st.cache_data
+def get_activities_on_period(auth, activities, start_date, end_date, page):
+    response = get_activities(auth, page)
+    number_added = 0
+    for activity in response:
+        strava_start_date = datetime.strptime(activity["start_date"], '%Y-%m-%dT%H:%M:%SZ').date()
+        if strava_start_date >= start_date and strava_start_date <= end_date:
+            activities.append(activity)
+            number_added += 1
+    if number_added == 0:
+        return activities
+    else:
+        return get_activities_on_period(auth, activities, start_date, end_date, page + 1)
+
+@st.cache_data
+def get_activities(auth, page=1):
+    return strava_call(auth, f"athlete/activities", params={"page": page})
+
+@st.cache_data
+def get_activity_zones(auth, activity_id):
+    return strava_call(auth, f"activities/{activity_id}/zones")
+
+@st.cache_data
+def get_athlete_zones(auth):
+    return strava_call(auth, f"athlete/zones")
+
+# @st.cache_data
+# def get_all_activities(auth):
+#     page = 1
+#     activities = []
+#     while True:
+#         new_activities = strava_call(auth, f"athlete/activities", params={
+#                 "page": page,
+#                 "per_page": 200
+#             })
+#         activities.append(new_activities)
+#         if len(new_activities) == 0:
+#             break
+#         page += 1
+#
+#     return activities
+
+# def export_all_activities(auth):
+#     activities = get_all_activities(auth=auth)
+#     st.write(f"You got {len(activities)} activities")
+#     with open('activities.json', 'w') as f:
+#         json.dump(activities, f)

utils.py

@@ -0,0 +1,11 @@
+from datetime import datetime, timedelta
+import streamlit as st
+
+@st.cache_data
+def find_default_publish_start_end_date():
+    today = datetime.now()
+    start = today - timedelta(days=today.weekday())
+    end = start + timedelta(days=6)
+    if end > today:
+        end = today
+    return start, end