Strava API とPythonを使ってSegmentの履歴を取り出す(7)-AWSクラウド(後編)

この記事では、前回までに作ったStrava APIで取得したサイクリングデータを可視化するStreamlitアプリをAWSクラウドで動かしてみます。以下の記事の続編になります。

atsushi3.hatenablog.com

github.com

前回は、

  • ステップ1: ネットワーク環境の構築 (VPC)

  • ステップ2: セキュリティ設定 (IAMロールとセキュリティグループ)

  • ステップ3: 秘密情報の管理 (Systems Manager Parameter Store)

  • ステップ4: EC2インスタンスの作成

  • ステップ5: EC2への接続

まで実装しました。今回は、EC2環境構築、Streamlitアプリデプロイ、動作確認、リソース削除についてです。

ステップ6: EC2環境構築

  • パッケージのインストールとアップデート (SSH接続したターミナル内で実行)
# システムを最新の状態に更新
sudo dnf update -y

# Git, PostgreSQLサーバー、そしてPythonから使うための開発パッケージをインストール
sudo dnf install -y git postgresql15-server postgresql-devel
  • pyenv + pyenv-virtualenv 環境構築

以下を参考にさせていただきました。

【Python】Amazon Linux 2023にPython 3.11 + Selenium環境を構築する(前編)|yuu

# Python環境の構築
curl https://pyenv.run | bash

# pyenvでインストールできるバージョンを確認
pyenv install -l | grep '3.13\.'

# python 3.13.5 をインストール
pyenv install 3.13.5

# virtualenv 環境を作成
pyenv virtualenv 3.13.5 strava-pg-env
# .bashrc に以下を追加
export PYENV_ROOT="$HOME/.pyenv"
command -v pyenv >/dev/null || export PATH="$PYENV_ROOT/bin:$PATH"
eval "$(pyenv init -)"
eval "$(pyenv virtualenv-init -)"

ステップ7: アプリケーションのデプロイ

  • アプリケーションのクローン、virtualenv有効化
cd /home/ec2-user

git clone https://github.com/atsushi3hsgw/strava_pg.git

# virtualenvを作成して有効化
cd strava_pg
pyenv local strava-pg-env

# virtualenvを確認
pyenv versions
  system
  3.13.5
  3.13.5/envs/strava-pg-env
* strava-pg-env --> /home/ec2-user/.pyenv/versions/3.13.5/envs/strava-pg-env (set by /home/ec2-user/strava_pg/.python-version)
  • Pythonライブラリのインストール
pip install -r requirements.txt

ステップ8: アプリの自動起動設定

起動スクリプトは、以下の様になります。

  • virtualenv有効化

  • Parameter Storeから秘密情報を取得して環境変数にセット

  • Streamlitアプリを起動

#!/bin/bash
# /home/ec2-user/strava_pg/aws/start_strava_segment_efforts.sh
set -e

# Enable the virtual environment
export PYENV_ROOT="$HOME/.pyenv"
export PATH="$PYENV_ROOT/bin:$PATH"
eval "$(pyenv init -)"
eval "$(pyenv virtualenv-init -)"

# Force application of the project's pyenv virtual environment
# (Unnecessary if .python-version exists)
pyenv activate strava-pg-env

# Configure AWS Region (Tokyo Region)
export AWS_REGION="ap-northeast-1"

# Retrieve confidential information from the Parameter Store and set it as an environment variable
export STRAVA_CLIENT_ID=$(aws ssm get-parameter --name "/streamlit-app/strava/client_id" --with-decryption --query "Parameter.Value" --output text)
export STRAVA_CLIENT_SECRET=$(aws ssm get-parameter --name "/streamlit-app/strava/client_secret" --with-decryption --query "Parameter.Value" --output text)
export STRAVA_REDIRECT_URI=$(aws ssm get-parameter --name "/streamlit-app/strava/redirect_uri" --with-decryption --query "Parameter.Value" --output text)

# Set the database connection information in the environment variables
export DB_USER=$(aws ssm get-parameter --name "/streamlit-app/strava/db_user" --with-decryption --query "Parameter.Value" --output text)
export DB_PASSWORD=$(aws ssm get-parameter --name "/streamlit-app/strava/db_passwd" --with-decryption --query "Parameter.Value" --output text)
export DB_HOST="localhost"
export DB_NAME="strava"

# Start the Streamlit application
streamlit run strava_segment_efforts.py --server.address 0.0.0.0 --server.port 8501
  • systemdサービスファイルの作成:
[Unit]
Description=Streamlit App Service
After=network.target postgresql.service

[Service]
User=ec2-user
Group=ec2-user
WorkingDirectory=/home/ec2-user/strava_pg
ExecStart=/home/ec2-user/strava_pg/aws/start_strava_segment_efforts.sh
Restart=always

[Install]
WantedBy=multi-user.target
  • サービスの有効化と起動:
sudo systemctl daemon-reload
sudo systemctl enable strava_segment_efforts.service
sudo systemctl start strava_segment_efforts.service

# サービスが正常に起動しているか確認 (qで終了)
sudo systemctl status strava_segment_efforts.service

ステップ9: ロードバランサーの作成 (ALB)

  • 「EC2」サービス > 「ロードバランサー」> 「ロードバランサーの作成」をクリックします。

  • Application Load Balancer の「作成」を選択します。

  • ロードバランサー名: streamlit-app-alb

  • スキーム: インターネット向け

  • VPC: ステップ1で作成したVPCを選択します。

  • マッピング: 作成したVPCのパブリックサブネットを2つとも選択します。

  • セキュリティグループ: ステップ2で作成した streamlit-alb-sg を選択します。

  • リスナーとルーティング:

    • デフォルトの HTTP:80 のままで、デフォルトアクションに「ターゲットグループの作成」を選択します。

    • ターゲットタイプ: インスタンス

    • ターゲットグループ名: streamlit-app-tg

    • ポート: 8501

    • 「次へ」進み、利用可能なインスタンスから streamlit-server を選択し、「保留中として以下を含める」をクリックします。

    • 「ターゲットグループの作成」をクリックします。

  • ロードバランサーの作成画面に戻り、ターゲットグループの「更新ボタン」を押し、今作成した streamlit-app-tg を選択します。

  • ロードバランサーの作成」をクリックします。

ステップ10: 動作確認(1)

  • ロードバランサー」の一覧で、作成した streamlit-app-alb を選択します。

  • 「詳細」タブにある DNS名 をコピーします。

  • WebブラウザのアドレスバーにDNS名を貼り付けてアクセスし、Streamlitアプリケーションが表示されることを確認します。ターゲットグループのヘルスチェックが healthy になるまで数分かかるみたいです。

ステップ11: 動作確認(2)

  • 今のままでは、Strava認証でエラーになってしまいます。
    • REDIRECT URL にステップ10で取得したDNS名を使っていないためです。「ステップ3: 秘密情報の管理」の要領で /streamlit-app/strava/redirect_url にDNS名を設定して下さい。
値の例:
http://streamlit-app-alb-2019454863.ap-northeast-1.elb.amazonaws.com
  • STRAVA 側の設定で「認証コールバックドメイン」をステップ10で取得したDNS名を使っていないためです。STRAVA のサイトのMy API アプリケーションで「認証コールバックドメイン」を設定して下さい。
値の例:
streamlit-app-alb-2019454863.ap-northeast-1.elb.amazonaws.com
  • サービス再起動
sudo systemctl restart strava_segment_efforts.service
sudo systemctl status strava_segment_efforts.service

WebブラウザのアドレスバーにDNS名を貼り付けてアクセスし、Streamlitアプリケーションが表示されることを確認します。Strava認証が完了してセグメントダッシュボード、ヒートマップが表示されたら成功です。

ステップ12: リソース削除チェックリスト

以下の順序でリソースを削除

  • [ ] EC2インスタンスの削除: streamlit-server を選択し、「インスタンスを終了」します。

  • [ ] ロードバランサー (ALB) の削除: streamlit-app-alb を選択し、「削除」します。

  • [ ] ターゲットグループ (TG) の削除: streamlit-app-tg を選択し、「削除」します。

  • [ ] NATゲートウェイの削除: 「VPC」>「NATゲートウェイ」から作成したものを削除します。

  • [ ] VPCの削除: 「VPC」>「お使いのVPC」から streamlit-app-vpc を選択し、「VPCを削除」します。(関連リソースも同時に削除されます)

  • [ ] IAMロールの削除: EC2-Streamlit-SSM-Role を削除します。

  • [ ] Parameter Store のパラメータ削除: 作成した2つのパラメータを削除します。

  • [ ] EC2キーペアの削除: streamlit-server で使用したキーペアを削除します。

  • [ ] セキュリティグループの削除: streamlit-alb-sg と streamlit-ec2-sg を削除します。

次は、AWS Lambda、AWS Rekognition, Line 連携を使ってイケメン判定アプリを作ってみようかな!!

Strava API とPythonを使ってSegmentの履歴を取り出す(6)-AWSクラウド(前編)

この記事では、前回までに作ったStrava APIで取得したサイクリングデータを可視化するStreamlitアプリをAWSクラウドで動かしてみます。

初めてのAWS挑戦メモです!!
Streamlitアプリは以下のことができます。

  • OAuth認証してStrava API を使用
  • サイクリングデータ(アクテビティ)をPostgresSQLに保存
  • サイクリングデータ(アクテビティ)を分析して可視化
  • パーソナルヒートマップを作成

atsushi3.hatenablog.com

atsushi3.hatenablog.com

github.com

システムの全体像

今回構築するシステムの全体像です。 - Streamlitアプリ、PostgreSQLはプライベートサブネット内のEC2インスタンスに配置します。

  • ユーザーからのアクセスはインターネットを経由し、ALB(ロードバランサー)が受け付けます。

  • ALBはトラフィックをプライベートサブネット内のEC2インスタンスに転送し、そこでStreamlitアプリが動作します。

  • EC2インスタンスは、外部のAPI(Stravaなど)やパッケージのダウンロードのために、NATゲートウェイを経由してインターネットに接続します。

  • EC2インスタンスには、ローカルPCからSystems Manager (SSM) + SSH で安全に接続できます。(22番ポートを開けない、踏み台使いたくない)

  • Strava Client Secret 等の秘密情報をSystems Manager (SSM)で管理します。

  • なぜこの構成? パブリックネットワークにアプリをデプロイ、プライベートネットワークにDBサーバ、あるいはRDSという構成は、あたりまえすぎて面白くない。

システム全体像

ステップ1: ネットワーク環境の構築 (VPC)

  • AWSマネジメントコンソールで「VPC」サービスに移動します。
  • VPCを作成」を選択し、「VPCなど」を選択します。
  • 名前タグの自動生成: streamlit-app-vpc など、わかりやすい名前を入力します。
  • アベイラビリティーゾーン (AZ) の数: 2 を選択します。(重要:ALBのために2つ必要です
  • パブリックサブネットの数: 2
  • プライベートサブネットの数: 2
  • NATゲートウェイ: 1 AZ 内 を選択します。(実験用のため1つで十分です)
  • VPCエンドポイント: なし
  • VPCを作成」をクリックします。これにより、必要なネットワークリソース(VPC、サブネット、インターネットゲートウェイ、NATゲートウェイ、ルートテーブル)が自動で作成されます。

ステップ2: セキュリティ設定 (IAMロールとセキュリティグループ)

EC2インスタンスに適切な権限を与え、通信を安全に制御します。

IAMロールの作成:

  • 「IAM」サービス > 「ロール」> 「ロールを作成」を選択します。
  • 信頼されたエンティティタイプ: 「AWS のサービス」、ユースケース: 「EC2」を選択し、「次へ」。
  • 許可ポリシー: 以下の2つを検索し、チェックを入れます。
  • AmazonSSMManagedInstanceCore (Session ManagerやSSHポートフォワーディングに必須)
  • AmazonSSMReadOnlyAccess (Parameter Storeから秘密情報を読み取るために必要)
  • 「次へ」進み、ロール名に EC2-Streamlit-SSM-Role などを入力し、ロールを作成します。

セキュリティグループの作成:

  • 「EC2」サービス > 「セキュリティグループ」に移動します。
  • ALB用セキュリティグループの作成:

    • セキュリティグループ名: streamlit-alb-sg
    • VPC: ステップ1で作成したVPCを選択します。
    • インバウンドルール: タイプ: HTTP、ソース: Anywhere-IPv4 (0.0.0.0/0) を設定します。
    • 「セキュリティグループを作成」します。
  • EC2用セキュリティグループの作成:

    • セキュリティグループ名: streamlit-ec2-sg
    • VPC: ステップ1で作成したVPCを選択します。
    • インバウンドルール: タイプ: カスタムTCP、ポート範囲: 8501 (Streamlitのデフォルトポート)、ソース: 上で作成した streamlit-alb-sg を選択します。(これによりALBからの通信のみ許可されます)
    • 「セキュリティグループを作成」します。

ステップ3: 秘密情報の管理 (Systems Manager Parameter Store)

Strava APIのキーなどを安全に保管します。

  • 「Systems Manager」サービス > 「パラメータストア」を選択します。
  • 「パラメータを作成」をクリックします。
  • 名前: /streamlit-app/strava/client_id
  • タイプ: 「安全な文字列 (SecureString)」を選択します。
  • 値: ご自身のStrava Client IDを入力します。
  • 「パラメータを作成」します。

同様に以下のパラメータも作成します。

  • Strava Client Secret:/streamlit-app/strava/client_secret
    ご自身のStrava Client Secretを入力します。

  • Strava Redirect URL:/streamlit-app/strava/redirect_url
    OAuth認証のリダイレクト先URIです。ここではとりあえず空欄とします。後のステップでALBを作成したらDNS名を入力します。

  • /streamlit-app/strava/db_user

  • /streamlit-app/strava/db_passwd

ステップ4: EC2インスタンスの作成

StreamlitアプリとPostgreSQLを動かすサーバーを起動します。

  • 「EC2」サービス > 「インスタンスを起動」をクリックします。
  • 名前: streamlit-server
  • アプリケーションおよび OS イメージ: Amazon Linux 2023 AMI を選択します。
  • インスタンスタイプ: t2.micro または t3.micro を選択します(無料利用枠対象)。
  • キーペア: 「新しいキーペアの作成」からキーペアを作成し、.pem ファイルを安全な場所にダウンロードしておきます。(SSH接続に必須です)
  • ネットワーク設定:

    • 「編集」をクリックします。
    • VPC: ステップ1で作成したVPCを選択します。
    • サブネット: private と名前のついたプライベートサブネットのいずれか1つを選択します。
    • ファイアウォール (セキュリティグループ): 「既存のセキュリティグループを選択する」を選び、ステップ2で作成した streamlit-ec2-sg を選択します。
  • 高度な詳細:

    • IAM インスタンスプロファイル: ステップ2で作成した EC2-Streamlit-SSM-Role を選択します。
  • インスタンスを起動」をクリックします。

ステップ5: EC2への接続

ローカルPCの設定

以下のAWS CLIをローカルPCにインストールします。詳細は、省略します。

  • awscli
  • aws-session-manager-plugin

Systems Manager Session Manager + SSH で接続

  • EC2のインスタンスID(例: i-0123...)をコピーしておきます。
  • ローカルPCで1つ目のターミナルを開き、以下のコマンドを実行してEC2との安全なトンネルを確立します。(このターミナルは接続中、開いたままにします)
  • 私は、Windowsを使っていますが、powershellとかを使うと文字エスケープとかわけわからなくなってはまるので、Git Bash を使うことをお勧めします。
aws ssm start-session --target i-0c486d8363b61e289 --document-name AWS-StartPortForwardingSession --parameters '{"portNumber":["22"],"localPortNumber":["2222"]}'
  • ローカルPCで2つ目のターミナルを開き、以下のコマンドでSSH接続します。
    • "path/to/your-key.pem" は、ステップ4でダウンロードしたキーペアファイルのパスに置き換えてください。
    • ポート番号は -p 2222 を指定します。
ssh -i "path/to/your-key.pem" -p 2222 ec2-user@localhost

長くなったの今回はここまでにします。 次回は、「EC2インスタンス環境、アプリケーションデプロイ」の予定です。

atsushi3.hatenablog.com

Strava API とPythonを使ってSegmentの履歴を取り出す(5)-Foliumヒートマップ作成

パーソナルヒートマップ

この記事では、Strava APIで取得したサイクリングデータを使って、 Foliumライブラリでパーソナルヒートマップを作成する方法を解説します。 よく走るルートが一目でわかる美しい可視化が実現できます。

前回ブログまででPython、Strava APIPostgreSQL JSONB、Streamlitを使って自分のアクテビティが分析できるアプリを作ってきました。 前回までは、こちら atsushi3.hatenablog.com

今回は、Foliumヒートマップ作成してみました。宮ケ瀬、裏ヤビツがお気に入りなのがわかりますね。

パーソナルヒートマップ
セグメント分析
ソースは↓にあります。 github.com

緯度, 経度を取得する

アクテビティ詳細には、polyline*がGoogleのポリラインエンコーディングアルゴリズムによってエンコードされた形式で格納されています。
以下のコードは、polylineをデコードして [[緯度, 経度], ...]で取り出す関数です。 アクテビティ詳細は、STRAVA APIで取得したデータをJSONのままPostgreSQLに保存してあります。パフォーマンス云々を考えなければ、JSON構造をそのまま検索できるので便利です。

import polyline

@st.cache_data(ttl=300)
def get_polylines(before=None, after=None):
    """
    Get all decoded polylines for heatmap as [[lat, lng], ...].
    
    Parameters
    ----------
    before : datetime | None
        Include activities with start_time <= before
    after : datetime | None
        Include activities with start_time >= after
    """

    engine = get_db_engine()
    base_query = """
    SELECT 
        data->'map'->>'polyline' AS polyline,
        (data->>'start_date_local')::timestamp AS start_time
    FROM
        activities
    WHERE 
        data->'map'->>'polyline' IS NOT NULL
    """
    conditions = []
    bind_params = {}
    if before is not None:
        conditions.append(" (data->>'start_date_local')::timestamp<= :before")
        bind_params["before"] = before
    if after is not None:
        conditions.append(" (data->>'start_date_local')::timestamp>= :after")
        bind_params["after"] = after

    if conditions:
        base_query += " AND " + " AND ".join(conditions)

    all_coords = []
    with engine.connect() as conn:
        result = conn.execute(text(base_query), bind_params)
        for row in result:
            poly = row[0]
            if not poly or not isinstance(poly, str):
                continue
            try:
                coords = polyline.decode(poly)
                all_coords.extend(coords)
            except Exception as e:
                st.warning(f"Failed to decode a polyline: {e}")
                logger.warning(f"Failed to decode polyline: {e}")
                continue
    return all_coords

Foliumヒートマップ作成

Foliumは、Pythonインタラクティブな地図を作成するためのライブラリです。 streamlit_foliumは、StreamlitにFoliumで作成したインタラクティブな地図を埋め込むためのコンポーネントライブラリです。 簡単に実装できましたが、なんか重くて表示がちらつきます。

import folium
from folium.plugins import HeatMap
from streamlit_folium import st_folium

def heatmap_dashboard():
    
    st.subheader("Personal Heatmap")
    st.info(f"Fetching coordinates...")
    coords = get_polylines()
    st.info(f"Fetched {len(coords)} coordinates.")
                    
    if coords:
        center_lat = sum(p[0] for p in coords) / len(coords)
        center_lon = sum(p[1] for p in coords) / len(coords)
    
        m = folium.Map(location=[center_lat, center_lon], zoom_start=12)
        HeatMap(coords, radius=8, blur=5, min_opacity=0.2).add_to(m)
        st_folium(m, use_container_width=True)
    else:
        st.warning("No coordinates to display for the selected date range.")
        
    if st.button("Segment Efforts Dashboard"):
        st.session_state['dashboard'] = 'segment_efforts'
        st.rerun()

Streamlit 最適化

座標データ (coords) の量が多い場合に処理が重くなり、表示がちらついていました。Streamlitはウィジェットを操作するたびにスクリプト全体を再実行するため、その都度重い地図の描画処理が走ってしまうからです。
Strava API とPythonを使ってSegmentの履歴を取り出す(3)- Streamlit データ可視化ダッシュボード - atsushi-h’s blog

毎回地図オブジェクトを生成するのではなく、一度生成した地図(のHTMLコード)をキャッシュしてしまうのが最も効果的です。foliumの地図は直接キャッシュできませんが、HTML文字列に変換すればキャッシュ可能です。

  • generate_heatmap_html 関数に @st.cache_data をつけることで、同じ座標データ (coords) で呼び出された場合は、中の重い処理(地図生成)をスキップして、キャッシュ済みのHTMLを即座に返します。
  • st_folium の代わりに st.components.v1.html で表示します。再描画のオーバーヘッドをなくすことができました。
import streamlit.components.v1 as components

@st.cache_data(ttl=3600)
def generate_heatmap_html(coords):

    if not coords:
        return None

    center_lat = sum(p[0] for p in coords) / len(coords)
    center_lon = sum(p[1] for p in coords) / len(coords)

    m = folium.Map(location=[center_lat, center_lon], zoom_start=12)
    HeatMap(coords, radius=8, blur=5, min_opacity=0.2).add_to(m)
    
    return m._repr_html_()

def heatmap_dashboard(before=None, after=None):
    
    logger.debug(f"heatmap_dashboard before={before}, after={after} ")
    
    st.subheader("Personal Heatmap")
    st.info(f"Fetching coordinates...")
    coords = get_polylines(before=before, after=after)
    st.info(f"Fetched {len(coords)} coordinates.")
                    
    if coords:
        # center_lat = sum(p[0] for p in coords) / len(coords)
        # center_lon = sum(p[1] for p in coords) / len(coords)
    
        # m = folium.Map(location=[center_lat, center_lon], zoom_start=12)
        # HeatMap(coords, radius=8, blur=5, min_opacity=0.2).add_to(m)
        # st_folium(m, use_container_width=True)
        map_html = generate_heatmap_html(coords)
        if map_html:
            components.html(map_html, height=500)
    else:
        st.warning("No coordinates to display for the selected date range.")
        
    if st.button("Segment Efforts Dashboard"):
        st.session_state['dashboard'] = 'segment_efforts'
        st.rerun()

まとめ

今回はFoliumを使ってStravaデータのヒートマップを作成しました。 次回は「AWSクラウド化してみる」です。

atsushi3.hatenablog.com

Strava API とPythonを使ってSegmentの履歴を取り出す(4)-Streamlitダッシュボード作成|OAuth認証実装

Streamlit アプリへの統合

この記事では、これまで個別に実装してきたStrava API機能を、 Streamlitを使って統合されたWebアプリケーションにまとめる方法を解説します。 OAuth認証からデータ可視化まで、一つのアプリで完結できます。

前回ブログまで

  1. Strava認証してアクセストーク を取得
  2. アクセストークンを使ってアクテビティをデータベース(PostgreSQL) に保存
  3. Streamlit データベース(PostgreSQL)に保存したアクテビティにアクセスして分析結果を可視化

を別々のコマンドで実装してきました。

atsushi3.hatenablog.com

今回は、これらをStreamit アプリに統合してみます。 出来上がりは↓のイメージになります。

セグメント分析

ソースは、↓にあります。   github.com

Strava OAuth認証の実装手順

Strava認証の部分は少し複雑なので整理します。
Strava認証の詳細ドキュメントは、
https://developers.strava.com/docs/getting-started/#oauth にあります。
ここでは、 stravalib を利用して認証する方法を説明します。  https://stravalib.readthedocs.io/en/latest/reference.html

OAuth認証フロー

  1. 認証URL生成:ユーザーをStravaの認証画面に誘導
  2. 認証コード取得:ユーザー承認後にコードを受け取り
  3. アクセストークン発行:コードをトークンに交換

認証URL生成:ユーザーをStravaの認証画面に誘導

stravalib.client.Client.authorization_url() でStrava認証(OAuth)URLを作成します。
redirect_uri には、認証が完了した時にリダイレクトされるurlを指定します。

# Generate and display authorization URL
try:
    authorize_url = client.authorization_url(
            client_id=os.environ['STRAVA_CLIENT_ID'],
            redirect_uri=os.environ['STRAVA_REDIRECT_URI'],
            scope=['read', 'activity:read_all', 'profile:read_all']
        )
    st.info("To connect to Strava and download your activities, please click the link below.")
    # Use explicit HTML to ensure link opens in the same tab
    st.markdown(f"<a href='{authorize_url}' target='_self'>Connect to Strava</a>", nsafe_allow_html=True)
    st.stop()
except Exception as e:
    st.error(f"Error generating authorization URL: {e}")
    st.stop()

Strava認証(OAuth)URLは次の様になります。

https://www.strava.com/oauth/authorize?client_id=[YOUR_CLIENT_ID]&redirect_uri=[YOUR_RIDIRECT_URL]&approval_prompt=auto&scope=read%2Cactivity%3Aread_all%2Cprofile%3Aread_all&response_type=code

認証コード取得:ユーザー承認後にコードを受け取り

Strava認証(OAuth)で認証が完了すると、先ほど指定した redirect_uri認証コード(authorization code) が付加されてリダイレクトされます。 認証コード(authorization code) は、この後 アクセストークン(access token) を発行するために必要となります。
リダイレクトURLは、次の様になります。

[YOUR_RIDIRECT_URL]/?state=&code=[your authorization code]&scope=read,activity:read_all,profile:read_all

アクセストークン発行:コードをトークンに交換

認証コード(authorization code)クライアントシークレット からアクセストークン(access token) を発行します。
stravalib.client.Client.exchange_code_for_token() を使います。

# Get authorization code from URL
query_params = st.query_params
auth_code = query_params.get("code")

# Check if token is in session state or has expired
if 'access_token' not in st.session_state or st.session_state['expires_at'] < time_module.time():
    client = Client()
    
    # Exchange authorization code for token
    if auth_code:
        logger.debug(f"Authorization Code: {auth_code}") # Debug log
        try:
            with st.spinner("Exchanging authorization code for token..."):
                token_info = client.exchange_code_for_token(
                    client_id=os.environ['STRAVA_CLIENT_ID'],
                    client_secret=os.environ['STRAVA_CLIENT_SECRET'],
                    code=auth_code
                )
            
            # Save token info to session state
            st.session_state['access_token'] = token_info['access_token']
            st.session_state['refresh_token'] = token_info['refresh_token']
            st.session_state['expires_at'] = token_info['expires_at']            
            st.rerun()

次回はこちら

atsushi3.hatenablog.com

Strava API とPythonを使ってSegmentの履歴を取り出す(3)- Streamlit データ可視化ダッシュボード

Strava のセグメント分析をしたくて

atsushi3.hatenablog.com

とやってきました。 この記事では、これまでPostgreSQLに保存したStravaデータを、 Streamlitを使って視覚的に分析できるWebアプリとして実装します。 データベース連携からリアルタイム可視化まで、実践的な開発手法を解説します。 今回は、Streamlit を使って視覚化してみようと思います。

ソースは

github.com

にあります。詳しくは、githubを参照ください。

こんな分析画面です。

STRAVA セグメント分析画面
セグメント分析(タイムと心拍数の関係)

ソースはざっと以下の通りです。結構、簡単に書けちゃいます。 Streamlitすばらしいです。簡単なものならサックとできちゃいます。

ポイント1(psycopg2の直接接続 → SQLAlchemy Engineを使用)

はじめはよくわからずに、 psycopg2 を直接使っていましたが、以下の警告が出たので SQLAlchemy Engine を使う様に修正しました。

UserWarning: pandas only supports SQLAlchemy connectable (engine/connection) or database string URI or sqlite3 DBAPI2 connection. Other DBAPI2 objects are not tested. Please consider using SQLAlchemy.
  df = pd.read_sql(query, conn)

ポイント2(Streamlitのキャッシュを利用)

Streamlitはユーザーの操作ごとにスクリプト全体を再実行します。キャッシュを使わないと、データベース接続やデータ取得が毎回実行されて遅くなります。

@st.cache_resource
def get_db_engine():
    return create_engine(database_url)
@st.cache_data(ttl=300)  # 5分間キャッシュ
def get_segments():
    return pd.read_sql(query, engine)
# 初回実行
get_segments()  # DB実行 → 結果をキャッシュ(5分間有効)

# 2回目実行(5分以内)
get_segments()  # キャッシュから取得(高速!)

# 6分後実行
get_segments()  # キャッシュ期限切れ → DB実行 → 再キャッシュ

Streamlitの動作実験

import streamlit as st
import time

print("🔄 スクリプト開始!")  # デバッグ用

st.title("再実行デモ")

# 時間のかかる処理
print("⏳ データベース接続中...")
time.sleep(2)  # 2秒待機(DB接続のシミュレーション)
print("✅ データ取得完了")

name = st.text_input("名前を入力:")
button = st.button("挨拶")

if button:
    st.write(f"こんにちは、{name}さん!")

print("✅ スクリプト終了")

初回ロード時:

🔄 スクリプト開始!
⏳ データベース接続中...
✅ データ取得完了
✅ スクリプト終了

テキスト入力時("hasegawa"と入力してEnterで確定):

🔄 スクリプト開始! ← 再実行!
⏳ データベース接続中...
✅ データ取得完了
✅ スクリプト終了

ボタンクリック時:

🔄 スクリプト開始! ← 再実行!
⏳ データベース接続中...
✅ データ取得完了
✅ スクリプト終了

入力が変わると自動で毎回全体再実行することがわかります。

import streamlit as st
import time
from datetime import datetime

# 実行カウンター(キャッシュなし)
if 'counter' not in st.session_state:
    st.session_state.counter = 0

st.session_state.counter += 1

st.title(f"再実行回数: {st.session_state.counter}")
st.write(f"現在時刻: {datetime.now()}")

# 何かインタラクションを追加
name = st.text_input("何か入力してみて:")
if name:
    st.write(f"入力内容: {name}")

button = st.button("ボタンを押してみて")
if button:
    st.write("ボタンが押されました!")

Strava のセグメント分析のソース

import streamlit as st
import psycopg2
import pandas as pd
from datetime import datetime
import os
from sqlalchemy import create_engine

# ページ設定
st.set_page_config(
    page_title="STRAVA セグメント分析",
    page_icon="🚴",
    layout="wide"
)

# データベース接続設定
@st.cache_resource
def get_db_engine():
    """SQLAlchemyエンジンを取得"""
    db_url = f"postgresql://{os.getenv('DB_USER', 'postgres')}:{os.getenv('DB_PASSWORD', 'postgres')}@{os.getenv('DB_HOST', 'localhost')}:{os.getenv('DB_PORT', '5432')}/{os.getenv('DB_NAME', 'strava')}"
    return create_engine(db_url)

# セグメント一覧を取得
@st.cache_data(ttl=300)  # 5分間キャッシュ
def get_segments():
    """利用可能なセグメント一覧を取得"""
    engine = get_db_engine()
    query = """
    SELECT DISTINCT
        (effort->'segment'->>'id')::bigint AS segment_id,
        effort->'segment'->>'name' AS segment_name,
        COUNT(*) AS effort_count
    FROM activities,
         jsonb_array_elements(data->'segment_efforts') AS effort
    WHERE effort->'segment'->>'name' IS NOT NULL
    GROUP BY segment_id, segment_name
    ORDER BY effort_count DESC, segment_name
    """
    df = pd.read_sql(query, engine)
    return df

# セグメントデータを取得
@st.cache_data(ttl=60)  # 1分間キャッシュ
def get_segment_data(segment_id):
    """指定セグメントのデータを取得"""
    engine = get_db_engine()
    query = """
    SELECT 
        (effort->>'start_date_local')::timestamp AS start_time,
        (effort->>'elapsed_time')::int AS elapsed_time_sec,
        (effort->>'average_heartrate')::numeric AS avg_heartrate,
        (effort->>'average_cadence')::numeric AS avg_cadence,
        (effort->>'distance')::numeric AS distance_m,
        data->>'name' AS activity_name,
        data->>'type' AS activity_type
    FROM activities,
         jsonb_array_elements(data->'segment_efforts') AS effort
    WHERE (effort->'segment'->>'id')::bigint = %(segment_id)s
      AND effort->>'elapsed_time' IS NOT NULL
    ORDER BY start_time DESC
    """
    df = pd.read_sql(query, engine, params={'segment_id': int(segment_id)})
    
    if not df.empty:
        # 時間を分:秒形式に変換
        df['elapsed_time_formatted'] = df['elapsed_time_sec'].apply(
            lambda x: f"{x//60}:{x%60:02d}" if pd.notnull(x) else "N/A"
        )
        # 平均速度を計算 (km/h)
        df['avg_speed_kmh'] = df.apply(
            lambda row: (row['distance_m'] / 1000) / (row['elapsed_time_sec'] / 3600) 
            if pd.notnull(row['distance_m']) and pd.notnull(row['elapsed_time_sec']) and row['elapsed_time_sec'] > 0
            else None, axis=1
        )
        # 日付フォーマット
        df['date'] = df['start_time'].dt.date
    
    return df

# メインアプリケーション
def main():
    st.title("🚴 STRAVA セグメント分析")
    st.markdown("---")
    
    try:
        # セグメント一覧を取得
        segments_df = get_segments()
        
        if segments_df.empty:
            st.warning("セグメントデータが見つかりません。アクティビティデータを確認してください。")
            return
        
        # サイドバーでセグメント選択
        with st.sidebar:
            st.header("セグメント選択")
            
            # セグメント選択ボックス
            segment_options = [
                f"{row['segment_name']} ({row['effort_count']}回)"
                for _, row in segments_df.iterrows()
            ]
            
            selected_option = st.selectbox(
                "分析するセグメントを選択:",
                segment_options
            )
            
            # 選択されたセグメントIDを取得
            selected_idx = segment_options.index(selected_option)
            selected_segment_id = int(segments_df.iloc[selected_idx]['segment_id'])  # int()で変換
            selected_segment_name = segments_df.iloc[selected_idx]['segment_name']
        
        # 選択されたセグメントの情報を表示
        st.header(f"📊 {selected_segment_name}")
        
        # セグメントデータを取得
        segment_data = get_segment_data(selected_segment_id)
        
        if segment_data.empty:
            st.warning("このセグメントのデータが見つかりません。")
            return
        
        # 統計情報を表示
        col1, col2, col3, col4, col5, col6 = st.columns(6)
        
        with col1:
            st.metric("総実施回数", len(segment_data))
        
        with col2:
            best_time = segment_data['elapsed_time_sec'].min()
            st.metric("ベストタイム", f"{best_time//60}:{best_time%60:02d}")
        
        with col3:
            avg_time = segment_data['elapsed_time_sec'].mean()
            st.metric("平均タイム", f"{avg_time//60:.0f}:{avg_time%60:02.0f}")
        
        with col4:
            if segment_data['avg_heartrate'].notna().any():
                avg_hr = segment_data['avg_heartrate'].mean()
                st.metric("平均心拍数", f"{avg_hr:.0f} bpm")
            else:
                st.metric("平均心拍数", "N/A")
        
        with col5:
            if segment_data['avg_cadence'].notna().any():
                avg_cadence = segment_data['avg_cadence'].mean()
                st.metric("平均ケイデンス", f"{avg_cadence:.0f} rpm")
            else:
                st.metric("平均ケイデンス", "N/A")
        
        with col6:
            if segment_data['avg_speed_kmh'].notna().any():
                avg_speed = segment_data['avg_speed_kmh'].mean()
                st.metric("平均速度", f"{avg_speed:.1f} km/h")
            else:
                st.metric("平均速度", "N/A")
        
        st.markdown("---")
        
        # タブで表示を切り替え
        tab1, tab2, tab3 = st.tabs(["📈 グラフ表示", "📋 データ表", "📊 統計"])
        
        with tab1:
            # グラフ表示
            st.subheader("時系列グラフ")
            
            # タイムグラフ
            st.write("**タイムの推移**")
            chart_data = segment_data.set_index('start_time')[['elapsed_time_sec']]
            st.line_chart(chart_data)
            
            # 心拍数グラフ(データがある場合のみ)
            if segment_data['avg_heartrate'].notna().any():
                st.write("**平均心拍数の推移**")
                hr_data = segment_data.dropna(subset=['avg_heartrate']).set_index('start_time')[['avg_heartrate']]
                st.line_chart(hr_data)
            
            # ケイデンスグラフ(データがある場合のみ)
            if segment_data['avg_cadence'].notna().any():
                st.write("**平均ケイデンスの推移**")
                cadence_data = segment_data.dropna(subset=['avg_cadence']).set_index('start_time')[['avg_cadence']]
                st.line_chart(cadence_data)
            
            # 速度グラフ(データがある場合のみ)
            if segment_data['avg_speed_kmh'].notna().any():
                st.write("**平均速度の推移**")
                speed_data = segment_data.dropna(subset=['avg_speed_kmh']).set_index('start_time')[['avg_speed_kmh']]
                st.line_chart(speed_data)
            
            # 散布図(タイム vs 心拍数)
            if segment_data['avg_heartrate'].notna().any():
                st.write("**心拍数 vs タイムの関係**")
                scatter_data = segment_data.dropna(subset=['avg_heartrate'])
                st.scatter_chart(
                    data=scatter_data,
                    x='avg_heartrate',
                    y='elapsed_time_sec'
                )
        
        with tab2:
            # データ表示
            st.subheader("詳細データ")
            
            # 表示用データフレームを準備
            display_df = segment_data.copy()
            display_df = display_df[[
                'date', 'elapsed_time_formatted', 'avg_heartrate', 'avg_cadence', 'avg_speed_kmh',
                'activity_name', 'activity_type'
            ]]
            display_df.columns = [
                '日付', 'タイム', '平均心拍数', '平均ケイデンス', '平均速度(km/h)', 'アクティビティ名', 'タイプ'
            ]
            
            # データフィルター
            st.write("**フィルター:**")
            col1, col2 = st.columns(2)
            
            with col1:
                date_range = st.date_input(
                    "日付範囲",
                    value=(segment_data['date'].min(), segment_data['date'].max()),
                    min_value=segment_data['date'].min(),
                    max_value=segment_data['date'].max()
                )
            
            with col2:
                activity_types = segment_data['activity_type'].unique().tolist()
                selected_types = st.multiselect(
                    "アクティビティタイプ",
                    activity_types,
                    default=activity_types
                )
            
            # フィルターを適用
            if len(date_range) == 2:
                filtered_data = segment_data[
                    (segment_data['date'] >= date_range[0]) &
                    (segment_data['date'] <= date_range[1]) &
                    (segment_data['activity_type'].isin(selected_types))
                ]
                
                filtered_display = filtered_data[[
                    'date', 'elapsed_time_formatted', 'avg_heartrate', 'avg_cadence', 'avg_speed_kmh',
                    'activity_name', 'activity_type'
                ]].copy()
                filtered_display.columns = [
                    '日付', 'タイム', '平均心拍数', '平均ケイデンス', '平均速度(km/h)', 'アクティビティ名', 'タイプ'
                ]
                
                st.dataframe(filtered_display, use_container_width=True)
                
                # CSVダウンロード
                csv = filtered_display.to_csv(index=False).encode('utf-8')
                st.download_button(
                    label="📥 CSVでダウンロード",
                    data=csv,
                    file_name=f"{selected_segment_name}_data.csv",
                    mime="text/csv"
                )
        
        with tab3:
            # 統計情報
            st.subheader("詳細統計")
            
            col1, col2, col3, col4 = st.columns(4)
            
            with col1:
                st.write("**タイム統計**")
                time_stats = segment_data['elapsed_time_sec'].describe()
                for stat, value in time_stats.items():
                    if stat == 'count':
                        st.write(f"- {stat}: {value:.0f}")
                    else:
                        st.write(f"- {stat}: {value//60:.0f}:{value%60:02.0f}")
            
            with col2:
                if segment_data['avg_heartrate'].notna().any():
                    st.write("**心拍数統計**")
                    hr_stats = segment_data['avg_heartrate'].describe()
                    for stat, value in hr_stats.items():
                        if stat == 'count':
                            st.write(f"- {stat}: {value:.0f}")
                        else:
                            st.write(f"- {stat}: {value:.1f} bpm")
            
            with col3:
                if segment_data['avg_cadence'].notna().any():
                    st.write("**ケイデンス統計**")
                    cadence_stats = segment_data['avg_cadence'].describe()
                    for stat, value in cadence_stats.items():
                        if stat == 'count':
                            st.write(f"- {stat}: {value:.0f}")
                        else:
                            st.write(f"- {stat}: {value:.1f} rpm")
            
            with col4:
                if segment_data['avg_speed_kmh'].notna().any():
                    st.write("**速度統計**")
                    speed_stats = segment_data['avg_speed_kmh'].describe()
                    for stat, value in speed_stats.items():
                        if stat == 'count':
                            st.write(f"- {stat}: {value:.0f}")
                        else:
                            st.write(f"- {stat}: {value:.1f} km/h")
            
            # 月別統計
            if len(segment_data) > 1:
                st.write("**月別パフォーマンス**")
                segment_data['year_month'] = segment_data['start_time'].dt.to_period('M')
                monthly_stats = segment_data.groupby('year_month').agg({
                    'elapsed_time_sec': ['count', 'mean', 'min'],
                    'avg_heartrate': 'mean',
                    'avg_cadence': 'mean',
                    'avg_speed_kmh': 'mean'
                }).round(1)
                st.dataframe(monthly_stats)
    
    except Exception as e:
        st.error(f"エラーが発生しました: {str(e)}")
        st.write("データベース接続設定を確認してください。")

if __name__ == "__main__":
    # 環境変数の設定例を表示
    with st.sidebar.expander("🔧 データベース設定"):
        st.write("""
        環境変数で設定してください:
        - DB_HOST (default: localhost)
        - DB_NAME (default: strava) 
        - DB_USER (default: postgres)
        - DB_PASSWORD (default: postgres)
        - DB_PORT (default: 5432)
        """)
    
    main()

次回はこちら

atsushi3.hatenablog.com

AWS学習費用実録|Udemy講座受講後の実際の料金とタイムラグ

AWS初心者がUdemy講座で学習した際の実際の費用と、 料金反映のタイムラグについて詳しくレポートします。 これからAWSを学習する方の参考になれば幸いです。

Udemyの個人向け定額プランを1か月だけ登録してみました。で、AWSを勉強しようと思い 「【2025年最新版】AWS 学習の決定版:手を動かしながら2週間で学ぶ AWS 基本から応用まで | Udemy」を受講してみました。 ハンズオン形式で楽しく受講できました。AWSEC2 RDS ELBとか妙な3文字の意味が分かるようになった程度ですが。。。 外人って3文字すきですよね。以前、かかわっていたプロジェクトで「ECO Server」(エコサーバー)というプログラムがありました。Exposure Control Server の略です。「CPS」はChange Propagation Server なんかかっこいいですよね。

受講が終わって利用したリソースを全部削除して翌日に料金を確認すると3.24ドルで、さらに翌日に確認すると3.45ドルでした。あれ、タイムラグがあるとしてもなんで微妙に0.21ドル増えてるんだ?もしかして「削除もれ」か?

料金(1)
料金(2)

さらに翌日に確認すると3.45ドルから増えていませんでした。ということで、安心しました。利用するサービス毎にタイムラグが違うみたいですね。タイムラグは,2,3日あるみたいですね。

8/29 追記:

Webセミナーとかに参加するとAmazonのクレジットがもらえるんですが、それが反映されているかどうかどこで見てよいかわかりませんでしたが判明しました。 「請求とコスト管理->請求書->サービス別料金」で折りたたまれた、No Region を展開すると値引きされているのがわかります。

クレジット反映

Stravaアプリをクラウド化の実験をする予定です。

atsushi3.hatenablog.com

Strava API とPythonを使ってSegmentの履歴を取り出す(2)-Python PostgreSQL JSONB データ保存

Strava APIには厳しい制限(100リクエスト/15分)があり、 大量のデータ分析には工夫が必要です。 この記事では、PostgreSQLのJSONB機能を活用して API データを効率的に保存・検索する方法を解説します。

前回の投稿(Strava API とPythonを使ってSegmentの履歴を取り出す - atsushi-h’s blog)では、アクテビティ詳細を取得する毎にSTRAVA APIを呼び出していました。 この方法だと、すぐにSTRAVAのアクセス制限(100リクエスト/15分)に引っ掛かってしまいます。なので、アクテビティ詳細をローカルなデータベースに保存しておくことにします。

github.com

どんなデータベースを使えばよいのかわからなかったのでchat-gptに聞いてみました。

STRAVA API を使ってアクテビティ詳細を取得してデータベースに保存したい。 
・PythonからSTRAVA APIを呼び出して アクテビティ詳細をデータベースに保存したい。
・アクテビティ詳細はオブジェクトだと思うが、カラムに変換して登録するのは面倒なので避けたい
・主にセグメント分析をしたい(例:ヒストリカルにタイム、ケイデンス、心拍数を見たい。最高タイムを見たい)
データベースは何を使えばよいかアドバイスしてください。

答えは、以下の通りでした。すばらしい。

選択肢1: PostgreSQL + JSONB
PostgreSQLはJSONB型をサポートしており、Strava APIのレスポンス(JSON)をそのまま保存可能。
SQLでJSONの中をクエリできる(-> / ->> 演算子や jsonb_path_query)。
後から特定のキー(例: average_cadence, max_heartrate)だけ取り出して集計できる。
安定性・拡張性ともに高い。
👉 おすすめNo.1(JSONのまま保存しても後から分析しやすい)

それでは、やってみましょう。以下のテーブルを作ります。

CREATE TABLE activities (
    id BIGINT PRIMARY KEY,
    athlete_id BIGINT,
    data JSONB,
    created_at TIMESTAMP DEFAULT now()
);

集めてきて保存するスクリプトは以下の通りです。前回の投稿(Strava API とPythonを使ってSegmentの履歴を取り出す - atsushi-h’s blog)をちょこっと変えたものです。

import time
from stravalib.client import Client
from stravalib.exc import RateLimitExceeded
import argparse
import os
import psycopg2
import json
import requests
import datetime

def save_activities_to_db(client, db_conn, before=None, after=None, limit=30):
    """
    Strava APIからアクティビティ詳細を取得し、PostgreSQLにJSONのまま保存
    """
    
    # https://stravalib.readthedocs.io/en/latest/reference/api/stravalib.client.Client.get_activities.html#stravalib.client.Client.get_activities
    activities = client.get_activities(before=before, after=after, limit=limit)
    cur = db_conn.cursor()

    for activity in activities:
        if activity.type != 'Ride':
            continue
        
        print(f"Fetched activity Date: {activity.start_date},ID: {activity.id}, Name: {activity.name}")    
        # https://stravalib.readthedocs.io/en/latest/reference/api/stravalib.client.Client.get_activity.html
        detailed_activity = client.get_activity(activity_id = activity.id, include_all_efforts=True)
        detailed_activity_json = detailed_activity.json()

        cur.execute("""
            INSERT INTO activities (id, athlete_id, data)
            VALUES (%s, %s, %s)
            ON CONFLICT (id) DO NOTHING
        """, (
             detailed_activity.id,
             detailed_activity.athlete.id,
             detailed_activity_json)
        )
        db_conn.commit()
        time.sleep(10.0) # A100 15分毎のリクエスト
    cur.close()

def str_to_local_datetime(date_str: str) -> datetime.datetime:
    d = datetime.datetime.strptime(date_str, "%Y-%m-%d").date()
    local_tz = datetime.datetime.now().astimezone().tzinfo
    return datetime.datetime.combine(d, datetime.time.min, tzinfo=local_tz)

if __name__ == '__main__':
    
    parser = argparse.ArgumentParser(description='Save your activities to database')
    parser.add_argument('--before', type=str_to_local_datetime, default=None, help='Result will start with activities whose start date is before specified date.')
    parser.add_argument('--after', type=str_to_local_datetime, default=None, help='Result will start with activities whose start date is after specified value')
    parser.add_argument('--limit', type=int, default=30, help='Maximum number of activities')
    args = parser.parse_args()
    before_utc = args.before.astimezone(datetime.timezone.utc) if args.before else None
    after_utc = args.after.astimezone(datetime.timezone.utc) if args.after else None
    
    client_id = os.getenv('STRAVA_CLIENT_ID')
    client_secret = os.getenv('STRAVA_CLIENT_SECRET')
    access_token = os.getenv('STRAVA_ACCESS_TOKEN')
    refresh_token = os.getenv('STRAVA_REFRESH_TOKEN')
    token_expires_at = int(os.getenv('STRAVA_TOKEN_EXPIRES_AT', '0'))

    client = Client(
        access_token=access_token,
        refresh_token=refresh_token,
        token_expires=token_expires_at
    )
    client.client_id = client_id
    client.client_secret = client_secret

    # DB接続
    db_conn = psycopg2.connect(
        host=os.getenv("PGHOST", "localhost"),
        port=os.getenv("PGPORT", "5432"),
        dbname=os.getenv("PGDATABASE", "strava"),
        user=os.getenv("PGUSER", "postgres"),
        password=os.getenv("PGPASSWORD", "postgres")
    )

    save_activities_to_db(client, db_conn, before_utc, after_utc, args.limit)
    db_conn.close()

できたみたいです。 セグメントの履歴をSQLで抽出してみましょう。 (t->>とかjsonb_array_elementsなど見慣れないですが)

SELECT
    (effort->>'start_date_local')::timestamp AS start_time,
    (effort->>'elapsed_time')::int AS elapsed_time_sec,
    (effort->>'average_heartrate')::numeric AS avg_heartrate
FROM activities,
     jsonb_array_elements(data->'segment_efforts') AS effort
WHERE (effort->'segment'->>'id')::bigint = 10869382
ORDER BY start_time DESC;

実行結果は、以下の通りです。すごいですね。JSON配列をテーブルに見立てて検索できます。驚きです!!

psql -U postgres -d strava -f sample_segment.sql
     start_time      | elapsed_time_sec | avg_heartrate
---------------------+------------------+---------------
 2025-08-17 06:49:55 |              428 |         165.4
 2025-08-15 07:12:22 |              474 |         156.1
 2025-08-09 17:18:22 |              476 |         169.9
 2025-07-24 06:19:50 |              429 |         163.5
 2025-07-23 06:22:27 |              484 |         154.8

PostgreSQLのJSONB機能により、Strava APIから取得した 複雑なJSONデータを効率的に保存・検索できるようになりました。 これで大量のアクティビティデータを蓄積し、 後から自由に分析できる基盤が整いました。

次回は、蓄積したデータをStreamlitで可視化します。 atsushi3.hatenablog.com