Home Assistant에서 MQTT를 사용하여 LG 에어컨을 제어하기 위해, MQTT 브로커를 설정하고, 해당 브로커와 통신할 수 있는 코드를 작성해야 합니다. MQTT 브로커로는 일반적으로 Mosquitto를 많이 사용합니다. 아래는 이를 설정하고 사용하는 방법에 대한 단계별 가이드입니다.
- MQTT 브로커 설치 및 설정 (Mosquitto 예시)
sudo apt-get update sudo apt-get install mosquitto mosquitto-clients - Home Assistant에 MQTT 통합 설정
Home Assistant 설정 파일 (configuration.yaml)에 MQTT 브로커 설정을 추가합니다.mqtt: broker: your_broker_ip port: 1883 username: your_username password: your_password - 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() - 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 }}' - 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
댓글