【ネタ】絶対に監視アラートに気づけるお風呂、できました

Posted at 2024-12-08

システムのアラートはメールや電話やSlackなどに届きますよね。

つまりスマホが手元にないと大ピンチ！入浴中にアラートが鳴ったら大変です。

1秒でも早く気づけるように、今日からアラートをお風呂に通知しましょう!!

お風呂にスマホを持ち込めば良いのでは？というツッコミはだめです。

準備

まずは電球をSwitchBotのスマート電球に取り替えます。

試しにSwithBotアプリから操作してみます。

準備はバッチリです。

アーキテクチャ

汎用性を考えてメールをトリガーにします。DatadogやNew Relicなどの監視SaaSで、アラートの通知先にメールアドレスを登録して使います。

ちなみにこんなことをしなくても、SwitchBotはIFTTTと連携できます。

実装

Lambda関数を作る

APIを叩いて電球の色をrgb(255,0,0)にするコードを書きます。

main.py（一部を抜粋し改変）

device_id, token, secret = _get_secrets()
api_header = _create_header(token, secret)

url = f'https://api.switch-bot.com/v1.1/devices/{device_id}/commands'
payload = {
  'command': 'setColor',
  'parameter': '255:0:0'
}

requests.post(url, headers=api_header, json=payload)

コード全文（クリックして展開）

main.py

import base64
import hashlib
import hmac
import json
import time
import uuid

import botocore
import botocore.session
import requests
from aws_secretsmanager_caching import SecretCache, SecretCacheConfig


# Secret Managerからシークレットを取得（面倒くさい場合は環境変数などでどうぞ）
# ref. https://github.com/aws/aws-secretsmanager-caching-python

client = botocore.session.get_session().create_client('secretsmanager')
cache_config = SecretCacheConfig()
cache = SecretCache( config = cache_config, client = client)

def _get_secrets():

  secret_string = cache.get_secret_string('switchbot-led')
  secret_json = json.loads(secret_string)
  device_id = secret_json['device_id']
  token = secret_json['token']
  secret = secret_json['secret']

  return device_id, token, secret


# リクエストヘッダーの作成（認証のため）
# ref. https://github.com/OpenWonderLabs/SwitchBotAPI

def _create_header(token, secret):

  api_header = {}

  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())

  api_header['Authorization']=token
  api_header['Content-Type']='application/json'
  api_header['charset']='utf8'
  api_header['t']=str(t)
  api_header['sign']=str(sign, 'utf-8')
  api_header['nonce']=str(nonce)

  return api_header


# メインの処理

def lambda_handler(event, context):

  device_id, token, secret = _get_secrets()

  api_header = _create_header(token, secret)

  url = f'https://api.switch-bot.com/v1.1/devices/{device_id}/commands'
  payload = {
    'command': 'setColor',
    'parameter': '255:0:0'
  }

  response = requests.post(url, headers=api_header, json=payload)
  return f"{response.status_code}: {response.json()['message']}"


if __name__ == "__main__":
  lambda_handler(None, None)