0%

建造集成

创建第一个集成

好了,现在是时候编写集成的第一个代码了。太棒了。别担心,我们已经尽力让它尽可能的简单。在HA开发环境中,输入以下内容并遵循以下说明:

1
python3 -m script.scaffold integration

这将为您提供所需的一切,以构建一个能够通过用户界面进行设置的集成。我们的示例存储库中提供了更广泛的集成示例。our example repository

The minimum

scaffold集成包含的不仅仅是最小值。最小值是定义一个包含集成域的DOMAIN常量。第二部分是,它需要定义一个setup方法,该方法在设置成功时返回一个布尔值。

1
2
3
4
5
6
7
8
DOMAIN = "hello_state"


def setup(hass, config):
hass.states.set("hello_state.world", "Paulus")

# Return boolean to indicate that initialization was successful.
return True

如果你更喜欢异步组件

1
2
3
4
5
6
7
8
DOMAIN = "hello_state"


async def async_setup(hass, config):
hass.states.async_set("hello_state.world", "Paulus")

# Return boolean to indicate that initialization was successful.
return True

要加载这个,将hello_state:添加到configuration.yaml中。创建一个<config_dir>/custom_components/hello_state/__init__.py文件,使用上面两个代码块中的一个在本地测试它。

scaffold提供什么

当使用脚手架脚本时,它将超过集成的最小限度。它将包括配置流、配置流测试和基本转换基础设施,以提供配置流的国际化。

文件结构

每个集成都存储在一个以集成域命名的目录中。域名是由字符和下划线组成的短名称。此域必须是唯一的,不能更改。移动应用程序集成的域示例:mobile_app。因此,这个集成的所有文件都在文件夹mobile_app/中。

这个文件夹的最小内容是这样的:

  • manifest.json,manifest文件描述了集成和它的依赖项,more info
  • __init__.py,集成只提供一个平台,你可以把这个文件限制在一个文档字符串中,介绍集成,"""The Mobile App integration."""

集成设备,light.py,switch.py

如果您的集成要集成一个或多个设备,您需要创建一个与实体集成交互的平台来实现这一点。例如,如果您想在Home Assistant中表示一个照明设备,您将创建light.py,它将包含一个用于照明集成的照明平台。

集成服务,services.yaml

如果您的集成要注册服务,它将需要提供可用服务的描述。描述存储在services.yaml中,More information about services.yaml.

HA在哪里查找集成

当Home Assistant看到配置文件中引用的域时(例如mobile_app:),或者它是另一个集成的依赖时,它将查找集成。HA将检查以下地点:

  • <config directory>/custom_components/<domain>
  • homeassistant/components/<domain>,内置集成

您可以通过在<config directory>/custom_components文件夹中拥有与相同域的集成来覆盖内置集成,The manifest.json file requires a version tag when you override a core integration.

覆盖的核心集成可以通过概览中集成框右上角的特定图标来标识。打开您的家庭助理实例并显示您的集成。注意,不建议覆盖内置集成,因为您将不再获得更新。建议选择一个惟一的名称

Manifest(清单)

发现即查找到的一个动作,decovery

每个集成都有一个清单文件来指定关于集成的基本信息。该文件以manifest.json文件的形式保存在集成目录。即需要添加这样一个文件

1
2
3
4
5
6
7
8
9
10
11
12
{
"domain": "hue",
"name": "Philips Hue",
"documentation": "https://www.home-assistant.io/components/hue",
"issue_tracker": "https://github.com/balloob/hue/issues",
"dependencies": ["mqtt"],
"after_dependencies": ["http"],
"codeowners": ["@balloob"],
"requirements": ["aiohue==1.9.1"],
"quality_scale": "platinum",
"iot_class": "local_polling"
}

或者一个你可以复制到你的项目中的最小的例子:

1
2
3
4
5
6
7
8
9
{
"domain": "your_domain_name",
"name": "Your Integration",
"documentation": "https://www.example.com",
"dependencies": [],
"codeowners": [],
"requirements": [],
"iot_class": "cloud_polling"
}

domain

域名是由字符和下划线组成的短名称。此域必须是唯一的,不能更改。移动应用程序集成的域示例:mobile_app。域密钥必须与文件所在的目录相匹配(The domain key has to match the directory this file is in.)。

name

集成的名称

version

对于核心集成,应该忽略这一点。

集成的版本是定制集成所必需的。该版本需要是被识别的有效版本AwesomeVersion like CalVer or SemVer

documention

包含如何使用您的集成的文档的网站。如果这个集成被提交给HA,它应该是https://www.home-assistant.io/integrations/<domain>

issue tracker

集成的问题跟踪器,当用户遇到问题时,可以报告问题。如果这一整合是提交纳入HA,它应该被省略。对于内置集成,Home Assistant将自动生成正确的链接。

dependencies

依赖项是在加载集成之前希望Home Assistant成功设置的其他Home Assistant集成。如果您想从其他集成中提供功能,比如使用webhook或MQTT连接,这可能是必要的。

内置集成只能在依赖关系中指定其他内置集成。自定义集成可以在依赖关系中指定内置集成和自定义集成。

after dependencies

此选项用于指定集成可能使用但不是必需的依赖项。当after_dependencies存在时,集成的设置将等待after_dependencies设置好后再进行设置。它还将确保安装了after_dependencies的需求,这样集成中的方法就可以安全地导入了。例如,如果camera集成可能在某些配置中使用stream集成,将stream添加到camera清单的after_dependencies中,将确保streamcamera配置之前加载。如果stream没有配置,camrea仍将加载。

内置集成只能在after_dependencies中指定其他内置集成。自定义集成可以在after_dependencies中指定内置和自定义集成。

code owners

负责此集成的GitHub用户名或团队名。你至少应该在这里添加你的GitHub用户名,以及任何帮助你编写代码的人被包括在内。

config flow

如果您的集成有一个创建配置条目的配置流,则指定config_flow键。当指定时,config_flow.py文件需要存在于您的集成中。

1
2
3
{
"config_flow": true
}

requirements

需求是通常使用pip为组件安装的Python库或模块。如果您没有使用venv,那么Home Assistant将尝试将需求安装到Home Assistantconfiguration directorydeps子目录中,或者如果您在虚拟环境中运行,则安装到/to/venv/lib/python3.6/site-packages之类的路径中。这将确保所有需求在启动时都存在。如果步骤失败,比如缺少用于编译模块的包或其他安装错误,则组件将无法加载。

需求是一个字符串数组。每个条目都是pip兼容的字符串。例如,媒体播放器Cast平台依赖于Python包PyChromecast v3.2.0:[" PyChromecast ==3.2.0"]

Custom requirements during development & testing

在组件的开发过程中,针对不同版本的需求进行测试是很有用的。这可以分两步完成,以pychromecast为例:

1
2
pip install pychromecast==3.2.0 --target ~/.homeassistant/deps
hass --skip-pip

这将使用指定的版本,并防止Home Assistant试图用requirments中指定的内容覆盖它

如果您需要对需求进行更改以支持您的组件,也可以使用pip install -e

1
2
3
git clone https://github.com/balloob/pychromecast.git
pip install -e ./pychromecast
hass --skip-pip

也可以使用公共git存储库来安装需求。这可能很有用,例如,在将需求依赖关系发布到PyPI之前,测试它的更改。下面的例子将直接从GitHub安装pycoolmaster库的except_connect分支,除非当前安装的是0.2.2版本:

1
2
3
{
"requirements": ["git+https://github.com/issacg/pycoolmaster.git@except_connect#pycoolmaster==0.2.2"]
}

Custom integration requirements

自定义集成应该只包含核心不需要的需求requirements.txt

Zeroconf

如果您的集成支持通过Zeroconf进行发现,您可以将该类型添加到manifest。如果用户已经加载了zeroconf集成,它将在发现集成配置流的zeroconf步骤时加载它。

Zeroconf是一个列表,因此您可以指定多个类型进行匹配。

1
2
3
{
"zeroconf": ["_googlecast._tcp.local."]
}

某些zeroconf类型是非常通用的(例如,_printer._tcp.local., _axis-video._tcp.local._http._tcp.local.)。在这种情况下,你应该包含一个Name (name)或Properties (properties)过滤器:

1
2
3
4
5
6
7
{
"zeroconf": [
{"type":"_axis-video._tcp.local.","properties":{"macaddress":"00408c*"}},
{"type":"_axis-video._tcp.local.","name":"example*"},
{"type":"_airplay._tcp.local.","properties":{"am":"audioaccessory*"}},
]
}

请注意,属性过滤器中的所有值必须是小写的,并且可以包含fnmatch类型通配符。

ssdp

如果您的集成支持通过SSDP进行发现,您可以将该类型添加到清单中。如果用户加载了ssdp集成,它将在发现集成配置流的ssdp步骤时加载该步骤。我们通过SSDP ST、USN、EXT和服务器报头(报头名称小写)以及 UPnP device description数据支持SSDP发现。manifest值是一个匹配器字典列表,如果在SSDP/UPnP数据中找到任何指定的匹配器的所有项目,就会发现集成。您的配置流可以过滤掉重复的内容。

下面的示例有一个由三个项组成的匹配器,所有项都必须匹配,才能通过此配置进行发现。

1
2
3
4
5
6
7
8
9
{
"ssdp": [
{
"st": "roku:ecp",
"manufacturer": "Roku",
"deviceType": "urn:roku-com:device:player:1-0"
}
]
}

homekit

如果您的集成支持通过HomeKit发现,您可以将支持的模型名称添加到清单中。如果用户加载了zeroconf集成,它将在发现集成配置流的homekit步骤时加载它。

HomeKit发现工作通过测试发现的模型名是否以manifest.json中指定的任何模型名开始。

1
2
3
4
5
6
7
{
"homekit": {
"models": [
"LIFX"
]
}
}

通过HomeKit进行发现并不意味着你必须使用HomeKit协议才能与你的设备进行通信。你可以用你认为合适的方式与这个设备通信。

由于清单中的这个条目,当一个发现信息路由到集成时,该发现信息不再路由到监听HomeKit zeroconf类型的集成。

mqtt

如果您的集成支持通过MQTT进行发现,那么您可以添加用于发现的主题。如果用户加载了mqtt集成,它将在发现集成配置流的mqtt步骤时加载该步骤。

MQTT发现通过订阅manifest.json中指定的MQTT主题工作。

1
2
3
4
5
{
"mqtt": [
"tasmota/discovery/#"
]
}

dhcp

如果您的集成支持通过dhcp发现,您可以将该类型添加到清单中。如果用户已经加载了dhcp集成,它将在发现集成配置流的dhcp步骤时加载它。我们支持通过主机名和OUI被动监听DHCP发现。manifest值是一个匹配器字典列表,如果在DHCP数据中找到任何指定的匹配器的所有项目,你的集成被发现。您的配置流可以过滤掉重复的内容。

如果集成支持zeroconf或ssdp,它们应该优于dhcp,因为它通常提供更好的用户体验。

下面的示例有三个匹配器,由两个项组成。这三个匹配器中的所有项都必须匹配,才能通过此配置进行发现。

示例:

  • 如果主机名是Rachio-XYZ, macaddress是00:9D:6B:55:12:AA,就会发现。
  • 如果主机名是Rachio-XYZ,而macaddress是00:00:00:55:12:AA,则不会发生发现。
  • 如果主机名是NotRachio-XYZ,而macaddress是00:9D:6B:55:12:AA,则不会发生发现。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
{
"dhcp": [
{
"hostname": "rachio-*",
"macaddress": "009D6B*"
},
{
"hostname": "rachio-*",
"macaddress": "F0038C*"
},
{
"hostname": "rachio-*",
"macaddress": "74C63B*"
}
]
}

usb

如果您的集成支持通过usb进行发现,您可以将该类型添加到清单中。如果用户已经加载了usb集成,它将在发现集成配置流时加载usb步骤。通过从USB描述符中提取这些值,我们支持通过VID(供应商ID)、PID(设备ID)、序列号、制造商和描述来发现。有关标识这些值的帮助,请参见 How To Identify A Device。清单值是一个匹配器字典列表。如果在USB数据中找到任何指定的匹配器的所有项,就会发现集成。您的配置流可以过滤掉重复的内容。

一些VID和PID组合被许多不相关的设备使用。例如VID 10C4和PID EA60匹配任何Silicon Labs CP2102 USB-Serial桥接芯片。在匹配这些类型的设备时,重要的是要匹配描述或其他标识符,以避免意外发现。

下面的示例有两个匹配器,由两个项组成。两个匹配器中的所有项都必须匹配,才能通过此配置进行发现。

例如:

  • 如果vid为AAAA, pid为AAAA,则会发生发现。
  • 如果vid为AAAA, pid为FFFF,则不会发现。
  • 如果vid为CCCC, pid为AAAA,则不会发现。
  • 如果vid为1234,pid为ABCD, serial_number为12345678,制造商为Midway USB,描述为Version 12 Zigbee Stick,则会发现。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
{
"usb": [
{
"vid": "AAAA",
"pid": "AAAA"
},
{
"vid": "BBBB",
"pid": "BBBB"
},
{
"vid": "1234",
"pid": "ABCD",
"serial_number": "1234*",
"manufacturer": "*midway*",
"description": "*zigbee*"
},
]
}

Integration Quality Scale

(集成质量量表)Integration Quality Scale对代码质量和用户体验的集成进行评分。质量等级的每一个层次都由一系列需求组成。如果一个集成满足了所有的需求,那么它就被认为达到了那个级别。

如果您的集成没有得分,那么不要将其添加到集成的清单中。但是,一定要查看需求的集成质量量表列表。它有助于极大地改进代码和用户体验。

我们强烈建议对您的集成进行评分。

1
2
3
{
"quality_scale": "silver"
}

iot class

(物联网类别)IoT Class描述集成如何与设备或服务相连接。有关物联网分类的更多信息,请阅读有关"Classifying the Internet of Things"的博客。

清单接受以下物联网类别:

  • assumed_state:我们无法获得设备的状态。我们能做的最好的事情就是根据最后一个命令来假设状态。
  • cloud_polling:该设备的集成是通过云实现的,并且需要一个活跃的互联网连接。轮询状态意味着稍后可能会注意到更新。
  • cloud_push:该设备的集成是通过云实现的,需要主动的互联网连接。一旦新州成立,家庭助理将被通知。
  • local_polling:提供与设备的直接通信。轮询状态意味着稍后可能会注意到更新。
  • local_push:提供与设备的直接通信。一旦新州成立,家庭助理将被通知。
  • calculate:集成本身并不处理通信,而是提供一个计算结果。

Configuration

通过添加对配置流的支持来创建配置条目,因此可以通过用户界面建立集成。希望支持配置项的组件需要定义一个配置流处理程序。这个处理程序将管理来自用户输入、发现或其他来源(如Home Assistant OS)的条目的创建。

配置流处理程序控制存储在配置项中的数据。这意味着在Home Assistant启动时,不需要验证配置是否正确。它还将防止破坏性的更改,因为如果版本更改,我们将能够将配置条目迁移到新的格式。

在实例化处理程序时,Home Assistant将确保加载所有依赖项并安装组件的需求。

更新manifest

您需要更新集成清单,以通知Home Assistant您的集成有一个配置流。这是通过向清单(docs)添加config_flow: true来实现的。

定义一个配置流

配置条目使用data flow entry framework来定义它们的配置流。配置流需要在集成文件夹中的config_flow.py文件中定义,扩展homeassistant.config_entries.ConfigFlow,并传递domainkey作为继承ConfigFlow的一部分。

1
2
3
4
5
6
from homeassistant import config_entries
from .const import DOMAIN


class ExampleConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Example config flow."""

一旦你更新了清单并创建了config_flow.py,你将需要运行为HA运行python3 -m script.hassfest(只需要一次)为您的集成激活配置条目。

定义步骤(steps)

配置流需要定义配置流的步骤。 Data Entry Flow 的文档描述了步骤的不同返回值。下面是一个关于如何定义用户步骤的示例。

1
2
3
4
5
6
7
8
9
10
import voluptuous as vol

class ExampleConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_user(self, info):
if info is not None:
pass # TODO: process info

return self.async_show_form(
step_id="user", data_schema=vol.Schema({vol.Required("password"): str})
)

有几个步骤名保留给系统使用:

步骤名称 描述
discovery 如果您的集成已经被发现,并且匹配的步骤还没有定义,则不赞成调用。
dhcp 如果您的集成通过指定的DHCP被发现,则调用;请见Manifest文档
hassio 如果您的集成通过指定的Supervisor add-on,则调用;
homekit 如果您的集成通过指定的被发现,则调用;请见Manifest文档
mqtt 如果您的集成通过指定的MQTT被发现,则调用;请见Manifest文档
ssdp 如果您的集成通过指定的被发现,则调用;请见Manifest文档
usb 如果您的集成通过指定的被发现,则调用;请见Manifest文档
user 当用户通过用户界面启动流时调用,或者当发现且未定义匹配和发现步骤时调用。
Zeroconf 如果您的集成通过指定的被发现,则调用;请见Manifest文档

Unique IDs

配置流可以为配置流附加一个唯一的ID,以避免同一设备被设置两次。当设置了一个唯一ID时,如果有另一个流正在处理这个唯一ID,它将立即中止。如果这个ID已经存在一个配置条目,您也可以快速中止。配置条目将获得创建它们的流的唯一ID。

在配置流步骤中调用:

1
2
await self.async_set_unique_id(device_unique_id)
self._abort_if_unique_id_configured()

通过设置唯一ID,用户可以选择忽略配置条目的发现。这样,他们就不会再为此烦恼了。如果集成使用DHCP、HomeKit、Zeroconf/mDNS、USB或SSDP/uPnP被发现,则需要提供一个唯一的ID。

如果一个唯一的ID不可用,那么可以省略dhcp、zeroconf、hassio、homekit、ssdp、usb和discovery步骤,即使它们在集成清单中进行了配置。在这种情况下,当发现项目时将调用user步骤。

或者,如果集成不能始终获得唯一的ID(例如,多个设备,有些只有一个,有些没有),则可以使用一个助手,它仍然允许发现,只要还没有配置集成的任何实例。

1
2
3
if device_unique_id:
await self.async_set_unique_id(device_unique_id)
await self._async_handle_discovery_without_unique_id()

Unique ID需求

​ 唯一ID用于匹配底层设备或API的配置条目。唯一ID必须是稳定的,不应该被用户改变。当设备访问细节发生变化时,可以使用Unique ID更新配置条目数据。例如,对于通过本地网络进行通信的设备,如果由于新的DHCP分配而改变了IP地址,集成可以使用Unique ID来更新主机,使用下面的代码片段:

1
2
await self.async_set_unique_id(serial_number)
self._abort_if_unique_id_configured(updates={CONF_HOST: host, CONF_PORT: port})

唯一ID的可接受源示例:

  • 设备序列号
  • MAC地址:使用homeassistant.helpers.device_registry.format_mac;只能从设备API或发现处理程序中获取MAC地址。依赖于读取arp缓存或本地网络访问的工具,如getmac,将不能在所有支持的网络环境中工作,这是不可接受的。
  • 经纬度或其他独特的地理位置
  • 在设备上打印或刻录到EEPROM中的唯一标识符

有时本地设备的唯一ID的可接受源

  • 主机名:如果主机名的子集包含一个可接受的来源,则可以使用这一部分

有时云服务的唯一ID的可接受源

  • 电子邮件地址:必须格式化为小写
  • 用户名:如果用户名不区分大小写,则必须格式化为小写。
  • 帐户ID:不能有冲突

无法接受的唯一ID来源

  • IP地址
  • 设备名称
  • 主机名,如果它可以被用户更改
  • URL

Unignoring

您的配置流可以通过实现配置流中的unignore步骤来添加支持,以重新发现以前被忽略的条目。

1
2
3
4
5
6
7
async def async_step_unignore(self, user_input):
unique_id = user_input["unique_id"]
await self.async_set_unique_id(unique_id)

# TODO: Discover devices and find the one that matches the unique ID.

return self.async_show_form(…)

decovery steps

当发现一个集成时,它们各自的发现步骤(即async_step_dhcp或async_step_zeroconf)与发现信息一起被调用。该步骤将必须检查以下事项:

  • 确保在设置已发现设备的过程中没有此配置流的其他实例。如果有多种方式发现一个设备在网络上,就会发生这种情况。
  • 请确保设备尚未设置好。
  • 调用发现步骤不应该导致完成的流和配置条目。始终与用户确认。

不需要身份验证的可发现集成

如果您的集成不需要任何身份验证就可以被发现,那么您将能够使用内置的discoverable Flow。该流程提供以下功能:

  • 在完成配置流程之前,检测网络上是否可以发现设备/服务。
  • 支持所有基于清单的发现协议。
  • 限制只有一个配置条目。由配置条目来发现所有可用的设备。

开始时,运行python3 -m script.scaffold config_flow_discovery,并按照说明操作。这将创建使用discovery配置集成所需的所有样板。

通过OAuth2配置

Home Assistant内置了对使用the OAuth2 authorization framework提供帐户链接的集成支持。为了能够利用这一点,您将需要以一种允许Home Assistant负责刷新令牌的方式来构造您的Python API库。请参阅我们的API library guide,了解如何做到这一点。

内置的OAuth2支持使用本地配置的客户端ID / secret和Home Assistant云帐户链接服务开箱即用。该服务允许用户将他们的帐户与一个集中管理的客户端ID/秘密联系起来。如果您希望您的集成成为该服务的一部分,请通过hello@home-assistant.io联系我们。

开始时,运行python3 -m script.scaffold config_flow_oauth2,并按照说明操作。这将创建使用OAuth2配置集成所需的所有样板。

翻译

配置流处理程序的转换定义在组件转换文件strings.json中的config键下。Hue组件的例子:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
{
"title": "Philips Hue Bridge",
"config": {
"step": {
"init": {
"title": "Pick Hue bridge",
"data": {
"host": "Host"
}
},
"link": {
"title": "Link Hub",
"description": "Press the button on the bridge to register Philips Hue with Home Assistant.\n\n![Location of button on bridge](/static/images/config_philips_hue.jpg)"
}
},
"error": {
"register_failed": "Failed to register, please try again",
"linking": "Unknown linking error occurred."
},
"abort": {
"discover_timeout": "Unable to discover Hue bridges",
"no_bridges": "No Philips Hue bridges discovered",
"all_configured": "All Philips Hue bridges are already configured",
"unknown": "Unknown error occurred",
"cannot_connect": "Unable to connect to the bridge",
"already_configured": "Bridge is already configured"
}
}
}

当翻译合并到HA,他们将自动上传到okalise,在那里翻译团队将帮助翻译他们在其他语言。在本地开发时,需要运行python3 -m script.translations develop。翻译可以看到对string.json的更改。More info on translating Home Assistant.

Config Entry Migration

如上所述,每个配置条目都分配了一个版本。这样可以在Config Entry模式更改时将Config Entry数据迁移到新格式。

迁移可以通过在组件的__init__.py文件中实现async_migrate_entry函数来实现。如果迁移成功,函数应该返回True。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# Example migration function
async def async_migrate_entry(hass, config_entry: ConfigEntry):
"""Migrate old entry."""
_LOGGER.debug("Migrating from version %s", config_entry.version)

if config_entry.version == 1:

new = {**config_entry.data}
# TODO: modify Config Entry data

config_entry.version = 2
hass.config_entries.async_update_entry(config_entry, data=new)

_LOGGER.info("Migration to version %s successful", config_entry.version)

return True

Reauthentication

优雅地处理身份验证错误,如无效、过期或撤销令牌,需要在集成质量尺度上取得进步。本示例演示如何将reauth添加到脚本创建的OAuth流中。脚手架遵循构建Python库中的模式。

这个例子在__init__.py的配置条目设置中捕获了一个身份验证异常,并指示用户访问集成页面,以便重新配置集成。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry
from homeassistant.core import HomeAssistant
from . import api

async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Setup up a config entry."""

# TODO: Replace with actual API setup and exception
auth = api.AsyncConfigEntryAuth(...)
try:
await auth.refresh_tokens()
except TokenExpiredError as err:
raise ConfigEntryAuthFailed(err) from err

# TODO: Proceed with component setup

config_flow.py中的流处理程序还需要一些额外的步骤来支持reauth,包括显示确认、启动reauth流、更新现有的配置条目以及重新加载以再次调用setup。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class OAuth2FlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
):
"""Config flow to handle OAuth2 authentication."""

async def async_step_reauth(self, user_input=None):
"""Perform reauth upon an API authentication error."""
return await self.async_step_reauth_confirm()

async def async_step_reauth_confirm(self, user_input=None):
"""Dialog that informs the user that reauth is required."""
if user_input is None:
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema({}),
)
return await self.async_step_user()

async def async_oauth_create_entry(self, data: dict) -> dict:
"""Create an oauth config entry or update existing entry for reauth."""
# TODO: This example supports only a single config entry. Consider
# any special handling needed for multiple config entries.
existing_entry = await self.async_set_unique_id(DOMAIN)
if existing_entry:
self.hass.config_entries.async_update_entry(existing_entry, data=data)
await self.hass.config_entries.async_reload(existing_entry.entry_id)
return self.async_abort(reason="reauth_successful")
return await super().async_oauth_create_entry(data)

根据集成的细节,可能会有一些额外的考虑,比如确保跨reauth使用相同的帐户,或者处理多个配置条目。

reauth确认对话框需要在string.json附加定义用于reauth确认和成功对话框:

1
2
3
4
5
6
7
8
9
10
11
12
13
{
"config": {
"step": {
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
# TODO: Replace with the name of the integration
"description": "The Example integration needs to re-authenticate your account"
}
},
"abort": {
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
}

参见翻译本地开发说明。

身份验证失败(例如撤销的oauth令牌)对于手动测试来说有点棘手。一个建议是复制config/.storage/core.Config_entries,并根据需要测试的场景手动更改access_token、refresh_token和expires_at的值。然后,您可以遍历reauth流,并确认这些值已被新的有效令牌替换。

自动化测试应验证reauth流更新了现有的配置条目,而没有创建其他条目。

测试配置流

与配置流集成需要对config_flow.py中的所有代码进行完整的测试覆盖,才能被核心接受。测试您的代码包括关于如何生成覆盖率报告的更多细节。

配置选项

通过配置条目配置的集成可以向用户公开一些选项,以允许对集成的行为进行调整,比如应该集成哪些设备或位置。

配置条目选项使用 Data Flow Entry framework来允许用户更新配置条目选项。希望支持配置入口选项的组件需要定义一个选项流处理程序。

选项支持

为了让集成支持选项,它需要在配置流处理程序中有一个async_get_options_flow方法。调用它将返回组件选项流处理程序的实例。

1
2
3
4
@staticmethod
@callback
def async_get_options_flow(config_entry):
return OptionsFlowHandler(config_entry)

Flow handler

流处理程序的工作方式与配置流处理程序类似,只不过流中的第一步始终是async_step_init

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
class OptionsFlowHandler(config_entries.OptionsFlow):
def __init__(self, config_entry):
"""Initialize options flow."""
self.config_entry = config_entry

async def async_step_init(self, user_input=None):
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Required(
"show_things",
default=self.config_entry.options.get("show_things"),
): bool
}
),
)

Signal updates

如果集成应该对更新的选项起作用,您可以向配置条目注册一个更新监听器,当条目更新时将调用它。监听器是通过在集成的__init__.py中添加以下async_setup_entry函数来注册的

1
entry.async_on_unload(entry.add_update_listener(update_listener))

使用上述方法意味着在加载条目时附加Listener,并在卸载时分离该条目。Listener应该是一个async函数,它接受与async_setup_entry相同的输入。然后可以从entry.options访问选项。

1
2
async def update_listener(hass, entry):
"""Handle options update."""

通过yaml配置

configuration.yaml是一个由用户定义的配置文件。它是由HA在第一次启动时自动创建的。它定义了要加载的组件。

与设备和/或服务通信的集成通过配置流进行配置。在极少数情况下,我们可以破例。允许并鼓励不应该有YAML配置的现有集成实现配置流并删除YAML支持。对于这些相同的现有集成,对现有YAML配置的更改将不再被接受。

详细信息请阅读ADR-0010

Pre-processing

HA将根据指定要加载的组件对配置进行一些预处理。

CONFIG_SCHEMA

如果组件定义了一个变量CONFIG_SCHEMA,则传入的配置对象将是通过CONFIG_SCHEMA运行配置的结果。CONFIG_SCHEMA应该是一个丰富的模式。

PLATFORM_SCHEMA

如果一个组件定义了一个变量PLATFORM_SCHEMA,那么这个组件将被视为一个实体组件。实体组件的配置是平台配置的列表。

家庭助理将收集此组件的所有平台配置。它将通过在组件的域(比如light)下查找配置条目,以及在任何域+额外文本的条目下查找配置条目来实现。

在收集平台配置时,Home Assistant将验证它们。它将查看平台是否存在,以及平台是否定义了一个PLATFORM_SCHEMA,然后根据该模式进行验证。如果没有定义,它将根据组件中定义的PLATFORM_SCHEMA验证配置。任何引用非现有平台或包含无效配置的配置将被删除。

以下configuration.yaml:

1
2
3
4
5
6
7
8
9
10
unrelated_component:
some_key: some_value

switch:
platform: example1

switch living room:
- platform: example2
some_config: true
- platform: invalid_platform

将传递给组件为

1
2
3
4
5
6
7
8
9
10
11
12
13
14
{
"unrelated_component": {
"some_key": "some_value"
},
"switch": [
{
"platform": "example1"
},
{
"platform": "example2",
"some_config": True
}
],
}

自定义服务

HA为很多事情提供现成的服务,但并不总是包罗万象。与其尝试更改Home Assistant,不如先将其作为服务添加到您自己的集成中。一旦我们看到这些服务中的模式,我们就可以讨论一般化它们。

这是一个简单的“hello world”示例,展示了注册服务的基础知识。要使用这个例子,创建文件<config dir>/custom_components/hello_service/__init__.py并复制下面的例子代码。

服务可以从自动化和前端的服务“开发人员工具”中调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
DOMAIN = "hello_service"

ATTR_NAME = "name"
DEFAULT_NAME = "World"


def setup(hass, config):
"""Set up is called when Home Assistant is loading our component."""

def handle_hello(call):
"""Handle the service call."""
name = call.data.get(ATTR_NAME, DEFAULT_NAME)

hass.states.set("hello_service.hello", name)

hass.services.register(DOMAIN, "hello", handle_hello)

# Return boolean to indicate that initialization was successfully.
return True

通过在您的configuration.yaml中添加以下内容来加载集成。当组件加载后,应该可以调用一个新服务。

1
2
# configuration.yaml entry
hello_service:

打开前端,在侧栏,单击开发人员工具部分中的第一个图标。这将打开Call Service开发人员工具。在右侧,找到您的服务并点击它。这将自动填写正确的值。

按下“Call Service”将现在没有任何参数调用您的服务。这将导致你的服务创建一个默认名称为“World”的状态。如果要指定名称,则必须通过Service Data提供一个参数。在YAML模式下,添加如下内容,再按“Call Service again”。

1
2
service: helloworld_service.hello
data: { "name": "Planet" }

该服务现在将用“Planet”覆盖以前的状态

服务描述

添加服务只有在用户知道的情况下才有用。在HA中,我们使用一种services.yaml作为集成的一部分来描述服务。

服务是在集成的域名下发布的,因此在services.yaml中也是如此我们只使用服务名作为base key。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
# Example services.yaml entry

# Service ID
set_speed:
# Service name as shown in UI
name: Set speed
# Description of the service
description: Sets fan speed.
# If the service accepts entity IDs, target allows the user to specify entities by entity, device, or area. If `target` is specified, `entity_id` should not be defined in the `fields` map. By default it shows only targets matching entities from the same domain as the service, but if further customization is required, target supports the entity, device, and area selectors (https://www.home-assistant.io/docs/blueprint/selectors/). Entity selector parameters will automatically be applied to device and area, and device selector parameters will automatically be applied to area.
target:
# Different fields that your service accepts
fields:
# Key of the field
speed:
# Field name as shown in UI
name: Speed
# Description of the field
description: Speed setting
# Whether or not field is required (default = false)
required: true
# Advanced fields are only shown when the advanced mode is enabled for the user (default = false)
advanced: true
# Example value that can be passed for this field
example: "low"
# The default field value
default: "high"
# Selector (https://www.home-assistant.io/docs/blueprint/selectors/) to control the input UI for this field
selector:
select:
options:
- "off"
- "low"
- "medium"
- "high"

实体服务

有时您希望提供额外的服务来控制您的实体。例如,Sonos集成为分组和取消分组设备提供服务。实体服务是特殊的,因为用户可以通过许多不同的方式指定实体。它可以使用areas、a group or 一组实体。

你需要在你的平台上注册实体服务,比如<your-domain>/media_player.py。这些服务将在您的域而不是media player域下提供。示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from homeassistant.helpers import config_validation as cv, entity_platform, service

async def async_setup_entry(hass, entry):
"""Set up the media player platform for Sonos."""

platform = entity_platform.async_get_current_platform()

# This will call Entity.set_sleep_timer(sleep_time=VALUE)
platform.async_register_entity_service(
SERVICE_SET_TIMER,
{
vol.Required('sleep_time'): cv.time_period,
},
"set_sleep_timer",
)

如果你需要对服务调用进行更多的控制,你也可以传递一个async函数来代替"set_sleep_timer":

平台(Platforms)

Home Assistant有各种内置的集成来抽象设备类型。有lights, switches, covers, climate devices, and many more. 。您的集成可以通过创建一个平台来挂钩到这些集成中。对于要集成的每个集成,您都需要一个平台。

要创建平台,您需要创建一个文件,该文件带有您正在为之构建平台的集成的域名。因此,如果您正在构建一个light,您将向您的集成文件夹中添加一个新文件light.py

我们已经创建了两个例子,应该让你看看这是如何工作的:

Interfacing with devices

HA的一个规则是,集成绝不能直接与设备连接。相反,它应该与第三方Python 3库交互。通过这种方式,Home Assistant可以与Python社区共享代码,并保持项目的可维护性。

准备好Python库并发布到PyPI之后,将其添加到manifest.中。现在是时候实现Entity基类了,它是由您正在为其创建平台的集成所提供的。

entity index处找到您的集成,查看有哪些方法和属性可以实现。

Multiple Platforms

大多数集成由单个平台组成。在这种情况下,只定义一个平台是可以的。但是,如果要添加第二个平台,则需要集中连接逻辑。这是在组件内部完成的(__init__.py)。

如果您的集成是通过configuration.yaml可配置的,它将导致配置的入口点发生更改,因为现在用户需要直接设置集成,而由集成来设置平台。

Loading platforms when configured via a config entry

如果您的集成是通过配置条目来设置的,那么您需要将配置条目转发到适当的集成来设置您的平台。有关更多信息,请参见 config entry documentation.

Loading platforms when configured via configuration.yaml

如果你的集成没有使用配置项,它将不得不使用我们的发现助手(discovery helper)来设置它的平台。注意,这种方法不支持卸载。

为此,您需要使用来自发现助手的load_platformasync_load_platform方法。

获取数据(Fetching Data)

您的集成将需要从API获取数据,以便能够向Home Assistant提供这些数据。这个API可以通过web(本地或云)、插座、通过USB棒暴露的串行端口等方式提供。

Push vs Poll

api有许多不同的shapes and forms,但其核心分为两类:push(推送)和poll(拉取)。

通过push,我们订阅了一个API,当有新数据可用时,我们会得到API的通知。它把变化推给我们。推送api很棒,因为它们消耗的资源更少。当发生更改时,我们可以得到更改的通知,而不必重新获取所有数据并查找更改。因为实体可以被禁用,所以你应该确保你的实体在async_added_to_hass回调函数中订阅,并在删除时取消订阅。

通过poll,我们将以指定的时间间隔从API获取最新的数据。然后,您的集成会将该数据提供给它的实体,该实体将被写入Home Assistant。

由于poll常见,Home Assistant默认假设您的实体基于poll,如果不是这样,从Engity.should_poll返回False。当您禁用poll时,您的集成将负责调用其中一个方法,以指示Home Assistant是时候将实体状态写入Home Assistant了:

  • 如果你在一个异步函数中执行,并且不需要调用你的实体更新方法,调用Entity.async_write_ha_state()。这是一个异步回调函数,它将把状态写入事件循环中的状态机。
  • Entity.schedule_update_ha_state(force_refresh=False)/Entity.async_schedule_update_ha_state(force_refresh=False)会安排实体的更新。如果force_refresh设置为True, Home Assistant将在写入状态之前调用你的实体更新方法(update()/async_update())。

Polling API endpoints

我们将在这里解释几种不同的API类型,以及在Home Assistant中集成它们的最佳方法。请注意,一些集成将遇到以下几种集成的组合

Coordinated, single API poll for data for all entities

这个API将有一个单一的方法来获取您在Home Assistant中拥有的所有实体的数据。在本例中,我们希望在这个端点上进行单个周期轮询,然后让实体在有新数据可用时立即知道。

Home Assistant提供了一个dataupdatcoordinator类,以帮助您尽可能有效地管理这个问题。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
"""Example integration using DataUpdateCoordinator."""

from datetime import timedelta
import logging

import async_timeout

from homeassistant.components.light import LightEntity
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
UpdateFailed,
)

from .const import DOMAIN

_LOGGER = logging.getLogger(__name__)


async def async_setup_entry(hass, entry, async_add_entities):
"""Config entry example."""
# assuming API object stored here by __init__.py
api = hass.data[DOMAIN][entry.entry_id]

async def async_update_data():
"""Fetch data from API endpoint.

This is the place to pre-process the data to lookup tables
so entities can quickly look up their data.
"""
try:
# Note: asyncio.TimeoutError and aiohttp.ClientError are already
# handled by the data update coordinator.
async with async_timeout.timeout(10):
return await api.fetch_data()
except ApiAuthError as err:
# Raising ConfigEntryAuthFailed will cancel future updates
# and start a config flow with SOURCE_REAUTH (async_step_reauth)
raise ConfigEntryAuthFailed from err
except ApiError as err:
raise UpdateFailed(f"Error communicating with API: {err}")

coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
# Name of the data. For logging purposes.
name="sensor",
update_method=async_update_data,
# Polling interval. Will only be polled if there are subscribers.
update_interval=timedelta(seconds=30),
)

#
# Fetch initial data so we have data when entities subscribe
#
# If the refresh fails, async_config_entry_first_refresh will
# raise ConfigEntryNotReady and setup will try again later
#
# If you do not want to retry setup on failure, use
# coordinator.async_refresh() instead
#
await coordinator.async_config_entry_first_refresh()

async_add_entities(
MyEntity(coordinator, idx) for idx, ent in enumerate(coordinator.data)
)


class MyEntity(CoordinatorEntity, LightEntity):
"""An entity using CoordinatorEntity.

The CoordinatorEntity class provides:
should_poll
async_update
async_added_to_hass
available

"""

def __init__(self, coordinator, idx):
"""Pass coordinator to CoordinatorEntity."""
super().__init__(coordinator)
self.idx = idx

@property
def is_on(self):
"""Return entity state.

Example to show how we fetch data from coordinator.
"""
self.coordinator.data[self.idx]["state"]

async def async_turn_on(self, **kwargs):
"""Turn the light on.

Example method how to request data updates.
"""
# Do the turning on.
# ...

# Update the data
await self.coordinator.async_request_refresh()

Separate polling for each individual entity

有些api将为每个设备提供一个端点。有时不可能将一个设备从你的API映射到一个单独的实体。如果您从一个API设备端点创建多个实体,请参阅前一节。

如果可以将一个设备端点映射到单个实体,则可以在update()/async_update()方法中获取该实体的数据。确保polling被设置为True,并且Home Assistant将定期调用此方法。

如果你的实体需要在第一次写入Home Assistant之前获取数据,传递Trueadd_entities方法:add_entities([MyEntity()], True)

您可以通过在平台中定义SCAN_INTERVAL常量来控制集成的轮询间隔。小心设置太低。它将占用HA的资源,可能会使托管API的设备不堪重负,或者会阻止你使用云API。

1
2
3
from datetime import timedelta

SCAN_INTERVAL = timedelta(seconds=5)

Request Parallelism(并行性)

这是一个高级话题

Home Assistant有内置的逻辑,以确保集成不会破坏APIs同时还可以利用Home Assistant中的所有可用资源。这种逻辑是围绕限制并行请求的数量而构建的。此逻辑在服务调用和实体更新期间自动使用。

Home Assistant通过维护每个集成的信号量(semaphore)来控制并行更新(对update()的调用)的数量。例如,如果信号量允许1个并行连接,更新和服务调用将等待一个正在进行的连接。如果值为0,则集成本身负责在必要时限制并行请求的数量。

平台的并行请求的默认值是根据添加到Home Assistant的第一个实体决定的。如果实体定义了async_update方法,则为0,否则为1。(这是遗留问题的决定)

平台可以通过在其平台中定义PARALLEL_UPDATES常量(比如rflink/light.py)来覆盖默认值。

触发事件(Firing event)

有些集成代表有事件的设备或服务,比如检测到移动或瞬间按钮被按下。通过在Home Assistant中将它们作为事件触发,集成可以使它们对用户可用。

您的集成应该触发类型为<domain>_event的事件。例如,ZHA集成会触发zha_event事件。

如果该事件与特定的设备/服务相关,则应正确定位。通过向包含设备注册表中的设备标识符的事件数据添加device_id属性来实现这一点。

1
2
3
4
5
event_data = {
"device_id": "my-device-id",
"type": "motion_detected",
}
hass.bus.async_fire("mydomain_event", event_data)

If a device or service only fires events, you need to manually register it in the device registry.

Making events accessible to users

可以根据payload,将Device trigger 附加到特定的事件,并使用户可以访问该事件。有了设备触发器,用户将能够看到设备的所有可用事件,并在其自动化中使用它。

What not to do

与事件相关的代码不应该是集成的实体逻辑的一部分。你想要启用从__init__.py中的async_setup_entry中转换集成事件到Home Assistant事件的逻辑。

实体状态不应该表示事件。例如,您不希望二进制传感器在事件发生时打开30秒。

网络和发现(Networking and Discovery)

一些集成可能需要通过启用mDNS/Zeroconf, SSDP或其他方法发现网络上的设备。主要的用例是寻找没有已知固定IP地址的设备,或者用于能够动态添加和删除任意数量的可发现设备的集成。

HA有内置的助手来支持mDNS/Zeroconf和SSDP。如果您的集成使用了另一种发现方法,需要确定使用哪个网络接口来广播流量,那么Network集成将提供一个帮助API来访问用户的界面首选项。

mDNS/Zeroconf

Home Assistant使用python-zeroconf包来支持mDNS。由于不推荐在一台主机上运行多个mDNS实现,Home Assistant提供了内部帮助器api来访问正在运行的ZeroconfAsyncZeroconf实例。

在使用这些助手之前,请确保在集成的manifest.json中将zeroconf添加到dependencies

Obtaining the AsyncZeroconf object

1
2
3
4
5
from homeassistant.components import zeroconf

...
aiozc = await zeroconf.async_get_async_instance(hass)

Obtaining the Zeroconf object

1
2
3
4
5
from homeassistant.components import zeroconf

...
zc = await zeroconf.async_get_instance(hass)

Using the AsyncZeroconf and Zeroconf objects

python-zeroconf provides examples on how to use both objects examples.

SSDP

通过SSDP提供内置的发现功能

Before using these helpers, be sure to add ssdp to dependencies in your integration’s manifest.json

获取已发现设备列表

发现的SSDP设备列表可以通过以下内置的帮助api获取。SSDP集成提供了以下帮助api来从缓存查找现有的SSDP发现:ssdp.async_get_discovery_info_by_udn_st, ssdp.async_get_discovery_info_by_st, ssdp.async_get_discovery_info_by_udn

查找指定设备

The ssdp.async_get_discovery_info_by_udn_st API returns a single discovery_info or None when provided an SSDP, UDN and ST.

1
2
3
4
5
from homeassistant.components import ssdp

...

discovery_info = ssdp.async_get_discovery_info_by_udn_st(hass, udn, st)

通过 ST查找设备

如果您想查找所发现的设备的特定类型,请调用ssdp.async_get_discovery_info_by_st将返回匹配SSDP st的所有已发现设备的列表。下面的例子返回网络上发现的每个Sonos播放器的发现信息列表。

1
2
3
4
5
6
7
8
from homeassistant.components import ssdp

...

discovery_infos = ssdp.async_get_discovery_info_by_st(hass, "urn:schemas-upnp-org:device:ZonePlayer:1")
for discovery_info in discovery_infos:
...

通过UDN查找设备

如果希望查看特定UDN提供的服务的列表,请调用ssdp.async_get_discovery_info_by_udn将返回匹配UPNP UDN的所有已发现设备的列表。

1
2
3
4
5
6
7
8
from homeassistant.components import ssdp

...

discovery_infos = ssdp.async_get_discovery_info_by_udn(hass, udn)
for discovery_info in discovery_infos:
...

订阅SSDP发现

一些集成可能需要知道设备何时被立即发现。SSDP集成提供了一个注册API,用于在发现匹配特定键值的新设备时接收回调。在 manifest.json中使用相同的ssdp格式用于匹配。

提供函数ssdp.async_register_callback来启用此功能。该函数返回一个回调函数,当调用时该函数将取消注册。

下面的例子展示了在网络上看到Sonos播放器时如何注册以获得回调。

1
2
3
4
5
6
7
8
9
from homeassistant.components import ssdp

...

entry.async_on_unload(
ssdp.async_register_callback(
hass, _async_discovered_player, {"st": "urn:schemas-upnp-org:device:ZonePlayer:1"}
)
)

下面的例子展示了当x-rincon-bootseq报头出现时如何注册以获得回调。

1
2
3
4
5
6
7
8
9
10
from homeassistant.components import ssdp
from homeassistant.const import MATCH_ALL

...

entry.async_on_unload(
ssdp.async_register_callback(
hass, _async_discovered_player, {"x-rincon-bootseq": MATCH_ALL}
)
)

Network

对于使用非内置的发现方法并需要访问用户的网络适配器配置的集成,应该使用以下帮助器API。

1
2
3
4
from homeassistant.components import network

...
adapters = await network.async_get_adapters(hass)

Example async_get_adapters data structure

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
[
{
"auto": True,
"default": False,
"enabled": True,
"ipv4": [],
"ipv6": [
{
"address": "2001:db8::",
"network_prefix": 8,
"flowinfo": 1,
"scope_id": 1,
}
],
"name": "eth0",
},
{
"auto": True,
"default": False,
"enabled": True,
"ipv4": [{"address": "192.168.1.5", "network_prefix": 23}],
"ipv6": [],
"name": "eth1",
},
{
"auto": False,
"default": False,
"enabled": False,
"ipv4": [{"address": "169.254.3.2", "network_prefix": 16}],
"ipv6": [],
"name": "vtun0",
},
]

Obtaining the IP Network from an adapter

1
2
3
4
5
6
7
8
9
10
11
12
from ipaddress import ip_network
from homeassistant.components import network

...

adapters = await network.async_get_adapters(hass)

for adapter in adapters:
for ip_info in adapater["ipv4"]:
local_ip = ip_info["address"]
network_prefix = ip_info["network_prefix"]
ip_net = ip_network(f"{local_ip}/{network_prefix}", False)