힐스테이트 광교산::엘리베이터 현재 층수 및 이동 방향 표시 엔티티 추가 (HomeAssistant)

 

현대통신 RS-485 연동코드: 엘리베이터 호기별 층수 및 이동방향 표시 엔티티 추가 (홈어시스턴트)

 

엘리베이터와 관련된 요구사항이 아래와 같이 추가되었다

 

요구사항: 엘리베이터의 층수 및 이동 방향을 HA에서 확인

 

어차피 엘리베이터 호출 후 RS-485 패킷에서 확인할 수 있는 층수랑 방향은 기존 코드에서도 모니터링할 수 있는데, HA랑 별도로 연동해두진 않았었다

 

이번 기회에 엘리베이터 호출 후 각 호기별 층수와 이동방향을 표기하는 HA 엔티티를 자동으로 추가하도록 코드를 조금 수정해봤다 

commit id: f2b6efc600b4e691f549a97a6df91728a1a997da

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

- 엘리베이터: 패킷 해석 및 상태 전이 구문 강화 · YOGYUI/HomeNetwork@f2b6efc

※ 기존 코드에서 시퀀스상 불합리했던 지점 (query - response 패킷간 state 전이 시 상태 꼬임)도 이번 기회에 state 변수를 나눠서 확실하게 해결했다

 

엘리베이터 관련 패킷은 아래 링크에서 확인할 수 있다

힐스테이트 광교산::엘리베이터 호출 패킷 추가 분석 및 코드 적용

힐스테이트 광교산::엘리베이터 호출 패킷 추가 분석 및 코드 적용

1. 엘리베이터 호기별 센서 엔티티 MQTT Discovery 토픽 추가

class Elevator(Device):
    def configMQTT(self, retain: bool = False):
        topic = f'{self.ha_discovery_prefix}/switch/{self.unique_id}_calldown/config'
        obj = {
            "name": self.name + " Call (Down)",
            "object_id": self.unique_id + "_calldown",
            "unique_id": self.unique_id + "_calldown",
            "state_topic": self.mqtt_publish_topic,
            "command_topic": self.mqtt_subscribe_topic,
            "value_template": '{ "state": {{ value_json.state }} }',
            "payload_on": '{ "state": 6 }',
            "payload_off": '{ "state": 0 }',
            "icon": "mdi:elevator"
        }
        self.mqtt_client.publish(topic, json.dumps(obj), 1, retain)

        topic = f'{self.ha_discovery_prefix}/sensor/{self.unique_id}_arrived/config'
        obj = {
            "name": self.name + " Arrived",
            "object_id": self.unique_id + "_arrived",
            "unique_id": self.unique_id + "_arrived",
            "state_topic": self.mqtt_publish_topic,
            "value_template": "{% if value_json.state == 0 %} \
                               IDLE \
                               {% elif value_json.state == 1 %} \
                               ARRIVED \
                               {% else %} \
                               MOVING \
                               {% endif %}",
            "icon": "mdi:elevator-passenger"
        }
        self.mqtt_client.publish(topic, json.dumps(obj), 1, retain)

기존에는 엘리베이터 호출 스위치 및 이동 상태 (IDLE, MOVING, ARRIVED)를 표기하는 센서 두 엔티티만 자동으로 추가하고 있었다 (개인적으로는 이걸로도 충분했지만 ㅋㅋ)

 

class Elevator(Device):
    ha_dev_config_list: List[dict]
    
    def configMQTTDevInfo(self, ev_dev_idx: int, retain: bool = False):
        find = list(filter(lambda x: x.get('index') == ev_dev_idx, self.ha_dev_config_list))
        if len(find) == 0:
            ev_info = {
                'index': ev_dev_idx,
                'config': False
            }
            self.ha_dev_config_list.append(ev_info)
        else:
            ev_info = find[0]

        if not ev_info.get('config', False):
            topic = f'{self.ha_discovery_prefix}/sensor/{self.unique_id}_{ev_dev_idx}_floor/config'
            obj = {
                "name": self.name + f" #{ev_dev_idx} Floor",
                "object_id": self.unique_id + f"_{ev_dev_idx}_floor",
                "unique_id": self.unique_id + f"_{ev_dev_idx}_floor",
                "state_topic": self.mqtt_publish_topic + f'/dev/{ev_dev_idx}',
                "value_template": "{{ value_json.floor }}",
                "icon": "mdi:counter"
            }
            self.mqtt_client.publish(topic, json.dumps(obj), 1, retain)

            topic = f'{self.ha_discovery_prefix}/sensor/{self.unique_id}_{ev_dev_idx}_direction/config'
            obj = {
                "name": self.name + f" #{ev_dev_idx} Direction",
                "object_id": self.unique_id + f"_{ev_dev_idx}_direction",
                "unique_id": self.unique_id + f"_{ev_dev_idx}_direction",
                "state_topic": self.mqtt_publish_topic + f'/dev/{ev_dev_idx}',
                "value_template": "{{ value_json.direction }}",
                "icon": "mdi:swap-vertical-bold"
            }
            self.mqtt_client.publish(topic, json.dumps(obj), 1, retain)

            ev_info['config'] = True
            writeLog(f"EV#{ev_dev_idx} HA entity configured", self)

 

엘리베이터 패킷은 엘리베이터가 2대 이상일 경우 각 호기별 층수와 이동방향, 명령상태를 각각의 RS-485 패킷으로 정보를 주고받는다

 

이를 반영해 각 호기별 현재 층수 및 이동방향에 대한 엔티티를 자동으로 추가하도록 위와 같이 MQTT discovery config 토픽 및 페이로드를 구현했다 (도착 여부 센서와 유사하게 각 호기별로 2개의  sensor 엔티티 추가)

2. 호기별 층수 및 이동방향 MQTT 페이로드 구현

class Elevator(Device):
    def publishMQTTDevInfo(self):
        for elem in self.ha_dev_config_list:
            ev_index = elem.get('index')
            dev_find = list(filter(lambda x: x.index == ev_index, self.dev_info_list))
            if len(dev_find) > 0:
                dev = dev_find[0]
                moving_state = dev.moving_state
                direction = moving_state.name if moving_state in [MovingState.MOVINGUP, MovingState.MOVINGDOWN] else ""
                floor = dev.floor if moving_state in [MovingState.MOVINGUP, MovingState.MOVINGDOWN] else ""
                obj = {
                    "direction": direction.replace("MOVING", ""),
                    "floor": floor
                }
            else:
                obj = {
                    "direction": "",
                    "floor": ""
                }
            topic = self.mqtt_publish_topic + f'/dev/{ev_index}'
            self.mqtt_client.publish(topic, json.dumps(obj), 1)

 

각 호기별 층수(floor) 및 이동 상태(moving_state)는 기존에도 Elevator 객체의 dev_info_list 리스트 멤버변수에 객체를 담아 정보를 저장하고 있었기 때문에, 각 호기별 인덱스(index)별 MQTT 메시지 발행(publish) 메서드 publishMQTTDevInfo 를 위와 같이 손쉽게 구현할 수 있었다

3. 호기별 엔티티 Discovery 토픽 발행 시퀀스 구현

class Elevator(Device):
    def updateState(self, _: int, **kwargs):
        data_type = kwargs.get('data_type')
        if data_type == 'query':
            command_state = CommandState(kwargs.get('command_state', 0))  # possible values: 0(idle), 5(command up), 6(command down)
            moving_state = MovingState(kwargs.get('moving_state', 0))  # possible values: 0(idle), 1(arrived), 5(moving upside), 6(moving downside)
            ev_dev_idx = kwargs.get('ev_dev_idx', 0)
            floor = kwargs.get('floor', '')
            if command_state != CommandState.IDLE:
                self.configMQTTDevInfo(ev_dev_idx, True)  # HA Config MQTT Each Elevators
                self.publishMQTTDevInfo()
            else:
                self.publishMQTTDevInfo()

 

RS-485 패킷 해석 후 상태 업데이트(updateState)시 각 호기별 정보 패킷을 읽었다면, 해당 정보를 MQTT로 발행해주면 된다 (위에서 구현한 publishMQTTDevInfo 메서드 호출)

또한, 호기별 MQTT Discovery 토픽이 발행되었는지 판단한 뒤, 발행되지 않았다면 configMQTTDevInfo 메서드를 호출해준다

[참고 사항]
- 엘리베이터 패킷은 평소에는 호기별 정보를 확인할 수 없다
- 사용자가 월패드 등에서 호출 후 호기별 정보가 포함된 RS-485 패킷을 볼 수 있다
- 따라서 앱 실행 후 최초에는 엘리베이터를 한번 호출해줘야 정상적으로 엔티티를 추가할 수 있다

 

어차피 HA에 엔티티를 추가하는 과정을 '완전 자동화'하는 것이 목표였기에 별도로 엘리베이터 호기 관련 config를 하지 않는 방향으로 구현했기 때문에, 최초 1번 호출하는 정도의 번거로움은 충분히 감내할만하지 않을까? ㅎㅎ

4. HA 엔티티 Discovery 결과

기존에도 추가되었던 switch.elevator_0_0_calldown 및 sensor.elevator_0_0_arrived 엔티티 외에도 호기별로 2개의 엔티티가 추가된다

• sensor.elevator_0_0_{호기}_floor
• sensor.elevator_0_0_{호기}_direction

 

우리 집은 6호기와 7호기가 패킷으로 보이기 때문에 위와 같이 4개의 엔티티가 추가되었다

06, 07호기 패킷 (패킷 인덱스 10)

 

대시보드에서 엘리베이터 관련 엔티티만 아래와 같이 모아줬다

5. DEMO

 

호출 후 패킷이 들어오는 즉시 각 호기별 이동방향(Direction) 및 현재 층수 (Floor)를 위와 같이 모니터링할 수 있다

※ 주의: Floor, Direction 모두 문자열(character)로 처리된다!

 

 

도착 후에는 Arrived 센서의 문자열이 'ARRIVED'로 바뀌며, 각 호기별 층수 및 이동방향 엔티티의 값은 빈 문자열로 Clear된다 (N/A보다는 빈 문자열이 더 보기가 좋다고 판단)

 

그리고 일정 시간(default 10초)이 지난 후에는 Arrived 센서의 값이 'IDLE'로 초기화된다

ARRIVED → IDLE 상태 전이 시 시간 지연(time delay)를 준 이유?
애플 홈킷이나 구글 홈 등 홈네트워크 플랫폼에서 arrived 센서의 값을 기준으로 '알람' 기능을 구현했는데, 시간 지연이 없으면 센서의 값이 ARRIVED로 유지되는 시간이 너무 짧아 플랫폼의 알람을 트리거할 수 없는 문제가 있다

 

끝~!