홈어시스턴트는 푸시(push)와 폴링(polling) 방식으로 API를 제공합니다. 이번 글에서는 푸시와 폴링을 통한 데이터 처리를 설명하겠습니다.
푸시(push) 방식은 서버에서 클라이언트로 데이터를 보내는 방식입니다. 서버 API를 구독하고 새로운 데이터가 사용가능할 때 API로부터 알림을 받는 방식으로 클라이언트의 리소스 소모가 적다는 장점이 있습니다.
폴링(polling) 방식은 지정된 간격으로 서버 API에서 최신 데이터를 가져오는 방식입니다. 폴링 방식이 매우 일반적이기때문에 홈어시스턴트의 구성요소는 폴링 방식을 기반으로 합니다.
우선 푸시 방식을 통한 데이터를 처리해보도록 하겠습니다.
example_integration 통합 구성요소 폴더로 이동합니다. $ cd /usr/share/hassio/homeassistant/custom_components/example_integration |
sensor 플랫폼을 다음과 같이 수정해주세요.
"""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)
|
필요한 모듈을 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:
|
저장하고 홈어시스턴트 메인화면에서 구성하기 > 서버 제어 > 다시 시작하기 를 누릅니다. 
개발자 도구 > 서비스 누르고, 서비스를 선택해줍니다. 대상은 지난번에 만든 예제 센서를 선택하고 이름을 스팟으로 변경해봅시다. 
서버로부터 푸시 데이터를 받아서 구성요소의 상태가 스팟으로 변경되는 것을 확인할 수 있습니다. 
이번에는 폴링 방식을 통한 데이터 처리를 해보겠습니다.
sensor 플랫폼을 다음과 같이 수정해주세요.
"""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
|
필요한 모듈을 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 파일을 생성합시다.
홈어시스턴트 메인화면에서 구성하기 > 서버 제어 > 다시 시작하기 를 누릅니다.
name 파일에 문자열을 입력하고 저장하면 5초 안에 상태가 변합니다. 
|
스팟 
