【AITRIOS開発】WebPageを開発し、WebAPI経由で、AITRIOSに準備されているConsole Rest APIと連動を完成

皆さんこんにちは、痛快技術株式会社の上田哲弘です。
Console Rest APIに接続してデバイスを操作し、HTTP Serverへデータを転送することができました。

前回の記事はコチラ

今回は、当初開発をおこないたかったAITRIOS ConsoleのUIを使うのではなく、自作のWEB画面からAITIRIOS（アイトリオス）でデプロイしたデバイスの起動を簡単に行いたいを開発していきます！

構成イメージ

完成イメージ

流れ

1. WebPageにて推論開始／推論停止というボタンをクリックするとWebAPIを飛び出す

2. WebAPI経由で、AITRIOSに準備されているConsole Rest APIへPOST

3. 推論開始のAPIが実行

4. ローカル環境のIPアドレス先のMeta/Imageに保存される

ファイル構成

|-server
| |_ webapp.py # HTTP Serverで保存をするファイル
| |_ image # 画像ファイル保存用ディレクトリ
| |_ meta  # メタデータ保存用ディレクトリ
|
|-web
| |_ api_services
| |   |_ service.py #推論開始・停止するために外部 API に対して POST リクエストを送信します。
| |_ templates
| |   |_ index.html
| |_ access_token.py #外部 API へのアクセスに必要なトークンを取得し、管理します。
| |_ app.py
|_ .env

前回に作成をしました HTTP Sereverで保存をする構成＋WebアプリケーションをWebフォルダにて追加する構成を考えました。

開発するポイント

1. Console Rest APIへの接続する準備
   前回、手動にておこなったConsole Rest APIへ接続する際に必要なアクセストークンを取得し、管理することが大きく大切になってきます。

2. APIに対してリクエストを操作行う関数の準備
   APIの開始・停止を操作する関数の開発が必要です。

WEB構成の開発

こちらについて、今回はPythonをベースにFastAPI を使用して API を呼び出す簡単な Web アプリケーションの構築を行います。
※下記サンプルコードは参考用であり、動作を保証するものではありません。

1.環境変数について

このファイルには以下のような形式でIDとシークレットが保存されています。

CLIENT_ID=your_client_id_here
CLIENT_SECRET=your_client_secret_here
DEVICE_ID=your_device_id

2. app.pyについて

from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse

from api_services.service import post_collectstart_to_api, post_collectstop_to_api
from access_token import load_access_token


# FastAPIのインスタンスを作成
app = FastAPI()

# Jinja2テンプレートの設定
templates = Jinja2Templates(directory="templates")

# ルートエンドポイントにアクセスした際に HTML を返す
@app.get("/", response_class=HTMLResponse)
async def read_root(request: Request):
    # 毎回トークンをロードして使用
    token = load_access_token()
    
    return templates.TemplateResponse("index.html", {
        "request": request, 
        "message": "Click the button to start the API call",
        "token": token
    })

# ボタンがクリックされた際に呼び出されるエンドポイント
@app.post("/start-api")
async def start_api():
     # 毎回トークンをロードして使用
    token = load_access_token()
    
    # APIを呼び出す
    api_response = post_collectstart_to_api(token)
    
    # APIレスポンスを返す
    return api_response

# ボタンがクリックされた際に呼び出されるエンドポイント
@app.post("/stop-api")
async def start_api():
    # 毎回トークンをロードして使用
    token = load_access_token()
    
    # APIを呼び出す
    api_response = post_collectstop_to_api(token)
    
    # APIレスポンスを返す
    return api_response

3. インポートする関数

■関数
今回、開発をする関数をインポートしていきます。

APIの開始・停止を操作するための関数

 from api_services.service import post_collectstart_to_api, post_collectstop_to_api 

APIにアクセスする際に必要なアクセストークンをロードする関数

from access_token import load_access_token

4. APIにアクセスする際に必要なアクセストークンをロードする関数

access_token.py

import requests
import base64
import json
import time
from dotenv import load_dotenv
import os

# .envファイルの内容を読み込む
load_dotenv()

# デバッグ用にプリント文を追加
print("スクリプトが開始されました")

# クライアントIDとシークレットをBase64エンコード
client_id = os.getenv("CLIENT_ID") #your_client_id_here
client_secret = os.getenv("CLIENT_SECRET") #your_client_secret_here

client_credentials = f"{client_id}:{client_secret}"
encoded_credentials = base64.b64encode(client_credentials.encode("utf-8")).decode("utf-8")

def get_access_token():
    url = "https://auth.aitrios.sony-semicon.com/oauth2/default/v1/token"
    
    # デバッグ用にプリント文を追加
    print("アクセストークンを取得中...")

    # ヘッダーの設定
    headers = {
        "accept": "application/json",
        "authorization": f"Basic {encoded_credentials}",
        "cache-control": "no-cache",
        "content-type": "application/x-www-form-urlencoded"
    }
    
    # データ（POSTリクエストのパラメータ）
    data = {
        "grant_type": "client_credentials",
        "scope": "system"
    }
    
    # POSTリクエストを送信してアクセストークンを取得
    response = requests.post(url, headers=headers, data=data)
    
    # ステータスコードが200（成功）ならトークンを保存
    if response.status_code == 200:
        print("アクセストークン取得に成功しました")
        token_data = response.json()
        access_token = token_data["access_token"]
        expires_in = token_data["expires_in"]
        
        # 現在の時刻 + 有効期限を保存（UNIXタイムスタンプ）
        expiration_time = time.time() + expires_in
        
        # アクセストークンと有効期限をテキストファイルに保存
        with open("access_token.txt", "w") as token_file:
            json.dump({"access_token": access_token, "expiration_time": expiration_time}, token_file)
        
        print("アクセストークンを保存しました")
        return access_token  # 新しいアクセストークンを返す
    else:
        print(f"トークンの取得に失敗しました。ステータスコード: {response.status_code}")
        print(response.text)

def load_access_token():
    try:
        with open("access_token.txt", "r") as token_file:
            token_data = json.load(token_file)

        # 有効期限をチェック
        if time.time() < token_data["expiration_time"]:
            return token_data["access_token"]
        else:
            print("アクセストークンの有効期限が切れています。再取得してください。")
            return get_access_token()  # トークンの有効期限が切れていたら、新しいトークンを取得して返す

    except FileNotFoundError:
        print("アクセストークンファイルが見つかりません。再取得が必要です。")
        return get_access_token()  # ファイルがない場合も新しいトークンを取得して返す

概要について
この access_token.py スクリプトは、Sony AITRIOS の OAuth 2.0 認証サーバーからアクセストークンを取得し、そのアクセストークンをローカルファイルに保存することで、再度必要なときに使い回すことができるようにするものです。また、アクセストークンの有効期限を確認し、期限が切れていれば自動的に新しいトークンを取得する機能も実装されています。

1.モジュールのインポート

import requests
import base64
import json
import time
from dotenv import load_dotenv
import os

このスクリプトでは、以下のモジュールを使用しています：

requests: HTTPリクエストを行うためのライブラリです。POSTリクエストを使用して、認証サーバーからトークンを取得します。
base64: クライアントIDとシークレットをBase64形式にエンコードするために使用します。
json: アクセストークンをファイルに保存する際に、JSON形式でデータを管理します。
time: アクセストークンの有効期限を管理するために現在の時刻を取得し、トークンの有効期限を計算します。
dotenv: .envファイルから環境変数をロードするために使用します。
os: 環境変数を読み込むために使用します。

2.環境変数の読み込み

load_dotenv()

.envファイルからクライアントIDとシークレットを読み込みます。

3.クライアントIDとシークレットをBase64エンコード

client_id = os.getenv("CLIENT_ID")
client_secret = os.getenv("CLIENT_SECRET")

client_credentials = f"{client_id}:{client_secret}"
encoded_credentials = base64.b64encode(client_credentials.encode("utf-8")).decode("utf-8")

認証サーバーに対してBasic認証を行うために、client_id と client_secret を Base64 形式にエンコードしています。

4.アクセストークンを取得する関数: get_access_token()

def get_access_token():
    url = "https://auth.aitrios.sony-semicon.com/oauth2/default/v1/token"
    
    headers = {
        "accept": "application/json",
        "authorization": f"Basic {encoded_credentials}",
        "cache-control": "no-cache",
        "content-type": "application/x-www-form-urlencoded"
    }
    
    data = {
        "grant_type": "client_credentials",
        "scope": "system"
    }
    
    response = requests.post(url, headers=headers, data=data)
    
    if response.status_code == 200:
        token_data = response.json()
        access_token = token_data["access_token"]
        expires_in = token_data["expires_in"]
        expiration_time = time.time() + expires_in
        
        with open("access_token.txt", "w") as token_file:
            json.dump({"access_token": access_token, "expiration_time": expiration_time}, token_file)
        
        return access_token
    else:
        print(f"トークンの取得に失敗しました。ステータスコード: {response.status_code}")
        print(response.text)

get_access_token() 関数では、認証サーバーに対してPOSTリクエストを送信し、アクセストークンを取得します。

url: Sony AITRIOS の認証サーバーのURL。
headers: 認証サーバーにリクエストを送信する際に必要なヘッダー情報。特に、authorization ヘッダーにはBase64エンコードされたクライアント情報が含まれます。
data: POSTリクエストのパラメータ。OAuth 2.0 の client_credentials フローを使用しています。

成功した場合 (status_code == 200)、取得したアクセストークンとその有効期限をローカルファイル access_token.txt に保存します。

5.アクセストークンを読み込む関数: load_access_token()

def load_access_token():
    try:
        with open("access_token.txt", "r") as token_file:
            token_data = json.load(token_file)

        if time.time() < token_data["expiration_time"]:
            return token_data["access_token"]
        else:
            print("アクセストークンの有効期限が切れています。再取得してください。")
            return get_access_token()

    except FileNotFoundError:
        print("アクセストークンファイルが見つかりません。再取得が必要です。")
        return get_access_token()

load_access_token() 関数は、アクセストークンがすでに保存されている場合、そのファイルを読み込んで有効期限を確認します。

ファイルが存在し、トークンがまだ有効であれば、そのトークンを返します。
有効期限が切れている場合やファイルが存在しない場合には、新しいトークンを取得し直します (get_access_token()を呼び出します)。

5. APIの開始・停止を操作するための関数

service.py

import os
import requests # type: ignore
from dotenv import load_dotenv

def post_collectstart_to_api(token):
    device_id = os.getenv('DEVICE_ID')
    url = f"https://console.aitrios.sony-semicon.com/api/v1/devices/{device_id}/inferenceresults/collectstart"
    
    # 認証のためのヘッダー
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    print(f"APIリクエストを送信中: {url}")
    print(f"使用するトークン: {token}")
    
    try:
        # POSTリクエストを送信
        response = requests.post(url, headers=headers, timeout=10)  # タイムアウトを10秒に設定

        print(f"レスポンスのステータスコード: {response.status_code}")

        # レスポンスの確認
        response.raise_for_status()  # ステータスコードが200番台以外なら例外を発生させる

        # JSONレスポンスを返す
        try:
            return response.json()
        except ValueError:
            return {"error": "Invalid JSON response"}
        
    except requests.exceptions.RequestException as e:
        return {"error": f"Request failed: {e}"}

def post_collectstop_to_api(token):
    device_id = os.getenv('DEVICE_ID')
    url = f"https://console.aitrios.sony-semicon.com/api/v1/devices/{device_id}/inferenceresults/collectstop"
    
    # 認証のためのヘッダー
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    try:
        # POSTリクエストを送信
        response = requests.post(url, headers=headers, timeout=10)  # タイムアウトを10秒に設定

        # レスポンスの確認
        response.raise_for_status()  # ステータスコードが200番台以外なら例外を発生させる

        # JSONレスポンスを返す
        try:
            return response.json()
        except ValueError:
            return {"error": "Invalid JSON response"}
        
    except requests.exceptions.RequestException as e:
        return {"error": f"Request failed: {e}"}

1.モジュールのインポート

import os
import requests
from dotenv import load_dotenv

os: 環境変数 (DEVICE_ID など) を読み込むために使用します。
requests: HTTPリクエストを実行するためのライブラリ。APIに対する POST リクエストを送信します。
dotenv: .env ファイルから環境変数をロードするためのライブラリ。

2.デバイスに対して推論結果の収集を開始するAPIエンドポイントにPOSTリクエスト関数

def post_collectstart_to_api(token):
    device_id = os.getenv('DEVICE_ID')
    url = f"https://console.aitrios.sony-semicon.com/api/v1/devices/{device_id}/inferenceresults/collectstart"
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    print(f"APIリクエストを送信中: {url}")
    print(f"使用するトークン: {token}")
    
    try:
        response = requests.post(url, headers=headers, timeout=10)
        print(f"レスポンスのステータスコード: {response.status_code}")
        response.raise_for_status()

        try:
            return response.json()
        except ValueError:
            return {"error": "Invalid JSON response"}
        
    except requests.exceptions.RequestException as e:
        return {"error": f"Request failed: {e}"}

環境変数の取得: device_id は .env ファイルから読み込みます。デバイスIDは API 呼び出しに必須で、どのデバイスに対してアクションを実行するかを指定します。
API エンドポイントの作成: url 変数に、device_id を含む API のエンドポイント URL を生成します。この URL は inferenceresults/collectstart というエンドポイントで、デバイスの推論結果の収集を開始するためのものです。
ヘッダーの設定: Authorization ヘッダーには、Bearer トークンとしてアクセストークンを渡します。Content-Type は application/json を指定しており、JSONデータを扱うことを示しています。
POST リクエストの送信: requests.post() を使って API にリクエストを送信し、タイムアウトは 10 秒に設定されています。
レスポンスの確認と処理:ステータスコードが200番台以外の場合、raise_for_status() によって例外が発生します。正常なレスポンスが返ってきた場合、レスポンスを JSON としてパースし、関数の返り値として返します。JSON が無効な場合には、エラーメッセージ {"error": "Invalid JSON response"} を返します。

3.推論結果の収集を停止するAPIエンドポイントにPOSTリクエスト関数

def post_collectstop_to_api(token):
    device_id = os.getenv('DEVICE_ID')
    url = f"https://console.aitrios.sony-semicon.com/api/v1/devices/{device_id}/inferenceresults/collectstop"
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    try:
        response = requests.post(url, headers=headers, timeout=10)
        response.raise_for_status()

        try:
            return response.json()
        except ValueError:
            return {"error": "Invalid JSON response"}
        
    except requests.exceptions.RequestException as e:
        return {"error": f"Request failed: {e}"}

この関数の動作は post_collectstart_to_api() とほぼ同じです。ただし、こちらは推論結果の収集を停止するために collectstop エンドポイントにリクエストを送信します。

HTMLについて

概要
HTMLファイルは、FastAPIアプリケーションと連携して、JavaScriptを使ってボタンのクリックイベントをトリガーにAPI呼び出しを行うページの構成となっています。ページが表示されると、ユーザーがボタンをクリックして、推論開始または 推論停止のリクエストをサーバーに送信し、その結果をページ上に表示する仕組みです。

index.html

<!DOCTYPE html>
<html>
<head>
    <title>Web Page</title>
    <!-- JavaScriptを使用してボタンがクリックされたときにAPIを呼び出す -->
    <script>
        // ページ読み込み時にイベントリスナーを設定
        document.addEventListener("DOMContentLoaded", function() {
            document.getElementById("apiButton_InferenceExecution").addEventListener("click", async function() {
                const response = await fetch("/start-api", { method: "POST" });
                const data = await response.json();
                document.getElementById("apiResponse").innerText = JSON.stringify(data, null, 2);
            });

            document.getElementById("apiButton_InferenceStopped").addEventListener("click", async function() {
                const response = await fetch("/stop-api", { method: "POST" });
                const data = await response.json();
                document.getElementById("apiResponse").innerText = JSON.stringify(data, null, 2);
            });
        });

    </script>
</head>
<body>
    トークン取得 : {{ token }}<br />
    <!-- API呼び出しボタン -->
    <button id="apiButton_InferenceExecution">推論実行</button>
    <br/>
    <button id="apiButton_InferenceStopped">推論停止</button>

    <!-- API呼び出し結果を表示する場所 -->
    <p id="apiResponse"></p>

</body>
</html>

全体的な流れ

1.ページ読み込み時:

{{ message }} と {{ token }} がJinja2テンプレートを通じてPython側から埋め込まれ、HTMLページが生成されます。
ページの読み込みが完了すると、JavaScriptでボタンのクリックイベントが監視されるようになります。

{{ token }}: Python側から渡されたアクセストークンを表示します。

2.ユーザーがボタンをクリック:

推論実行または推論停止ボタンがクリックされると、それに応じたAPIリクエストが送信されます（/start-api または /stop-api に対して POST リクエスト）。リクエストが成功すれば、サーバーから返されたレスポンスが p タグ内に表示されます。

3.レスポンスの表示:

APIの結果は、JSON形式で整形されて表示され、ユーザーはレスポンス内容を確認できます。

まとめ

完成図

今回、上記を完成図ができました。
実際にWebPageから推論開始をクリックするとWebAPIへリクエストを行い、前回同様に推論が開始されました。
この様に、WebPageの管理画面として使うことで実際にPortalからの操作ではなくAPIを中心に開発ができることが分かりました！

著者 : 上田　哲弘 (痛快技術株式会社)
中学生の時に「時代の寵児」と話題になったホリエモンに憧れIT業界に興味を持つ。その後、WEB開発を学びWEBディレクションを経験するが、より開発をしたく再度エンジニアとして粉骨砕身で挑んでいる。

---

痛快技術株式会社について

痛快技術株式会社　：　https://www.tsu-x.com/
当社は、技術の力で社会の問題を解決することを目的として設立されました。私たちは、技術の楽しさを広め、社会のニーズに合わせた製品を開発し、幸せな未来を実現することを使命としています。

痛快技術株式会社

痛快技術ではAIの技術やを使った開発を行っております。
他の記事も読んでください！