Как я разрабатывал интеграцию для Home Assistant12.04.2022 20:31

Так сложилось что недавно я поставил себе Home Assistant (далее HA) для управления всем моим зоопарком устройств из одного места, что оказалось довольно удобно. Но без ложки дегтя никуда и нашлось все таки одно устройство, интеграции для которого в HA не было, а привязать его хотелось. Было решено написать собственную интеграцию. Если интересно, что из этого вышло, добро пожаловать под кат.

Дисклеймер

Статья является описанием моего пути разработки, возможно он не верный, так же я не являюсь Python-разработчиком, писал на нем первый раз, поэтому возможно множество ошибок или использования антипаттернов.

Прочитав инструкцию по разработке интеграций на официальном сайте, стало ясно что придется писать на Python. Хоть и инструкция довольно подробная, некоторые моменты мне показались не очевидными и пришлось столкнуться со многими трудностями, искать решения в других кастомных компонентах, поэтому решил описать свой опыт в статье, возможно он будет кому то полезен. В статье постараюсь описать обобщенный процесс разработки интеграции, и не сосредотачиваться на том устройстве для которого я ее разрабатывал.

Структура файлов

Модуль кастомной интеграции для HA состоит из каталога с несколькими файлами, который нужно поместить в каталог HA custom_components. Сразу важный момент, название каталога должно полностью соответствовать названию компонента — в терминологии HA называется Domain.

В каталоге должны как минимум содержаться следующие файлы:

  • __init__.py — Файл инициализации, выполняется при загрузке компонента.

  • config_flow.py — Файл не обязательный, но используется для конфигурирования процесса настройки компонента.

  • const.py — не обязательный файл, который используется для обозначения констант.

  • Файлы объектов — для каждого типа объектов, которое может реализовывать интеграция, нужно создать файл, например:

    • sensor.py — для описания объектов типа сенсор, для сбора каких либо метрик.

    • switch.py — для описания объектов типа переключатель.

    • binary_sensor.py — бинарный сенсор, имеет только два положения.

      Полный список поддерживаемых в HA объектов можно найти на официальном сайте.

  • Какие — либо другие файлы необходимые для реализации интеграции.

В моем случае было необходимо разработать компонент, который обращается во внешнее API для получения данных с сенсоров или управления устройством. В статье буду публиковать только важные, на мой взгляд, части кода, чтобы не вызвать путаницы полные листинги кода можно посмотреть по ссылке на github.

Пример файла manifest.json из моей интеграции: 

{
  "domain": "sst_cloud",
  "name": "SST Cloud integration",
  "config_flow": true,
  "documentation": "https://github.com/sergeylysov/sst_cloud",
  "requirements": ["requests","requests"],
  "ssdp": [],
  "zeroconf": [],
  "homekit": {},
  "dependencies": [],
  "codeowners": ["@sergeylysov"],
  "iot_class": "cloud_polling",
  "version": "0.1.0"
}

Описание полей хорошо представлено на официальном сайте

Отмечу важный момент, поле domain должно обязательно совпадать с названием каталога и с именем которое будет использоваться в других файлах.

Чтобы не запутаться можно создать константу с именем домена в файле const.py, затем импортировать его в других файлах.

Содержимое моего файла const.py.

DOMAIN = "sst_cloud"

В моем проекте был еще один дополнительный файл sst.py (назвать его можно как угодно, я называл исходя из разрабатываемой интеграции), в котором я описал интеграцию. Т.е. в этом файле выполняются все запросы, полученная из API информация преобразуется в классы, описанные в файле. А к этому файлу в свою очередь уже обращаются остальные части интеграции. Судя по найденным мною примерам, это общепринятая практика и сделана для того, чтобы избежать дублирования запросов, т. к. одно устройство в терминологии HA может представлять из себя одновременно несколько объектов, например сенсор и переключатель. Поэтому данные запрашиваются один раз сохранятся в объектах классов из файла sst.py, а затем вызовами из файлов объектов (sensor.py, switch.py и т. д.) получается информация из них или отправляются команды.

Схема взаимодействия интеграцииСхема взаимодействия интеграции

Чуть подробнее расскажу про структуру моего файла для интеграции, он состоит из нескольких классов. 

import requests
import json

import logging
from homeassistant.core import HomeAssistant

class SST:
#Общий класс интеграции, используется для инициализации, а так же как объект хранящий информацию обо всех устройствах интеграции
    def __init__(self, hass: HomeAssistant, username: str, password: str) -> None:
	self.devices = []
        	…

	 def pull_data(self):
		#Метод получения информации из API и создания объектов и сохранение их в массив self.devices  на основе этой информации,  специально вынесен из конструктора в отдельный метод из-за ограничений HA на вызов запросов в не асинхронных методах. В моем случае я вызывал методы API с помощью модуля request, парсил с помощью json, затем передавал нужную информацию в конструктор класса устройства.
#Далее описываются классы устройств. 
class LeakModule:
    def __init__(self, moduleDescription: json, sst: SST):
	…
	@property
    	def get_device_name(self) -> str:
        		return self._device_name
	def update(self) -> None:
		…
class Counter:
    def __init__(self, id: int, name: str, value: int):
        self._id = id
        self.name = name
        self._value = value

    @property
    def counter_id(self) -> int:
        return self._id

    @property
    def counter_name(self) -> str:
        return self.name

    @property
    def counter_value(self) -> int:
        return self._value
    …

Для всех классов устройств нужно реализовать конструктор, в котором сохраняется информация о объекте, а так же методы для получения из объекта информации, которую вы хотите выводить в HA и методы для управления например метод включения и выключения для switch.

Так же обязательно нужно реализовать метод обновления информации, может быть как обновление одного устройства так и обновление всех устройств интеграции сразу. Зависит от возможности API и структуры устройств.

Этот файл не используется напрямую HA поэтому к нему каких то особенных требований нет, просто необходимо реализовать работу с API.

Дальше рассмотрим уже использование этого файла файлом инициализации. В этом файле ключевое это вызов создания родительского объекта интеграции и метода заполнения его данными. Файл запускается в момент загрузки модуля, после загрузки HA либо после настройки интеграции.

from __future__ import annotations
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
import asyncio
#Не забываем импортировать наш основной файл
from . import sst
#И файл с константами
from .const import DOMAIN
import logging
_LOGGER = logging.getLogger(__name__)
#Перечисляем типы устройств, которое поддерживает интеграция
PLATFORMS: list[str] = ["sensor", "binary_sensor", "switch"]

async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
    #Создаем объект с подключением к сервису
    sst1 = sst.SST(hass, entry.data["username"], entry.data["password"])
    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = sst1
#Вызываем метод получения данных в асинхронной джобе
    await hass.async_add_executor_job(
             sst1.pull_data
         )

    hass.config_entries.async_setup_platforms(entry, PLATFORMS)
    return True

async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
    if unload_ok:
        hass.data[DOMAIN].pop(entry.entry_id)

    return unload_ok

Файл config_flow.py используется для описания процесса конфигурирования, можно обойтись без него, если конфигурацию устройства описывать в файле configuration.yaml. В моем примере ему нужно было передать логин и пароль для подключения к сервису. 

from __future__ import annotations

import logging
from typing import Any
import voluptuous as vol
from homeassistant import config_entries, exceptions
from homeassistant.core import HomeAssistant
from .const import DOMAIN  
from .sst import SST

_LOGGER = logging.getLogger(__name__)
#Схема данных необходимых для интеграции
DATA_SCHEMA = vol.Schema({("username"): str, ("password"): str})

class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
    VERSION = 1

    CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL

    async def async_step_user(self, user_input=None):
            
                return self.async_create_entry(title=info["title"], data=user_input)
       
       …

Дальше необходимо создать файлы с описанием объектов, которые будет использовать HA. Здесь появляются уже требования к файлам, для каждого типа объектов есть требуемый список свойств, со списком можно ознакомиться на официальном сайте.

В каждом файле каждого типа может быть описана реализация нескольких типов устройств, главное чтобы они соответствовали типу файла.

Рассмотрим пример реализации сенсора: 

#Импортируем  единицу измерения для датчика, не знаю описаны ли они где то, я брал из исходников HA
from homeassistant.const import (VOLUME_CUBIC_METERS,PERCENTAGE)
#Импортируем классы классификации устройства
from homeassistant.components.sensor import (
    SensorDeviceClass,
    SensorEntity,
    SensorStateClass,
)
from homeassistant.helpers.entity import Entity
#Не забываем импортировать константы
from .const import DOMAIN
#И основной файл интеграции
from . import sst
import logging
_LOGGER = logging.getLogger(__name__)

#Метод настройки устройств
async def async_setup_entry(hass, config_entry, async_add_entities):
#Получаем ссылку на родительский класс нашей интеграции из конфигурации модуля.
    sst1 = hass.data[DOMAIN][config_entry.entry_id]
    new_devices = []
    #Перебираем все объекты и создаем нужные на основе классов из этого файла, все созданные объекты складываем в массив.

    for module in sst1.devices:
        for counter in module.counters:
            new_devices.append(Counter(counter,module))

       
class Counter(Entity):
#Описываем используемые единицы измерения 
    _attr_unit_of_measurement = VOLUME_CUBIC_METERS
#И тип значения, которое передает сенсор
    _attr_state_class = SensorStateClass.TOTAL

    def __init__(self,counter: sst.Counter, module: sst.LeakModule):
        self._counter = counter
        self._module = module
        #Уникальный идентификатор
        self._attr_unique_id = f"{self._counter.counter_id}_WaterCounter"
        #Отображаемое имя
        self._attr_name = f"WaterCounter {self._counter.counter_name}"
        #Текущее значение
        self._state = self._counter.counter_value/1000
#Обязательное поле — информация о устройстве	
    @property
    def device_info(self):
        return {"identifiers": {(DOMAIN, self._module.get_device_id)}}
#Иконка объекта — не обязательное поле
    @property
    def icon(self):
        return "mdi:counter"
#Самое важное поле, значение объекта.
    @property
    def state(self):
        self._state = self._counter.counter_value/1000
        return self._state

Тут стоит обратить внимание на иерархию устройств и объектов в HA. Каждое устройство может содержать один или несколько объектов. Например рассмотрим гипотетический выключатель, у него есть объекты:

  • Переключатель (switch) — собственно для управления включением и выключением.

  • Уровень заряда батареи (sensor) — отображает процент заряда.

  • Состояние (binary_sensor) — текущее состояние выключателя.

В интерфейсе HA объекты отображаются как элементы устройства, поэтому связанные объекты необходимо связать между собой, для этого у них указывается одинаковый идентификатор объекта.

В одном объекте необходимо указать полную информацию об устройстве 

@property
def device_info(self):
    return {
            "identifiers": {(DOMAIN, self._module.get_device_id)},
            "name": self._module.get_device_name,
            "sw_version": "none",
            "model": "Model name",
            "manufacturer": "manufacturer",
    }

В остальных достаточно указывать только идентификатор, такой же как у основного объекта

@property
def device_info(self):
    return {"identifiers": {(DOMAIN, self._module.get_device_id)}}

Так же необходимо реализовать метод update для обновления информации о состоянии, HA будет его периодически вызывать. Т.к. у меня информация обновляется целиком для всего устройства, метод update я реализовал только в одном объекте. 

def update(self) -> None:
    self._module.update()

Который вызывает метод обновления в основном файле интеграции. HA периодически вызывает метод обновления только у одного объекта, но при его вызове обновляется информация по всем.

После разработки хотелось упростить установку модуля, реализовать установку через HACS, для этого в соответствии с инструкцией нужно выложить код на GitHub и добавить файл hacs.json с описанием интеграции, затем репозиторий можно добавить через интерфейс HACS в пункте «Пользовательские репозитории» и установить его одной кнопкой.

На этом все, всем успехов в домашней автоматизации!

© Habrahabr.ru