aiboのWeb APIを試すスクリプト

aibo.py

#!/bin/env python
#
# aibo Developer Site: https://developer.aibo.com/jp/home
# https://developer.aibo.com/jp/docs#api-%E3%81%AE%E5%AE%9F%E8%A3%85%E4%BE%8B を書き替えてみた
#
# 環境変数「AIBO_TOKEN」にトークンを設定
# 環境変数「AIBO_DEVICE_ID」に aibo のデバイス ID を設定（一匹前提）
#
import json
import os
import requests
import time

AIBO_BASE_URL = 'https://public.api.aibo.com/v1'
MAX_RETRIES = 10

ENV_AIBO_TOKEN = 'AIBO_TOKEN'
ENV_AIBO_DEVICE_ID = 'AIBO_DEVICE_ID'

deviceId = os.environ[ENV_AIBO_DEVICE_ID]
token = os.environ[ENV_AIBO_TOKEN]
headers = {
    'Authorization': 'Bearer ' + token,
}

def dump_json(str):
    return json.dumps(str, indent=4, ensure_ascii=False)
    
def get_result(exec_id):
    url = AIBO_BASE_URL + '/executions/' + exec_id
    for _ in range(MAX_RETRIES):
        response = requests.get(url, headers=headers)
        result = json.loads(response.text)
        status = result['status']
        if status == 'SUCCEEDED' or status == 'FAILED':
            return result
        time.sleep(1)
    else:
        return {'status': 'TIMEOUT'}

def post(api, args):
    url = AIBO_BASE_URL + '/devices/' + deviceId + '/capabilities/' + api + '/execute'
    data = '{"arguments":' + args + '}'
    response = requests.post(url, headers=headers, data=data)
    result = json.loads(response.text)
    return result['executionId']
    
if __name__ == '__main__':
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument('api')
    parser.add_argument('params', nargs='?', default='{}')
    args = parser.parse_args()

    print(dump_json(get_result(post(args.api, args.params))))