광교아이파크::엘리베이터 Apple 홈킷 연동 (4) - Final

4. Homebridge

애플 홈킷에는 엘리베이터와 관련된 액세서리가 별도로 존재하는 것 같지는 않다 (검색 능력의 한계...)

npm 액세서리 패키지 하나 개발할까 하다가 귀차니즘이 발동해서 그냥 MQTT 기반 '스위치'로 구현하기로 결심했다

엘리베이터 도착 시 아이폰으로 알림 기능을 구현하기 위해 '인체 감지 센서' 기능도 함께 구현하면 좋을 것 같다

{
    "accessory": "mqttthing",
    "type": "switch",
    "name": "Elevator Down (MQTT)",
    "url": "mqtt:://localhost:1883",
    "username": "yogyui",
    "password": "12345678",
    "topics": {
        "getOn": {
            "topic": "home/ipark/elevator/state",
            "apply": "return JSON.parse(message).state;"
        },
        "setOn": {
            "topic": "home/ipark/elevator/command",
            "apply": "return JSON.stringify({state: message});"
        }
    },
    "integerValue": true,
    "onValue": 1,
    "offValue": 0,
    "logMqtt": true
},
{
    "accessory": "mqttthing",
    "type": "occupancySensor",
    "name": "Elevator Down Occupancy (MQTT)",
    "url": "mqtt:://localhost:1883",
    "username": "yogyui",
    "password": "12345678",
    "topics": {
        "getOccupancyDetected": {
            "topic": "home/ipark/elevator/state/occupancy",
            "apply": "return JSON.parse(message).state;"
        }
    },
    "integerValue": true,
    "onValue": 1,
    "offValue": 0,
    "logMqtt": true
}

엘리베이터 호출 스위치, 도착 알림 센서 액세서리

occupancy sensor로 엘리베이터가 도착하면 '사람이 감지되었음' 을 아이폰에 notification할 수 있다 (이 iPhone에서 알림 기능을 활성화해줘야 된다)

 

서버 코드도 MQTT를 Publish, Subscribe할 수 있도록 수정해줘야 된다

# HomeDef.py
import json
import ctypes
from typing import List
import paho.mqtt.client as mqtt
from common import Callback
from Serial485.SerialComm import SerialComm
from Serial485.SmartParser import SmartParser

class Elevator:
    init: bool = False
    state: int = -1
    state_prev: int = -1
    my_floor: int = -1
    current_floor: int = -1
    mqtt_client: mqtt.Client = None
    
    def __init__(self, **kwargs):
	self.mqtt_client = kwargs.get('mqtt_client')
        self.sig_call_up = Callback()
        self.sig_call_down = Callback()		

    def call_up(self):
        self.sig_call_up.emit()

    def call_down(self):
        self.sig_call_down.emit()

    def publish_mqtt(self):
        obj = {
            "state": int(self.state == 4 and self.current_floor == self.my_floor)
        }
        self.mqtt_client.publish("home/ipark/elevator/state", json.dumps(obj), 1)
        self.mqtt_client.publish("home/ipark/elevator/state/occupancy", json.dumps(obj), 1)

class Home:
    name: str = 'Home'
    device_list: List[Device]
    elevator: Elevator

    serial_baud: int = 9600
    serial_485_smart_port1: str = ''
    serial_485_smart_port2: str = ''

    mqtt_client: mqtt.Client
    mqtt_host: str = 'localhost'
    mqtt_port: int = 1883

    def __init__(self, room_info: List, name: str = 'Home'):
        self.name = name
        self.device_list = list()

        self.mqtt_client = mqtt.Client()
        self.mqtt_client.on_message = self.onMqttClientMessage
		
        self.elevator = Elevator(mqtt_client=self.mqtt_client)
        self.elevator.sig_call_down.connect(self.onElevatorCallDown)

        self.device_list.append(self.elevator)

        self.mqtt_client.connect(self.mqtt_host, self.mqtt_port)
        self.mqtt_client.loop_start()

        self.serial_485_smart1 = SerialComm('Smart1')
        self.serial_485_smart2 = SerialComm('Smart2')
        self.parser_smart = SmartParser(self.serial_485_smart1, self.serial_485_smart2)
        self.parser_smart.sig_parse1.connect(self.onParserSmartResult1)
        # self.parser_smart.sig_parse2.connect(self.onParserSmartResult2)

    def release(self):
        self.mqtt_client.loop_stop()
        self.mqtt_client.disconnect()
        self.serial_485_smart1.release()
        self.serial_485_smart2.release()

    def initDevices(self):
        self.serial_485_smart1.connect(self.serial_485_smart_port1, self.serial_baud)
        self.serial_485_smart2.connect(self.serial_485_smart_port2, self.serial_baud)

    def sendSerialSmartPacket(self, packet: str):
        if self.serial_485_smart2.isConnected():
            self.serial_485_smart2.sendData(bytearray([int(x, 16) for x in packet.split(' ')]))

    def onParserSmartResult1(self, chunk: bytearray):
        header = chunk[1]  # [0xC1]
        packetLen = chunk[2]
        cmd  = chunk[3]
        if header == 0xC1 and packetLen == 0x13 and cmd == 0x13:
            dev = self.elevator
            if len(chunk) >= 13:
                dev.state = chunk[11]
                dev.current_floor = ctypes.c_int8(chunk[12]).value
                if dev.state != dev.state_prev or not dev.init:
                    dev.publish_mqtt()
                    dev.init = True
                dev.state_prev = dev.state

    def onElevatorCallDown(self):
        self.parser_smart.flag_send_down_packet = True

    def startMqttSubscribe(self):
        self.mqtt_client.subscribe("home/ipark/elevator/command")

    def onMqttClientMessage(self, client, userdata, message):
        writeLog('Mqtt Client Message: {}, {}'.format(userdata, message), self)
        topic = message.topic
        msg_dict =  json.loads(message.payload.decode("utf-8"))
        if 'elevator/command' in topic:
            if 'state' in msg_dict.keys():
                self.onElevatorCallDown()

# app.py
from flask import Flask
from HomeDef import Home

app = Flask(__name__)

if __name__ == '__main__':
    home = Home()
    home.initDevices()
    app.run(host='0.0.0.0'm port=9999, debug=False)
    home.release()

동작테스트 >>

 

호출도 잘 되고, 도착 시 알림도 정상적으로 동작한다 

당연히 Siri랑도 잘 연동된다 ("시리야, 엘리베이터 호출해줘" 정도면 충분히 알아듣는다)

 

끝!

 

[시리즈 링크]

광교아이파크::엘리베이터 Apple 홈킷 연동 (1)

광교아이파크::엘리베이터 Apple 홈킷 연동 (2)

광교아이파크::엘리베이터 Apple 홈킷 연동 (3)

광교아이파크::엘리베이터 Apple 홈킷 연동 (4)