사물인터넷

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

제목홈어시스턴트 응용편 : 코로나 현황 센서 만들기2022-01-21 01:00
작성자user icon Level 4
첨부파일cov_situation.zip (4.6KB)

88x31.png


이제 실제로 통합 구성요소(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

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

requirements에는 코어에 존재하지 않는 의존성 pip 모듈 리스트를 작성합니다.

크롤링에 필요한 beautifulsoup4 모듈을 입력해줍시다.


다음으로 상수를 const.py 파일에 따로 작성하겠습니다.

 $ sudo touch const.py && sudo xed const.py

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



컴포넌트 파일(__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]


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

필요한 모듈을 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)

            })

        )


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


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%]"

}

}

}

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

저장하고 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"

}

}

}

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


마지막으로 센서 플랫폼 파일(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


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

의존성 모듈을 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


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

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


홈어시스턴트 메인화면에서 구성하기 > 통합 구성요소 > 통합 구성요소 추가하기 ​를 클릭합니다.

mb-file.php?path=2022%2F01%2F18%2FF4497_7.png
mb-file.php?path=2022%2F01%2F18%2FF4498_8.png


서울 특별시 코로나 현황을 클릭하고 행정구를 선택 후 확인을 누르면 엔트리가 생성됩니다.

mb-file.php?path=2022%2F01%2F21%2FF4529_8.png
mb-file.php?path=2022%2F01%2F21%2FF4530_9.png
mb-file.php?path=2022%2F01%2F21%2FF4531_10.png


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

센서를 만들때 device_info 필드의 identifiers 값은 서비스나 디바이스의 수를 결정합니다. 2개의 센서가 동일한 identifiers 값을 가지도록 생성했으므로 1개의 서비스가 형성된 것을 확인할 수 있습니다.


해당 서비스 정보를 확인해보세요.

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


동일하게 강동구 엔트리도 만들어 보세요.

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

#홈어시스턴트# HomeAssistant# 컴포넌트# component# 플랫폼# platform# config entry# ConfigFlow
댓글
자동등록방지
(자동등록방지 숫자를 입력해 주세요)