This article is Chapter 4 of a four-chapter article.
You can check the operation image on YouTube, so please have a look.
I wanted to be able to interactively check the data collected by SORACOM devices and operate home appliances with LINE, which is used by 80 million people in Japan and is familiar with the UI.
--GPS multi-unit SORACOM Edition
$ pip install flask
$ pip install line-bot-sdk
$ pip install firebase-admin
$ pip install pillow
$ pip install paramiko
--Display temperature and humidity --View the temperature and humidity measured by GPS multi-unit SORACOM Edition on LINE --Display motion sensor data --Fetch the data stored in Firebase Realtime Database and browse it on LINE --Send a Push message in the event of an emergency --Sends a Push message to LINE when the SORACOM LTE-M Button is pressed --Air conditioner control --When the risk of heat stroke increases, the air conditioner is automatically turned on. --The conditions for turning on the air conditioner are "temperature 30 ° C, humidity 60% or higher" and "human sensor responds within 15 minutes". --You can also turn off the air conditioner in case of malfunction.
I will explain each of them together with the source code.
It will be the entire source. Please note that the import and access token parts of the module will not be posted after that.
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 #Self-made module
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 of the user pushing the message
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>' #Relative path
@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": "The emergency button was pressed",
"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": "place",
"color": "#aaaaaa",
"size": "sm",
"flex": 1
},
{
"type": "text",
"text": "Dressing room",
"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": "I unlocked the front door as an emergency measure",
"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": "First aid",
"text": "First aid"
},
"color": "#E9462B"
},
{
"type": "spacer",
"size": "sm"
}
],
"flex": 0
}
}
messages = FlexSendMessage(alt_text='[SOS]The emergency button was pressed', 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": "I turned on the air conditioner",
"weight": "bold",
"size": "xl",
"color": "#7077BE"
},
{
"type": "box",
"layout": "vertical",
"contents": [
{
"type": "text",
"text": "Temperature and humidity with a high risk of heat stroke.",
"size": "xs",
"wrap": True
},
{
"type": "text",
"text": "Since the motion sensor responded within 15 minutes, I decided that I was at home and turned on the air conditioner.",
"size": "xs",
"wrap": True
}
],
"margin": "sm"
}
]
},
"footer": {
"type": "box",
"layout": "vertical",
"spacing": "sm",
"contents": [
{
"type": "button",
"style": "primary",
"height": "sm",
"action": {
"type": "message",
"label": "See the current temperature and humidity",
"text": "temperature humidity"
},
"color": "#6fb1bf"
},
{
"type": "button",
"style": "primary",
"height": "sm",
"action": {
"type": "uri",
"label": "Check with SORACOM Lagoon",
"uri": "https://jp.lagoon.soracom.io/"
},
"color": "#34CDD7"
},
{
"type": "button",
"style": "secondary",
"height": "sm",
"action": {
"type": "message",
"label": "Turn off the air conditioner",
"text": "Turn off the air conditioner"
},
"color": "#DDDDDD"
},
{
"type": "spacer",
"size": "sm"
}
],
"flex": 0
}
}
messages = FlexSendMessage(alt_text='I turned on the air conditioner', 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 == "temperature humidity":
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": "temperature&Humidity",
"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": "temperature",
"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": "Humidity",
"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": "Check with SORACOM Lagoon",
"uri": "https://jp.lagoon.soracom.io/"
},
"color": "#34CDD7"
},
{
"type": "button",
"style": "secondary",
"height": "sm",
"action": {
"type": "message",
"label": "Human Sensor",
"text": "Human Sensor"
},
"color": "#DDDDDD"
},
{
"type": "spacer",
"size": "sm"
}
],
"flex": 0
}
}
messages = FlexSendMessage(alt_text='temperature&Humidity', contents=json_message)
line_bot_api.reply_message(event.reply_token, messages)
elif event.message.text == "Human Sensor":
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 month%d day%H o'clock%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": "Human Sensor",
"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": "Number of detections within 1 hour",
"color": "#aaaaaa",
"size": "sm",
"flex": 10
},
{
"type": "text",
"text": str(count) + "Times",
"wrap": True,
"color": "#666666",
"size": "sm",
"flex": 4,
"align": "end"
}
]
},
{
"type": "box",
"layout": "baseline",
"spacing": "sm",
"contents": [
{
"type": "text",
"text": "Last detected time",
"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='Human Sensor', contents=json_message)
line_bot_api.reply_message(event.reply_token, messages)
elif event.message.text == "First aid":
messages = "Calm down and call 119"
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=messages))
elif event.message.text == "Turn off the air conditioner":
messages = "I turned off the air conditioner"
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)
Get the Key and Token required to use the API in the following part. Password is set as an environment variable just in case.
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']
And get the latest temperature and humidity. If you want to use other APIs, you can refer to it from API Reference. Since the cURL command is written, convert it to Python format at this site.
Since request_body
is returned as a list, take out content
in it and convert payload
to a dictionary with a module called ʻast`. You can get the temperature and humidity by decoding the message with 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']
Once you have the data, use Pillow to create an image with the temperature and humidity. I referred to this article. Create the following Base image with PowerPoint and write the temperature and humidity in it with the DSEG font.
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
Finally, send a Flex Message. You can easily create it using the Flex Message Simulator (https://developers.line.biz/flex-simulator/).
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": "temperature&Humidity",
"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": "temperature",
"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": "Humidity",
"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": "Check with SORACOM Lagoon",
"uri": "https://jp.lagoon.soracom.io/"
},
"color": "#34CDD7"
},
{
"type": "button",
"style": "secondary",
"height": "sm",
"action": {
"type": "message",
"label": "Human Sensor",
"text": "Human Sensor"
},
"color": "#DDDDDD"
},
{
"type": "spacer",
"size": "sm"
}
],
"flex": 0
}
}
messages = FlexSendMessage(alt_text='temperature&Humidity', contents=json_message)
line_bot_api.reply_message(event.reply_token, messages)
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 month%d day%H o'clock%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
The Flex Message part is long, so I omitted it, but it is the same as ➊.
I also used Paramiko
to ssh connect to another Raspberry Pi and execute commands in Python.
Change the ```ECDSAKey`` part to the key type you have set as appropriate. You should be able to use either RSA or 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
is a simple program that just turns the servo motor.
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()
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:
#Flex Message part omitted
subprocess.run("python3 IR-remocon02-commandline.py t cat `filename4.dat`", shell = True, cwd="/home/pi/I2C0x52-IR")
I was able to browse data and operate home appliances by making full use of the services of LINE Bot and SORACOM, two Raspberry Pis, and other external modules. I haven't seen many articles about touching the SORACOM API in Python, or articles linking SORACOM and LINE Bot, so I hope it helps someone. If you have any suggestions or questions, please feel free to comment.
-Implement push notifications using LINE Bot API in Python
Recommended Posts