AndroidスクリーンキャプチャをリアルタイムでストリーミングするApollonStreamAPIの使い方

こんにちは！今回は、Androidデバイスのスクリーンショットをリアルタイムでストリーミングするためのツール「ApollonStreamAPI」の使い方を初心者向けに解説します。

こちらの記事もおすすめ

Android Studio で Geminiアプリを作成して実機で動作確認してみる

はじめにAndroidアプリの開発は、技術の進歩とともによりアクセスしやすくなっています。特にAndroid Studioは、開発者にとって非常に強力なツールです。この記事では、Android Studioを使用してGeminiアプリを作成...

【Fate・東方ネタで学ぶ】日本語版Tavily APIの完全ガイド

IntroductionTavily Search APIの特徴目的と特徴 : Tavily Search APIは、LLM（大規模言語モデル: Large Language Models）やRAG（検索強化生成: Retrieval-Au...

ApollonStreamAPIとは？
機能
使用方法
1. 1. サーバー側の設定
2. 2. クライアント側の設定
疎通確認
ADBコマンドのヘルプ
注意事項
リポジトリ
1. 関連

ApollonStreamAPIとは？

ApollonStreamAPIは、FastAPIを使用してWebSocketサーバーを構築し、ADBコマンドを使用してAndroidデバイスのスクリーンショットを取得するツールです。取得したスクリーンショットはリアルタイムでストリーミングされ、JPEG形式で保存することができます。

機能

ApollonStreamAPIには以下のような機能があります：

Androidデバイスのスクリーンショットをリアルタイムでストリーミング
取得したスクリーンショットをJPEG形式で保存
デモ画像のストリーミングにも対応

使用方法

1. サーバー側の設定

まず、FastAPIとUvicornをインストールします。

pip install fastapi uvicorn

次に、api/screencap_server.pyを実行してサーバーを起動します。

uvicorn api.screencap_server:app --host 0.0.0.0 --port 8000

サーバーのコードは以下のようになっています：

from fastapi import FastAPI, WebSocket
import subprocess
import asyncio
from PIL import Image
import io

app = FastAPI()

async def capture_and_send_screenshot(websocket: WebSocket, path: str = None):
    """ADBコマンドまたは指定されたパスからスクリーンショットを取得し、JPEGとしてWebSocketを通じて送信する"""
    if path:
        # 指定されたパスから画像を読み込む
        with open(path, "rb") as image_file:
            result = image_file.read()
    else:
        # ADBを使用してスクリーンショットを取得
        result = subprocess.check_output(['adb', 'exec-out', 'screencap', '-p'])

    image = Image.open(io.BytesIO(result))
    if image.mode == 'RGBA':
        image = image.convert('RGB')

    with io.BytesIO() as output:
        image.save(output, format="JPEG")
        jpeg_data = output.getvalue()
        await websocket.send_bytes(jpeg_data)

@app.websocket("/stream")
async def stream_screen(websocket: WebSocket):
    """Androidデバイスのスクリーンショットをリアルタイムでストリーミングするエンドポイント"""
    await websocket.accept()
    while True:
        await capture_and_send_screenshot(websocket)
        await asyncio.sleep(0.1)

@app.websocket("/demo")
async def stream_demo_image(websocket: WebSocket):
    """デモ画像をリアルタイムでストリーミングするエンドポイント"""
    await websocket.accept()
    while True:
        await capture_and_send_screenshot(websocket, "demo/demo.jpg")
        await asyncio.sleep(0.1)

2. クライアント側の設定

demo/demo_screencap_client.pyを実行してクライアントを起動します。

python demo/demo_screencap_client.py

クライアントのコードは以下のようになっています：

import asyncio
import websockets
import io
from PIL import Image

async def receive_and_save_image():
    uri = "ws://localhost:8000/stream"  # WebSocket接続のURI
    async with websockets.connect(uri) as websocket:
        # サーバーからのバイナリデータ（JPEG画像）を受信
        data = await websocket.recv()

        # 受信したデータをファイルに保存
        with open("received_image.jpg", "wb") as image_file:
            image_file.write(data)
            print("画像を受信し、保存しました。")

# イベントループを実行して、画像受信関数を呼び出す
asyncio.run(receive_and_save_image())

クライアントは、サーバーからストリーミングされたスクリーンショットを受信し、received_image.jpgとして保存します。

疎通確認

スタンドアロンでのスクリーンショット取得と保存の処理時間を確認するには、demo/demo_screenshot_standalone.pyを実行します。

python demo/demo_screenshot_standalone.py

このスクリプトは、ADBコマンドを使用してスクリーンショットを取得し、画像データの変換と保存を行います。処理時間が出力されます。

ADBコマンドのヘルプ

ADBのscreencapコマンドのヘルプは以下の通りです。

(ApollonStreamAPI) C:\Prj\_ApollonStreamAPI\demo>adb shell screencap -h    
usage: screencap [-hp] [-d display-id] [FILENAME]
   -h: this message
   -p: save the file as a png.
   -d: specify the physical display ID to capture (default: 0)
       see "dumpsys SurfaceFlinger --display-id" for valid display IDs.
If FILENAME ends with .png it will be saved as a png.
If FILENAME is not given, the results will be printed to stdout.