AITRIOSのConsole REST APIを利用してメタデータを取得してみた

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

今回はAITRIOSのDeveloper Siteで公開されている[Console REST API Specification](https://developer.aitrios.sony-semicon.com/edge-ai-sensing/documents/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 してメタデータを取得してみた](https://qiita.com/SSS-NorimitsuTakahashi/items/3c11544610fcc8f8d4c1)を一通り読んでからこちらの記事を見ていただくとより理解が深まると思います。
![image-1.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3724589/62517029-5d51-36c9-5fd8-8425403f9851.png)



## 実行環境に関して

[Get Started](https://developer.aitrios.sony-semicon.com/edge-ai-sensing/documents/get-started/)のCloud SDKチュートリアル/チュートリアルに記載のある、[Tutorial Checkdata: Sample Application on AITRIOS](https://github.com/SonySemiconductorSolutions/aitrios-sample-application-check-data-tool-python)のCodespaces環境を利用しました。

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

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

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


## Utilsクラスの作成
本チュートリアルで使用するAITRIOSとは無関係のクラスを作成します。  
UtilsクラスはデータのBase64エンコード、デコードに用います。
以下のコードは、[Tutorial Checkdata: Sample Application on AITRIOS](https://github.com/SonySemiconductorSolutions/aitrios-sample-application-check-data-tool-python)のCodespaces環境上でJupyter Notebookのファイルをjupyter_notebookディレクトリ以下に`aitrions_rest_api_tutorial.ipynb`を作成し、セルに記載しています。
同じ配置にすることでパスの修正の必要がなく実施いただけます。
![aitrios-qiita-rest-api-tutorial-codespaces.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3724589/b461f271-51dc-cd41-53f8-aacb0b499b22.png)



```python
# 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エンドポイント情報](https://developer.aitrios.sony-semicon.com/file/download/edge-ai-sensing-portal-console-end-point-info)から取得可能な`ClientID`, `ClientSecret`、`エンドポイント情報`を引数に初期化を行います。

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

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

:::info note
[OpenAPI Generator](https://github.com/OpenAPITools/openapi-generator)などを利用することでAITRIOSのConsole REST APIから直接下記のクライアントライブラリを作成可能です。要望があれば記事公開させていただこうと思います。
:::

```python
!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一覧情報を取得する

1. AITRIOSに接続されているDevice一覧情報を取得します
2. responseからDeviceID情報のみ取り出し、listに格納します
:::note info
```
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](https://developer.aitrios.sony-semicon.com/en/file/download/rest-api-authentication)を参考に取得してください  
`baseURL` = `Console Endopoint`、`gcs_oktadomain` = `Portal Endpoint`です。
また、`client_id`、`client_scret`については[Get Started 3.6.2. AITRIOSにアクセスするためのクライアントを取得する](https://developer.aitrios.sony-semicon.com/edge-ai-sensing/documents/get-started/?version=2024-01-23&progLang=#_cloud_sdk%E3%82%92%E7%94%A8%E3%81%84%E3%81%9F%E3%83%87%E3%83%BC%E3%82%BF%E3%81%AE%E5%8F%AF%E8%A6%96%E5%8C%96:~:text=2.6.2.%20AITRIOS%E3%81%AB%E3%82%A2%E3%82%AF%E3%82%BB%E3%82%B9%E3%81%99%E3%82%8B%E3%81%9F%E3%82%81%E3%81%AE%E3%82%AF%E3%83%A9%E3%82%A4%E3%82%A2%E3%83%B3%E3%83%88%E3%82%92%E5%8F%96%E5%BE%97%E3%81%99%E3%82%8B)を参考にしてください。
:::

```python
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における画像ディレクトリ情報を取得する

1. 前セルで取得したDeviceIDを`GetImageDirectories`の引数にして実行し、DeviceIDに紐づいた画像フォルダ名を取得します
2. 私の環境ではデバイスを1台接続していたので下記のコマンドを実行すると1台出力に表示されました
![aitrios-qiita-rest-api-tutorial-devices.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3724589/b50208df-4c2a-c6e7-2e4c-d7d0ddba32dc.png)


```python
# 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 選択されたディレクトリ内の画像情報を取得し表示する

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

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

### 画像の取得

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

### 画像の描画

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

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

![apple.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3724589/a3570bfd-f869-23c1-912a-f67581176a8a.jpeg)



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

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

```python
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](https://github.com/SonySemiconductorSolutions/smart-camera-tutorial-checkdata-python)で提供されているDeserializeコードを用いてデータをbase64のデコードおよび、FlatBuffersによるデシリアライズを行います。

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

```python
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のセルにコピーいただく必要はありません。**

<details><summary>get_deserialize_data.py</summary>

```python
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
```
</details>

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

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

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

```python
buf_decode = base64.b64decode(serialize_data)
```

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

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

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

```python
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ライブラリを用いて、画像に矩形として描画します。

```python
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)
```

![image-1.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3724589/c87f5e76-b1f5-9fd4-5b04-db23f027df1e.png)


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

## 困った時は
もし、記事の途中でうまくいかなかった場合は、気軽にこの記事にコメントいただいたり、以下のサポートのページもご覧ください。  
コメントのお返事にはお時間を頂く可能性もありますがご了承ください。
 
- [Support (FAQ)](
https://developer.aitrios.sony-semicon.com/edge-ai-sensing/support)
 
また、記事の内容以外で AITRIOS についてお困りごとなどあれば以下よりお問い合わせください。
- [Contact Us](
https://developer.aitrios.sony-semicon.com/edge-ai-sensing/contact-us)


## さいごに

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