「FIWAREでスマートホームする」 シリーズ第三弾として、SwicthBotデバイスをFIWARE対応させるプログラムコードを作成しました。

できること

様々なNGSIでの操作をトリガーに、SwitchBotデバイス（赤外線リモコン）を操作できます。例えば「FIWAREでスマートホームする」第二弾でご紹介したBluetoothプッシュボタンと連携させて、ボタン押下で照明をつける、といったことが実現できます。

今回のコードは赤外線リモコンの操作を実施してますが、コードを少し改造することで、任意のSwitchBotデバイスを操作可能です。
なお今回のコードは、SwitchBotデバイスを操作する方向を実装していますが、SwitchBotデバイス側からのデータ通知には対応していません（例：室温センサからのデータ通知）。

使い方

前提

FIWARE環境がある（Context Brokerが動いている）
SwitchBotデバイスを設置・稼働している
- インターネットと接続するため、HubMini(Plus)が必要です。赤外線リモコン機能もHubMiniにありますし。
SwitchBotアプリの開発者向けオプションで、トークンとシークレットキーを取得している

SwitchBotデバイスをAPI経由で操作するためのトークンとシークレットキーの取得方法については、こちらを参照してください。

設定の仕方

FIWARE環境がorion.example.comで稼働しているとします。

orion.example.com上で、ngsi_switchbotapi_converter.pyを適当なディレクトリに配置します。
コードの中で呼び出しているrequestsパッケージ等をインストールするために、$ pip install requestsなどの操作をします。

環境変数をセットします

$ export SWITCHBOT_DEVICE=12-202001011234-56(対象SwitchBotデバイスのID、後述)
$ export SWITCHBOT_TOKEN=yyyyyyyy(SwitchBotアプリから取得)
$ export SWITCHBOT_SECRET=zzzzzz(SwitchBotアプリから取得)

NGSI-SwitchBotAPI変換プログラム(下記項目)を実行させます
```
$ python3 ./ngsi_switchbotapi_converter.py
```

FIWARE環境での使い方

NGSI-SwitchBotAPI変換プログラムは、コンテキストブローカーの特定のエンティティを監視(Subscription)し、値が変更された場合にSwitchBotAPIを叩いてSwitchBotデバイスを制御します。

以下の環境変数をセットしてから、

$ export ORION_URL=https://orion.example.com
$ export CONTEXT_PROVIDER_URL=http://orion.example.com:8082
$ export ORION_TOKEN=xxxxxxxxxx(認証系がある場合)

以下のスクリプトを実行して、Subscription登録してください。

create-subscription_switchbot.sh

#!/bin/bash
: ${ORION_URL:?Not found}
: ${CONTEXT_PROVIDER_URL:?Not found}
: ${ORION_TOKEN:?Not found}
curl -iX POST \
  "$ORION_URL/v2/subscriptions" \
  --header "Authorization: Bearer $ORION_TOKEN" \
  --header "Content-Type: application/json" \
  --data "{
    \"description\": \"Notify to SwichBot API Converter\",
    \"subject\": {
      \"entities\": [
        {
	        \"idPattern\": \"urn:ngsi-ld:Light:001\",
          \"type\": \"Light\"
        }
      ],
      \"condition\": {
        \"attrs\": [ \"power\" ],
        \"alterationTypes\": [ \"entityUpdate\" ]
      }
    },
    \"notification\": {
      \"http\": {
        \"url\": \"$CONTEXT_PROVIDER_URL/v1/Light/command\",
        \"attrs\": [ \"power\" ]
      }
    }
  }"

実際に実行させるにはこんな感じ。
$ bash ./create-subscription_switchbot.sh

orion context brokerをローカル環境で動かしているなど、認証系がない場合は、ORION_TOKENの設定は不要です。スクリプトの３行目と６行目は削除してください。

照明の点灯・消灯

サブスクリプション登録した状態で、urn:ngsi-ld:Light:001のpowerアトリビュート値を0もしくは1に変更すると、それに対応してSwitchbotデバイス(赤外線リモコン)がTurnOn/TurnOffされます。

以下はngsi goコマンドを利用する場合の例です。

点灯させる場合

$ ngsi update attr --id urn:ngsi-ld:Light:001 --attr power --data 1

消灯させる場合

$ ngsi update attr --id urn:ngsi-ld:Light:001 --attr power --data 0

SwitchBotのデバイスIDを取得する方法

SwitchBotのデバイスをAPI経由で操作するためには、対象装置のデバイスIDを知る必要があります。
デバイスIDを取得・表示するコードを書きましたのでご利用ください。

switchbot_devicelist.py

import os
import requests
import json
import time
import hashlib
import hmac
import base64
import uuid

def getDeviceList():
  token = os.environ.get('SWITCHBOT_TOKEN')
  secret = os.environ.get('SWITCHBOT_SECRET')

  # Declare empty header dictionary
  apiHeader = {}
  # open token
  #token = '' # copy and paste from the SwitchBot app V6.14 or later
  # secret key
  #secret = '' # copy and paste from the SwitchBot app V6.14 or later
  nonce = uuid.uuid4()
  t = int(round(time.time() * 1000))
  string_to_sign = '{}{}{}'.format(token, t, nonce)

  string_to_sign = bytes(string_to_sign, 'utf-8')
  secret = bytes(secret, 'utf-8')

  sign = base64.b64encode(hmac.new(secret, msg=string_to_sign, digestmod=hashlib.sha256).digest())

  #print ('Authorization: {}'.format(token))
  #print ('t: {}'.format(t))
  #print ('sign: {}'.format(str(sign, 'utf-8')))
  #print ('nonce: {}'.format(nonce))

  #Build api header JSON
  apiHeader['Authorization']=token
  apiHeader['Content-Type']='application/json'
  apiHeader['charset']='utf8'
  apiHeader['t']=str(t)
  apiHeader['sign']=sign
  apiHeader['nonce']=str(nonce)

  response = requests.get("https://api.switch-bot.com/v1.1/devices", headers=apiHeader)
  devices  = json.loads(response.text)

  if(debug):
    print(devices)

  for device in devices['body']['deviceList']:
    print(device)
  for pseudoDevice in devices['body']['infraredRemoteList']:
    print(pseudoDevice)

if __name__ == '__main__':
  #debug = True
  debug = False

  print(getDeviceList())

コード

NGSIのSubscription通知を受けて、SwitchBotAPIを操作するプログラムコードです。

ngsi_switchbotapi_converter.py

import os
import sys
import signal
from http.server import BaseHTTPRequestHandler, HTTPServer
import requests
import json
import time
import hashlib
import hmac
import base64
import uuid

def send_command(command):
  token = os.environ.get('SWITCHBOT_TOKEN')
  secret = os.environ.get('SWITCHBOT_SECRET')

  # Declare empty header dictionary
  apiHeader = {}
  nonce = uuid.uuid4()
  t = int(round(time.time() * 1000))
  string_to_sign = '{}{}{}'.format(token, t, nonce)

  string_to_sign = bytes(string_to_sign, 'utf-8')
  secret = bytes(secret, 'utf-8')

  sign = base64.b64encode(hmac.new(secret, msg=string_to_sign, digestmod=hashlib.sha256).digest())

  #Build api header JSON
  apiHeader['Authorization']=token
  apiHeader['Content-Type']='application/json'
  apiHeader['charset']='utf8'
  apiHeader['t']=str(t)
  apiHeader['sign']=sign
  apiHeader['nonce']=str(nonce)

  body={"command" : command,
        "parameter" : "default",
        "commandType" : "command"}

  if(debug):
    print(body)

  response = requests.post("https://api.switch-bot.com/v1.1/devices/"+ switchBotDeviceId + "/commands", data=json.dumps(body), headers=apiHeader)
  result = json.loads(response.text)

  if(debug):
    print(result)

def handler(signum, frame):
  sys.exit(0)

class SwicthBotConverterHandler(BaseHTTPRequestHandler):
  def do_POST(self):
    content_length = int(self.headers['content-length'])
    path = self.path
    if (path != '/v1/Light/command'):
      self.send_response(404)
      return

    body = json.loads(self.rfile.read(content_length).decode('utf-8'))
    lightId = body['data'][0]['id']
    lightPower = body['data'][0]['power']['value']

    if (lightId != ngsiDeviceId):
      self.send=response(404)
      return

    if (lightPower == 1):
      send_command('turnOn')
    elif (lightPower == 0):
      send_command('turnOff')

    if (debug):
      print('path = {}'.format(path))
      print('body = {}'.format(body))
      print('id = {}'.format(lightId))
      print('power = {}'.format(lightPower))

    response = [{"Status":"OK"}]

    self.send_response(200)
    self.send_header('Content-Type', 'application/json; charset=utf-8')
    self.end_headers()
    self.wfile.write(json.dumps(response).encode('utf-8'))

if __name__ == '__main__':
  #debug = True
  debug = False

  signal.signal(signal.SIGTERM, handler)
  signal.signal(signal.SIGINT, handler)

  ngsiDeviceId = "urn:ngsi-ld:Light:001"
  switchBotDeviceId = os.environ.get('SWITCHBOT_DEVICE')

  print('Start SwitchBot API Converter')
  address = ('0.0.0.0', 8082)
  with HTTPServer(address, SwicthBotConverterHandler) as server:
    server.serve_forever()

今後

SwitchBotデバイスからのデータ通知（コンテキストブローカーへの登録）は近いうちに実装したいと思っています。

まとめ

SwitchBotの装置を制御できるとスマートホームとしてできることがかなり広がります。
NGSIを使うと、特定のメーカーの装置で固める必要がなく、いろいろなものを組み合わせて使えますので、便利だと思います。
是非試してみてくださいね。

参考文献

ここ↓に書いてあるやり方は古くなっています。現在はSwitchBotAPIはV1.1に上がっています。(互換性維持のため、現状ではv1でも動きます。)

SwitchBot APIの本家ドキュメントはこちら。

NGSI、OrionのAPI詳細についてはこちら。

「FIWAREでスマートホームする」第一弾

「FIWAREでスマートホームする」第二弾