이제 실제로 통합 구성요소(Integration)을 만들어서 홈어시스턴트(HomeAssistant)의 기능을 확장해보도록 하겠습니다.
이번 글에서는 예전에 제가 만든 통합 구성요소(https://github.com/huimang2/covid19_kr)를 참고하여 코로나 현황을 sensor 플랫폼을 통해 표시하는 통합 구성요소를 만들어 보도록 하겠습니다.
첨부파일에 완성된 통합 구성요소 압축 파일을 올려두었습니다.
도메인명은 cov_situation으로 하고 api 데이터는 서울특별시 코로나 발생현황 페이지(https://www.seoul.go.kr/coronaV/coronaStatus.do)를 6시간에 한번씩 크롤링하도록 하겠습니다.
우선 커스텀 컴포넌트 폴더로 이동하고 cov_situation 폴더를 만듭시다.
$ cd /usr/share/hassio/homeassistant/custom_components && sudo mkdir cov_situation |
통합 구성요소 설정 파일(manifest.json)을 만들고 다음과 같이 작성합니다.
$ sudo touch manifest.json && sudo xed manifest.json |
requirements에는 코어에 존재하지 않는 의존성 pip 모듈 리스트를 작성합니다. 크롤링에 필요한 beautifulsoup4 모듈을 입력해줍시다.
다음으로 상수를 const.py 파일에 따로 작성하겠습니다.
$ sudo touch const.py && sudo xed const.py |
컴포넌트 파일(__init__.py)을 만들고 다음과 같이 작성합니다.
$ sudo touch __init__.py && sudo xed __init__.py |
"""cov_situation integration component""" from __future__ import annotations
import async_timeout import logging
from datetime import datetime, timedelta
from bs4 import BeautifulSoup
from homeassistant.core import HomeAssistant from homeassistant.config_entries import ConfigEntry from homeassistant.helpers.update_coordinator import DataUpdateCoordinator from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import ( ATTR_CUMULATIVE_CASE, ATTR_NEW_CASE, CONF_CITY, DOMAIN, PLATFORMS, URL )
_LOGGER = logging.getLogger(__name__)
async def async_setup( hass: HomeAssistant, config: dict) -> bool: """Set up the cov_situation integration component.""" hass.data.setdefault(DOMAIN, {}) return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up the cov_situation integration component from a config entry."""
if not entry.unique_id: hass.config_entries.async_update_entry(entry, unique_id=entry.data.get(CONF_CITY))
coordinator = await get_coordinator(hass)
if not coordinator.last_update_success: await coordinator.async_config_entry_first_refresh()
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Unload a config entry.""" return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def get_coordinator(hass: HomeAssistant) -> DataUpdateCoordinator: """Get the data update coordinator."""
if hass.data[DOMAIN]: return hass.data[DOMAIN]
async def async_get_data(): with async_timeout.timeout(30):
try: hdr = { "User-Agent": ( "mozilla/5.0 (windows nt 10.0; win64; x64) applewebkit/537.36 (khtml, like gecko) chrome/78.0.3904.70 safari/537.36" ) }
session = async_get_clientsession(hass)
res = await session.get(URL, headers=hdr, timeout=30) res.raise_for_status()
cr = BeautifulSoup(await res.text(), "html.parser")
except Exception as ex: _LOGGER.error("Failed to crawling Error: %s", ex) raise
selector = lambda a, b: [x.text for x in cr.select( "#move-cont1 > div:nth-of-type(2) > table.tstyle-status.pc.pc-table > tbody > tr:nth-of-type({0}) > {1}".format(a,b))]
city_list = selector("3n-2", "th") cumulative_cases = selector("3n-1", "td") new_cases = selector("3n", "td")
data = { city_list[n]: { ATTR_CUMULATIVE_CASE: cumulative_cases[n], ATTR_NEW_CASE: new_cases[n] } for n in range(len(city_list)) }
return data
hass.data[DOMAIN] = DataUpdateCoordinator( hass, logging.getLogger(__name__), name=DOMAIN, update_method=async_get_data, update_interval=timedelta(hours=6), )
await hass.data[DOMAIN].async_refresh()
return hass.data[DOMAIN]
|
필요한 모듈을 import 합니다.
from __future__ import annotations
import async_timeout import logging
from datetime import datetime, timedelta
from bs4 import BeautifulSoup
from homeassistant.core import HomeAssistant from homeassistant.config_entries import ConfigEntry from homeassistant.helpers.update_coordinator import DataUpdateCoordinator from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import ( ATTR_CUMULATIVE_CASE, ATTR_NEW_CASE, CONF_CITY, DOMAIN, PLATFORMS, URL )
_LOGGER = logging.getLogger(__name__) |
홈어시스턴트가 시작되면 컴포넌트의 async_setup 함수가 실행됩니다. setdefault 함수를 통해 코어의 data 필드를 비어있는 딕셔너리로 초기화합니다.
async def async_setup( hass: HomeAssistant, config: dict) -> bool: """Set up the cov_situation integration component.""" hass.data.setdefault(DOMAIN, {}) return True |
엔트리가 추가될때 async_setup_entry 함수가 실행되고, 삭제될때 async_unload_entry 함수가 실행됩니다. 엔트리의 unique_id가 없으면 async_update_entry 함수로 unique_id를 설정해줍니다. get_coordinator 함수를 통해 데이터 코디네이터 객체를 형성합니다. 코디네이터에 대해서는 아래에 설명하도록 하겠습니다.
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up the cov_situation integration component from a config entry."""
if not entry.unique_id: hass.config_entries.async_update_entry(entry, unique_id=entry.data.get(CONF_CITY))
coordinator = await get_coordinator(hass)
if not coordinator.last_update_success: await coordinator.async_config_entry_first_refresh()
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Unload a config entry.""" return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) |
데이터 코디네이터를 생성하기 위한 함수를 정의합니다. 함수 이름은 get_coordinator이고, 홈어시스턴트 객체를 매개변수로 받으며, 데이터 코디네이터 객체를 리턴합니다. async def get_coordinator(hass: HomeAssistant) -> DataUpdateCoordinator: """Get the data update coordinator.""" |
데이터 코디네이터 객체는 코어의 DataUpdateCoordinator 모듈을 통해 생성할 수 있습니다. polling 방식의 api 데이터는 코디네이터 객체를 통해 관리가능합니다. 일정 시간을 간격으로 지정된 함수를 실행하여 데이터를 갱신할 수 있습니다.
우선 데이터 코디네이터 객체가 존재하는지 확인하고 존재한다면 해당 객체를 리턴합니다.
if hass.data[DOMAIN]: return hass.data[DOMAIN] |
코디네이터에서 실행할 함수를 작성합시다. 함수 이름은 async_get_data이고 비동기 함수입니다.
async def async_get_data(): with async_timeout.timeout(30): |
코어의 async_get_clientsession 함수를 통해 세션을 생성하고 크롤링 페이지에 요청을 보내고 응답 객체를 받습니다. BeautifulSoup 모듈을 통해 응답 객체에서 필요한 정보를 추출하고 해당 정보를 리턴합니다. BeautifulSoup 모듈에 대해서는 자세히 설명하지 않겠습니다.
try: hdr = { "User-Agent": ( "mozilla/5.0 (windows nt 10.0; win64; x64) applewebkit/537.36 (khtml, like gecko) chrome/78.0.3904.70 safari/537.36" ) }
session = async_get_clientsession(hass)
res = await session.get(URL, headers=hdr, timeout=30) res.raise_for_status()
cr = BeautifulSoup(await res.text(), "html.parser")
except Exception as ex: _LOGGER.error("Failed to crawling Error: %s", ex) raise
selector = lambda a, b: [x.text for x in cr.select( "#move-cont1 > div:nth-of-type(2) > table.tstyle-status.pc.pc-table > tbody > tr:nth-of-type({0}) > {1}".format(a,b))]
city_list = selector("3n-2", "th") cumulative_cases = selector("3n-1", "td") new_cases = selector("3n", "td")
data = { city_list[n]: { ATTR_CUMULATIVE_CASE: cumulative_cases[n], ATTR_NEW_CASE: new_cases[n] } for n in range(len(city_list)) }
return data |
DataUpdateCoordinator 함수를 통해 데이터 코디네이터 객체를 생성하고 해당 객체를 리턴해줍시다. 리턴하기전에 코디네이터 객체의 async_refresh 메쏘드를 통해 데이터를 갱신 해줍시다. name에는 코디네이터를 구분하기 위한 이름을, update_method에는 실행할 함수를, update_interval에는 갱신 간격을 매개변수로 입력합니다.
hass.data[DOMAIN] = DataUpdateCoordinator( hass, logging.getLogger(__name__), name=DOMAIN, update_method=async_get_data, update_interval=timedelta(hours=6), )
await hass.data[DOMAIN].async_refresh()
return hass.data[DOMAIN] |
다음으로 엔트리 생성 파일(config_flow.py)을 작성합시다.
$ sudo touch config_flow.py && sudo xed config_flow.py |
"""cov_situation integration config entry"""
import voluptuous as vol
from homeassistant.helpers import config_validation as cv from homeassistant import config_entries
from .const import DOMAIN, CITY_LIST, CONF_CITY
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): """cov_situation integration ConfigFlow"""
async def async_step_user(self, user_input):
if user_input is not None:
city = user_input.get(CONF_CITY)
await self.async_set_unique_id(city) self._abort_if_unique_id_configured()
return self.async_create_entry(title=city, data=user_input)
return self.async_show_form( step_id="user", data_schema=vol.Schema({ vol.Required(CONF_CITY): vol.In(CITY_LIST) }) )
|
city값을 설정값으로 받도록 작성합니다. 스키마를 작성할때 voluptuos 모듈의 In 함수에 리스트를 넘겨주면 목록형식의 폼이 생성됩니다. 아래쪽에 엔트리를 생성할때 사진을 참고해주세요.
저장하고 backend localization을 위해 strings.json 파일을 다음과 같이 작성하세요.
$ sudo touch strings.json && sudo xed strings.json |
{ "title": "COVID-19 Situation", "config": { "step": { "user": { "title": "Pick a city to monitor", "data": { "city": "City" } } }, "abort": { "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", "already_configured": "[%key:common::config_flow::abort::already_configured_service%]" } } } |
저장하고 frotend localization을 위해 translations/ko.json 파일을 다음과 같이 작성하세요.
{ "title": "\uC11C\uC6B8\uD2B9\uBCC4\uC2DC \uCF54\uB85C\uB098 \uD604\uD669", "config": { "step": { "user": { "title": "\uC11C\uC6B8\uD2B9\uBCC4\uC2DC \uD589\uC815\uAD6C\uB97C \uC120\uD0DD\uD558\uC138\uC694.", "data": { "city": "\uD589\uC815\uAD6C" } } }, "abort": { "cannot_connect": "\uc5f0\uacb0\ud558\uc9c0 \ubabb\ud588\uc2b5\ub2c8\ub2e4", "already_configured": "\uc11c\ube44\uc2a4\uac00 \uc774\ubbf8 \uad6c\uc131\ub418\uc5c8\uc2b5\ub2c8\ub2e4" } } } |
마지막으로 센서 플랫폼 파일(sensor.py)을 작성해주세요.
$ sudo touch sensor.py && sudo xed sensor.py |
"""cov_situation integration sensor platform""" from __future__ import annotations
import re
from homeassistant.components.sensor import SensorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import get_coordinator from .const import ( ATTR_CUMULATIVE_CASE, ATTR_NEW_CASE, BRAND, CONF_CITY, DOMAIN, MODEL, SW_VERSION )
async def async_setup_entry(hass, config_entry, async_add_entities): """Set up the cov_situation integration sensor platform from a config entry."""
coordinator = await get_coordinator(hass)
city = config_entry.data.get(CONF_CITY) async_add_entities( CoronaSensor(coordinator, city, sensor_type) for sensor_type in [ATTR_CUMULATIVE_CASE, ATTR_NEW_CASE] if coordinator.data.get(city) )
class CoronaSensor(CoordinatorEntity, SensorEntity): """sensor platform class.""" def __init__(self, coordinator, city, sensor_type): """Initialize coronavirus sensor.""" super().__init__(coordinator)
self.city = city self.sensor_type = sensor_type
self._attr_unit_of_measurement = "명" self._attr_unique_id = f"{city}-{sensor_type}"
@property def device_info(self): """Return device registry information for this entity.""" return { "connections": {(self.city, self.unique_id)}, "identifiers": { ( DOMAIN, self.city, ) }, "manufacturer": BRAND, "model": f"{MODEL} {SW_VERSION}", "name": f"{self.city} 확진자 현황", "sw_version": SW_VERSION, "via_device": (DOMAIN, self.city), "entry_type": "service", }
@property def icon(self): """Return the sensor icon.""" if self.sensor_type == ATTR_CUMULATIVE_CASE: return "mdi:emoticon-neutral-outline" else: return "mdi:emoticon-sad-outline"
@property def name(self): """Return the sensor name.""" if self.sensor_type == ATTR_CUMULATIVE_CASE: return f"{self.city} 코로나 누적 확진자" else: return f"{self.city} 신규 확진자"
@property def state(self): """Return the state of the sensor.""" return "".join(re.split("[^0-9]", self.coordinator.data[self.city][self.sensor_type])) or 0
|
의존성 모듈을 import 합니다.
"""cov_situation integration sensor platform""" from __future__ import annotations
import re
from homeassistant.components.sensor import SensorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import get_coordinator from .const import ( ATTR_CUMULATIVE_CASE, ATTR_NEW_CASE, BRAND, CONF_CITY, DOMAIN, MODEL, SW_VERSION ) |
엔트리를 통해 플랫폼을 생성할 때 async_setup_entry 함수가 실행됩니다. async_add_entities 함수를 통해 구성요소(entity)를 생성하는데 누적 확진자와 신규 확진자 2개의 센서를 생성하도록 합시다.
async def async_setup_entry(hass, config_entry, async_add_entities): """Set up the cov_situation integration sensor platform from a config entry."""
coordinator = await get_coordinator(hass)
city = config_entry.data.get(CONF_CITY) async_add_entities( CoronaSensor(coordinator, city, sensor_type) for sensor_type in [ATTR_CUMULATIVE_CASE, ATTR_NEW_CASE] if coordinator.data.get(city) ) |
CoordinatorEntity 클래스를 상속하는 센서 클래스를 만듭시다. CoordinatorEntity 클래스는 코디네이터 객체를 통해 구성요소 상태를 관리할 수 있도록 하는 클래스입니다. 생성자 함수 __init__에서 super 함수를 통해 부모 클래스 생성자에 코디네이터 객체를 넘겨주면 coordinator 필드에 코디네이터 객체가 할당됩니다. 아래 프로퍼티에 대해서는 자세히 설명하지 않겠습니다.
class CoronaSensor(CoordinatorEntity, SensorEntity): """sensor platform class.""" def __init__(self, coordinator, city, sensor_type): """Initialize coronavirus sensor.""" super().__init__(coordinator)
self.city = city self.sensor_type = sensor_type
self._attr_unit_of_measurement = "명" self._attr_unique_id = f"{city}-{sensor_type}"
@property def device_info(self): """Return device registry information for this entity.""" return { "connections": {(self.city, self.unique_id)}, "identifiers": { ( DOMAIN, self.city, ) }, "manufacturer": BRAND, "model": f"{MODEL} {SW_VERSION}", "name": f"{self.city} 확진자 현황", "sw_version": SW_VERSION, "via_device": (DOMAIN, self.city), "entry_type": "service", }
@property def icon(self): """Return the sensor icon.""" if self.sensor_type == ATTR_CUMULATIVE_CASE: return "mdi:emoticon-neutral-outline" else: return "mdi:emoticon-sad-outline"
@property def name(self): """Return the sensor name.""" if self.sensor_type == ATTR_CUMULATIVE_CASE: return f"{self.city} 코로나 누적 확진자" else: return f"{self.city} 신규 확진자"
@property def state(self): """Return the state of the sensor.""" return "".join(re.split("[^0-9]", self.coordinator.data[self.city][self.sensor_type])) or 0 |
저장하고 홈어시스턴트 메인화면에서 구성하기 > 서버 제어 > 다시 시작하기 를 누릅니다. 
홈어시스턴트 메인화면에서 구성하기 > 통합 구성요소 > 통합 구성요소 추가하기 를 클릭합니다.
서울 특별시 코로나 현황을 클릭하고 행정구를 선택 후 확인을 누르면 엔트리가 생성됩니다. 
센서를 만들때 device_info 필드의 identifiers 값은 서비스나 디바이스의 수를 결정합니다. 2개의 센서가 동일한 identifiers 값을 가지도록 생성했으므로 1개의 서비스가 형성된 것을 확인할 수 있습니다.
해당 서비스 정보를 확인해보세요. 
동일하게 강동구 엔트리도 만들어 보세요.
|
스팟 
