Associez SORACOM aux appareils ménagers et au LINE Bot [Python / Flask / Raspberry Pi]
introduction
Cet article est le chapitre 4 d'un article de quatre chapitres.
- Création d'un système "non seulement" pour surveiller les maisons de personnes âgées en utilisant pleinement l'IoT [SORACOM Summer Challenge 2020]
- Envoyer un message Push au LINE Bot lorsque le bouton LTE-M est enfoncé [SORACOM]
- [Raspberry Pi] Lorsque le capteur humain le détecte, stockez l'horodatage dans la base de données Firebase Realtime
- Liez SORACOM avec les appareils électroménagers et LINE Bot [Python / Flask / Raspberry Pi] ** Toutes les sources publiées **: arrow_backward: Maintenant ici
Vous pouvez vérifier l'image de l'opération sur YouTube, alors jetez un œil.
Déclencheur
Je voulais pouvoir vérifier de manière interactive les données collectées par les appareils SORACOM et faire fonctionner les appareils électroménagers avec LINE, qui est utilisé par 80 millions de personnes au Japon et connaît l'interface utilisateur.
Ce que j'ai utilisé
- Édition SORACOM multi-unités GPS
- SORACOM LTE-M Button Plus
- Puisque l'entrée de contact n'est pas utilisée, le bouton LTE-M est en fait une seule unité.
- Raspberry Pi 3 Model A+ --Connectez un capteur humain (capteur infrarouge focal) --Connectez le servomoteur
- Raspberry Pi 3 Model B+ --Utiliser comme serveur --Connectez la télécommande d'apprentissage infrarouge (ADRSIR)
environnement
- 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
Installation
$ pip install flask
$ pip install line-bot-sdk
$ pip install firebase-admin
$ pip install pillow
$ pip install paramiko
Fonctions principales
- Afficher la température et l'humidité
- Visualisez la température et l'humidité mesurées par le multi-unité GPS SORACOM Edition sur LINE --Afficher les données du capteur humain
- Récupérez les données stockées dans Firebase Realtime Database et parcourez-les sur LINE --Envoyer un message Push en cas d'urgence
- Envoie un message Push à LINE lorsque le bouton SORACOM LTE-M est enfoncé
- Contrôle du climatiseur
- Le climatiseur est automatiquement mis en marche lorsque le risque de coup de chaleur augmente.
- Les conditions de mise en marche du climatiseur sont "température 30 ° C, humidité 60% ou plus" et "capteur humain réagit dans les 15 minutes".
- Vous pouvez également éteindre le climatiseur en cas de dysfonctionnement.
Je vais expliquer chacun d'eux avec le code source.
D'abord de la vue d'ensemble
Ce sera la source entière. Veuillez noter que les parties de jeton d'importation et d'accès du module ne seront plus publiées après cela.
Code source
<détails> Obtenez la clé et le jeton nécessaires pour utiliser l'API dans la partie suivante. Le mot de passe est défini comme une variable d'environnement au cas où. Et obtenez la dernière température et humidité. Si vous souhaitez utiliser d'autres API, vous pouvez vous y référer depuis API Reference. Puisque la commande cURL est écrite, convertissez-la au format Python sur ce site.
Puisque «request_body» est renvoyé sous forme de liste, retirez le «contenu» et convertissez «payload» en un dictionnaire avec un module appelé «ast ». Vous pouvez obtenir la température et l'humidité en décodant le message avec base64. Une fois que vous avez les données, utilisez Pillow pour créer une image avec la température et l'humidité. J'ai fait référence à cet article. Créez l'image de base suivante avec PowerPoint et écrivez-y la température et l'humidité avec la police DSEG. Enfin, envoyez un message Flex. Il est facile de créer avec le Flex Message Simulator (https://developers.line.biz/flex-simulator/). La partie Flex Message est longue, donc je l'ai omise, mais c'est la même chose que ➊. J'ai également utilisé J'ai pu parcourir les données et faire fonctionner les appareils ménagers en utilisant pleinement les services de LINE Bot et SORACOM, deux Raspberry Pis et d'autres modules externes.
Je n'ai pas vu beaucoup d'articles sur l'utilisation de l'API SORACOM en Python ou sur la liaison de SORACOM avec LINE Bot, alors j'espère que cela aidera quelqu'un.
Si vous avez des suggestions ou des questions, n'hésitez pas à commenter.
Recommended Posts
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 #Module fait maison
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..." #ID de l'utilisateur qui envoie le message
FQDN = 'https://xxx.ngrok.io' #URL ngrok
# 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>' #Chemin relatif
@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": "Le bouton d'urgence a été enfoncé",
"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": "endroit",
"color": "#aaaaaa",
"size": "sm",
"flex": 1
},
{
"type": "text",
"text": "Vestiaire",
"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": "J'ai déverrouillé la porte d'entrée comme mesure d'urgence",
"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": "Premiers secours",
"text": "Premiers secours"
},
"color": "#E9462B"
},
{
"type": "spacer",
"size": "sm"
}
],
"flex": 0
}
}
messages = FlexSendMessage(alt_text='[SOS]Le bouton d'urgence a été enfoncé', 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": "J'ai allumé le climatiseur",
"weight": "bold",
"size": "xl",
"color": "#7077BE"
},
{
"type": "box",
"layout": "vertical",
"contents": [
{
"type": "text",
"text": "Température et humidité avec un risque élevé de coup de chaleur.",
"size": "xs",
"wrap": True
},
{
"type": "text",
"text": "Comme le capteur humain a répondu dans les 15 minutes, j'ai décidé que j'étais à la maison et j'ai mis le climatiseur en marche.",
"size": "xs",
"wrap": True
}
],
"margin": "sm"
}
]
},
"footer": {
"type": "box",
"layout": "vertical",
"spacing": "sm",
"contents": [
{
"type": "button",
"style": "primary",
"height": "sm",
"action": {
"type": "message",
"label": "Voir température / humidité actuelle",
"text": "humidité de la température"
},
"color": "#6fb1bf"
},
{
"type": "button",
"style": "primary",
"height": "sm",
"action": {
"type": "uri",
"label": "Vérifier avec SORACOM Lagoon",
"uri": "https://jp.lagoon.soracom.io/"
},
"color": "#34CDD7"
},
{
"type": "button",
"style": "secondary",
"height": "sm",
"action": {
"type": "message",
"label": "Éteignez le climatiseur",
"text": "Éteignez le climatiseur"
},
"color": "#DDDDDD"
},
{
"type": "spacer",
"size": "sm"
}
],
"flex": 0
}
}
messages = FlexSendMessage(alt_text='J'ai allumé le climatiseur', 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 == "humidité de la température":
headers = {
'Content-Type': 'application/json',
}
data = '{"email": "[email protected]", "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": "Température&Humidité",
"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": "Température",
"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": "Humidité",
"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": "Vérifier avec SORACOM Lagoon",
"uri": "https://jp.lagoon.soracom.io/"
},
"color": "#34CDD7"
},
{
"type": "button",
"style": "secondary",
"height": "sm",
"action": {
"type": "message",
"label": "Capteur humain",
"text": "Capteur humain"
},
"color": "#DDDDDD"
},
{
"type": "spacer",
"size": "sm"
}
],
"flex": 0
}
}
messages = FlexSendMessage(alt_text='Température&Humidité', contents=json_message)
line_bot_api.reply_message(event.reply_token, messages)
elif event.message.text == "Capteur humain":
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 mois%jour j%H heure%M minutes')
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": "Capteur humain",
"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": "Nombre de détections en 1 heure",
"color": "#aaaaaa",
"size": "sm",
"flex": 10
},
{
"type": "text",
"text": str(count) + "Fois",
"wrap": True,
"color": "#666666",
"size": "sm",
"flex": 4,
"align": "end"
}
]
},
{
"type": "box",
"layout": "baseline",
"spacing": "sm",
"contents": [
{
"type": "text",
"text": "Dernière heure détectée",
"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='Capteur humain', contents=json_message)
line_bot_api.reply_message(event.reply_token, messages)
elif event.message.text == "Premiers secours":
messages = "Calme-toi et appelle le 119"
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=messages))
elif event.message.text == "Éteignez le climatiseur":
messages = "J'ai éteint le climatiseur"
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)
➊ Affichage de la température et de l'humidité
[SORACOM Harvest](https://soracom.jp/services/harvest/) stocke les données envoyées depuis l'appareil SORACOM. Vous pouvez obtenir ces données avec l'API.
line_bot.py
password = os.environ["soracom_pass"]
headers = {
'Content-Type': 'application/json',
}
data = '{"email": "[email protected]", "password": "' + password + '"}'
response = requests.post('https://api.soracom.io/v1/auth', headers=headers, data=data)
apikey = response.json()['apiKey']
token = response.json()['token']
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']
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
line_bot.py
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": "Température&Humidité",
"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": "Température",
"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": "Humidité",
"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": "Vérifier avec SORACOM Lagoon",
"uri": "https://jp.lagoon.soracom.io/"
},
"color": "#34CDD7"
},
{
"type": "button",
"style": "secondary",
"height": "sm",
"action": {
"type": "message",
"label": "Capteur humain",
"text": "Capteur humain"
},
"color": "#DDDDDD"
},
{
"type": "spacer",
"size": "sm"
}
],
"flex": 0
}
}
messages = FlexSendMessage(alt_text='Température&Humidité', contents=json_message)
line_bot_api.reply_message(event.reply_token, messages)
➋ Afficher les données du capteur humain
Obtenez la valeur de la base de données Firebase Realtime. L'horodatage du serveur est l'heure UNIX (millisecondes), il est donc converti en conséquence. Pour obtenir la valeur de Firebase, je me suis référé au [Document officiel](https://firebase.google.com/docs/database/admin/retrieve-data?hl=ja).
line_bot.py
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 mois%jour j%H heure%M minutes')
count = 0
data = ref.order_by_key().get()
for key, val in data.items():
timestamp = val['timestamp']
if timestamp >= one_hour_ago:
count += 1
➌ Envoyez un message push en cas d'urgence
Vous pouvez voir le contenu du webhook reçu de SORACOM Lagoon avec `ʻobject = request.get_json ()` `. Cliquez ici pour plus de détails]()
La ligne suivante indique `ʻif object ['title'] ==" [Alerting] Emergency alert ":` `, qui est aussi un webhook quand Lagoon passe à [No Data] ou [OK]. C'est à envoyer.
Paramiko '' pour SSH dans un autre Raspberry Pi et exécuter des commandes en Python. Remplacez la partie ```ECDSAKey par le type de clé que vous avez défini comme il convient. Vous devriez pouvoir utiliser RSA ou Ed25519.line_bot.py
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')
key_open.py est un programme simple qui ne fait que tourner le servomoteur.key_open.py
import time
import RPi.GPIO as GPIO
GPIO.setmode(GPIO.BCM)
GPIO.setup(4, GPIO.OUT)
p = GPIO.PWM(4, 50)
p.start(0.0)
p.ChangeDutyCycle(3.0)
time.sleep(0.4)
p.ChangeDutyCycle(0.0)
GPIO.cleanup()
➍ Contrôle du climatiseur
Le climatiseur est automatiquement mis en marche lorsque les conditions de «température 30 ° C, humidité 60% ou plus» et «capteur humain réagit dans les 15 minutes» sont satisfaites.
Le programme d'exploitation de la télécommande peut être téléchargé sur le site officiel de [Bit Trade One](https://bit-trade-one.co.jp/blog/20180515/).
line_bot.py
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:
#Partie du message Flex omise
subprocess.run("python3 IR-remocon02-commandline.py t cat `filename4.dat`", shell = True, cwd="/home/pi/I2C0x52-IR")
Résumé
Site de référence