현대통신 월패드 RS-485 상태 조회 패킷 주기적 전송 기능 추가 (깃허브, HA 애드온)

Add Hyundai HT Wallpad periodic sending RS-485 query (device state) packet

얼마 전 홈어시스턴트(Home Assistant, HA)용 현대통신 월패드 RS-485 연동 애드온 베타 버전을 출시(?)했다

Home Assistant add-on 베타버전 릴리즈

Home Assistant add-on 베타버전 릴리즈

 

별도로 홍보 활동을 한 적이 없는데, 내가 알 수 없는 경로로 알려져서인지 관련 문의가 1주일에 1~2건씩 이메일로 오고 있다 

 

대부분 문의는 사용방법에 대한 문의인지라 신속하게 처리가 가능한데, 최근 문의 중 하나는 신규 기능 개발에 대한 요구 사항이었다 (단순 피드백보다는 더 선호하는 종류의 메일)

 

월패드의 RS-485 기반 홈네트워크에 대한 배경 지식이 해박하신 분이라 빠르게 요구사항을 접수할 수 있었고, 별도의 설계 작업 없이 곧바로 구현 작업을 시작할 수 있었다

[요구사항]
카테고리: 기능 추가
요구사항 상세: 주기적으로 상태 조회(쿼리) 패킷을 전송하지 않는 구형 월패드와 연동한 환경에서, 일정 주기로 각 디바이스별 쿼리 패킷을 전송하는 기능을 추가

 

내가 거주 중인 힐스테이트 광교산에 장착된 HDHN-2000의 경우 전등, 아울렛, 난방 및 에어컨, 환기 등 연결된 각 기기별로 월패드가 수백ms 간격으로 쿼리 패킷을 전송하고 상태 응답 패킷을 해석하고 있기 때문에 따로 구현해두지 않았던 기능인데, 구형 모델과의 연동을 위해서는 꼭 필요한 기능이기에 이 기회에 빠르게 구현하고 커밋, 애드온 버전 업데이트까지 2일만에 완료!

1. 소스코드 구현 및 커밋 (깃허브)

커밋 아이디는 cd2a6e84221cad3dfb161ef646687128ff53f7dd

https://github.com/YOGYUI/HomeNetwork/commit/cd2a6e84221cad3dfb161ef646687128ff53f7dd

feature: periodic send query packet · YOGYUI/HomeNetwork@cd2a6e8

어떤 기능이 추가되었는지 간단하게 살펴보자

1.1. 상태 조회 스레드 구현

기존의 타이머 스레드 내부에 추가하면 코드가 너무 지저분해져서 가독성이 떨어질 것 같아 디바이스 별로 쿼리 패킷을 일정 간격으로 전송하는 스레드 객체를 따로 만들었다

 

[ThreadQueryState.py]

class ThreadQueryState(threading.Thread):
    _keepAlive: bool = True

    def __init__(
        self, 
        device_list: list, 
        parser_mapping: dict, 
        rs485_info_list: list,
        period: int,
        verbose: bool
    ):
        threading.Thread.__init__(self, name='Query State Thread')
        self.device_list = device_list
        self.parser_mapping = parser_mapping
        self.rs485_info_list = rs485_info_list
        self.period = period
        self.verbose = verbose
        self.sig_terminated = Callback()
        self.available = True
    
    def run(self):
        writeLog('Started', self)
        while self._keepAlive:
            for dev in self.device_list:
                if not self._keepAlive:
                    break

                dev_type: DeviceType = dev.getType()
                index = self.parser_mapping.get(dev_type)
                info: RS485Info = self.rs485_info_list[index]
                packet_query = dev.makePacketQueryState()
                while not self.available:
                    if not self._keepAlive:
                        break
                    time.sleep(1e-3)
                while info.parser.isRS485LineBusy():
                    if not self._keepAlive:
                        break
                    time.sleep(1e-3)
                if self.verbose:
                    writeLog(f'sending query packet for {dev_type.name}/idx={dev.index}/room={dev.room_index}', self)
                info.parser.sendPacket(packet_query, self.verbose)
                
                time.sleep(self.period * 1e-3)
        writeLog('Terminated', self)
        self.sig_terminated.emit()
    
    def stop(self):
        self._keepAlive = False

    def setAvailable(self, value: bool):
        self.available = value

 

Home 객체는 설정파일(config.xml)에 주기적 쿼리 기능 (query_state_period)이 활성화된 경우 위에서 구현한 스레드를 생성하도록 만들었다 (굳이 필요없는 환경에서는 전송하지 않아야 하기 때문)

[Home.py]

class Home:
    enable_periodic_query_state: bool = False
    query_state_period: int = 1000
    verbose_periodic_query_state: bool = False
    
    def initialize(self, init_service: bool, connect_rs485: bool):
        self.loadConfig()
        if init_service:
            if self.enable_periodic_query_state:
                self.startThreadQueryState()
    
    def release(self):
        self.stopThreadQueryState()
    
    def loadConfig(self):
        self.config_tree = ET.parse(self.config_file_path)
        root = self.config_tree.getroot()
        node = root.find('device')
        self.loadDeviceConfig(node)
    
    def loadDeviceConfig(self, node: ET.Element):
        periodic_query_state_node = node.find('periodic_query_state')
        enable_node = periodic_query_state_node.find('enable')
        self.enable_periodic_query_state = bool(int(enable_node.text))
        period_node = periodic_query_state_node.find('period')
        self.query_state_period = int(period_node.text)
        verbose_node = periodic_query_state_node.find('verbose')
        self.verbose_periodic_query_state = bool(int(verbose_node.text))
    
    def startThreadQueryState(self):
        if self.thread_query_state is None:
            self.thread_query_state = ThreadQueryState(
                self.device_list,
                self.parser_mapping,
                self.rs485_info_list,
                self.query_state_period,
                self.verbose_periodic_query_state
            )
            self.thread_query_state.sig_terminated.connect(self.onThreadQueryStateTerminated)
            self.thread_query_state.setDaemon(True)
            self.thread_query_state.start()

    def stopThreadQueryState(self):
        if self.thread_query_state is not None:
            self.thread_query_state.stop()

    def onThreadQueryStateTerminated(self):
        del self.thread_query_state
        self.thread_query_state = None

1.2. 디바이스별 상태 조회 패킷 생성

디바이스 객체에는 초창기부터 makePacketQueryState 메서드를 구현해뒀고 패킷 명세에 따라 바이트어레이를 생성하게만 구현해뒀는데, 이 참에 문제없이 동작할 수 있도록 깔끔하게 코드를 정리해줬다

 

1.2.1. 디바이스 공통 부모 클래스 (Device.py)

class Device:
    @abstractmethod
    def makePacketQueryState(self) -> bytearray:
        return bytearray()

 

1.2.2. 에어컨 (AirConditioner.py)

class AirConditioner(Device):
    def makePacketQueryState(self) -> bytearray:
        packet = bytearray([0xF7, 0x0B, 0x01, 0x1C, 0x01, 0x40])
        packet.append((self.room_index << 4) + (self.index + 1))
        packet.extend([0x00, 0x00])
        packet.append(self.calcXORChecksum(packet))
        packet.append(0xEE)
        return packet

 

1.2.3. 가스 밸브 (GasValve.py)

class GasValve(Device):
    def makePacketQueryState(self) -> bytearray:
        packet = bytearray([0xF7, 0x0B, 0x01, 0x1B, 0x01, 0x43])
        packet.append(0x11)
        packet.extend([0x00, 0x00])
        packet.append(self.calcXORChecksum(packet))
        packet.append(0xEE)
        return packet

 

1.2.4. 조명 (Light.py)

class Light(Device):
    def makePacketQueryState(self) -> bytearray:
        packet = bytearray([0xF7, 0x0B, 0x01, 0x19, 0x01, 0x40])
        packet.append((self.room_index << 4) + (self.index + 1))
        packet.extend([0x00, 0x00])
        packet.append(self.calcXORChecksum(packet))
        packet.append(0xEE)
        return packet

 

1.2.5. 대기전력 콘센트 (Outlet.py)

class Outlet(Device):
    def makePacketQueryState(self) -> bytearray:
        packet = bytearray([0xF7, 0x0B, 0x01, 0x1F, 0x01, 0x40])
        packet.append((self.room_index << 4) + (self.index + 1))
        packet.extend([0x00, 0x00])
        packet.append(self.calcXORChecksum(packet))
        packet.append(0xEE)
        return packet

 

1.2.6. 난방 (Thermostat.py)

class Thermostat(Device):
    def makePacketQueryState(self) -> bytearray:
        packet = bytearray([0xF7, 0x0B, 0x01, 0x18, 0x01, 0x46])
        packet.append(0x10 + self.room_index)
        packet.extend([0x00, 0x00])
        packet.append(self.calcXORChecksum(packet))
        packet.append(0xEE)
        return packet

 

1.2.7. 전열교환기 (Ventilator.py)

class Ventilator(Device):
    def makePacketQueryState(self) -> bytearray:
        packet = bytearray([0xF7, 0x0B, 0x01, 0x2B, 0x01, 0x40])
        packet.append(0x11)
        packet.extend([0x00, 0x00])
        packet.append(self.calcXORChecksum(packet))
        packet.append(0xEE)
        return packet

 

1.2.8. 기타 

• 쿼리 패킷은 공간 인덱스 및 디바이스 인덱스를 정보로 담고 있어야 하는데, 위에서 구현한 것과는 형태가 상이할 수 있다
• 엘리베이터, HEMS의 쿼리 패킷은 통신 방향에 따른 해석법이 미묘하게 달라 본문에서는 다루지 않도록 한다
• 또한, 일괄소등 스위치는 쿼리 패킷을 보낼 때 현재 '주방 가스 밸브'가 열려있는지 여부를 함께 보내야 하는데, 해당 기능은 아직 구현해놓지 않았다 (만약 다른 유저로부터 이슈 제기가 있다면 수정할 예정)

1.3. 설정 파일 템플릿 추가 (xml)

설정 파일(config.xml)에는 주기적 쿼리 기능을 활성화할 지 여부 (enable 태그), 디바이스 간 쿼리 패킷 전송 간격 (period 태그, 단위=ms), 어떤 쿼리 패킷을 보내는지 로그로 남길지 여부 (verbose 태그)를 <device> 태그의 자식 태그로 추가해줬다

<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<config>
    <device>
        <periodic_query_state>
            <enable>0</enable>
            <period>100</period>
            <verbose>0</verbose>
        </periodic_query_state>
    </device>
</config>

2. 홈어시스턴트 애드온 옵션 추가

HA 애드온 저장소의 커밋 아이디는 6783a937f256c1b7e50c1080df50b0d7c844e6e6

https://github.com/YOGYUI/homeassistant-addons/commit/6783a937f256c1b7e50c1080df50b0d7c844e6e6

1.0.7 - periodic send query state packet · YOGYUI/homeassistant-addons@6783a93

options:
  periodic_query_state:
    enable: false
    period: 1000
    verbose: false
schema:
  periodic_query_state:
    enable: bool
    period: int
    verbose: bool

 

위에서 설명한 기능을 동일하게 애드온 옵션에도 추가해서 config.xml을 애드온 시작 시 변경할 수 있도록 만들어줬다

CONFIG_FILE_PATH="$(bashio::config 'app_config_file_path')"
CONFIG_MQTT_BROKER="$(bashio::config 'mqtt_broker')"
CONFIG_RS485="[$(echo $(bashio::config 'rs485') | sed -e 's/} /},/g')]"  # parse dict list to json list format
CONFIG_DISCOVERY="$(bashio::config 'discovery')"
CONFIG_PARSER_MAPPING="$(bashio::config 'parser_mapping')"
CONFIG_PERIODIC_QUERY_STATE="$(bashio::config 'periodic_query_state')"
CONFIG_ETC="$(bashio::config 'etc')"

uwsgi --ini ${repo_path}/Hillstate-Gwanggyosan/uwsgi.ini \
  --pyargv "--config_file_path=$CONFIG_FILE_PATH \
  --mqtt_broker=$CONFIG_MQTT_BROKER \
  --rs485=$CONFIG_RS485 \
  --discovery=$CONFIG_DISCOVERY \
  --parser_mapping=$CONFIG_PARSER_MAPPING \
  --periodic_query_state=$CONFIG_PERIODIC_QUERY_STATE \
  --etc=$CONFIG_ETC"

 

중요: 해당 기능은 애드온 1.0.7 이후 버전에 적용된다

3. 테스트

다음과 같이 애드온 옵션 설정 후 각 디바이스 별로 어떤 쿼리 패킷을 보내는지 애드온 로그로 확인할 수 있다

아래 사진은 애드온 로그 예시 (sending query packet for XXX 다음에 디바이스에 해당하는 쿼리 패킷을 확인할 수 있다)

 

애초에 우리 집에서는 평소에 쿼리 패킷을 주고 받고 있기 때문에 제대로 동작하는지 별도로 확인하지는 않았고, 애드온 버전 업데이트에 대해 메일 작성자 분께 알려드렸다

 

다행히도(?) 원하셨던 대로 잘 동작한다고 한다!

 

애초에 모듈화 코드 설계 및 구현을 해뒀기 때문에 스레드 및 관련 옵션 추가를 제외하고는 코드 작성에 시간이 거의 소요되지 않았기에 빠르게 버전 업데이트를 할 수 있었다 (즉, 전혀 귀찮은 요청이 아니었다는 말 ^^)

 

왠만한 기능은 다 개발 및 적용할 수 있는 코딩 능력을 갖추고 있으니, 애드온이나 소스코드 사용자 분들께서는 추가 개발 요구사항에 대해서는 얼마든지 문의를 주시기를 바란다 ^^