Go to Qiita Advent Calendar 2024 Top
LoginSignup
22
20

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 1 year has passed since last update.

@TakashiM

LINEで我が家のスマートロックを管理する

Posted at

目的と概要

我が家はよく人を招いたり、友人や家族の宿泊場所になったりするので、物理鍵ではなくスマートロックで管理しています。遠隔でも開閉可能なので。
ちなみに一般的なマンションなので、僕の部屋に入るには共有部エントランスと個別の玄関扉を突破しないといけません。

共有部エントランスはこんな感じでSwtichBotを使って開いています。

玄関扉はSesame4を使っています。（後述しますが、QrioLockにしなかったのはAPIが使えるからです。）

アプリで鍵が扱えて便利！！！

しかしここで問題が…

自分で使う分にはいいんですが、友人たち以外に共有するには

  1. SwitchBot/Sesameのアプリをそれぞれインストールさせる
  2. SwitchBot/Sesameでそれぞれ招待リンクやそれに付随する期間や権限の設定をする

といろいろと面倒です。やっぱりアプリを新たにインストールさせるのはできるだけ避けたい。
だったら国民アプリであるLINEをプラットフォームにおんぶにだっこさせてもらおうと考えました。

使ったもの

  • LINE Messanger API ※メッセージの受信と返信
  • Sesame RESTful webAPI ※Sesameの操作
  • SwtichBot API ※SwitchBotの操作
  • GCP
    • Functions ※LINEから受け取ったリクエストの処理
    • Firestore ※LINE登録ユーザーの管理（主に鍵権限管理）

やりたいこと

特定のメッセージを送って開け閉めする。

unlock mainentrance → 共有エントランスのSwitchBotを起動し「開錠」を押す
unlock doorway → 玄関扉を開ける
unlock all → 上二つの操作を連続して行う
lock → 玄関扉を閉める

オーナー（僕）以外のユーザーは承認リクエストを送る（それを承認・拒否する）

request → 承認リクエスト
auth userId → 認証
unauth userId → 認証解除

※userIdは特定の文字列だと思ってください。

ディレクトリ構造

中身は後で説明するとして、このような構造をしています。

line_channel
├── modules
│   ├── entrance_util.py # 鍵の開け閉め
│   ├── firestore_util.py # Firestoreへのアクセスなど
│   └── line_channel_util.py # LINEの操作
├── secrets
│   ├── .env.yaml # .gcloudignoreに記述してfunctionsにはアップしない
│   └── your_credential.json # firestoreのアクセス権限
├── static
│   ├── greeting.txt # LINE登録時のあいさつ文
│   └── request_flex_message.json # Flex Message Simulatorで作成
├── .gcloudignore # gcloudコマンドでアップするために使っています
├── main.py # リクエストの受け取りなど
└── requirements.txt # pip installで必要なもの

SesameAPI

Sesameの操作はこちらを参考にしました。APIKEYの取得などはこちらから。
API開放してくれるとエンジニアにはとてもうれしいしそもそも安い！（このためにQrioから乗り換えました。）

./modules/entrance_util.py
import os

class Doorway:
    def __init__(self, history_tag='My Script'):
        from pysesame3.lock import CHSesame2
        self.device = CHSesame2(
            authenticator=self._get_auth(),
            device_uuid=os.environ.get('SESAME_UUID'),
            secret_key=os.environ.get('SESAME_SECRET')
        )
        self.history_tag = history_tag

    def _get_auth(self):
        from pysesame3.auth import WebAPIAuth
        return WebAPIAuth(apikey=os.environ.get('SESAME_APIKEY'))

    def unlocked(self):
        return self.device.unlock(history_tag=self.history_tag)

    def locked(self):
        return self.device.lock(history_tag=self.history_tag)

以下同様ですが、環境変数（os.environ.get('XXX')で取得している変数）に関しては.env.yamlで管理設定します。

Swtichbot API

SwitchBotに関してもこちらを参考にしてください。ちなみにSwitchBot系のスマートロックもあるので一元化できたらいいなあと思う。

./modules/entrance_util.py
import os
import json
import requests

class MainEntrance:
    def __init__(self):
        self.headers = {
            "Authorization": os.environ.get('SWITCHBOT_API_KEY'),
            "content-type": "application/json; charset: utf8"
        }

        deviceid = os.environ.get('SWITCHBOT_ENTRANCE_DEVICE_ID')
        self.url = f"https://api.switch-bot.com/v1.0/devices/{deviceid}/commands"

        body = {
            "command": "press", "parameter": "default", "commandType": "command"
        }
        self.data = json.dumps(body)

    def unlocked(self):
        response = requests.post(self.url, data=self.data, headers=self.headers)
        return response.text

Google Cloud Firestore

GCPから適当に1つコレクションを作ってください。
コレクション名は「line_user」にしています。（.env.yamlに記述）
ドキュメント名はLINEから取得したuserIdにしています。（※セキュリティ的にNG？笑）
フィールドは

  • authorized（数値）...鍵の認証をもらっていると1、ないと0になります。
  • creationDatetime（タイムスタンプ）...ドキュメント作成日時
  • displayName（文字列）...LINEの表示名
  • follow（数値）...登録ユーザーは1、ブロックされると0になります。
  • updateDatetime（タイムスタンプ）...ドキュメント更新日時
  • userId:（文字列）...ドキュメント名と同じLINEのuserId

またFunctionsからアクセスする際に（多分）credentialが必要なので、頑張ってjsonとして吐き出しておいてください。（./your_crediantial.json）

Firestoreにデータをインサートするときに使います。汎用的に使えるようにあえてふわっと書いています。

./modules/firestore_util.py
import os
import firebase_admin
from firebase_admin import firestore, credentials

class MyFirestore:
    def __init__(self, credential_path="./credential.json"):
        os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = credential_path
        if not firebase_admin._apps:
            # 初期済みでない場合は初期化処理を行う
            cred = credentials.Certificate(credential_path)
            default_app = firebase_admin.initialize_app(cred)
        # firebase_admin.initialize_app()
        self.DB = firestore.client()

    def set_data(self, collection, document, data, merge):
        """
        firestoreのcollectionにデータを追加（更新）する
        """
        self.DB.collection(collection).document(document).set(data, merge=merge)
        return None

    def get_data(self, collection, document):
        """
        firestoreのcollectionからデータを取得する
        """
        return self.DB.collection(collection).document(document).get().to_dict()

LINE Messanger API

こちらのソースコードの内容としては受けたリクエストの種類に対していろんな操作をしています。

./modules/line_channel_util.py
import os
import json
import codecs
from datetime import datetime, timedelta, timezone

from linebot.models import TextSendMessage, FlexSendMessage
from linebot.models import (
    MessageEvent, FollowEvent, UnfollowEvent
)

from modules.firestore_util import MyFirestore
from modules.entrance_util import MainEntrance, Doorway

JST = timezone(timedelta(hours=+9), 'JST')

ADMIN_USER_ID = os.environ.get('LINE_ADMIN_USER_ID')

FIRESTORE_CREDENTIAL_PATH = os.environ.get('FIRESTORE_CREDENTIAL_PATH')
DEFAULT_IMAGE_URL = 'https://scdn.line-apps.com/n/channel_devcenter/img/fx/01_1_cafe.png'

class LineChannel:
    def __init__(self, line_bot_api, event):
        self.line_bot_api = line_bot_api
        self.event = event
        self.event_type = self._event_type()
        self.userId = event.source.user_id
        self.dt = self._timestamp2datetime(event.timestamp)

        self._is_admin = self.userId == ADMIN_USER_ID

    def _event_type(self):
        if isinstance(self.event, FollowEvent):
            return "follow"
        elif isinstance(self.event, UnfollowEvent):
            return "unfollow"
        elif isinstance(self.event, MessageEvent):
            return f'{self.event.message.type}_message'
        else:
            return "unknown"

    def send_messages_to_admin(self, messages):
        message_objs = [self._message2object(*m) for m in messages]
        self.line_bot_api.push_message(to=ADMIN_USER_ID, messages=message_objs)
        return None

    def _message2object(self, message, alt_text='message'):
        if isinstance(message, dict):
            message_obj = FlexSendMessage(alt_text=alt_text, contents=message)
        elif isinstance(message, str):
            message_obj = TextSendMessage(text=message)
        else:
            return None
        return message_obj

    def _timestamp2datetime(self, timestamp):
        return datetime.fromtimestamp(timestamp / 1000, tz=JST)

    def get_profile(self, del_key=['language', 'statusMessage']):
        profile = self.line_bot_api.get_profile(self.userId).as_json_dict()
        for key in del_key:
            profile.pop(key, None)
        return profile

    def is_followed(self):
        data = self.get_profile()
        data.update({
            'updateDatetime': self.dt,
            'creationDatetime': self.dt,
            'follow': 1,
            'authorized': self._is_admin
        })

        fs = MyFirestore(credential_path=FIRESTORE_CREDENTIAL_PATH)
        collection = os.environ.get('FIRESTORE_COLLECTION_USER')
        fs.set_data(
            collection=collection, document=self.userId, data=data, merge=False
        )
        return None

    def is_unfollowed(self):
        data = {'updateDatetime': self.dt, 'follow': 0, 'authorized': 0}

        fs = MyFirestore(credential_path=FIRESTORE_CREDENTIAL_PATH)
        collection = os.environ.get('FIRESTORE_COLLECTION_USER')
        fs.set_data(
            collection=collection, document=self.userId, data=data, merge=True
        )
        return None

    def greet(self):
        with open('./static/greeting.txt') as f: # read greeting message
            greeting = f.read()
        self.reply(greeting) # send message
        return None

    def reply(self, message, alt_text=None):
        message_obj = self._message2object(message, alt_text)
        self.line_bot_api.reply_message(self.event.reply_token, message_obj)
        return None

    def _check_user_authority(self, firestore, collection):
        data = firestore.get_data(collection=collection, document=self.userId)
        return data['authorized'] == 1

    def check_authority(self):
        fs = MyFirestore(credential_path=FIRESTORE_CREDENTIAL_PATH)
        col = os.environ.get('FIRESTORE_COLLECTION_USER')
        if self.event.source.type == 'user':
            auth = self._check_user_authority(fs, col)
        else:
            auth = False
        return auth

    def unlock(self, item):
        if not self.check_authority():
            return {'message': 'unauthorized'}

        if item == "doorway":
            Doorway().unlocked()
        elif item == "mainentrance":
            MainEntrance().unlocked()
        elif item == "all":
            MainEntrance().unlocked()
            Doorway().unlocked()
        return {'message': 'unlock'}

    def lock(self):
        if not self.check_authority():
            return {'message': 'unauthorized'}
        Doorway().locked()
        return {'message': 'lock'}

    def recieve_auth_request(self):
        path = './static/request_flex_message.json'
        with codecs.open(path, 'r', 'utf-8') as f:
            content = json.load(f)
        content_str = json.dumps(content)

        prof = self.get_profile()
        if not 'pictureUrl' in prof:
            prof['pictureUrl'] = DEFAULT_IMAGE_URL
        trans_dict = {
            'displayName': prof['displayName'],
            'https://image_url.png': prof['pictureUrl'],
            'client_userId': self.userId,
        }
        for before, after in trans_dict.items():
            content_str = content_str.replace(before, after)
        content = json.loads(content_str)

        messages = [
            [content, u'承認リクエスト']
        ]
        self.send_messages_to_admin(messages)

        return {'message': u'requested authorization'}

    def authorize(self, client_userId):
        if not self._is_admin:
            return {'message': u'You are not admin'}
        data = {'updateDatetime': self.dt, 'authorized': 1}

        fs = MyFirestore(credential_path=FIRESTORE_CREDENTIAL_PATH)
        col = os.environ.get('FIRESTORE_COLLECTION_USER')
        fs.set_data(collection=col, document=client_userId, data=data, merge=True)

        messages = [TextSendMessage(text='承認されました。')]
        self.line_bot_api.push_message(to=client_userId, messages=messages)
        return {'message': u'Done'}

    def disapprove(self, userId):
        if not self._is_admin:
            return {'message': u'You are not admin'}
        data = {'updateDatetime': self.dt, 'authorized': 0}

        fs = MyFirestore(credential_path=FIRESTORE_CREDENTIAL_PATH)
        col = os.environ.get('FIRESTORE_COLLECTION_USER')
        fs.set_data(collection=col, document=userId, data=data, merge=True)

        messages = [TextSendMessage(text='承認が解除されました。')]
        self.line_bot_api.push_message(to=userId, messages=messages)
        return {'message': u'Done'}

Google Cloud Functions

メインとなる処理です。

./main.py
import os
import json
import codecs
import base64, hashlib, hmac
from flask import abort, jsonify

from linebot import LineBotApi, WebhookParser
from linebot.exceptions import InvalidSignatureError
from modules.line_channel_util import LineChannel

CHANNEL_ACCESS_TOKEN = os.environ.get('LINE_CHANNEL_ACCESS_TOKEN')
CHANNEL_SECRET = os.environ.get('LINE_CHANNEL_SECRET')

line_bot_api = LineBotApi(CHANNEL_ACCESS_TOKEN)
parser = WebhookParser(CHANNEL_SECRET)

def handle_event(line_bot_api, event):
    line_channel = LineChannel(line_bot_api, event)
    # channel is followed
    if line_channel.event_type == 'follow':
        line_channel.is_followed()
        line_channel.greet()
    # channel is unfollowed
    elif line_channel.event_type == 'unfollow':
        line_channel.is_unfollowed()

    # channel received a text message
    elif line_channel.event_type == 'text_message':
        message_recd = event.message.text
        if message_recd.startswith('unlock'):
            ret = line_channel.unlock(item=message_recd.split(' ')[-1])
            line_channel.reply(ret['message'])

        elif message_recd == 'lock':
            ret = line_channel.lock()
            line_channel.reply(ret['message'])

        elif message_recd == 'request':
            ret = line_channel.recieve_auth_request()
            line_channel.reply(ret['message'])

        elif message_recd.startswith('auth'):
            client_userId = message_recd.split(':')[-1]
            ret = line_channel.authorize(client_userId)
            line_channel.reply(ret['message'])

        elif message_recd.startswith('unauth'):
            client_userId = message_recd.split(':')[-1]
            ret = line_channel.disapprove(client_userId)
            line_channel.reply(ret['message'])

    else:
        line_channel.reply(u'I don\'t understand')

def main(request):
    # LINEからのリクエストを受け取る
    body = request.get_data(as_text=True)
    hash = hmac.new(
        CHANNEL_SECRET.encode('utf-8'), body.encode('utf-8'), hashlib.sha256
    ).digest()
    signature = base64.b64encode(hash).decode()
    if signature != request.headers['X_LINE_SIGNATURE']:
        return abort(405)

    try:
        events = parser.parse(body, signature)
    except InvalidSignatureError:
        return abort(405)

    # イベントを処理する
    for event in events:
        handle_event(line_bot_api, event)
    return jsonify({'message': 'ok'})

LINEチャンネル×Cloud Functions

これを参考に適当に作ってください。Webhockなどの設定を一つでもミスると処理が届きません。
またオーナーのuserIdも記述されているはずです。

環境変数

今までのことを行っていると以下の変数は埋まると思います。

./secrets/.env.yaml
FIREBASE_PROJECT_ID : your_project_id
FIRESTORE_CREDENTIAL_PATH : secrets/your_credential.json
LINE_CHANNEL_ACCESS_TOKEN : your_line_channel_access_token
LINE_CHANNEL_SECRET : your_line_channel_secret
LINE_ADMIN_USER_ID : your_line_admin_user_id
FIRESTORE_COLLECTION_USER : line_user # ここだけGoogle Cloud Firestoreの設定に依存します
SWITCHBOT_API_KEY : your_switchbot_api_key
SWITCHBOT_ENTRANCE_DEVICE_ID : your_switchbot_entrance_device_id
SESAME_UUID : your_sesame_uuid
SESAME_SECRET : your_sesame_secret
SESAME_APIKEY : youe_sesame_apikey

デプロイ

この辺の記事を見てうまいことやりましょう。もちろん手動でもいいです。

pip freeze > requirements.txt

DEPLOY_FUNCTION_NAME=line_channel
gcloud functions deploy ${DEPLOY_FUNCTION_NAME} \
    --region asia-northeast1 \
    --entry-point main \
    --env-vars-file="secrets/.env.yaml" \
    --runtime python38 \
    --trigger-http \
    --allow-unauthenticated

試す！

291284.jpg
（イメージつかないですよね…）

とりあえずできた！

22
20
1

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
22
20

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?