스마트폰 (Mobile)

LG 에어컨 MQTT 브로커 통해서 홈어시스턴트 연동 자동화 스마트홈

날으는물고기 2024. 10. 20. 00:07

Home Assistant에서 MQTT를 사용하여 LG 에어컨을 제어하기 위해, MQTT 브로커를 설정하고, 해당 브로커와 통신할 수 있는 코드를 작성해야 합니다. MQTT 브로커로는 일반적으로 Mosquitto를 많이 사용합니다. 아래는 이를 설정하고 사용하는 방법에 대한 단계별 가이드입니다.

  1. MQTT 브로커 설치 및 설정 (Mosquitto 예시) 
    sudo apt-get update
sudo apt-get install mosquitto mosquitto-clients
  2. Home Assistant에 MQTT 통합 설정
    Home Assistant 설정 파일 (configuration.yaml)에 MQTT 브로커 설정을 추가합니다. 
    mqtt:
  broker: your_broker_ip
  port: 1883
  username: your_username
  password: your_password
  3. Python 스크립트를 통해 MQTT 메시지 전송 및 수신
    아래는 기존의 TCP 소켓 통신 코드를 MQTT를 사용하도록 수정한 Python 코드입니다. 
    import paho.mqtt.client as mqtt
import sys
import binascii

mqtt_broker = 'your_broker_ip'
mqtt_port = 1883
mqtt_topic = 'home/lg_airconditioner/command'
mqtt_response_topic = 'home/lg_airconditioner/response'

packet = '8000A3' + sys.argv[1]

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("Connected to MQTT Broker!")
    else:
        print("Failed to connect, return code %d\n", rc)

def on_message(client, userdata, msg):
    print("Received message: ", binascii.hexlify(msg.payload).decode())
    client.disconnect()

client = mqtt.Client()
client.username_pw_set('your_username', 'your_password')
client.on_connect = on_connect
client.on_message = on_message

client.connect(mqtt_broker, mqtt_port, 60)

client.loop_start()

client.subscribe(mqtt_response_topic)
client.publish(mqtt_topic, bytes.fromhex(packet))

try:
    client.loop_forever()
except KeyboardInterrupt:
    client.disconnect()
  4. Home Assistant 자동화 설정
    Home Assistant에서 MQTT 메시지를 수신하고, 에어컨 제어 명령을 처리하도록 자동화를 설정합니다. 
    automation:
  - alias: 'LG Air Conditioner Command'
    trigger:
      platform: mqtt
      topic: 'home/lg_airconditioner/command'
    action:
      - service: script.lg_airconditioner_control
        data_template:
          command: '{{ trigger.payload }}'

script:
  lg_airconditioner_control:
    sequence:
      - service: mqtt.publish
        data_template:
          topic: 'home/lg_airconditioner/response'
          payload: '{{ response_payload }}'
  5. Home Assistant 커스텀 스크립트 작성
    command 변수를 처리하여 에어컨 제어 명령을 수행하는 스크립트를 작성합니다. 
    import paho.mqtt.publish as publish

def handle_command(command):
    # 에어컨 제어 로직을 여기에 구현
    # 예시로 단순히 받은 명령을 그대로 응답하는 로직
    response_payload = command

    publish.single('home/lg_airconditioner/response', payload=response_payload, hostname='your_broker_ip')

if __name__ == '__main__':
    import sys
    command = sys.argv[1]
    handle_command(command)

위 단계를 따라 설정을 완료하면, Home Assistant를 통해 LG 에어컨을 MQTT로 제어할 수 있습니다.

1. 요구 사항 이해

  • lgac/state/01부터 lgac/state/04까지의 MQTT 토픽에서 메시지를 수신하고, 이를 변환하여 다른 토픽으로 전송하는 시스템을 구축해야 합니다.
  • 이전에 수신된 값을 기억하고, 값이 변경되었을 때만 변환 및 전송하도록 설계해야 합니다.
  • 변환 과정에서 발생할 수 있는 오류를 처리하고, 시스템이 종료되지 않도록 예외 처리를 추가해야 합니다.
  • Home Assistant의 MQTT Discovery 기능을 활용하여 자동으로 센서를 등록하는 기능을 구현해야 합니다.
  • 로깅을 설정하여 시스템의 동작을 모니터링할 수 있도록 해야 합니다.

2. 기능별 설계

  • MQTT 클라이언트 초기화
    • MQTT 브로커에 연결하고, 필요한 토픽을 구독합니다.
    • 메시지 수신 시 호출되는 콜백 함수들을 설정합니다.
  • 이전 값 기억
    • 각 토픽에 대해 이전에 수신한 값을 기억하기 위해 딕셔너리를 사용합니다.
    • 새로운 값이 수신되면 이전 값과 비교하여 변경된 경우에만 메시지를 변환하고 전송하도록 합니다.
  • 메시지 변환 및 전송
    • 수신된 메시지를 16진수 문자열로 변환합니다.
    • 변환된 메시지를 필요한 토픽으로 전송합니다.
  • 예외 처리
    • 메시지 변환 과정에서 발생할 수 있는 오류를 처리하기 위해 예외 처리를 추가합니다.
    • 오류가 발생하더라도 시스템이 종료되지 않도록 합니다.
  • Home Assistant 통합
    • Home Assistant의 MQTT Discovery 기능을 활용하여 센서를 자동으로 등록합니다.
    • 초기 실행 시 각 센서에 대한 설정 메시지를 Home Assistant로 전송하여 센서를 등록합니다.
  • 로깅 설정
    • 시스템의 동작 상태를 모니터링할 수 있도록 로깅을 설정합니다.
    • 로그 파일은 일정 주기로 회전되도록 설정하고, 로그는 콘솔에도 출력되도록 합니다.

3. 구현 순서

  • 로깅 설정
    • 로그 파일 경로와 포맷터를 설정하고, 로그 파일을 일정 주기로 회전하도록 TimedRotatingFileHandler를 사용하여 설정합니다.
  • MQTT 클라이언트 초기화
    • MQTT 브로커에 연결하고, 필요한 토픽을 구독합니다.
    • 메시지 수신 시 호출될 콜백 함수들을 설정합니다.
  • 이전 값 기억 로직 추가
    • last_values 딕셔너리를 사용하여 각 토픽의 최근 값을 기억하도록 합니다.
    • 새로운 값이 수신될 때 이전 값과 비교하여 값이 변경된 경우에만 변환 및 전송이 이루어지도록 합니다.
  • 메시지 변환 및 전송
    • 수신된 메시지를 16진수 문자열로 변환하고, 변환된 메시지를 필요한 토픽으로 전송하는 로직을 구현합니다.
  • 예외 처리 추가
    • 메시지 변환 과정에서 발생할 수 있는 오류를 처리하는 예외 처리 로직을 추가하여 시스템이 종료되지 않도록 합니다.
  • Home Assistant 통합
    • Home Assistant의 MQTT Discovery 기능을 사용하여 센서를 자동으로 등록하는 기능을 구현합니다.
    • 초기 실행 시 필요한 설정 메시지를 Home Assistant로 전송하여 센서를 등록합니다.

4. 테스트 및 디버깅

  • 구현된 코드를 테스트하여 모든 기능이 정상적으로 작동하는지 확인합니다.
  • 메시지 변환 및 전송, 값의 변경 여부 판단, Home Assistant와의 통합, 예외 처리, 로깅 등의 기능을 테스트합니다.
  • 필요에 따라 디버깅을 통해 오류를 수정하고, 로깅을 통해 시스템의 동작을 모니터링합니다.

5. 최종 검토 및 배포

  • 모든 기능이 정상적으로 동작하는 것을 확인한 후, 코드를 최종 검토합니다.
  • 배포 환경에서 시스템을 실행하고, 로그를 통해 시스템이 정상적으로 작동하는지 지속적으로 모니터링합니다.
import paho.mqtt.client as mqtt
import binascii
import logging
import os
from logging.handlers import TimedRotatingFileHandler
import json

# 로깅 설정
logger = logging.getLogger('lgac_forward')
logger.setLevel(logging.DEBUG)

# 현재 파이썬 파일의 절대 경로를 기준으로 로그 파일 경로 설정
current_file_path = os.path.abspath(__file__)
current_dir = os.path.dirname(current_file_path)
log_dir = os.path.join(current_dir, 'log')
os.makedirs(log_dir, exist_ok=True)
log_file = os.path.join(log_dir, 'lgac_forward.log')

# 파일 핸들러와 포매터 설정
formatter = logging.Formatter(fmt="%(asctime)s %(levelname)-8s %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
file_handler = TimedRotatingFileHandler(log_file, when="midnight", backupCount=7)
file_handler.setFormatter(formatter)
file_handler.suffix = "%Y%m%d"
logger.addHandler(file_handler)

# 콘솔 핸들러 설정 (필요할 경우)
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)

# 바이너리 데이터를 16진수 문자열로 변환하는 함수
def binary_to_hex(packet):
    return binascii.hexlify(packet).decode('utf-8')

# 16진수 문자열을 바이너리 데이터로 변환하는 함수
def hex_to_binary(hex_str):
    try:
        return binascii.unhexlify(hex_str)
    except binascii.Error as e:
        logger.error(f"Error converting hex to binary: {e}")
        return None

# 최근 값을 저장할 딕셔너리
last_values = {
    "lgac/state/01": None,
    "lgac/state/02": None,
    "lgac/state/03": None,
    "lgac/state/04": None,
}

# MQTT 메시지를 수신할 때 호출되는 콜백 함수 (ew11b/recv)
def on_message_recv(client, userdata, msg):
    logger.debug(f"Received message on topic {msg.topic}:")
    logger.debug(f"Raw binary data: {msg.payload}")
    hex_data = binary_to_hex(msg.payload)
    logger.debug(f"Converted Hex data: {hex_data}")

    if len(hex_data) < 32:
        logger.error("Error: Hex data is less than 32 characters.")
        return
    elif len(hex_data) > 32:
        hex_data = hex_data[:32]

    sequence_number = hex_data[8:10]
    state_topic = f"lgac/state/{sequence_number}"

    # 값이 변경된 경우에만 전송
    if last_values.get(state_topic) != hex_data:
        last_values[state_topic] = hex_data
        logger.debug(f"Publishing to topic: {state_topic}")
        client.publish(state_topic, hex_data)
        logger.debug(f"Published converted message to {state_topic}")

# MQTT 메시지를 수신할 때 호출되는 콜백 함수 (lgac/send)
def on_message_send(client, userdata, msg):
    logger.debug(f"Received message on topic {msg.topic}:")
    hex_data = msg.payload.decode('utf-8')
    logger.debug(f"Received Hex data: {hex_data}")

    binary_data = hex_to_binary(hex_data)
    if binary_data is None:
        logger.error("Error: Failed to convert hex data to binary.")
        return

    logger.debug(f"Converted Binary data: {binary_data}")

    # 변환된 메시지를 다른 토픽으로 전송
    client.publish("ew11b/send", binary_data)
    logger.debug(f"Published converted message to ew11b/send")

# MQTT 클라이언트 초기화 함수
def init_connect():
    client = mqtt.Client()

    # 콜백 함수 설정
    client.message_callback_add("ew11b/recv", on_message_recv)
    client.message_callback_add("lgac/send", on_message_send)

    client.connect("your_broker_address")  # MQTT 브로커 주소 입력
    client.subscribe("ew11b/recv")
    client.subscribe("lgac/send")

    return client

# Home Assistant에 MQTT Discovery 메시지 전송 함수
def send_homeassistant_discovery(client):
    discovery_prefix = "homeassistant"
    sensors = [
        {"topic": "lgac/state/01", "sensor_id": "lgac_01_state"},
        {"topic": "lgac/state/02", "sensor_id": "lgac_02_state"},
        {"topic": "lgac/state/03", "sensor_id": "lgac_03_state"},
        {"topic": "lgac/state/04", "sensor_id": "lgac_04_state"},
    ]

    for sensor in sensors:
        config_topic = f"{discovery_prefix}/sensor/{sensor['sensor_id']}/config"
        config_payload = {
            "name": sensor['sensor_id'],
            "state_topic": sensor['topic'],
            "unique_id": sensor['sensor_id'],
            "device": {
                "identifiers": [sensor['sensor_id']],
                "name": sensor['sensor_id'],
                "model": "LGAC",
                "manufacturer": "LG"
            }
        }
        client.publish(config_topic, json.dumps(config_payload), retain=True)
        logger.debug(f"Published Home Assistant discovery message for {sensor['sensor_id']}")

# 메인 함수
if __name__ == "__main__":
    client = init_connect()
    send_homeassistant_discovery(client)  # Home Assistant MQTT Discovery 메시지 전송
    try:
        client.loop_forever()
    except Exception as e:
        logger.exception("Daemon finished!")

Home Assistant에서 4개의 센서를 5분 단위로 반복해서 실행하는 자동화를 설정하려면, time_pattern 트리거와 for 루프를 사용하여 각 센서를 개별적으로 처리할 수 있습니다. 그러나 Home Assistant 자동화 YAML에서는 직접적인 반복문을 사용할 수 없으므로, 여러 자동화를 설정하거나 Python 스크립트를 사용하여 반복적인 작업을 수행할 수 있습니다.

1. Home Assistant 자동화 설정

다음은 time_pattern 트리거를 사용하여 5분마다 자동화가 실행되도록 설정하는 예제입니다.

automation:
  - alias: "LGAC 01-04 Command Send Every 5 Minutes"
    trigger:
      - platform: time_pattern
        minutes: "/5"
    action:
      - service: mqtt.publish
        data:
          topic: "home/lgac/send"
          payload_template: >
            {{ '8000A3' + state_attr('sensor.lgac_01_state', 'operscan') + state_attr('sensor.lgac_01_state', 'opermode') + state_attr('sensor.lgac_01_state', 'temperature') + state_attr('sensor.lgac_01_state', 'checksumscan') + (states('binary_sensor.ew11b_01') if is_state_attr('sensor.ew11b_01_status', 'value', 'true') else '') }}
      - service: mqtt.publish
        data:
          topic: "home/lgac/send"
          payload_template: >
            {{ '8000A3' + state_attr('sensor.lgac_02_state', 'operscan') + state_attr('sensor.lgac_02_state', 'opermode') + state_attr('sensor.lgac_02_state', 'temperature') + state_attr('sensor.lgac_02_state', 'checksumscan') + (states('binary_sensor.ew11b_02') if is_state_attr('sensor.ew11b_02_status', 'value', 'true') else '') }}
      - service: mqtt.publish
        data:
          topic: "home/lgac/send"
          payload_template: >
            {{ '8000A3' + state_attr('sensor.lgac_03_state', 'operscan') + state_attr('sensor.lgac_03_state', 'opermode') + state_attr('sensor.lgac_03_state', 'temperature') + state_attr('sensor.lgac_03_state', 'checksumscan') + (states('binary_sensor.ew11b_03') if is_state_attr('sensor.ew11b_03_status', 'value', 'true') else '') }}
      - service: mqtt.publish
        data:
          topic: "home/lgac/send"
          payload_template: >
            {{ '8000A3' + state_attr('sensor.lgac_04_state', 'operscan') + state_attr('sensor.lgac_04_state', 'opermode') + state_attr('sensor.lgac_04_state', 'temperature') + state_attr('sensor.lgac_04_state', 'checksumscan') + (states('binary_sensor.ew11b_04') if is_state_attr('sensor.ew11b_04_status', 'value', 'true') else '') }}
  • 트리거: time_pattern 트리거를 사용하여 매 5분마다 자동화를 실행합니다.
  • 작업: 각 센서(lgac_01에서 lgac_04)에 대해 개별적으로 MQTT 메시지를 발행합니다.

이 예제는 각 센서에 대해 개별적으로 mqtt.publish 서비스를 호출하여 5분마다 메시지를 발행합니다.

2. Python 스크립트를 사용한 반복 처리

Python 스크립트를 사용하여 반복적인 작업을 처리하고 Home Assistant의 shell_command를 통해 스크립트를 실행하는 방법도 있습니다. Python 스크립트를 작성하여 5분마다 센서 값을 MQTT로 전송합니다.

import paho.mqtt.client as mqtt
import time
import binascii

broker_address = "your_mqtt_broker_address"
broker_port = 1883
mqtt_topic_send = "home/lgac/send"
sensor_ids = ['01', '02', '03', '04']

client = mqtt.Client("LGAC_Sender")
client.connect(broker_address, broker_port)
client.loop_start()

def send_sensor_data(sensor_id):
    # 이 부분은 실제 센서 값을 가져오는 코드로 교체해야 합니다.
    operscan = "example_operscan"
    opermode = "example_opermode"
    temperature = "example_temperature"
    checksumscan = "example_checksumscan"
    ew11b_status = "example_ew11b_status"

    packet = f"8000A3{operscan}{opermode}{temperature}{checksumscan}{ew11b_status}"
    client.publish(mqtt_topic_send, packet)
    print(f"Sent data for sensor {sensor_id}: {packet}")

try:
    while True:
        for sensor_id in sensor_ids:
            send_sensor_data(sensor_id)
        time.sleep(300)  # 5분 대기
except KeyboardInterrupt:
    client.loop_stop()
    client.disconnect()

Python 스크립트를 /config/python_scripts/lgac_sender.py에 저장하고, shell_command를 통해 실행합니다. configuration.yaml에 다음을 추가합니다.

shell_command:
  start_lgac_sender: 'python3 /config/python_scripts/lgac_sender.py'

다음으로, Home Assistant가 시작될 때 Python 스크립트를 실행하도록 자동화를 설정합니다.

automation:
  - alias: "Start LGAC Sender"
    trigger:
      - platform: homeassistant
        event: start
    action:
      - service: shell_command.start_lgac_sender
  • Home Assistant 자동화: time_pattern 트리거를 사용하여 매 5분마다 각 센서의 데이터를 MQTT로 발행합니다.
  • Python 스크립트: 5분마다 반복적으로 각 센서의 데이터를 MQTT로 발행하는 스크립트를 작성하고, Home Assistant의 shell_command를 통해 실행합니다.

 

이렇게 하면 4개의 센서에 대해 5분 단위로 반복적으로 데이터를 전송하는 작업을 효과적으로 처리할 수 있습니다.

728x90
저작자표시 비영리 동일조건