MQTT 중계 프로그램 제작 - 2부

안드로이드와 스위치 간 MQTT를 이용한 통신을 서버에서 중계를 하고자 한다. 그러면 어떻게 해야 중계를 할 수 있을까?

 

이번에는 MQTT_JSON.py, Network.py와 Reserve_Main.py까지 보도록 하겠습니다.

바로 코드 보시죠. (저번 글에 이어 바로 3번부터 시작하겠습니다.)

3. 코드 살펴보기

3번. MQTT_JSON.py

import json


def JSON_Parser(object):
    #print(object)
    jsonObject = json.loads(object)
    dic = {'sender': jsonObject['sender'], 'message': jsonObject['message'], 'room': jsonObject['room']}
    return dic


def JSON_Parser_android(object):
    jsonObject = json.loads(object)
    dic = {'sender': jsonObject['Light']['sender'], 'message': jsonObject['Light']['message'],
           'room': jsonObject['Light']['room'], 'destination': jsonObject['Light']['destination']}
    return dic


def JSON_Parser_Main(object):
    json_object = json.loads(object)
    if json_object[0] == 'Light':
        dic = JSON_Parser_android(object)
    else:
        dic = JSON_Parser(object)
    return dic


def JSON_ENCODE_TOSERVER(dic):
    object = {"Light": {"sender": dic[0][1], "message": dic[1][1], "room": dic[2][1], "destination": dic[3][1]}}
    jsonObject = json.dumps(object)
    return jsonObject


def JSON_ENCODE(dic):
    object = {"sender": dic['sender'], "message": dic['message'], "room": dic['room']}
    jsonObject = json.dumps(object)
    return jsonObject


def JSON_ENCODE_android(dic):
    object = {"Light": {"sender": dic[0][1], "message": dic[1][1], "destination": dic[2][1]}}
    jsonObject = json.dumps(object)
    return jsonObject

각 함수별 역할을 아래와 같습니다.

•  JSON_Parser
  • 스위치에서 보낸 JSON 파일을 파싱하는 함수
• JSON_Parser_android
  • 안드로이드에서 보낸 JSON 파일을 파싱하는 함수
• JSON_Parser_Main
  • "Light" 라는 헤더값이 있으면 안드로이드에서 보낸 값이므로 JSON_Parser_android로 보내 파싱
  • 없을 경우 JSON_Parser로 보내 파싱
• JSON_ENCODE_TOSERVER
  • 서버로 안드로이드 형식으로 보낼때 쓰는 함수 (예약 기능에서 사용)
• JSON_ENCODE
  • 스위치로 보낼 데이터를 JSON 파일로 인코딩하는 함수
• JSON_ENCODE_android
  • 안드로이드로 보낼 데이터를 JSON 파일로 인코딩하는 함수

 

4번. Network.py

# import MySQLdb # need to work on linux
import pymysql as MySQLdb # need to work on windows
import requests
import time

# DB ifno
User = 'DB ID'
Password = "DB Password"
Host = "192.168.0.254"
DB = "table name"
Port = 3306
Mode_O = "'OFF'"


def SQL_Def(mode, dic):
    conn = MySQLdb.connect(host=Host, port=Port, user=User, passwd=Password, db=DB, charset='utf8')
    if conn.open:
        curs = conn.cursor()
        # print(mode)
        if mode == "Light":
            if dic['message'] == "On" or dic['message'] == "Off":
                SQL_Update(dic['message'], dic['room'], curs, conn, "Room_Light", "State", 'room')
        elif mode == "LightList":
            result = SQL_Select("Room_Light", curs, conn)
            return result
        elif mode == "LightRecord":
            now = time.localtime()
            date = str(now.tm_year) + '-' + str(now.tm_mon) + '-' + str(now.tm_mday)
            hour = str(now.tm_hour) + ':' + str(now.tm_min)
            data = (hour, dic['room'], dic['message'], date, dic['sender'])
            SQL_Insert("LightRecord", curs, conn, data)
        elif mode == "Connect":
            SQL_Update(dic[0][1], dic[1][1], curs, conn, "Room_Light", "Connect", 'room')
        elif mode == "ReserveSelect":
            result = SQL_Select("Light_Reserve", curs, conn)
            return result
        elif mode == "ReserveUpdate":
            SQL_Update(dic[0][1], dic[1][1], curs, conn, "Light_Reserve", "Do", "num")
        elif mode == "ReserveUpdateActivated":
            SQL_Update(dic[0][1], dic[1][1], curs, conn, "Light_Reserve", "Activated", "num")
            # state = SQL_State(curs)


def SQL_Insert(table, cur, connect, data):
    sql = 'INSERT INTO ' + table + ' ( Time, Room, Do, Day, User ) VALUES (%s, %s, %s, %s, %s)'
    res = cur.execute(sql, data)
    connect.commit()


def SQL_Update(state, condition, cur, connect, table, column, mode):
    if mode == 'room':
        sql = 'UPDATE ' + table + ' SET ' + column + ' = "' + state + '" WHERE Room = "' + condition + '"'
    else:
        if mode == 'num':
            sql = 'UPDATE ' + table + ' SET ' + column + ' = "' + state + '" WHERE Num = "' + condition + '"'
    # print(sql)
    res = cur.execute(sql)
    connect.commit()
    # print("sql update")
    # print(res)


def SQL_Select(table, cur, connect):
    sql = 'SELECT * FROM ' + table
    # print(sql)
    cur.execute(sql)
    res = cur.fetchall()
    connect.commit()
    return res

스위치 제어 결과를 DB에 갱신하기 위한 코드입니다.

• import MySQLdb, pymysql
  • MySQLdb는 리눅스에서 DB 접근하기 위한 라이브러리 입니다. (윈도우에서 사용 불가)
  • pymysql은 윈도우에서 사용하기 위한 라이브러리 입니다. (리눅스에서 사용 불가)
• SQL_Def
  • 전등 상태 : 전등의 on /off 상태를 DB에 갱신합니다.
  • 전등 리스트 : 전등 리스트에 대한 정보를 DB로부터 가져옵니다.
  • 전등 컨트롤 내역 : 컨트롤 한 전등, 컨트롤 한 행동(on/off), 컨트롤 한 시간 등 정보를 담아 DB에 저장합니다.
  • 연결 상태 : 각 전등들과 서버의 연결 상태를 DB에 갱신합니다.
  • 예약 내역 조회 : 현재 예약이 되어있는 내역들을 DB로부터 가져옵니다.
  • 예약 활성화 : 원하는 예약을 실행하도록 활성화 여부를 DB에 갱신합니다.
  • 예약 실행 현황 업데이트 : 원하는 예약이 실행되었을 경우 실행했다고 DB에 갱신합니다.
  • 이렇게 상황에 맞는 DB에 값 제어를 하는 함수입니다.
• SQL_Insert
  • 전등 컨트롤 내역 insert 하는 함수입니다.
• SQL_Update
  • 전등 상태값(on/off) 업데이트 하는 함수입니다.
• SQL_Select
  • 원하는 테이블 값 불러오는 함수입니다.

5번. Reserve_Main.py

import time
import datetime

import paho.mqtt.client as mqtt
import schedule

import MQTT.MQTT_JSON as mqtt_json

from Network import SQL_Def

Run_able = 0
DoW = ['월', '화', '수', '목', '금', '토', '일']


def ReserveMain(queueToMain):
    print('ReserveMain')
    while True:
        ReserveAdd()
        while True:
            if not queueToMain.empty():
                if queueToMain.get() == "restart":
                    print("In Reserve")
                    break
            schedule.run_pending()
            time.sleep(1)


def ReserveAdd():
    info = ReserveSQL("ReserveSelect")
    run_recode = 0
    for i in range(0, len(info)):
        if info[i][5] == 'False':
            if info[i][6] == 'False':
                run_recode = JobDef(info[i], run_recode)
            else:
                days = info[i][4]
                day = days.split(",")
                n = time.localtime().tm_wday
                for j in range(0, len(day)):
                    if DoW[n] == day[j]:
                        run_recode = JobDef(info[i], run_recode)
    timeinfo = datetime.time(00, 00)
    schedule.every().day.at(str(timeinfo)).do(JobClear)
    for i in range(0, len(schedule.jobs)):
        print(schedule.jobs[i].at_time)


def ReserveMQTT(action, room, repeat, num):
    print("MQTT Reserve")
    result_list = ReserveSQL("LightList")
    cate = "no data"
    for i in range(0, len(result_list)):
        if room == result_list[i][0]:
            cate = result_list[i][3]
            break
    dic_object = [('name', 'ServerReserve'), ('message', action), ('room', cate), ('destination', room)]
    object = mqtt_json.JSON_ENCODE_TOSERVER(dic_object)

    client = mqtt.Client()
    client.connect("192.168.0.254", 1883)
    client.loop()
    client.publish("MyHome/Light/Pub/Server", object)
    client.loop_stop()

    print(dic_object)
    num = str(num)
    diction = [('message', action), ('room', num)]
    SQL_Def("ReserveUpdate", diction)
    print(repeat)
    if repeat == 'False':
        diction = [('activated', "True"), ('room', num)]
        SQL_Def("ReserveUpdateActivated", diction)
        ReserveAdd()


def ReserveSQL(mode):
    result_str = SQL_Def(mode, None)
    list_tmp = []
    for i in result_str:
        tmp = i
        count = (int(len(i) / 9))
        list_tmp.insert(count, tmp)
    return list_tmp


def JobDef(info, first):
    if first == 0:
        JobClear()
    JobAdd(info[1], info[3], info[2], info[6], info[8])  # time, action, room, repeat, num
    return 1


def JobAdd(time, action, room, repeat, num):
    time_list = time.split(':')
    time_hour = int(time_list[0])
    time_min = int(time_list[1])
    timeinfo = datetime.time(time_hour, time_min)
    schedule.every().day.at(str(timeinfo)).do(ReserveMQTT, action, room, repeat, num).tag(num)


def JobClear():
    schedule.clear()

예약과 관련된 기능들을 관리하는 코드입니다. 크게 통신 파트 와 스케쥴러 파트 로 나눠져있습니다.

통신 파트 : MQTT, Process Queue

스케쥴러 파트 : Job

함수 별 설명 드리겠습니다.

• ReserveMain : 무한 루프를 통해 계속 실행하며, 내부 무한루프는 Queue메세지로 "restart"가 올 경우 break하여 ReserveAdd()를 실행합니다.
• ReserveAdd :  ReserveSQL()을 실행시켜 DB로부터 예약된 값들을 받고 결과를 처리합니다.
  • run_recode : 최초 실행시, 기존 예약된 job 다 제거하기 위한 값. 한번이라도 실행 시 1로 변경
  • info[i][5] : 활성화가 되어있는지 (예약 실행을 해야하는지) 저장 되어 있는 값 [True, False] 
  • info[i][6] : 일회성 예약인지 주기적으로 작동하는 예약인지 저장 되어 있는 값 [True, False]
  • 이 둘다 False 일 경우 예약 실행 주기적으로 되면 안되기에 별도 요일 설정 없이 job 추가
  • info[i][4] : 예약 실행할 요일
  • Dow[i] == day[j] : 예약된 요일을 하나씩 비교해서 맞을 경우 job 추가
  • 마무리로 매일 00시에 DB값 갱신을 통해 매일 작동하는 job을 새로 갱신
• ReserveMQTT : job이 실행되었을때 서버로 MQTT를 보내 스위치를 제어하도록 하는 함수
  • job으로부터 받은 room 정보와 DB에서 가져온 전등 리스트 정보와 비교하여 카테고리와 방 정보를 가져옵니다.
  • dic_object에 JSON 타입 데이터를 정립합니다. 이후 JSON_ENCODE_TOSERVER함수를 통해 인코딩합니다.
  • MQTT를 통해 publish합니다.
  • 실행한 결과를 예약 테이블에 갱신합니다.
  • 만약 주기적으로 실행하는 job이 아니였다면 활성화를 비활성화로 변경하여 DB로 갱신하고 ReserveAdd()를      실행하여 job 리스트에서 제거합니다
• ReserveSQL : DB에서 가져온 정보를 리턴해줍니다.
  • ※ for문은 역순 정렬을 위함인데 제 집의 스위치 설치 조건 때문에 한거라서 여러분들은 안쓰셔도 됩니다.
• JobDef : first가 0 일 경우 기존 job 리스트를 초기화 합니다. 들어온 info를 기반으로 JobAdd를 실행합니다.
• JobAdd : 받은 데이터 기반하여 스케쥴을 추가합니다.
• JobClear : 등록된 스케쥴들을 초기화 합니다.

풀 소스는 아래 링크에 올려두었습니다.

https://github.com/sonjuhy/Myhome_Server

 

4. 기타

아래는 위 내용과 관련 궁금할 내용들입니다.

 

● SERVER_URL이라던가 DB ID 이런것들은 어디에 오픈되어 있나요?

· 여러분들의 서버 주소와 DB 정보를 작성하시면 됩니다. 서버 구축 방법은 아래 링크 글 시리즈에 있습니다.

2019.01.28 - [서버(공부중)] - 우분투 서버 설치하기 (Ubuntu Server)

 

● DB 구조가 궁금합니다 어떻게 구성되어있나요?

· DB 테이블 구조 정리해서 올리도록 하겠습니다.

 

 

추가적으로 궁금하신 점은 댓글 주시면 답해드리겠습니다.