こんにちは。
ソニーセミコンダクタソリューションズの細井と申します。

今回はAITRIOSのDeveloper Siteで公開されているConsole REST API Specificationを用いてエッジAIデバイスからAITRIOSのクラウドに送信された推論結果のデータの取得をしてみたという記事を投稿します。
ついでにAITRIOSから公開されているGitHubリポジトリのCodespaces環境で可視化まで行ってみましたので合わせて投稿させていただきます。

この記事は2024年02月29日時点の情報に基づいて作成されています。

この記事の要約

AITRIOSから提供されるConsole REST APIを利用することでUIを用いず、自分でコマンドを打ちこむことでメタデータを取得する
公開されているSDKを利用し簡単に可視化を行う

本記事ではPythonを利用します。ほぼコピペで済むようにしていますが、Pythonの知識が多少必要となります。最終的に下記のように取得した推論データをPythonのJupyter notebook上で可視化するのが本記事のゴールです。
AITRIOS のデバイスで Object Detection してメタデータを取得してみたを一通り読んでからこちらの記事を見ていただくとより理解が深まると思います。

実行環境に関して

Get StartedのCloud SDKチュートリアル/チュートリアルに記載のある、Tutorial Checkdata: Sample Application on AITRIOSのCodespaces環境を利用しました。

この記事に添付しているコードブロックをそれぞれjupyter notebook上のセルに張り付いていただくことで楽に実行いただけます。

始めに

下記の流れで今回REST APIを利用しています。
すでに撮影・推論を実施していること、およびその推論時のcommand parameterを"Mode": 1（Input Image & Inference Result） 、NumberOfInferencesPerMessage: 1になっていることを前提としています。

Console REST APIを使うためのUtilsクラス、ConsoleRESTAPIクラスを作成します。
Console REST APIを使うために、AITRIOSプロジェクトのClient ID, Client Secretを設定します。
GetDevices関数をConsoleRESTAPIクラス内に作成、実行しAITRIOSプロジェクトに紐づいたDeviceIDを取得します。
GetImageDirecotries関数をConsoleRESTAPIクラス内に作成して、DeviceIDに紐づいた画像ディレクトリ一覧を取得します。
GetImages関数をConsoleRESTAPIクラス内に作成して、画像ディレクトリに紐づいた画像ファイル一覧を取得します。
画像ファイルをBase64でデコードして、可視化します。
GetInferenceResults関数をConsoleRESTAPIクラス内に作成して、画像に紐づいた推論結果を取得します。
推論結果をBase64デコードとFlatbuffersスキーマによるデシリアライズを実行し、可視化します。

Utilsクラスの作成

本チュートリアルで使用するAITRIOSとは無関係のクラスを作成します。
UtilsクラスはデータのBase64エンコード、デコードに用います。
以下のコードは、Tutorial Checkdata: Sample Application on AITRIOSのCodespaces環境上でJupyter Notebookのファイルをjupyter_notebookディレクトリ以下にaitrions_rest_api_tutorial.ipynbを作成し、セルに記載しています。
同じ配置にすることでパスの修正の必要がなく実施いただけます。

# opencvと依存ライブラリのインストール
!pip install opencv-contrib-python
!sudo apt-get update
!sudo apt-get install -y libgl1-mesa-dev
!pip install matplotlib

import base64
import cv2
import numpy as np
import matplotlib.pyplot as plt

class Utils:
    @staticmethod
    def Base64Decoder(data):
        if type(data) is str:
            data = data.encode("utf-8")
        _decoded_data = base64.decodebytes(data)
        return _decoded_data
    @staticmethod
    def Base64EncodedStr(data):
        if type(data) is str:
            data = data.encode("utf-8")
        _encoded_data = base64.b64encode(data)
        _encoded_str = str(_encoded_data).replace("b'", "").replace("'", "")
        return str(_encoded_str)
    @staticmethod
    def Base64ToCV2(img_str):
    
        if "base64," in img_str:
            # DATA URI の場合、data:[<mediatype>][;base64], を除く
            _img_str = img_str.split(",")[1]
        else:
            _img_str = img_str
            
        # Base64文字列をデコードし、バイト列に変換
        _img_data = base64.b64decode(_img_str)
        
        # バイト列をNumPy配列に変換
        _nparr = np.frombuffer(_img_data, np.uint8)
        
        # NumPy配列をOpenCVの画像形式に変換
        _img = cv2.imdecode(_nparr, cv2.IMREAD_COLOR)
        image_rgb = cv2.cvtColor(_img, cv2.COLOR_BGR2RGB)
        return image_rgb

ConsoleRESTAPIの作成

Console REST APIを利用するためのクラスを作成します。
ConsoleRESTAPIクラスではインスタンス作成時に、Portal/Consoleエンドポイント情報から取得可能なClientID, ClientSecret、エンドポイント情報を引数に初期化を行います。

上記のファイルアクセスのためには、AITRIOSアカウントへの登録が必要です。

また、HTTP Requestを送るためのLow Level APIである、GetToken, GetHeaders, Requestを作成します。
こちら行数が長いので、コピペしてJupyter notebookに貼るだけでも問題ないです！

OpenAPI Generatorなどを利用することでAITRIOSのConsole REST APIから直接下記のクライアントライブラリを作成可能です。要望があれば記事公開させていただこうと思います。

!pip install requests
import requests
import json
class ConsoleRESTAPI:
    def __init__(self, baseURL, client_id, client_secret, gcs_okta_domain):
        # Project information

        self.BASE_URL = base_url
        CLIENT_ID = client_id
        CLIENT_SECRET = client_secret
        self.GCS_OKTA_DOMAIN = gcs_okta_domain
        self.AUTHORIZATION_CODE = Utils.Base64EncodedStr(CLIENT_ID + ":" + CLIENT_SECRET)
        
    ##########################################################################
    # Low Level APIs
    ##########################################################################
    def GetToken(self):
        headers = {
            "accept": "application/json",
            "authorization": "Basic " + self.AUTHORIZATION_CODE,
            "cache-control": "no-cache",
            "content-type": "application/x-www-form-urlencoded",
        }

        data = {
            "grant_type": "client_credentials",
            "scope": "system",
        }

        response = requests.post(
            url=self.GCS_OKTA_DOMAIN,
            data=data,
            headers=headers,
        )
        analysis_info = json.loads(response.text)
        token = analysis_info["access_token"]
        return token

    def GetHeaders(self, payload):
        token = self.GetToken()
        headers = {"Accept": "application/json", "Authorization": "Bearer " + token}
        if payload != {}:
            headers.setdefault("Content-Type", "application/json")
        return headers

    def Request(self, url, method, **kwargs):
        params = {}
        payload = {}
        files = {}
        url = self.BASE_URL + url

        # set parameters
        for key, val in kwargs.items():
            if val != None:
                if key == "payload":
                    # payload
                    payload = json.dumps(val)
                elif key == "files":
                    # multipart/form-data
                    files = val
                else:
                    # check parameters
                    if "{" + key + "}" in url:
                        # path parameter
                        url = url.replace("{" + key + "}", val)
                    else:
                        # query parameter
                        params.setdefault(key, str(val))

        # create header
        headers = self.GetHeaders(payload=payload)

        # call request
        try:
            response = requests.request(
                method=method, url=url, headers=headers, params=params, data=payload, files=files
            )
            analysis_info = json.loads(response.text)
        except Exception as e:
            return response.text
        return analysis_info

    def GetDevices(
        self, connectionState=None, device_name=None, device_id=None, device_group_id=None
    ):
        ret = self.Request(
            url="/devices",
            method="GET",
            connectionState=connectionState,
            device_name=device_name,
            device_id=device_id,
            device_group_id=device_group_id,
        )
        return ret

    def GetImageDirectories(self, device_id=None):
        ret = self.Request(url="/devices/images/directories", method="GET", device_id=device_id)
        return ret
    
    def GetImages(
        self, device_id, sub_directory_name, order_by=None, number_of_images=None, skip=None
    ):
        ret = self.Request(
            url="/devices/{device_id}/images/directories/{sub_directory_name}",
            method="GET",
            device_id=device_id,
            sub_directory_name=sub_directory_name,
            order_by=order_by,
            number_of_images=number_of_images,
            skip=skip,
        )
        return ret

    def GetInferenceResults(
        self, device_id, NumberOfInferenceresults=None, filter=None, raw=None, time=None
    ):
        ret = self.Request(
            url="/devices/{device_id}/inferenceresults",
            method="GET",
            device_id=device_id,
            NumberOfInferenceresults=NumberOfInferenceresults,
            filter=filter,
            raw=raw,
            time=time,
        )
        return ret

Step1 AITRIOSに接続されているDevice一覧情報を取得する

AITRIOSに接続されているDevice一覧情報を取得します
responseからDeviceID情報のみ取り出し、listに格納します

base_url = "{base_url}"
client_id = "{client_id}"
client_secret = "{client_secret}"
gcs_okta_domain = "{gcs_okta_domain}"

baseURLとgcs_oktadomainはそれぞれ、AITRIOS Developer SiteのPortal / Console Endpoint Informationを参考に取得してください
baseURL = Console Endopoint、gcs_oktadomain = Portal Endpointです。
また、client_id、client_scretについてはGet Started 3.6.2. AITRIOSにアクセスするためのクライアントを取得するを参考にしてください。

base_url = "{base_url}"
client_id = "{client_id}"
client_secret = "{client_secret}"
gcs_okta_domain = "{gcs_okta_domain}"

console_api = ConsoleRESTAPI(base_url, client_id, client_secret, gcs_okta_domain)
response = console_api.GetDevices()
devices = {"devices": []}
for device in response["devices"]:
    devices["devices"].append(device["device_id"])
print(devices)

Step2 選択されたDeviceにおける画像ディレクトリ情報を取得する

前セルで取得したDeviceIDをGetImageDirectoriesの引数にして実行し、DeviceIDに紐づいた画像フォルダ名を取得します
私の環境ではデバイスを1台接続していたので下記のコマンドを実行すると1台出力に表示されました

# devices_list配列から適当なdevice_idを選択してください。
# 本チュートリアルでは、配列の先頭を選択していますが、device_idの文字列指定でも選択できます。
device_id = devices["devices"][0]
#device_id = "sid-xxxxxxxxxxxxxxxxxxxxxxx"

response = console_api.GetImageDirectories(device_id)
directories = {"directories": []}
for directory in response[0]["devices"][0]["Image"]:
    directories["directories"].append(directory)
print(directories)

Step3 選択されたディレクトリ内の画像情報を取得し表示する

DeviceIDと前セルで取得した画像フォルダの中の1つを引数とし、フォルダ内の画像群を取得するGetImagesの実行を実行します
device_imagesを実行後、画像群から1枚選択し、描画します

本記事ではbase64でエンコードされた画像の文字列をpythonの標準ライブラリbase64でデコード後、OpenCVを使用して画像を表示しています。
開発者の用途に応じて適切なライブラリを使用してください。

画像の取得

sub_direcotory = directories["directories"][0]
response = console_api.GetImages(device_id, sub_direcotory)
image_data = response["images"]
print("保存画像枚数: ", response["total_image_count"])

画像の描画

# 画像群から最新の1枚を選択
image_name = image_data[-1]["name"]
img_base64 = image_data[-1]["contents"]

img = Utils.Base64ToCV2(img_base64)
plt.imshow(img)

Step4 画像に紐づいた推論結果を取得する

device_idと取得する推論結果個数、タイムスタンプを引数に該当する推論結果を取得するGetInferenceResultsを実行します
推論結果がlistに格納されているため、inference_list[0]で先頭の推論結果を取得後、Oのkeyを指定することで推論結果の中身を可視化します

latest_image_ts = image_name.replace(".jpg", "")
print(latest_image_ts)
response = console_api.GetInferenceResults(device_id=device_id, NumberOfInferenceresults=1, time=latest_image_ts)
print(response)
inference_list = response[0]["inferences"]
print(inference_list[0]["O"])

Step5 推論結果をDeserializeする

smart-camera-tutorial-checkdata-pythonで提供されているDeserializeコードを用いてデータをbase64のデコードおよび、FlatBuffersによるデシリアライズを行います。

get_deserialize_dataを実行するためにはworking directoryの移動が必要なためos.chdirでwordking directoryを変更しています

import os
print (os.getcwd())
os.chdir("/workspace/src")
print (os.getcwd())

from data_deserializer import get_deserialize_data

deserialize_data = get_deserialize_data.get_deserialize_data(inference_list[0]["O"])
print(deserialize_data)

なんでFlatbuffersを使っているの？ということについては今後別の記事で投稿させていただければと思います！

get_desesrialize_dataについて

get_deserialize_data.pyに関して詳細な説明を記載します。
こちらはjupyter notebookのセルにコピーいただく必要はありません。

get_deserialize_data.py

import base64
from src.common.deserialize import ObjectDetectionTop, BoundingBox, BoundingBox2d


def get_deserialize_data(serialize_data):
    """Get access information from yaml and generate ConsoleAccess client
    Returns:
        ConsoleAccessClient: CosoleAccessClient Class generated from access information.
    """
    buf = {}
    buf_decode = base64.b64decode(serialize_data)
    ppl_out = ObjectDetectionTop.ObjectDetectionTop.GetRootAsObjectDetectionTop(buf_decode, 0)
    obj_data = ppl_out.Perception()
    res_num = obj_data.ObjectDetectionListLength()
    for i in range(res_num):
        obj_list = obj_data.ObjectDetectionList(i)
        union_type = obj_list.BoundingBoxType()
        if union_type == BoundingBox.BoundingBox.BoundingBox2d:
            bbox_2d = BoundingBox2d.BoundingBox2d()
            bbox_2d.Init(obj_list.BoundingBox().Bytes, obj_list.BoundingBox().Pos)
            buf[str(i + 1)] = {}
            buf[str(i + 1)]['C'] = obj_list.ClassId()
            buf[str(i + 1)]['P'] = obj_list.Score()
            buf[str(i + 1)]['X'] = bbox_2d.Left()
            buf[str(i + 1)]['Y'] = bbox_2d.Top()
            buf[str(i + 1)]['x'] = bbox_2d.Right()
            buf[str(i + 1)]['y'] = bbox_2d.Bottom()

    return buf

初めに、src.common.deserialize以下にあるFlatBuffersスキーマをimportします。
- AITRIOS Console上のpresetで提供されているobject detectionのpplを使用する場合はこちらのファイルの修正は不要です。

from src.common.deserialize import ObjectDetectionTop, BoundingBox, BoundingBox2d

AITRIOS Consoleから取得した推論結果データをbase64でデコードします。

buf_decode = base64.b64decode(serialize_data)

FlatBuffersのライブラリがWrapされているObjectDetectionTopを使用して指定された前段でデコードされたデータbuf_decodeをデシリアライズします。その後、Perceptionによってデータをpythonのlist型に変換します。

ppl_out = ObjectDetectionTop.ObjectDetectionTop.GetRootAsObjectDetectionTop(buf_decode, 0)
obj_data = ppl_out.Perception()

listの個数分だけ、データのフィールドにアクセスする処理を行い、データをdict型に変換します。

res_num = obj_data.ObjectDetectionListLength()
for i in range(res_num):
    obj_list = obj_data.ObjectDetectionList(i)
    union_type = obj_list.BoundingBoxType()
    if union_type == BoundingBox.BoundingBox.BoundingBox2d:
        bbox_2d = BoundingBox2d.BoundingBox2d()
        bbox_2d.Init(obj_list.BoundingBox().Bytes, obj_list.BoundingBox().Pos)
        buf[str(i + 1)] = {}
        buf[str(i + 1)]['C'] = obj_list.ClassId()
        buf[str(i + 1)]['P'] = obj_list.Score()
        buf[str(i + 1)]['X'] = bbox_2d.Left()
        buf[str(i + 1)]['Y'] = bbox_2d.Top()
        buf[str(i + 1)]['x'] = bbox_2d.Right()
        buf[str(i + 1)]['y'] = bbox_2d.Bottom()

以上がget_deserialize_data.pyの説明です。

Step6 推論結果を画像に重ね合わせる

前項でデシリアライズした結果をopencvライブラリを用いて、画像に矩形として描画します。

import cv2
import numpy as np

colors = [
(255, 0, 0),
(255, 128, 0),
(255, 255, 0),
(128, 255, 0),
(0, 255, 0),
(0, 255, 255),
(0, 128, 255),
(0, 0, 255),
(128, 0, 255),
(255, 0, 255),
(255, 0, 128),
(255, 255, 255),
(192, 192, 192),
(128, 128, 128),
(0, 0, 0),
  ]

img_np = np.array(img)
for index in range(len(deserialize_data)):
    color = colors[index]
    # 矩形の描画
    cv2.rectangle(img_np, (deserialize_data[str(index+1)]['X'], deserialize_data[str(index+1)]['Y']), (deserialize_data[str(index+1)]['x'], deserialize_data[str(index+1)]['y']), color\
              ,thickness=1, lineType=cv2.LINE_4, shift=0)
    # 文字列の描画 {class id} : {score}
    class_text = str(deserialize_data[str(index+1)]['C']) + ': ' + str(deserialize_data[str(index+1)]['P'])
    cv2.putText(img_np,
            text=class_text,
            org=(deserialize_data[str(index+1)]['X'], deserialize_data[str(index+1)]['Y']),
            fontFace=cv2.FONT_HERSHEY_SIMPLEX,
            fontScale=0.3,
            color=color,
            thickness=1,
            lineType=cv2.LINE_4)
    if index == 1:
        break
plt.imshow(img_np)

これで画像上にメタデータが出力されました！
長い記事になりましたがお付き合いいただきありがとうございました。

困った時は

もし、記事の途中でうまくいかなかった場合は、気軽にこの記事にコメントいただいたり、以下のサポートのページもご覧ください。
コメントのお返事にはお時間を頂く可能性もありますがご了承ください。

Support (FAQ)

また、記事の内容以外で AITRIOS についてお困りごとなどあれば以下よりお問い合わせください。

Contact Us

さいごに

今回はAITRIOSのDeveloper Siteで公開されているConsole REST APIを用いてデータの取得から可視化まで試してみました。
また、これらの技術を応用することでアプリケーション上でAITRIOSのデータ取得と解析などもできるようになります。
また近々、そういった内容の記事も出していければと思います。
長い記事となりましたが改めて、見ていただきありがとうございました！