사물인터넷

서브 주제인 사물인터넷에 관한 다양한 이야기를 올려주세요.

제목홈어시스턴트 : 데이터 처리2022-01-19 02:40
작성자user icon Level 4

88x31.png


홈어시스턴트는 푸시(push)와 폴링(polling) 방식으로 API를 제공합니다.

이번 글에서는 푸시와 폴링을 통한 데이터 처리를 설명하겠습니다.


푸시(push) 방식은 서버에서 클라이언트로 데이터를 보내는 방식입니다. 서버 API를 구독하고 새로운 데이터가 사용가능할 때 API로부터 알림을 받는 방식으로 클라이언트의 리소스 소모가 적다는 장점이 있습니다.


폴링(polling) 방식은 지정된 간격으로 서버 API에서 최신 데이터를 가져오는 방식입니다. 폴링 방식이 매우 일반적이기때문에 홈어시스턴트의 구성요소는 폴링 방식을 기반으로 합니다.


우선 푸시 방식을 통한 데이터를 처리해보도록 하겠습니다.


example_integration 통합 구성요소 폴더로 이동합니다.

 $ cd /usr/share/hassio/homeassistant/custom_components/example_integration


sensor 플랫폼을 다음과 같이 수정해주세요.

 $ sudo xed sensor.py

 """Example Integration sensor platform"""


 from homeassistant.components.sensor import SensorEntity

 from homeassistant.helpers import entity_platform

 from homeassistant.helpers.dispatcher import async_dispatcher_connect, async_dispatcher_send



 async def async_setup_entry(hass, entry, async_add_entities):

     """example_integration sensor platform entry setup"""


     platform = entity_platform.async_get_current_platform()

     platform.async_register_entity_service(

         "set_name", 

         {vol.Required("name"): str}, 

         "async_set_name"

     )

    

     async_add_entities([ExampleSensor(entry)])



 class ExampleSensor(SensorEntity):

     """sensor platform class"""


     def __init__(self, entry):

         """sensor init"""

         self._state = entry.data.get("name")

         self._attr_unique_id = f"example_integration.{self._state}"

         self._attr_should_poll = False


     async def async_added_to_hass(self):

         """called when entity is created"""


         async def async_update_name(name):

             """dispatcher callback"""

             self._state = name

             await self.async_write_ha_state()


         self._dispatcher_remove = async_dispatcher_connect(

             self.hass, "update_name", async_update_name)


     async def async_will_remove_from_hass(self):

         """called when entity is removed"""

         if self._dispatcher_remove:

             self._dispatcher_remove()


     @property

     def name(self):

         """return sensor name"""

         return '예제 센서'


     @property

     def state(self):

         """return sensor state"""

         return self._state


     async def async_set_name(self, **kwargs):

         """push API example"""

         name = kwargs.get("name")

         async_dispatcher_send(self.hass, "update_name", name)


mb-file.php?path=2022%2F01%2F19%2FF4512_1.png

mb-file.php?path=2022%2F01%2F19%2FF4513_2.png
 


필요한 모듈을 import합니다.

  """Example Integration sensor platform"""


 import voluptuous as vol


 from homeassistant.components.sensor import SensorEntity

 from homeassistant.helpers import config_validation as cv, entity_platform

 from homeassistant.helpers.dispatcher import async_dispatcher_connect, async_dispatcher_send


async_setup_entry함수를 정의합니다. config entry가 설정되면 실행됩니다.

 async def async_setup_entry(hass, entry, async_add_entities):

     """example_integration sensor platform entry setup"""


entity_platform 모듈의 async_get_current_platform 함수를 통해 현재 플랫폼을 불러오고 async_register_entity_service함수를 통해 서비스를 등록합니다. 서비스에 대한 내용은 생략하겠습니다. 서비스를 통해 푸시 API 데이터를 받겠습니다. 서비스 이름은 "set_name"이고 name키값을 받으며, 구성요소(entity) 객체의 async_set_name 함수를 실행합니다. 

     platform = entity_platform.async_get_current_platform()

     platform.async_register_entity_service(

         "set_name", 

         {vol.Required("name"): cv.string}, 

         "async_set_name"

     )


구성요소(entity)를 추가합니다.

     async_add_entities([ExampleSensor(entry)])


구성요소 클래스를 정의합니다.

 class ExampleSensor(SensorEntity):

     """sensor platform class"""


클래스 생성자를 작성합니다. should_poll 프로퍼티가 False를 리턴하도록 _attr_should_poll 필드를 False로 설정해줍니다.

      def __init__(self, entry):

         """sensor init"""

         self._state = entry.data.get("name")

         self._attr_unique_id = f"example_integration.{self._state}"

         self._attr_should_poll = False


이부분이 중요합니다. 구성요소가 생성될때 async_added_to_hass 함수가 실행되고, 구성요소가 제거될때 async_will_remove_from_hass 함수가 실행되는데, 구성요소를 생성할때 async_dispatcher_connect 함수를 통해 디스패처에 연결하고, 구성요소를 제거할때 디스패처 연결을 해제하면 됩니다.

async_dispatcher_connect 함수는 첫번째 인수로 홈어시스턴트 객체를, 두번째 인수로 신호문자열을, 세번째 인수로 콜백 함수를 받고 디스패쳐 연결 해제 함수를 리턴합니다. "update_name"이라는 신호를 받으면 async_update_name 함수가 실행됩니다. 콜백함수는 전달된 데이터를 인수로 받는데 이를 통해 구성요소 상태를 변환하고 async_write_ha_state 함수를 통해 홈어시스턴트 상태머신에 구성요소의 변경된 상태를 반영합니다. 

     async def async_added_to_hass(self):

         """called when entity is created"""


         async def async_update_name(name):

             """dispatcher callback"""

             self._state = name

             await self.async_write_ha_state()


         self._dispatcher_remove = async_dispatcher_connect(

             self.hass, "update_name", async_update_name)


     async def async_will_remove_from_hass(self):

         """called when entity is removed"""

         if self._dispatcher_remove:

             self._dispatcher_remove()


name 프로퍼티와 state 프로퍼티를 작성합니다.

     @property

     def name(self):

         """return sensor name"""

         return '예제 센서'


     @property

     def state(self):

         """return sensor state"""

         return self._state


서비스를 통해 실행될 async_set_name 함수를 작성합니다. async_dispatcher_send 함수를 통해 디스패처에 "update_name" 신호를 보냅니다. 전달할 데이터는 서비스를 통해 전달받은 name키값 입니다.

     async def async_set_name(self, **kwargs):

         """push API example"""

         name = kwargs.get("name")

         async_dispatcher_send(self.hass, "update_name", name)


서비스 설정 파일(services.yaml)을 작성합시다. 서비스에 대한 내용은 이번 글에서 설명하지 않겠습니다.

 $ sudo touch services.yaml && sudo xed services.yaml

 set_name:

   name: 이름 설정

   description: 이름을 설정합니다.

   target:

     entity:

       domain: sensor

   fields:

     name:

       name: 이름

       description: 변경할 이름을 입력합니다.

       required: true

       selector:

         text:


mb-file.php?path=2022%2F01%2F19%2FF4514_3.png

 

저장하고 홈어시스턴트 메인화면에서 구성하기 > 서버 제어 > 다시 시작하기 를 누릅니다.

mb-file.php?path=2022%2F01%2F01%2FF4448_3.png


개발자 도구 > 서비스 누르고, 서비스를 선택해줍니다. 대상은 지난번에 만든 예제 센서를 선택하고 이름을 스팟으로 변경해봅시다.

mb-file.php?path=2022%2F01%2F19%2FF4515_4.png
 

서버로부터 푸시 데이터를 받아서 구성요소의 상태가 스팟으로 변경되는 것을 확인할 수 있습니다.

mb-file.php?path=2022%2F01%2F19%2FF4516_5.gif
 


이번에는 폴링 방식을 통한 데이터 처리를 해보겠습니다.


sensor 플랫폼을 다음과 같이 수정해주세요.

 $ sudo xed sensor.py

 """Example Integration sensor platform"""


 from datetime import timedelta

 from homeassistant.components.sensor import SensorEntity


 SCAN_INTERVAL = timedelta(seconds=1)

 

 async def async_setup_entry(hass, entry, async_add_entities):

     """example_integration sensor platform entry setup"""


     async_add_entities([ExampleSensor(entry)], True)



 class ExampleSensor(SensorEntity):

     """sensor platform class"""


     def __init__(self, entry):

         """sensor init"""

         self._state = entry.data.get("name")

         self._attr_unique_id = f"example_integration.{self._state}"


     @property

     def name(self):

         """return sensor name"""

         return '예제 센서'


     @property

     def state(self):

         """return sensor state"""

         return self._state


     def update(self):

         """polling API example"""


         f = open("/config/custom_components/example_integration/name", 'r')

         name = f.readline()

         f.close()


         if name:

             self._state = name


mb-file.php?path=2022%2F01%2F19%2FF4517_6.png
 

필요한 모듈을 import합니다.

  """Example Integration sensor platform"""


 from datetime import timedelta

 from homeassistant.components.sensor import SensorEntity


업데이트 간격을 설정하기 위해 SCAN_INTERVAL 필드를 설정합니다. timedelta 모듈을 통해 업데이트 간격을 5초로 설정했습니다. 5초마다 구성요소(entity)의 update 함수를 실행하게 됩니다.

 SCAN_INTERVAL = timedelta(seconds=5)


async_setup_entry함수를 정의합니다. config entry가 설정되면 실행됩니다.

 async def async_setup_entry(hass, entry, async_add_entities):

     """example_integration sensor platform entry setup"""



구성요소(entity)를 추가합니다. 2번째 인수로 True를 주면 구성요소가 생성될때 update 함수가 실행됩니다.

     async_add_entities([ExampleSensor(entry)])


구성요소 클래스를 정의하고 생성자와 name 프로퍼티, state 프로퍼티를 작성합니다.

 class ExampleSensor(SensorEntity):

     """sensor platform class"""


     def __init__(self, entry):

         """sensor init"""

         self._state = entry.data.get("name")

         self._attr_unique_id = f"example_integration.{self._state}"


     @property

     def name(self):

         """return sensor name"""

         return '예제 센서'


     @property

     def state(self):

         """return sensor state"""

         return self._state


update 함수를 작성합니다. 해당 통합 구성요소 폴더의 name 파일 첫줄을 읽고 구성요소 상태에 반영합니다.

     def update(self):

         """polling API example"""


         f = open("/config/custom_components/example_integration/name", 'r')

         name = f.readline()

         f.close()


         if name:

             self._state = name


저장하고 name 파일을 생성합시다.

 $ sudo touch name


홈어시스턴트 메인화면에서 구성하기 > 서버 제어 > 다시 시작하기 를 누릅니다.

mb-file.php?path=2022%2F01%2F01%2FF4448_3.png


name 파일에 문자열을 입력하고 저장하면 5초 안에 상태가 변합니다.

mb-file.php?path=2022%2F01%2F19%2FF4518_7.gif
 

#홈어시스턴트# HomeAssistant# 푸시# push# 폴링# polling# 데이터 처리
댓글
자동등록방지
(자동등록방지 숫자를 입력해 주세요)