記事一覧に戻る

FitbitのWeb APIを実行する方法

はじめに

Fitbit Sense2を購入しました。はじめてのスマートウォッチです。

Fitbitデバイスでは、心拍数や歩数等、収集したデータをWeb APIで取得することが可能です。さっそく使って遊んでみようと思ったら、Web APIの認証がなかなか通らない、、、ドキュメントはとても充実しているのですが、OAuth2.0の認証パターンがImplicit Grant Flowの場合、Authorization Code Grant Flowの場合、PKCEを使う場合、、、などなど、情報量がとにかく多く混乱してしまいました。

何はともあれ、何とか認証を通して、こんな感じで歩数などのアクティビティ情報や、心拍数や血中酸素濃度（SpO2）を取得することが出来ました。

example

毎晩23時に、その日のデータを集計してTwitterに自動投稿しています。良かったらフォローしてください。 ソースはGithubにも上げています。

心拍数やSpO2等は、分刻みのデータが取れるので、アプリより細かい情報を確認することができます。上の画像のようにグラフ化すると、1日の中でどういうタイミングで心拍数が上がっているかも分かります。私はこれで、仕事中は思いのほか緊張していることが確認できました。

取得できるデータはデバイスによって異なりますが、Fitbit Web APIは結構充実しているので、うまく活用すれば色々面白いことが出来ると思います。

この記事の内容

Web APIを実行するために必要な認証用トークン（以下、アクセス・トークン）の取得方法が、情報量が多くて分かりにくいです。加えて、ドキュメントは全て英語で記載されており、日本語ページも無いため、一層ハードルが上がってしまっています。

私も苦戦したので、少しでも同じ苦労をする人が減るように、現時点（2023.3.25）のアクセス・トークンの取得手順について説明します。

加えて、実際にPythonでWeb APIを実行するサンプルも提示し、Fitbit Web APIのクセとか注意点についても触れます。

前提

Web APIの用途について

今回前提にしているWeb APIの使い方は、自分の持っているFitbitデバイスに紐づけたアカウントを使って、自分のデータを取得する使い方です。 例えば、Fitbit利用者向けのモバイルアプリを作ったり、従業員が利用しているFitbitデバイスを会社のアカウントで管理する、といった使い方は想定していません。

プログラムの実装部分について

他にWeb APIの実行をされたことがある方を想定していますが、そんなに難しくないので初めてでも問題ありません。しかし、アクセス・トークンを取得し、Web APIの実行が可能になるところまでを中心に解説するため、プログラムの細かい部分の解説や、Fitbit Web APIの機能の細かい解説まではカバーしておりませんので、ご了承ください。

利用までの流れ

  • Fitbitアカウントでログイン
  • Web APIを利用するアプリ情報の登録
  • アクセス・トークン等の取得
  • Web API利用可能！

アクセス・トークンの取得手順

以下の手順に出てくる例示の画像ですが、テスト用に作成したもので、記事公開後削除しています。いずれも機密な情報になるため、ご自身のものは他人に見られないように管理してください。

1.fitbitアカウントを作成

https://accounts.fitbit.com/signupでアカウントを作成します。Fitbitデバイスの初期設定でアカウント作成が必要なので、ほとんどの場合既に作成済みかと思います。そのアカウントでログインしてください。

2.Web APIを利用するアプリを登録

https://dev.fitbit.com/appsでアプリを登録します。後からアプリ情報の確認が出来るので、ブックマークしておくと良いです。

右上のRegister a new appをクリックします。

app-manage-page

アプリの情報を入力していきます。URLを入力する欄がたくさんありますが、今回のようにrequestsを使ってAPIを実行するだけのプログラムなら、Redirect URL以外のURL項目はhttp://localhostで問題ありません。

app-register
  • Application Name: 何でもOK！
  • Description: 10文字以上なら何でもOK!
  • OAuth2.0 Application Type: 個人データを取得するのでPersonalを選択
  • Redirect URL: 初回のアクセス・トークンはここに送られます。http://localhost:8000にしておきます。
  • Default Access Type: APIでデータの書込みをしたい場合はRead & Writeを選択。私はRead Onlyにしています。

I have read and agree to the terms of serviceをチェックしてピンクのRegisterボタンを押すと、以下のページが表示されます。

app-setting

登録された情報が表示されます。Client IDとClient Secretが発行されています。この画面は後からいつでも確認できるので、控えなくて大丈夫です。

左下のOAuth2.0 Tutorialをクリックして、アクセス・トークン等の発行を行います。

3.アクセス・トークンの発行(OAuth2.0 tutorial)

ここから、チュートリアルのWeb画面の操作に沿ってアクセストークンを発行します。様々な情報が表示されますが、メモっておく必要があるのは最後に出てくるAccess TokenとRefresh Tokenの２つのみです。

■ App Settings

Client IDは自動で表示されているはずなのでそのままにしておきます。Application TypeはClient (for client-side access)を選択します。

app-type

説明書きでは、「アプリの登録時にPersonalを選択している場合はどちらでも良い」と書いてあるので、どちらでも問題はないと思いますが、今回はClientで説明を進めます。ServerとClientで、エンドポイントによって認証の仕方が異なる場合があるのでご留意下さい。

■ Step1: Generate PKCE and State Values

アクセス・トークンを発行するために必要な情報を生成します。緑色のGenerateボタンを2か所押します。

文字列が2つ生成されますが、一時的なものなので控えておく必要はありません。

key-generate-guide-page

■ Step 2: Display Authorization Page

APIから取得を認めない項目からチェックを外します。私はLocationとSocialを外しました。

そして、Authorization URLをクリックします。

authorization-guide-page

自動で認可ページが開きます。先ほど除外した項目が無いことを確認し、「すべて許可する」をチェックし、ピンクの「許可」ボタンを押します。

authorize-page

すると、Redirect URLで指定したhttp://localhost:8000/が開きます。「このサイトにアクセス出来ません」と表示されていると思いますが、閉じずにURLをコピーしてください。

redirect-page

■ Step 3: Handle the Redirect

下の画像の赤枠のところに、先ほどのURLを貼り付けます。すると、URLからAuthorization CodeとStateを抽出して表示してくれます。いずれの値も、控えておく必要はありません。

authorization-code

■ Step 4: Get Tokens

上記のURLを元にリクエストを自動生成してくれています。緑色のボタンを押します。下のテキストエリアに、エラーが表示されていなければOKです。

request

このすぐ下に、レスポンスを見やすく表示している箇所があります。Access TokenとRefresh Tokenは必ず控えてください。

Access TokenはAPIを実行するのに使います。このTokenは8時間の寿命なので、期限が切れた場合はRefresh Tokenを使ってAccess Tokenの再発行を行います。

tokens

■ チュートリアルの残りについて

後は、profile取得のAPIが通るかのテストと、Refresh Tokenを使って再発行をするチュートリアルなので、割愛します。Profileはそもそも権限を与えていない場合はエラーになるので、試す場合はご留意ください。また、Refresh Tokenを使って再発行をすると、Access TokenとRefresh Tokenともに更新されるので、更新後の値を控えておいてください。

PythonでWeb APIを実行

1.バージョンや外部パッケージについて

PythonのバージョンはPython 3.9.6です。外部パッケージはhttpリクエストを実行するためにrequests 2.26.0を利用します。極端に古いバージョンでなければ問題ないはずです。

2.Web APIについて

Web APIのドキュメントページはこちらです。

特記事項

  • 1時間あたり150リクエストが上限です。
  • アクセス・トークンは8時間で失効します。失効時はリフレッシュ・トークンを使って再取得が必要です。

実行するエンドポイント

今回は「Get Heart Rate Time Series by Date（日付指定の心拍数）」を例にします。 また、アクセス・トークンは8時間で失効するため、「Refresh Token（トークン再取得）」も実装します。

それぞれのエンドポイントのドキュメントです：

3.実行前の準備

APIを実行する前に、アクセス・トークンの読取りや、各エンドポイントに設定するヘッダの取得部分を実装します。

アクセス・トークン等の読取りについて

認証情報はJSONファイルから読み取る方式にします。以下の形でtest_conf.jsonの名前で保存しておきます。

{
  "client_id": "client_id",
  "access_token": "access_token",
  "refresh_token": "refresh_token"
}

いずれも取得した値で埋めてください。client_idはアプリの登録ページで、登録したアプリをクリックすると確認できます。

読み取る部分をPythonで実装します。グローバル変数にしてしまいます。

with open("./test_conf.json", "r", encoding="utf-8") as f:
    conf = json.load(f)

認証の肝の部分です。httpのヘッダに、authorization:Bearer アクセス・トークンと設定します。Bearerの後に半角スペースが必要です。

私もまだ全てのエンドポイントを確認できてはいませんが、アプリの種類がServerの場合はBasic認証が必要なものもありますが、clientの場合は基本はBearer認証です。

エンドポイントごとにヘッダを書いてもいいのですが、手間なので関数化します。

def bearer_header():
    """Bearer認証用ヘッダ
    Returns:
        dict: {"Authorization":"Bearer " + your-access-token}
    """
    return {"Authorization": "Bearer " + conf["access_token"]}

4.Refresh Token（アクセス・トークンの再取得）

上述のとおり、アクセス・トークンは8時間で失効します。失効する度、Webページで再発行するのは手間ですので、リフレッシュ・トークンを使ってアクセス・トークンを再取得する処理を作ります。

エンドポイント等

refresh-token
  • メソッド: POST
  • エンドポイント: https://api.fitbit.com/oauth2/token
  • 認証: Bearer

Body部に設定するデータは以下の通りです。

grant_type: "refresh_token"
refresh_token: あなたのリフレッシュ・トークン
client_id: あなたのclient_id

ちなみに、Content-Typeはjsonではなくx-www-form-urlencodedです。

Pythonで実装

再取得の部分を実装します。認証が切れていると、以下のエラーメッセージが返ってきます。

{
  'errors': [{
  'errorType': 'expired_token',
  'message': 'Access token expired: 旧アクセス・トークン
            Visit https://dev.fitbit.com/docs/oauth2 for more 
            information on the Fitbit Web API authorization process.'
}],
　'success': False
}

「リクエストを実行し、このエラーがあったら再取得処理を行い、再取得したトークンで再度リクエストを投げる」処理にします。

今回実行するのは心拍数取得だけですが、今後SpO2や他のデータ取得の処理を追加していった時に、エンドポイントごとにこの処理の記載をするのは面倒です。なので、どのエンドポイントでも共通して呼び出すように作ります。

アクセス・トークンを再発行すると、リフレッシュ・トークンも再発行されます。両方控えて必要があるので、取得後はconf.jsonを更新する処理も加えます。

見にくいので、ちぎって記載します。

再取得の処理
# Sessionを使わず、import requestsでもOKです
from requests import Session
from pprint import pprint
import json

# requests.Session初期化
session = Session()

# 認証ファイルの読み取り
with open("./test_conf.json", "r", encoding="utf-8") as f:
    conf = json.load(f)

def bearer_header():
    """Bearer認証用ヘッダ
    Returns:
        dict: {"Authorization":"Bearer " + your-access-token}
    """
    return {"Authorization": "Bearer " + conf["access_token"]}

def refresh():
    """
    access_tokenを再取得し、conf.jsonを更新する。
    refresh_tokenは再取得に必要なので重要。
    access_tokenが失効している時に呼ぶ。
    """

    url = "https://api.fitbit.com/oauth2/token"

    # client typeなのでclient_idが必要
    params = {
        "grant_type": "refresh_token",
        "refresh_token": conf["refresh_token"],
        "client_id": conf["client_id"],
    }

    # POST実行。 Body部はapplication/x-www-form-urlencoded。
    # requestsなら何も指定しなくてOK。
    res = session.post(url, data=params)
    
    # responseをパース
    res_data = res.json()

    # errorあり
    if res_data.get("errors") is not None:
        emsg = res_data["errors"][0]
        print(emsg)
        return


    # errorなし。confを更新し、ファイルを更新
    conf["access_token"] = res_data["access_token"]
    conf["refresh_token"] = res_data["refresh_token"]
    with open("./test_conf.json", "w", encoding="utf-8") as f:
        json.dump(conf, f, indent=2)
失効チェック処理

アクセス・トークン失効のエラーメッセージがあるかをチェックします。仮引数はAPIのレスポンスをパースしたものを想定しています。

def is_expired(resObj) -> bool:
    """
    Responseから、accesss-tokenが失効しているかチェックする。
    失効ならTrue、失効していなければFalse。Fitbit APIでは8時間が寿命。
    Args:
        reqObj (_type_): response.json()したもの

    Returns:
        boolean: 失効ならTrue、失効していなければFalse
    """

    errors = resObj.get("errors")

    # エラーなし。
    if errors is None:
        return False

    # エラーあり
    for err in errors:
        etype = err.get("errorType")
        if (etype is None):
            continue
        if etype == "expired_token":
            pprint("TOKEN_EXPIRED!!!")
            return True

    # 失効していないのでFalse。エラーありだが、ここでは制御しない。
    return False
リクエスト実行処理（共通）

エンドポイントごとに再発行処理を書くのは面倒なので、リクエストを処理する関数を外だしします。

  • method: requestsのgetやpostメソッド
  • url: APIのエンドポイント
  • kw: requestsのgetやpostメソッドに渡すパラメタ（params={},headers={}等）を想定

methodを実行し、アクセス・トークンが失効していたら再取得を行います。再取得した場合、更新したトークンを使って再度methodを実行し、test_conf.jsonを更新します。

def request(method, url, **kw):
    """
    sessionを通してリクエストを実行する関数。
    アクセストークンが8Hで失効するため、失効時は再取得し、
    リクエストを再実行する。
    レスポンスはparseしないので、呼ぶ側で.json()等すること。

    Args:
        method (function): session.get,session.post...等
        url (str): エンドポイント
        **kw: headers={},params={}を想定

    Returns:
        session.Response: レスポンス
    """

    # パラメタで受け取った関数を実行し、jsonでパース
    res = method(url, **kw)
    res_data = res.json()

    if is_expired(res_data):
        # 失効していしている場合、トークンを更新する
        refresh()
        # headersに設定されているトークンも
        # 新しい内容に更新して、methodを再実行
        kw["headers"] = bearer_header()
        res = method(url, **kw)
    # parseせずにレスポンスを返す
    return res

再取得の部分はこれで以上です！次は実際に心拍数を取得します。

5.心拍数取得

エンドポイント等

  • ドメイン: https://api.fitbit.com
  • ディレクトリ: /1/user/[user-id]/activities/heart/date/[date]/[period].json
request-example

英語で記載されていますが、ディレクトリの可変部分（[]書きのuser-id、date、period）は留意が必要です。

  • user-id: デバイスに紐づけたアカウントでアプリ登録しているので、"-"を設定します。
  • date: 抽出する日付です。"today"、もしくはyyyy-mm-ddの形式で指定します。
  • period: 指定した日付から、何日前分のデータを取得するか指定します。1d,7d,30d,1w,1mのいずれかで指定します。

「今日1日分の心拍データを取得する」場合、エンドポイントはhttps://api.fitbit.com/1/user/-/activities/heart/date/today/1d.jsonになります。

Pythonで実装

dateとperiodを仮引数で指定します。デフォルトが、「今日1日分の心拍データ」になるようにしています。

def heartbeat(date: str = "today", period: str = "1d"):
    """心拍数を取得しレスポンスを返す。パースはしない。

    Args:
        date (str, optional): 取得する日付。yyyy-mm-ddで指定も可能。Defaults to "today".
        period (str, optional): 取得する範囲。1d,7d,30d,1w,1m。 Defaults to "1d".

    Returns:
        session.Response: レスポンス
    """
    # パラメタを埋め込んでエンドポイント生成
    url = f"https://api.fitbit.com/1/user/-/activities/heart/date/{date}/{period}.json"
    
    # 認証ヘッダ取得
    headers = bearer_header()
    
    # 自作のリクエスト関数に渡す
    res = request(session.get, url, headers=headers)
    
    return res

長くなってしまうのでパーツごとに記載していきました。後は全部繋げて、以下のように実行すればOKです。

# レスポンスをパースしていないので、呼んだ後にパース
res = heartbeat()
data = res.json()
pprint(data)

レスポンスについて

レスポンスは以下のような形になります。

{
  "activities-heart": [
    {
      "dateTime": "2023-03-20",
      "value": {
        "customHeartRateZones": [],
        "heartRateZones": [
          {
            "caloriesOut": 945.4251199999999,
            "max": 114,
            "min": 30,
            "minutes": 1409,
            "name": "Out of Range"
          },
          {
            "caloriesOut": 207.3862400000001,
            "max": 138,
            "min": 114,
            "minutes": 31,
            "name": "Fat Burn"
          }
        ],
        "restingHeartRate": 68
      }
    }
  ],
  "activities-heart-intraday": {
    "dataset": [
      {
        "time": "00:24:00",
        "value": 62
      },
      {
        "time": "00:25:00",
        "value": 65
      },
      {
        "time": "00:26:00",
        "value": 71
      }
    ],
    "datasetInterval": 1,
    "datasetType": "minute"
  }
}

activities-heart-intradayが、分単位の詳細情報です。 心拍数以外にも、BreathingRateやSpO2等、他のエンドポイントでもこのintradayの取得が可能なデータもあります。

アプリを登録するときに、Personalタイプで登録したと思いますが、Personalの場合は普通にリクエストを投げれば、intradayのデータが取れます。 Serverタイプを選んでいる場合は、intraday専用のエンドポイントを実行する必要があり、 その前に取得する人の許諾を取らなければならないようですね。

なので、Personalタイプを選んでいる限り、intradayのエンドポイントを使う必要性はないと思います。

ドキュメントから明確に読み取れませんが、リクエストのperiodに1dを指定した場合は抽出されますが、1wとか長い期間を指定すると省略されるようです。

最後に

Fitbit Web APIのアクセス・トークン等の取得方法と、Web APIでアクセス・トークンを更新する方法・心拍数取得方法を解説しました。 私はトークンを取得するところに時間がかかってしまったので、参考になると嬉しいです。

スマートウォッチは初めてですが、身体のデータをデバイスで収集するのは近未来感があって良いですね。毎日のデータが貯まっていくので、 健康診断のようにある時点の状態を見るだけでなく、自分の体の状態の推移を確認出来るのは有益だと感じます。

Web APIだけでなく、Device APIのように、Fitbitアプリ開発用のAPIも公開されており、こちらも楽しそうに見えます。しかし、非常に 残念ですが、最新のSense2 / Versa4では、アプリの開発自体ができなくなっています。これは何とかして欲しいですね！

【コピペ用】スクリプト

Pythonのスクリプトを全てマージしたものです。


# Sessionを使わず、import requestsでもOKです
from requests import Session
from pprint import pprint
import json

session = Session()

with open("./test_conf.json", "r", encoding="utf-8") as f:
    conf = json.load(f)


def bearer_header():
    """Bearer認証用ヘッダ
    Returns:
        dict: {"Authorization":"Bearer " + your-access-token}
    """
    return {"Authorization": "Bearer " + conf["access_token"]}


def refresh():
    """
    access_tokenを再取得し、conf.jsonを更新する。
    refresh_tokenは再取得に必要なので重要。
    is_expiredがTrueの時のみ呼ぶ。
    False時に呼んでも一式更新されるので、実害はない。
    """

    url = "https://api.fitbit.com/oauth2/token"

    # client typeなのでclient_idが必要
    params = {
        "grant_type": "refresh_token",
        "refresh_token": conf["refresh_token"],
        "client_id": conf["client_id"],
    }

    # POST実行。 Body部はapplication/x-www-form-urlencoded。requestsならContent-Type不要。
    res = session.post(url, data=params)

    # responseをパース
    res_data = res.json()

    # errorあり
    if res_data.get("errors") is not None:
        emsg = res_data["errors"][0]
        print(emsg)
        return

    # errorなし。confを更新し、ファイルを更新
    conf["access_token"] = res_data["access_token"]
    conf["refresh_token"] = res_data["refresh_token"]
    with open("./test_conf.json", "w", encoding="utf-8") as f:
        json.dump(conf, f, indent=2)


def is_expired(resObj) -> bool:
    """
    Responseから、accesss-tokenが失効しているかチェックする。
    失効ならTrue、失効していなければFalse。Fitbit APIでは8時間が寿命。
    Args:
        reqObj (_type_): response.json()したもの

    Returns:
        boolean: 失効ならTrue、失効していなければFalse
    """

    errors = resObj.get("errors")

    # エラーなし。
    if errors is None:
        return False

    # エラーあり
    for err in errors:
        etype = err.get("errorType")
        if (etype is None):
            continue
        if etype == "expired_token":
            pprint("TOKEN_EXPIRED!!!")
            return True

    # 失効していないのでFalse。エラーありだが、ここでは制御しない。
    return False


def request(method, url, **kw):
    """
    sessionを通してリクエストを実行する関数。
    アクセストークンが8Hで失効するため、失効時は再取得し、
    リクエストを再実行する。
    レスポンスはパースしないので、呼ぶ側で.json()なり.text()なりすること。

    Args:
        method (function): session.get,session.post...等
        url (str): エンドポイント
        **kw: headers={},params={}を想定

    Returns:
        session.Response: レスポンス
    """

    # パラメタで受け取った関数を実行し、jsonでパース
    res = method(url, **kw)
    res_data = res.json()

    if is_expired(res_data):
        # 失効していしている場合、トークンを更新する
        refresh()
        # headersに設定されているトークンも
        # 新しい内容に更新して、methodを再実行
        kw["headers"] = bearer_header()
        res = method(url, **kw)
    # parseしていないほうを返す
    return res


def heartbeat(date: str = "today", period: str = "1d"):
    """心拍数を取得しレスポンスを返す。パースはしない。

    Args:
        date (str, optional): 取得する日付。yyyy-mm-ddで指定も可能。Defaults to "today".
        period (str, optional): 取得する範囲。1d,7d,30d,1w,1m。 Defaults to "1d".

    Returns:
        session.Response: レスポンス
    """
    # パラメタを埋め込んでエンドポイント生成
    url = f"https://api.fitbit.com/1/user/-/activities/heart/date/{date}/{period}.json"

    # 認証ヘッダ取得
    headers = bearer_header()

    # 自作のリクエスト関数に渡す
    res = request(session.get, url, headers=headers)

    return res


# 実行例
res = heartbeat()
data = res.json()
pprint(data)
記事一覧に戻る
全力君より、愛を込めて。
2024 全力君
ABOUT
削り削って研ぎ澄ませ、最先端の感性を。
そしてLANケーブルに、注ぎ込め！
Feel free to follow me.
Site Policy