km2-system

明日の天気が雨か晴れかによってアクションを起こせないか、という要望をとある掲示板で目にしたので調べたところ、お天気APIというのがあるのを知りました。
自分にはこれといって必要性はないものの、面白そうなので勝手に作ってみました。（特に周知せず）

参考にしたページ
天気予報 API (livedoor 天気互換)のJSONを表示させてみた
https://four4to6.x0.com/800/

【Python】無料で使える天気予報APIで降水確率を取得する
https://creepfablic.site/2021/11/14/python-weather-api/

step1:仮想センサーの増設

雨の場合と晴れの場合で別の定型アクションを動かそうと考え、「スイッチでAlexa定型アクションを起動～スキル作成編～」で作成した仮想センサーを増やしました。
詳細は以下を参照してください。ここでは簡潔に記述します。
スイッチでAlexa定型アクションを起動～スキル作成編～
https://km2-system.com/2022/06/28/alexa_switch_2/

スキルのAWS関数のindex.js内記載のセンサー数を増やします。
sensor_num=10 を sensor_num=30 に変更して「Deploy」します。（数値は必要数にします。自分の場合、他にもやりたいことがあるため、30まで増やしました）

Alexaアプリでデバイスの検出を行うと追加したダミーセンサーが検出されます。既にスキルが有効になっているのであれば無効/有効等の操作は必要ありません。
※デバイスの検出方法：Alexaアプリの[デバイス]タブで一番上の右側＋ – [デバイスの追加] – [その他]

step2:pythonプログラム

プログラムは参照ページとほとんど同じですが、もし他の方が使うときに変更する可能性があるところを中心に若干の説明を書きます。

今回、雨の時に起動するセンサー番号は22、晴れの場合は23としました。
rain_sensor_no = 22
shine_sensor_no = 23

地点コードは以下から取得します。
https://weather.tsukumijima.net/primary_area.xml
札幌の地点コードは016010でした。
city_code = “016010” #札幌市

天気は日中の予報で判断します。６時から１２時までの予報と１２時から１８時までの予報どちらかが30%以上の場合、雨と判断するようにしました。
threshold = 30

確認用に取得した天気情報はresponse.txtというファイルに保存しておきます。
response_file = ‘./response.txt’

仮想センサーの起動は「スイッチでAlexa定型アクションを起動～スキルテスト編～(https://km2-system.com/2022/06/29/alexa_switch_4/)」で作成したalexa_trigger.pyをちょっとだけ改造し、関数化してimport可能なファイルにしました。
仮想センサーを起動するプログラムは以下のalexa_trigger_func.pyファイルをimportし、センサー番号を引数に渡して関数alexa_trigger()を呼べば良いだけになります。
※client_ID = 、client_secret = にはスキルビルド画面のアクセス権限メニューから取得したAlexaクライアントIDとAlexaクライアントシークレットを貼り付けます。(index.jsと同じもの) 「スイッチでAlexa定型アクションを起動～スキルテスト編～」参照

alexa_trigger_func.py

import sys
import os
import json
import datetime
from trace import Trace
import requests

client_ID = ''
client_secret = ''
token_file = './token.txt'

def alexa_trigger(no):
  sensor_name = 'dummy_sensor_' + format(no, '0>3')
  # アクセストークンが有効かどうかチェックする
  access_token = ''
  access_token_valid = False
  if os.path.exists(token_file):
    #ファイル有、データ読み込み
    f = open(token_file)
    data = f.read()
    f.close()
    jdata = json.loads(data)
    expires = jdata['expires_in']
    f_time = os.path.getmtime(token_file)
    file_time = datetime.datetime.fromtimestamp(f_time)
    expiring_date = datetime.datetime.fromtimestamp(f_time + expires)
    today = datetime.datetime.today()
    if today < expiring_date:
      # アクセストークン有効、取得
      access_token_valid = True
      access_token = jdata['access_token']

  # アクセストークンが無効の場合はoa2サーバーから取得
  if not access_token_valid:
    header_str = {
      'Content-Type': 'application/x-www-form-urlencoded;Accept-Charset=UTF-8'
    }
    body_str = {
      "grant_type": "refresh_token",
      "refresh_token": refresh_token,
      "client_id": client_ID,
      "client_secret": client_secret
    }
    response = requests.post('https://api.amazon.com/auth/o2/token', data = body_str, headers = header_str)
    if response.status_code != 200 :
      print(response.status_code)
      exit()
    jdata = json.loads(response.text)
    access_token = jdata['access_token']
    if len(access_token) > 0:
      access_token_valid = True
      # ファイルに書き込み
      f = open(token_file, 'w')
      f.write(response.text)
      f.close()

  if not access_token_valid:
    print('can not get access_token')
    exit()

  # イベントゲートウェイにイベントを送信する
  message_id = sensor_name + '_' + str(today)
  timeOfSample = today.strftime('%Y-%m-%dT%H:%M:%S.00Z')
  header_str = {
      'Authorization': 'Bearer ' + access_token,
      'Content-Type': 'application/json'
  }
  json_str = {
    "event":{
      "header":{
        "namespace":"Alexa",
        "name":"ChangeReport",
        "messageId":message_id,
        "payloadVersion":"3"
      },
      "endpoint":{
        "scope":{
          "type":"BearerToken",
          "token":access_token
        },
        "endpointId":sensor_name
      },
      "payload":{
        "change":{
          "cause":{
            "type":"PHYSICAL_INTERACTION"
          },
          "properties":[
            {
              "namespace":"Alexa.MotionSensor",
              "name":"detectionState",
              "value":"DETECTED",
              "timeOfSample":timeOfSample,
              "uncertaintyInMilliseconds":0
            }
          ]
        }
      }
    },
    "context":{
      "properties": [
        {
          "namespace": "Alexa.EndpointHealth",
          "name": "connectivity",
          "value": {
            "value": "OK"
          },
          "timeOfSample": timeOfSample,
          "uncertaintyInMilliseconds": 0
        }
      ]
    }
  }

  response = requests.post('https://api.fe.amazonalexa.com/v3/events', json = json_str, headers = header_str)
  if response.status_code == 202:
    print('success')
  else:
    print(response)

天気を予報するプログラムは以下のようになります。
weather.py

import requests
import json
import alexa_trigger_func

city_code = "016010"    #札幌市
threshold = 30
rain_sensor_no = 22
shine_sensor_no = 23
response_file = './response.txt'

# URL
url = "https://weather.tsukumijima.net/api/forecast/city/" + city_code
try:
    # 天気予報取得
    response = requests.get(url)
    # 通信が正常でなければ例外発生
    response.raise_for_status()
except requests.exceptions.RequestException as e:
    # エラー処理
    res_text = "Error:{}".format(e)
    print(res_text)
else:
    res_text = response.text
    # json解析
    weather_json = response.json()
    # データ取得
    yesterday = weather_json["forecasts"][1]["date"]
    chanceOfRain1 = weather_json["forecasts"][1]["chanceOfRain"]["T06_12"]
    chanceOfRain2 = weather_json["forecasts"][1]["chanceOfRain"]["T12_18"]

    am_per = int(chanceOfRain1[:-1])
    pm_per = int(chanceOfRain2[:-1])
    if am_per >= threshold or pm_per >= threshold:
        alexa_trigger_func.alexa_trigger(rain_sensor_no)
    else:
        alexa_trigger_func.alexa_trigger(shine_sensor_no)
finally:
    # fileに保存
    f = open(response_file, 'w')
    f.write(res_text)
    f.close()

step3:起動される定型アクションの作成

雨の場合と晴れの場合でそれぞれ仮想センサーの検知に対応する定型アクションを作成します。
実行条件で[スマートホーム]から該当するダミーセンサーを選択します。
アクションはとりあえずアレクサが「明日は雨です」、「明日は晴れです」と喋るだけにします。

step4:pythonプログラムの動作確認

同じフォルダにweather.pyをalexa_trigger_func.pyを置き、以下のコマンドを打ち込み動作確認します。pythonがインストールされているならばlinuxでもWindowsでも動作します。
$ python weather.py⏎
Alexaが「明日は雨です」、「明日は晴れです」などと喋れば成功です。
response.txtの内容も確認します。
$ nano response.txt

step5:pythonプログラムの起動設定について

24時間起動したラズベリーパイにプログラムを置き、毎日決まった時間にプログラムを起動するようにします。
crontabなどlinuxのシステムでそれを実現することは可能ですが、メンテナンス性という観点ではAlexaアプリの定型アクションを使う方が良いと考えました。
時刻で起動する定型アクションからNode-REDを使ってpythonプログラムを起動します。全体の流れは以下のようになります。
定型アクション(時刻起動) -> Node-RED -> pythonプログラム -> (晴/雨)定型アクション

step6:Node-REDにデバイス作成

詳細は以下を参照して下さい。ここでは簡潔に書きます。
Alexaによる野球チャンネルの自動化2(Nature API)
https://km2-system.com/2022/06/13/auto_tv_channnel_2/

Node-RED Alexa Home Skill Bridgeのページ(https://alexa-node-red.bm.hardill.me.uk/)にログインします。
[Add Device]ボタンを押下して新しいデバイスを作成します。

NameはここではRaspberryPiLiving2としました。
起動するだけなのでActionsにはOnだけにチェックします。
Action TypeはSWITCHにします。
[OK]ボタンを押下して設定します。

Alexaアプリでデバイスの検出を行い、Node-REDのデバイスを追加します。

step7:ラズベリーパイのNode-REDでフロー作成

ラズベリーパイにNode-REDをインストールしていない場合はインストールします。
※参照；Alexaによる野球チャンネルの自動化2(Nature API)

ブラウザを起動し、http://xxx.xxx.xxx.xxx:1880を開きます。xxx.xxx.xxx.xxxはラズベリーパイのIPアドレスです。ラズベリーパイでも別のPCからでも構いません。
Node-REDにalexa-home-skillを追加していない場合は追加します。
※参照；Alexaによる野球チャンネルの自動化2(Nature API)

パレットから[alexa-home]を配置し、追加したデバイス(上記の例ではRaspberryPiLiving2)を設定します。nodeの表示が[alexa-home] から[RaspberryPiLiving2]に変わります。

パレットから[exec]を配置し、[RaspberryPiLiving2]と繋げます。
execノードをダブルクリックし、コマンドを設定します。
python /home/pi/weather.py ディレクトリ’/home/pi/’の部分は自分の環境に合わせます。
名前にweather triggerなどと入れ[完了]ボタンを押します。

「デプロイ」ボタンでフローを送信します。

step8:alexa定型アクションでフローの起動を設定

定型アクションを作成します。
実行条件の設定で[スケージュール]を選択し、[繰り返し]と[時刻]を設定します。
アクションを追加で[スマートホーム]-[全てのデバイス]-[RaspberryPiLiving2]を選択します。

私はこれを夜寝る時間に設定し、そろそろ寝るだよという合図に使っております。

ダイソーのテンキーで定型アクションを起動することはできました。
注文したラズベリーパイゼロ２も到着しました。モバイル機器にできないかと思案したところ、キャン★ドゥのスマホ充電器（税込550円）を使うことを思いつきました。ラズパイゼロ２に繋げたところ、問題なく動きます。

良い箱がないかと物色したところ、キャン★ドゥに丁度良い木箱がありました。

テンキーがピタリと入ります。

木箱にゼロ２の基盤を押し付け、ネジ穴をペンでなぞり、プロクソンのテーブルドリルで2.5mmの穴をあけました。

ゼロ２をネジで止めます。

テンキーはUSB-Aとmicro-Bへの変換コネクタ経由でゼロ２に接続しているのですが、ケーブルが長すぎ、コネクタが邪魔すぎるので切ってしまいました。

テンキー側は赤黒白青、変換ケーブル側は赤黒白緑です。赤-赤、黒-黒、白-白、青-緑と繋げれば良いのかなと、試しに繋いでみましたが動きません。テンキーのLEDすら点きません。他の組み合わせの接続も試してみたのですが、うまくいきませんでした。
最後に思いついたのは、切り取ってしまったコネクタ同士を繋いで導通を測る、でした。

赤-黒、黒-緑、白-赤、青-白を繋げばよいことがわかりました。
変換コネクタの方が普通の色の使い方をしていないです。
テスト後はんだ付けして動作確認しましたが、うまく動きません。調べたところ、キャン★ドゥのSDカードが壊れていることがわかりました。不良品と言うより、接続を試しているときにショートさせてしまい、それが原因で壊してしまった感じがします。
Amazon最安値32GBのものに交換して再セットアップし、動作確認しました。

テンキーで箱の蓋をして使おうということで、テンキーと電池のがたつき防止のために、紙粘土を使うことにしました。これもキャン★ドゥで購入。

ピンク色のケーブルは充電用です。

テンキー裏の針金状のスタンドは電気的に危なそうなので外しました。

テンキーを上に乗せると意外とイイ感じです。

輪ゴムでテンキーを押さえて完成です。

やっつけ感が凄いですが、運用にいろいろ懸念事項があるので当面これで使ってみます。

懸念事項
・電源スイッチはあった方がいいのでは？⇒必要。
・電池の持ち、充電間隔など⇒24時間くらいしか持たない。充電時間は６時間ほど。
・蓋開けが必要となる頻度⇒充電中はバッテリのインジケーターが見たい。

(2022/8/8追記)
電源をブチっと切るとSDカードが壊れて起動できなくなる可能性があることがわかりました。
LegacyのBuster版では対策されているようですが、Bullseye版でもそれらしいメニューがあるのでSDを書き込み禁止にし、一応の電源ぶつ切り対策をしてみました。

あらかじめ、/home/pi/にあるtoken.txtを削除します。
sudo raspi-config⏎
[Performance Option]-[Overlay File System]を選択し、Would you like the overlay file system to be enabled?をにします。
再起動すると、SDは書き込み禁止になっています。
token.txtは作成されますがRAM上にあり、再起動すると元のファイルがない状態に戻っています。

キーボード入力から定型アクションが起動できたので、GPIO入力からも当然同じことができる筈です。防犯カメラの時に試しに買っていた人感センサがあるのでそれで試してみました。

以下のページを参照しました。

【Raspberry Pi】自作人感センサーの使い方と活用法
https://chasuke.com/motionsensor/

メスーメスのケーブルを持ってなかったのでブレッドボードを使いました。

ページに掲載したプログラムを動かすとちゃんと動きました。
alexa_trigger.pyは関数化して同じプログラム内に入れ込みました。問題なく動作しました。
以下、プログラムです。

import sys
import os
import json
import datetime
from trace import Trace
import requests
from datetime import datetime
import time
import RPi.GPIO as GPIO

def detect_sensor( no ):
  client_ID = ''
  client_secret = ''
  refresh_token=''
  sensor_name = 'dummy_sensor_' + format(no, '0>3')
  token_file = './token.txt'
  today = datetime.today()
  
  # アクセストークンが有効かどうかチェックする
  access_token = ''
  access_token_valid = False
  if os.path.exists(token_file):
    #ファイル有、データ読み込み
    f = open(token_file)
    data = f.read()
    f.close()
    jdata = json.loads(data)
    expires = jdata['expires_in']
    f_time = os.path.getmtime(token_file)
    file_time = datetime.fromtimestamp(f_time)
    expiring_date = datetime.fromtimestamp(f_time + expires)
    if today < expiring_date:
      # アクセストークン有効、取得
      access_token_valid = True
      access_token = jdata['access_token']
  
  # アクセストークンが無効の場合はoa2サーバーから取得
  if not access_token_valid:
    header_str = {
      'Content-Type': 'application/x-www-form-urlencoded;Accept-Charset=UTF-8'
    }
    body_str = {
      "grant_type": "refresh_token",
      "refresh_token": refresh_token,
      "client_id": client_ID,
      "client_secret": client_secret
    }
    response = requests.post('https://api.amazon.com/auth/o2/token', data = body_str, headers = header_str)
    if response.status_code != 200 :
      print(response.status_code)
      exit()
    jdata = json.loads(response.text)
    access_token = jdata['access_token']
    if len(access_token) > 0:
      access_token_valid = True
      # ファイルに書き込み
      f = open(token_file, 'w')
      f.write(response.text)
      f.close()

  if not access_token_valid:
    print('can not get access_token')
    exit()
  
  # イベントゲートウェイにイベントを送信する
  message_id = sensor_name + '_' + str(today)
  timeOfSample = today.strftime('%Y-%m-%dT%H:%M:%S.00Z')
  header_str = {
      'Authorization': 'Bearer ' + access_token,
      'Content-Type': 'application/json'
  }
  json_str = {
    "event":{
      "header":{
        "namespace":"Alexa",
        "name":"ChangeReport",
        "messageId":message_id,
        "payloadVersion":"3"
      },
      "endpoint":{
        "scope":{
          "type":"BearerToken",
          "token":access_token
         },
         "endpointId":sensor_name
      },
      "payload":{
        "change":{
          "cause":{
            "type":"PHYSICAL_INTERACTION"
          },
          "properties":[
            {
              "namespace":"Alexa.MotionSensor",
              "name":"detectionState",
              "value":"DETECTED",
              "timeOfSample":timeOfSample,
              "uncertaintyInMilliseconds":0
            }
          ]
        }
      }
    },
    "context":{
      "properties": [
        {
          "namespace": "Alexa.EndpointHealth",
          "name": "connectivity",
          "value": {
            "value": "OK"
          },
          "timeOfSample": timeOfSample,
          "uncertaintyInMilliseconds": 0
        }
      ]
    }
}
  
  response = requests.post('https://api.fe.amazonalexa.com/v3/events', json = json_str, headers = header_str)
  if response.status_code == 202:
    print('success')
  else:
    print(response)


# インターバル
INTERVAL = 3
# スリープタイム
SLEEPTIME = 20
# 使用するGPIO
GPIO_PIN = 18

GPIO.setmode(GPIO.BCM)
GPIO.setup(GPIO_PIN, GPIO.IN)

if __name__ == '__main__':
    try:
        print ("処理キャンセル：CTRL+C")
        cnt = 1
        while True:
            # センサー感知
            if(GPIO.input(GPIO_PIN) == GPIO.HIGH):
                print(datetime.now().strftime('%Y/%m/%d %H:%M:%S') +
                "：" + str("{0:05d}".format(cnt)) + "回目の人感知")
                detect_sensor( '0' )
                cnt = cnt + 1
                time.sleep(SLEEPTIME)
            else:
                print(GPIO.input(GPIO_PIN))
                time.sleep(INTERVAL)
    except KeyboardInterrupt:
        print("終了処理中...")
    finally:
        GPIO.cleanup()
        print("GPIO clean完了")

タクトスイッチ等を使っても当然できることでしょう。pico Wの発売が待たれます。

ラズベリーパイをセットアップします。

Step1:OSのインストール

ラズベリーパイのホームページ(https://www.raspberrypi.com/software/)からimagerをダウンロードします。imagerはOSのバージョンが固定されているようなのでOSがバージョンアップされるとimagerもダウンロードしなければならないようです。v.1.7.2で説明します。
imagerを起動し、OSを書き込みます。
CLI版を使用するのでOSは[Raspberry Pi OS(other)]-[Raspberry Pi OS Lite(32bit)]を選択します。
書き込みが終わったらSDをラズベリーパイに差し替え、電源を入れます。
キーボードの設定画面(Configuring Keyboard-Configuration)が現れます。日本語の普通のキーボードの場合、[Other]-[Japanese]-[Japanese – Japanese(OADG 109A)]を選択します。
ユーザー名とパスワード入力します。
コマンドプロンプトが出来たら設定を行います。
$ sudo raspi-config⏎

wifi設定:[1 System Options]-[S1 Wireless LAN]-[JP Japan]-[SSIDの入力]-[パスワードの入力]
SSH設定:[3 Interface Options]-[I2 SSH]-[YES]
ファイルシステムの拡張:[6 Advanced Options]-[A1 Expand Filesystem]

最新バージョンへアップデートします。
$ sudo apt update⏎
$ sudo apt upgrade⏎ Y/nでY

Step2:ファイルの転送

スキルテスト編で作成したpythonプログラム(alexa_trigger.py)とシェルスクリプトをラズベリーパイのホームディレクトリに転送します。
シェルスクリプトは全く単純です。
ファイル名はalexa_key.shとしました。

#!/usr/bin/bash

while :
do
  read CMD
  case "$CMD" in
  "0") python alexa_trigger.py 0;;
  "1") python alexa_trigger.py 1;;
  "2") python alexa_trigger.py 2;;
  "3") python alexa_trigger.py 3;;
  "4") python alexa_trigger.py 4;;
  "5") python alexa_trigger.py 5;;
  "6") python alexa_trigger.py 6;;
  "7") python alexa_trigger.py 7;;
  "8") python alexa_trigger.py 8;;
  "9") python alexa_trigger.py 9;;
esac
done

シェルスクリプトに実行権限を与えます。
$ chmod 777 alexa_key.sh
試しにテストしてみます。
$ ./alexa_key.sh
0⏎
success

成功したら、起動時にシェルスクリプトが起動されるようにします。
ホームディレクトリにある.profileファイルを編集します。
$ nano .profile⏎
最後の行に１行追加します。
./alexa_key.sh

Ctrl + O⏎
Ctrl + W⏎
で保存します。

キーボードをテンキーに付け替え、再起動します。
これで完了です。

定型アクションを作成すれば10個のアクションがキーで動作します。
Lambda関数とシェルスクリプトに少し修正を加えれば動作するアクションを増やすことができるはずです。

.profileに変更を加えたことでSSH接続時にもシェルスクリプトalexa_key.shが実行された状態になっています。まず最初にCtrl+C⏎でシェルスクリプトを終了させる必要があります。