予測のアップロード

7 分

はじめに

この例では、既存のデータセットにAPIを使用して予測をアップロードする方法を説明します。

開始する前に、はじめにはじめにの前提条件を確認し、予測フォーマット予測フォーマットについて学んでください。

手順

新しいPythonファイルを作成し、以下のライブラリをインポートしてください：

import requests

from kognic.auth.requests.auth_session import RequestsAuthSession

base_url = "https://dataset.app.kognic.com/v1/"
client = RequestsAuthSession()

1. データセットのUUIDを取得する

ツールにアクセスしてURLのdataset/に続くUUIDをコピーするか、datasetsエンドポイントを利用してデータセットのUUIDを取得できます：

client.session.get("https://dataset.app.kognic.com/v2/datasets")

2. 既存の予測グループのUUIDを取得するか、新しいものを作成する

2.a 既存の予測グループのUUIDを取得する

予測をアップロードするには、予測グループが存在する必要があります。予測は任意の目的でグループに整理できます。既存の予測グループのUUIDは、URLのpredictions/に続く部分で確認するか、以下のエンドポイントを使用して取得できます：

client.session.get(base_url + f"/datasets/{datasetUuid}/predictions-groups")

2.b 予測グループの作成（オプション）

セグメンテーションタスクを含まないデータセットの場合、新しい予測グループは、アプリ内で操作する（右上のManage Predictionsをクリックし、+ Create predictions groupを選択）か、以下のコードスニペットを使用することで作成できます：

path = base_url + f"/datasets/{datasetUuid}/predictions-groups"
body = {"name": "My predictions group", "description": "A description of my new predictions group"}
try:
    response = client.session.post(path, json=body)
    response.raise_for_status()
    response_json = response.json()
    print(f"Created predictions group with uuid {response_json['data']}")
except requests.exceptions.RequestException as e:
    msg = e.response.text
    print(f"Request error: {e}. {msg}")

特殊なケース：セグメンテーションデータセット

セグメンテーションデータセットに関連付けられた予測グループは、コードスニペットを使用して作成する必要があり、classMappingという追加パラメータも必要です。このマッピングは、予測とアノテーション間の不一致を計算する際に使用され、ソート順やギャラリーでの不一致の表示方法に影響します。classMappingパラメータは辞書のリストで、各辞書にはannotatedとpredictedのキーが含まれます。annotatedキーはアノテーション内のクラス名、predictedキーは予測内のクラス名です。例えば、異なる樹種をアノテーションしているが、木かどうかのみを予測する場合は{"annotated": "oak", "predicted": "tree"}のようになります。

予測とアノテーションのすべてのクラス名は、マッピングが不要な場合でも、クラスマッピングに含まれている必要があります。アノテーションでは、セグメンテーションされていない領域には_backgroundというクラス名が付けられます。

example_body = {
    "name": "My predictions group",
    "description": "A description of my new predictions group",
    "classMapping": [
        {"annotated": "oak", "predicted": "tree"},
        {"annotated": "_background", "predicted": "not_tree"},
        {"annotated": "only_in_annotations"}
    ]
}

3. 予測のアップロード

少量の予測の場合は、同期呼び出しでも問題ない場合があります：

import requests

from kognic.auth.requests.auth_session import RequestsAuthSession

base_url = "https://dataset.app.kognic.com/v1/"
client = RequestsAuthSession()

predictions_group_uuid = "..."
openlabel_content = {"openlabel": ...}
data = {
    "sceneUuid": "...",
    "openlabelContent": openlabel_content,
}

try:
    response = client.session.post(
        base_url + f"predictions-groups/{predictions_group_uuid}/predictions",
        json=data
    )
    response.raise_for_status()
    response_json = response.json()
    print(f"Created prediction with uuid {response_json['data']}")
except requests.exceptions.RequestException as e:
    msg = e.response.text
    print(f"Request error: {e}. {msg}")

大量の予測の場合は、非同期呼び出しを推奨します。以下の例では、kognic-authライブラリの非同期クライアントを使用して100件の非同期呼び出しを行います：

import asyncio

from kognic.auth.httpx.async_client import HttpxAuthAsyncClient

base_url = "https://dataset.app.kognic.com/v1/"
predictions_group_uuid = "..."
url = base_url + f"predictions-groups/{predictions_group_uuid}/predictions"
openlabel_content = {"openlabel": ...}

MAX_CONNECTIONS = 10


async def upload_prediction(payload, session, sem):
    async with sem:
        response = await session.post(url, json=payload)
        response.raise_for_status()
        return response.json().get("data")


async def main(n_runs: int):
    client = HttpxAuthAsyncClient()
    session = await client.session

    sem = asyncio.Semaphore(MAX_CONNECTIONS)
    tasks = []
    for i in range(n_runs):
        payload = {"sceneUuid": "...", "openlabelContent": openlabel_content}
        task = upload_prediction(payload, session, sem)
        tasks.append(task)

    responses = await asyncio.gather(*tasks)
    await session.aclose()

    print(responses)


if __name__ == '__main__':
    asyncio.run(main(100))

MAX_CONNECTIONSを10より大きい値に設定すると動作しない場合があり、推奨されません。