はじめに

この記事は全4章からなる記事の第4章です。

動作イメージはYouTubeで確認できるので是非ご覧ください。

きっかけ

日本で8000万人が利用し、UIに馴染みのあるLINEでインタラクティブにSORACOMデバイスで収集したデータを確認したり、家電の操作ができたらと思いました。

使ったもの

GPSマルチユニットSORACOM Edition
SORACOM LTE-M Button Plus
- 接点入力は使用していないので実質LTE-M Button単体
Raspberry Pi 3 Model A+
- 人感センサー（焦電型赤外線センサー）を接続
- サーボモータを接続
Raspberry Pi 3 Model B+
- サーバーとして使用
- 赤外線学習リモコン(ADRSIR)を接続

環境

Python 3.7.3
Flask==1.1.2
line-bot-sdk==1.16.0
firebase-admin==4.3.0
Pillow==7.2.0
paramiko==2.7.1

インストール

$ pip install flask
$ pip install line-bot-sdk
$ pip install firebase-admin
$ pip install pillow
$ pip install paramiko

主な機能

温度と湿度を表示
- GPSマルチユニットSORACOM Editionで測った温度と湿度をLINEで閲覧します
人感センサーデータを表示
- Firebase Realtime Databaseに格納されたデータをフェッチしてLINEで閲覧します
緊急事態発生時にPushメッセージを送信
- SORACOM LTE-M Buttonが押されるとLINEにPushメッセージを送ります
エアコンの制御
- 熱中症の危険が高くなると自動でエアコンをONにします
- エアコンをONにする条件は「気温30℃・湿度60%以上」かつ「人感センサーが15分以内に反応」です
- 誤作動の場合に備えてエアコンOFF操作もできます

それぞれについてソースコードとともに解説します。

まずは全体像から

ソース全体になります。以降、モジュールのimportやアクセストークンの部分は掲載しないので気を付けてください。

ソースコード

クリックして開く

line_bot.py

from flask import Flask, request, abort
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage, FlexSendMessage
import subprocess
import os
import json
import time
import datetime
import base64
import requests
import ast
import paramiko
import firebase_admin
from firebase_admin import credentials
from firebase_admin import db
from PIL import ImageFont, Image, ImageDraw
from image import add_text_to_image #自作モジュール

app = Flask(__name__)

# LINE Messaging API Settings
LINE_BOT_ACCESS_TOKEN = os.environ["LINE_BOT_ACCESS_TOKEN"]
LINE_BOT_CHANNEL_SECRET = os.environ["LINE_BOT_CHANNEL_SECRET"]

line_bot_api = LineBotApi(LINE_BOT_ACCESS_TOKEN)
handler = WebhookHandler(LINE_BOT_CHANNEL_SECRET)

user_id = "U0..." #メッセージをPushするユーザーのID

FQDN = 'https://xxx.ngrok.io' #ngrokのURL

# Firebase Settings
cred = credentials.Certificate("<secret key file>.json")
firebase_admin.initialize_app(cred, {
    'databaseURL': 'https://xxx.firebaseio.com/'
})

ref = db.reference('data')

# ssh settings
HOST = '192.168.11.xxx'
PORT = 22
USER = 'username'
KEY_FILE = '../.ssh/<secret_key_file>' #相対パス

@app.route("/webhook", methods=['POST'])
def webhook():
    print(json.dumps(request.get_json(), indent=2))
    object = request.get_json()
    if object['title'] == "[Alerting] Emergency alert":
        json_message = {
            "type": "bubble",
            "hero": {
                "type": "image",
                "url": "https://xxxx.ngrok.io/static/sos.png",
                "size": "full",
                "aspectRatio": "16:9",
                "aspectMode": "cover"
            },
            "body": {
                "type": "box",
                "layout": "vertical",
                "contents": [
                {
                    "type": "text",
                    "text": "緊急ボタンが押されました",
                    "weight": "bold",
                    "size": "lg",
                    "color": "#E9462B",
                    "align": "center"
                },
                {
                    "type": "box",
                    "layout": "vertical",
                    "margin": "lg",
                    "spacing": "sm",
                    "contents": [
                    {
                        "type": "box",
                        "layout": "baseline",
                        "spacing": "sm",
                        "contents": [
                        {
                            "type": "text",
                            "text": "場所",
                            "color": "#aaaaaa",
                            "size": "sm",
                            "flex": 1
                        },
                        {
                            "type": "text",
                            "text": "脱衣所",
                            "wrap": True,
                            "color": "#666666",
                            "size": "sm",
                            "flex": 5
                        }
                        ]
                    },
                    {
                        "type": "box",
                        "layout": "vertical",
                        "margin": "lg",
                        "spacing": "sm",
                        "contents": [
                        {
                            "type": "box",
                            "layout": "baseline",
                            "spacing": "sm",
                            "contents": [
                            {
                                "type": "text",
                                "text": "緊急措置として玄関の鍵を開錠しました",
                                "color": "#4764a6",
                                "size": "md",
                                "flex": 1,
                                "wrap": True
                            }
                            ]
                        }
                        ]
                    }
                    ]
                }
                ]
            },
            "footer": {
                "type": "box",
                "layout": "vertical",
                "spacing": "sm",
                "contents": [
                {
                    "type": "button",
                    "style": "primary",
                    "height": "sm",
                    "action": {
                    "type": "message",
                    "label": "救急",
                    "text": "救急"
                    },
                    "color": "#E9462B"
                },
                {
                    "type": "spacer",
                    "size": "sm"
                }
                ],
                "flex": 0
            }
        }
        messages = FlexSendMessage(alt_text='[SOS] 緊急ボタンが押されました', contents=json_message)
        line_bot_api.push_message(user_id, messages=messages)

        key = paramiko.ECDSAKey.from_private_key_file(KEY_FILE)
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(HOST, PORT, USER, pkey=key)
        ssh.exec_command('python3 key_open.py')

    elif object['title'] == "[Alerting] Temperature & Humidity alert":
        current_time = int(time.time()*1000)
        fifteen_minutes_ago = current_time - 900000

        data = ref.order_by_key().limit_to_last(1).get()
        for key, val in data.items():
            if val['timestamp']  >= fifteen_minutes_ago:
                json_message = {
                    "type": "bubble",
                    "hero": {
                        "type": "image",
                        "url": "https://xxx.ngrok.io/static/aircon.png",
                        "size": "full",
                        "aspectRatio": "16:9",
                        "aspectMode": "cover"
                    },
                    "body": {
                        "type": "box",
                        "layout": "vertical",
                        "contents": [
                        {
                            "type": "text",
                            "text": "エアコンをつけました",
                            "weight": "bold",
                            "size": "xl",
                            "color": "#7077BE"
                        },
                        {
                            "type": "box",
                            "layout": "vertical",
                            "contents": [
                            {
                                "type": "text",
                                "text": "熱中症の危険性が高い温度・湿度です。",
                                "size": "xs",
                                "wrap": True
                            },
                            {
                                "type": "text",
                                "text": "15分以内に人感センサーが反応したため、家にいると判断し冷房をつけました。",
                                "size": "xs",
                                "wrap": True
                            }
                            ],
                            "margin": "sm"
                        }
                        ]
                    },
                    "footer": {
                        "type": "box",
                        "layout": "vertical",
                        "spacing": "sm",
                        "contents": [
                        {
                            "type": "button",
                            "style": "primary",
                            "height": "sm",
                            "action": {
                            "type": "message",
                            "label": "現在の温度・湿度を見る",
                            "text": "温度・湿度"
                            },
                            "color": "#6fb1bf"
                        },
                        {
                            "type": "button",
                            "style": "primary",
                            "height": "sm",
                            "action": {
                            "type": "uri",
                            "label": "SORACOM Lagoonで確認する",
                            "uri": "https://jp.lagoon.soracom.io/"
                            },
                            "color": "#34CDD7"
                        },
                        {
                            "type": "button",
                            "style": "secondary",
                            "height": "sm",
                            "action": {
                            "type": "message",
                            "label": "エアコンを消す",
                            "text": "エアコンを消す"
                            },
                            "color": "#DDDDDD"
                        },
                        {
                            "type": "spacer",
                            "size": "sm"
                        }
                        ],
                        "flex": 0
                    }
                    }
                messages = FlexSendMessage(alt_text='エアコンをつけました', contents=json_message)
                line_bot_api.push_message(user_id, messages=messages)
                subprocess.run("python3 IR-remocon02-commandline.py t `cat filename4.dat`", shell = True, cwd="/home/pi/I2C0x52-IR")


    return request.get_data()

@app.route("/callback", methods=['POST'])
def callback():
    signature = request.headers['X-Line-Signature']

    body = request.get_data(as_text=True)
    app.logger.info("Request body: " + body)

    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        abort(400)

    return 'OK'


@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):

    password = os.environ["soracom_pass"]

    if event.message.text == "温度・湿度":
        headers = {
            'Content-Type': 'application/json',
        }

        data = '{"email": "test@example.jp", "password": "' + password + '"}'

        response = requests.post('https://api.soracom.io/v1/auth', headers=headers, data=data)

        apikey = response.json()['apiKey']
        token  = response.json()['token']

        current_time = int(time.time()*1000)
        headers = {
            'Accept': 'application/json',
            'X-Soracom-API-Key': apikey,
            'X-Soracom-Token': token,
        }

        params = (
            ('to', current_time),
            ('sort', 'desc'),
            ('limit', '1'),
        )

        response = requests.get('https://api.soracom.io/v1/data/Subscriber/44xxxxxxxxxxxxx', headers=headers, params=params)
        request_body = response.json()

        content = [d.get('content') for d in request_body]

        payload = content[0]
        payload_dic = ast.literal_eval(payload)
        message = base64.b64decode(payload_dic['payload']).decode()
        temp = ast.literal_eval(message)['temp']
        humi = ast.literal_eval(message)['humi']

        base_image_path = './image.png'
        base_img = Image.open(base_image_path).copy()
        base_img = base_img.convert('RGB')

        temperature = str(temp)
        font_path = "/usr/share/fonts/downloadfonts/DSEG7-Classic/DSEG7Classic-Regular.ttf"
        font_size = 80
        font_color = (255, 255, 255)
        height = 90
        width = 180
        img = add_text_to_image(base_img, temperature, font_path, font_size, font_color, height, width)

        humidity = str(humi)
        height = 330
        img = add_text_to_image(base_img, humidity, font_path, font_size, font_color, height, width)

        img_path = 'static/{}.png'.format(datetime.datetime.now().strftime('%H-%M-%S'))
        img.save(img_path)

        json_message = {
            "type": "bubble",
            "hero": {
                "type": "image",
                "url": FQDN + '/' + img_path,
                "size": "full",
                "aspectRatio": "1:1",
                "aspectMode": "fit",
            },
            "body": {
                "type": "box",
                "layout": "vertical",
                "contents": [
                {
                    "type": "text",
                    "text": "温度 & 湿度",
                    "weight": "bold",
                    "size": "xl",
                    "color": "#6fb1bf"
                },
                {
                    "type": "box",
                    "layout": "vertical",
                    "margin": "lg",
                    "spacing": "sm",
                    "contents": [
                    {
                        "type": "box",
                        "layout": "baseline",
                        "spacing": "sm",
                        "contents": [
                        {
                            "type": "text",
                            "text": "温度",
                            "color": "#aaaaaa",
                            "size": "sm",
                            "flex": 1
                        },
                        {
                            "type": "text",
                            "text": temperature + '℃',
                            "wrap": True,
                            "color": "#666666",
                            "size": "sm",
                            "flex": 5
                        }
                        ]
                    },
                    {
                        "type": "box",
                        "layout": "baseline",
                        "spacing": "sm",
                        "contents": [
                        {
                            "type": "text",
                            "text": "湿度",
                            "color": "#aaaaaa",
                            "size": "sm",
                            "flex": 1
                        },
                        {
                            "type": "text",
                            "text": humidity + "％",
                            "wrap": True,
                            "color": "#666666",
                            "size": "sm",
                            "flex": 5
                        }
                        ]
                    }
                    ]
                }
                ]
            },
            "footer": {
                "type": "box",
                "layout": "vertical",
                "spacing": "sm",
                "contents": [
                {
                    "type": "button",
                    "style": "primary",
                    "height": "sm",
                    "action": {
                    "type": "uri",
                    "label": "SORACOM Lagoonで確認する",
                    "uri": "https://jp.lagoon.soracom.io/"
                    },
                    "color": "#34CDD7"
                },
                {
                    "type": "button",
                    "style": "secondary",
                    "height": "sm",
                    "action": {
                    "type": "message",
                    "label": "人感センサー",
                    "text": "人感センサー"
                    },
                    "color": "#DDDDDD"
                },
                {
                    "type": "spacer",
                    "size": "sm"
                }
                ],
                "flex": 0
            }
        }

        messages = FlexSendMessage(alt_text='温度 & 湿度', contents=json_message)
        line_bot_api.reply_message(event.reply_token, messages)

    elif event.message.text == "人感センサー":
        current_time = int(time.time()*1000)
        one_hour_ago = current_time - 3600000

        data = ref.order_by_key().limit_to_last(1).get()
        for key, val in data.items():
            timestamp = datetime.datetime.fromtimestamp(int(val['timestamp']/1000))
            last_time = timestamp.strftime('%m月%d日 %H時%M分')
        count = 0
        data = ref.order_by_key().get()
        for key, val in data.items():
            timestamp = val['timestamp']
            if timestamp >= one_hour_ago:
                count += 1
        json_message = {
            "type": "bubble",
            "hero": {
                "type": "image",
                "url": "https://xxx.ngrok.io/static/sensors.png",
                "size": "full",
                "aspectRatio": "16:9",
                "aspectMode": "cover"
            },
            "body": {
                "type": "box",
                "layout": "vertical",
                "contents": [
                {
                    "type": "text",
                    "text": "人感センサー",
                    "weight": "bold",
                    "size": "xl",
                    "color": "#72D35B"
                },
                {
                    "type": "box",
                    "layout": "vertical",
                    "margin": "lg",
                    "spacing": "sm",
                    "contents": [
                    {
                        "type": "box",
                        "layout": "baseline",
                        "spacing": "sm",
                        "contents": [
                        {
                            "type": "text",
                            "text": "1時間以内に検知した回数",
                            "color": "#aaaaaa",
                            "size": "sm",
                            "flex": 10
                        },
                        {
                            "type": "text",
                            "text": str(count) + "回",
                            "wrap": True,
                            "color": "#666666",
                            "size": "sm",
                            "flex": 4,
                            "align": "end"
                        }
                        ]
                    },
                    {
                        "type": "box",
                        "layout": "baseline",
                        "spacing": "sm",
                        "contents": [
                        {
                            "type": "text",
                            "text": "最後に検知した時間",
                            "color": "#aaaaaa",
                            "size": "sm",
                            "flex": 5
                        },
                        {
                            "type": "text",
                            "text": last_time,
                            "wrap": True,
                            "color": "#666666",
                            "size": "sm",
                            "flex": 5,
                            "align": "end"
                        }
                        ]
                    }
                    ]
                }
                ]
            },
            "footer": {
                "type": "box",
                "layout": "vertical",
                "spacing": "sm",
                "contents": [
                {
                    "type": "spacer",
                    "size": "sm"
                }
                ],
                "flex": 0
                }
            }
        messages = FlexSendMessage(alt_text='人感センサー', contents=json_message)
        line_bot_api.reply_message(event.reply_token, messages)

    elif event.message.text == "救急":
        messages = "落ち着いて 119 に電話してください"
        line_bot_api.reply_message(event.reply_token, TextSendMessage(text=messages))

    elif event.message.text == "エアコンを消す":
        messages = "エアコンを消しました"
        line_bot_api.reply_message(event.reply_token, TextSendMessage(text=messages))
        subprocess.run("python3 IR-remocon02-commandline.py t `cat filename5.dat`", shell = True, cwd="/home/pi/I2C0x52-IR")


if __name__ == "__main__":
    port = int(os.getenv("PORT", 6000))
    app.run(host="0.0.0.0", port=port)

➊ 温度と湿度を表示

SORACOM HarvestにはSORACOMデバイスから送信されたデータが保存されています。そのデータをAPIで取得することができます。

以下の部分でAPIを使うのに必要なKeyとTokenを取得します。Passwordは念のため環境変数に設定しています。

line_bot.py

password = os.environ["soracom_pass"]
headers = {
            'Content-Type': 'application/json',
        }

data = '{"email": "test@example.jp", "password": "' + password + '"}'

response = requests.post('https://api.soracom.io/v1/auth', headers=headers, data=data)

apikey = response.json()['apiKey']
token  = response.json()['token']

そして、最新の温度と湿度を取得します。これ以外のAPIを利用したい場合はAPI Referenceから参照できます。cURLのコマンドが書かれてあるので、それをこのサイトでPythonの形式に変換します。
request_bodyはリストで返ってくるのでその中のcontentを取り出し、payloadをastというモジュールで辞書に変換します。そのメッセージをbase64でデコードすることで気温と湿度が得られます。

line_bot.py

current_time = int(time.time()*1000)
headers = {
    'Accept': 'application/json',
    'X-Soracom-API-Key': apikey,
    'X-Soracom-Token': token,
}

params = (
    ('to', current_time),
    ('sort', 'desc'),
    ('limit', '1'),
)

response = requests.get('https://api.soracom.io/v1/data/Subscriber/44xxxxxxxxxxxxx', headers=headers, params=params)
request_body = response.json()

content = [d.get('content') for d in request_body]

payload = content[0]
payload_dic = ast.literal_eval(payload)
message = base64.b64decode(payload_dic['payload']).decode()
temp = ast.literal_eval(message)['temp']
humi = ast.literal_eval(message)['humi']

データを取得したら、温度と湿度が書かれた画像をPillowで作成します。こちらの記事を参考にさせていただきました。次のようなBaseイメージをPowerPointで作成し、そこにDSEGフォントで温度・湿度を書き込みます。

line_bot.py

temperature = str(temp)
font_path = "/usr/share/fonts/downloadfonts/DSEG7-Classic/DSEG7Classic-Regular.ttf"
font_size = 80
font_color = (255, 255, 255)
height = 90
width = 180
img = add_text_to_image(base_img, temperature, font_path, font_size, font_color, height, width)

humidity = str(humi)
height = 330
img = add_text_to_image(base_img, humidity, font_path, font_size, font_color, height, width)

img_path = 'static/{}.png'.format(datetime.datetime.now().strftime('%H-%M-%S'))
img.save(img_path)

image.py

from PIL import ImageFont, Image, ImageDraw

def add_text_to_image(img, text, font_path, font_size, font_color, height, width, max_length=740):
    position = (width, height)
    font = ImageFont.truetype(font_path, font_size)
    draw = ImageDraw.Draw(img)
    if draw.textsize(text, font=font)[0] > max_length:
        while draw.textsize(text + '…', font=font)[0] > max_length:
            text = text[:-1]
        text = text + '…'

    draw.text(position, text, font_color, font=font)

    return img

最後に、Flex Messageを送ります。Flex Message Simulatorを使うと簡単に作成することができます。

注意: Flex Message Simulatorではtrueとなっていますが、Pythonでは先頭が大文字のTrueなので注意してください。