指文字認識 API

wss://prediction.ubigram.jp/api/prediction

Authorization: Bearer ubg_live_xxxxxxxxxxxxxxxx

クライアント → サーバ

// フレーム送信。1 メッセージに数フレームまとめて送ってよい。
// data は上記 wire schema (FramePayload) の配列。
{ "type": "send_frames", "data": [ /* FramePayload の配列 */ ] }

// セッション終了。サーバは最終確定を送ってから接続を閉じる。
{ "type": "disconnect_request" }

サーバ → クライアント

// 接続確立 = 送信開始の合図。時間上限プランでは session_time_limit_seconds が付く。
{ "type": "information", "message": "Start session.", "session_time_limit_seconds": 180 }

// 認識結果。差分でなく毎回全体を置き換える。表示する文字列は confirmed + tentative。
//   confirmed  … 仮名漢字変換で確定した部分。以後書き換わらない
//   tentative  … 確定以降の未確定かな。認識のやり直しで書き換わり得る
//   is_converting … 仮名漢字変換の実行中 (「変換中」表示に使える)
{ "type": "recognition_result",
  "data": { "display": { "confirmed": "今日は", "tentative": "あめ", "buffer": "", "latest_char": null },
            "is_converting": false } }

// エラー / 上限超過。error コードで理由を判別する。
{ "type": "error", "data": { "error": "session_time_limit_exceeded", "session_time_limit_seconds": 180 } }

// API キー利用時。api プランの契約が無い / 月間上限に達した場合。
{ "type": "error", "data": { "error": "api_subscription_required" } }
{ "type": "error", "data": { "error": "usage_limit_exceeded", "monthly_minute_limit": 3000 } }

[
  {
    "created_at": "2024-09-01T12:34:56.789+09:00",
    "image_size": { "width": 1280, "height": 720 },
    "is_front_camera": true,
    "mirrored": false,
    // MediaPipe に渡した映像が左右反転 (セルフィー表示) なら true。
    // サーバが座標と handedness を物理空間に戻します
    "dominant_hand": "right",
    // 指文字を出す手。サーバはこの手だけを認識に使い、
    // 写っていないフレームは捨てます (別の手で誤認識しないため)
    "hands": [
      {
        "handedness": { "label": "Right", "score": 0.99 },
        "landmarks": [
          { "x": 0.51, "y": 0.62, "z": 0.0 }
          // … 21 点 ([0,1] 正規化。x,y は画像内比率、z は相対深度)
        ],
        "world_landmarks": [
          { "x": 0.02, "y": -0.01, "z": 0.0 }
          // … 21 点 (任意。手中心原点のメートル単位 3D 座標)
        ]
      }
    ]
  }
]

インストール

# macOS / Windows / Linux (x86_64) の Python 3.9〜3.14 で動作確認済み
pip install opencv-contrib-python mediapipe==0.10.35 websocket-client

# 手のランドマーク検出モデル (hand_landmarker.task) を作業ディレクトリに置く
# (1 行のまま貼れば bash / zsh / PowerShell / コマンドプロンプトのどれでも動く)
curl -o hand_landmarker.task https://storage.googleapis.com/mediapipe-models/hand_landmarker/hand_landmarker/float16/1/hand_landmarker.task

# Linux で libGLESv2.so.2 / libEGL.so.1 が見つからないと言われたら (Ubuntu / Debian)
# sudo apt install libgles2 libegl1

hand_tracking.py — カメラ → MediaPipe Tasks Vision → wire schema

import time
from typing import Dict, List

import cv2
import mediapipe as mp
from mediapipe.tasks.python import BaseOptions
from mediapipe.tasks.python.vision import HandLandmarker, HandLandmarkerOptions, RunningMode


class HandTracker:
    """MediaPipe HandLandmarker で 1 フレームを推論し、ubigram の wire schema に変換する。"""

    def __init__(self, model_path: str = "hand_landmarker.task"):
        self.landmarker = HandLandmarker.create_from_options(
            HandLandmarkerOptions(
                base_options=BaseOptions(model_asset_path=model_path),
                running_mode=RunningMode.VIDEO,
                num_hands=2,  # 両手送って、どちらを使うかはサーバに決めさせる
                min_hand_detection_confidence=0.7,
                min_tracking_confidence=0.5,
            )
        )

    def process_frame(self, frame) -> List[Dict]:
        rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        image = mp.Image(image_format=mp.ImageFormat.SRGB, data=rgb)
        result = self.landmarker.detect_for_video(image, int(time.monotonic() * 1000))

        hands: List[Dict] = []
        for i, lms in enumerate(result.hand_landmarks):
            hand: Dict = {
                "landmarks": [{"x": lm.x, "y": lm.y, "z": lm.z} for lm in lms],
            }
            if i < len(result.hand_world_landmarks):
                hand["world_landmarks"] = [
                    {"x": lm.x, "y": lm.y, "z": lm.z} for lm in result.hand_world_landmarks[i]
                ]
            if i < len(result.handedness):
                top = result.handedness[i][0]
                hand["handedness"] = {"label": top.category_name, "score": float(top.score)}
            hands.append(hand)
        return hands

    def close(self):
        self.landmarker.close()

main.py — WebSocket 送受信

import cv2
import json
import threading
import time
from datetime import datetime

import websocket  # websocket-client
from hand_tracking import HandTracker

WS_ENDPOINT = "wss://prediction.ubigram.jp/api/prediction"
API_KEY = "ubg_live_..."  # LP「アカウント → API キー管理」で発行
FRAMES_PER_MESSAGE = 5


def on_message(ws, message):
    data = json.loads(message)
    if data.get("type") == "recognition_result":
        display = data["data"]["display"]
        print("認識結果:", display["confirmed"] + display["tentative"])
    elif data.get("type") == "error":
        print("エラー:", data["data"].get("error"))


def main():
    ws = websocket.WebSocketApp(
        WS_ENDPOINT,
        header={
            "Authorization": f"Bearer {API_KEY}",
            "User-Agent": "ubigram-sample/1.0",  # UA なしは WAF が 403 で弾く
        },
        on_message=on_message,
    )
    threading.Thread(target=ws.run_forever, daemon=True).start()
    time.sleep(1)  # 接続確立を待つ

    tracker = HandTracker()
    cap = cv2.VideoCapture(0)
    buffer = []
    try:
        while cap.isOpened():
            ok, frame = cap.read()
            if not ok:
                break
            height, width = frame.shape[:2]
            hands = tracker.process_frame(frame)
            if hands:
                buffer.append({
                    "created_at": datetime.now().astimezone().isoformat(),
                    "image_size": {"width": float(width), "height": float(height)},
                    "is_front_camera": False,
                    # 生のカメラ映像をそのまま MediaPipe に渡しているので物理空間
                    "mirrored": False,
                    "dominant_hand": "right",
                    "hands": hands,
                })
            if len(buffer) >= FRAMES_PER_MESSAGE:
                ws.send(json.dumps({"type": "send_frames", "data": buffer}))
                buffer = []
            if cv2.waitKey(5) & 0xFF == 27:  # ESC で終了
                break
    finally:
        if buffer:
            ws.send(json.dumps({"type": "send_frames", "data": buffer}))
        ws.send(json.dumps({"type": "disconnect_request"}))
        tracker.close()
        cap.release()


if __name__ == "__main__":
    main()