jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 17:27:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add config and options flow to KNX integration (#59377)

Browse Source
This commit is contained in:
Marvin Wichmann 2021-11-20 11:30:41 +01:00 committed by GitHub
parent 40104de0bf
commit e5c33474e3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
25 changed files with 1469 additions and 244 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
.coveragerc
View File

@ -540,7 +540,18 @@ omit =
homeassistant/components/keyboard_remote/*
homeassistant/components/kira/*
homeassistant/components/kiwi/lock.py
homeassistant/components/knx/*
homeassistant/components/knx/__init__.py
homeassistant/components/knx/climate.py
homeassistant/components/knx/const.py
homeassistant/components/knx/cover.py
homeassistant/components/knx/expose.py
homeassistant/components/knx/knx_entity.py
homeassistant/components/knx/light.py
homeassistant/components/knx/notify.py
homeassistant/components/knx/scene.py
homeassistant/components/knx/schema.py
homeassistant/components/knx/switch.py
homeassistant/components/knx/weather.py
homeassistant/components/kodi/__init__.py
homeassistant/components/kodi/browse_media.py
homeassistant/components/kodi/const.py

224
homeassistant/components/knx/__init__.py
View File

@ -21,28 +21,30 @@ from xknx.telegram.address import (
)
from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite
from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_EVENT,
CONF_HOST,
CONF_PORT,
CONF_TYPE,
EVENT_HOMEASSISTANT_STOP,
SERVICE_RELOAD,
)
from homeassistant.core import Event, HomeAssistant, ServiceCall
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import async_get_platforms
from homeassistant.helpers.reload import async_integration_yaml_config
from homeassistant.helpers.service import async_register_admin_service
from homeassistant.helpers.typing import ConfigType
from .const import (
CONF_KNX_CONNECTION_TYPE,
CONF_KNX_EXPOSE,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_ROUTING,
CONF_KNX_TUNNELING,
DATA_KNX_CONFIG,
DOMAIN,
KNX_ADDRESS,
SupportedPlatforms,
@ -87,6 +89,13 @@ CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.All(
# deprecated since 2021.12
cv.deprecated(ConnectionSchema.CONF_KNX_STATE_UPDATER),
cv.deprecated(ConnectionSchema.CONF_KNX_RATE_LIMIT),
cv.deprecated(CONF_KNX_ROUTING),
cv.deprecated(CONF_KNX_TUNNELING),
cv.deprecated(CONF_KNX_INDIVIDUAL_ADDRESS),
cv.deprecated(ConnectionSchema.CONF_KNX_MCAST_GRP),
cv.deprecated(ConnectionSchema.CONF_KNX_MCAST_PORT),
cv.deprecated(CONF_KNX_EVENT_FILTER),
# deprecated since 2021.4
cv.deprecated("config_file"),
@ -185,35 +194,73 @@ SERVICE_KNX_EXPOSURE_REGISTER_SCHEMA = vol.Any(
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the KNX integration."""
try:
knx_module = KNXModule(hass, config)
hass.data[DOMAIN] = knx_module
await knx_module.start()
except XKNXException as ex:
_LOGGER.warning("Could not connect to KNX interface: %s", ex)
hass.components.persistent_notification.async_create(
f"Could not connect to KNX interface: <br><b>{ex}</b>", title="KNX"
"""Start the KNX integration."""
conf: ConfigType | None = config.get(DOMAIN)
if conf is None:
# If we have a config entry, setup is done by that config entry.
# If there is no config entry, this should fail.
return bool(hass.config_entries.async_entries(DOMAIN))
conf = dict(conf)
hass.data[DATA_KNX_CONFIG] = conf
# Only import if we haven't before.
if not hass.config_entries.async_entries(DOMAIN):
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=conf
)
)
if CONF_KNX_EXPOSE in config[DOMAIN]:
for expose_config in config[DOMAIN][CONF_KNX_EXPOSE]:
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Load a config entry."""
conf = hass.data.get(DATA_KNX_CONFIG)
# When reloading
if conf is None:
conf = await async_integration_yaml_config(hass, DOMAIN)
if not conf or DOMAIN not in conf:
return False
conf = conf[DOMAIN]
# If user didn't have configuration.yaml config, generate defaults
if conf is None:
conf = CONFIG_SCHEMA({DOMAIN: dict(entry.data)})[DOMAIN]
config = {**conf, **entry.data}
try:
knx_module = KNXModule(hass, config, entry)
await knx_module.start()
except XKNXException as ex:
raise ConfigEntryNotReady from ex
hass.data[DATA_KNX_CONFIG] = conf
hass.data[DOMAIN] = knx_module
if CONF_KNX_EXPOSE in config:
for expose_config in config[CONF_KNX_EXPOSE]:
knx_module.exposures.append(
create_knx_exposure(hass, knx_module.xknx, expose_config)
)
for platform in SupportedPlatforms:
if platform.value not in config[DOMAIN]:
continue
hass.config_entries.async_setup_platforms(
entry,
[platform.value for platform in SupportedPlatforms if platform.value in config],
)
# set up notify platform, no entry support for notify component yet,
# have to use discovery to load platform.
if NotifySchema.PLATFORM_NAME in conf:
hass.async_create_task(
discovery.async_load_platform(
hass,
platform.value,
DOMAIN,
{
"platform_config": config[DOMAIN][platform.value],
},
config,
hass, "notify", DOMAIN, conf[NotifySchema.PLATFORM_NAME], config
)
)
@ -247,39 +294,53 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
schema=SERVICE_KNX_EXPOSURE_REGISTER_SCHEMA,
)
async def reload_service_handler(service_call: ServiceCall) -> None:
"""Remove all KNX components and load new ones from config."""
# First check for config file. If for some reason it is no longer there
# or knx is no longer mentioned, stop the reload.
config = await async_integration_yaml_config(hass, DOMAIN)
if not config or DOMAIN not in config:
return
await asyncio.gather(
*(platform.async_reset() for platform in async_get_platforms(hass, DOMAIN))
)
await knx_module.xknx.stop()
await async_setup(hass, config)
async_register_admin_service(
hass, DOMAIN, SERVICE_RELOAD, reload_service_handler, schema=vol.Schema({})
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unloading the KNX platforms."""
# if not loaded directly return
if not hass.data.get(DOMAIN):
return True
knx_module: KNXModule = hass.data[DOMAIN]
for exposure in knx_module.exposures:
exposure.shutdown()
unload_ok = await hass.config_entries.async_unload_platforms(
entry,
[
platform.value
for platform in SupportedPlatforms
if platform.value in hass.data[DATA_KNX_CONFIG]
],
)
if unload_ok:
await knx_module.stop()
hass.data.pop(DOMAIN)
hass.data.pop(DATA_KNX_CONFIG)
return unload_ok
async def async_update_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Update a given config entry."""
return await hass.config_entries.async_reload(entry.entry_id)
class KNXModule:
"""Representation of KNX Object."""
def __init__(self, hass: HomeAssistant, config: ConfigType) -> None:
def __init__(
self, hass: HomeAssistant, config: ConfigType, entry: ConfigEntry
) -> None:
"""Initialize KNX module."""
self.hass = hass
self.config = config
self.connected = False
self.exposures: list[KNXExposeSensor | KNXExposeTime] = []
self.service_exposures: dict[str, KNXExposeSensor | KNXExposeTime] = {}
self.entry = entry
self.init_xknx()
self.xknx.connection_manager.register_connection_state_changed_cb(
@ -292,64 +353,49 @@ class KNXModule:
self.register_event_callback()
)
self.entry.async_on_unload(
self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self.stop)
)
self.entry.async_on_unload(self.entry.add_update_listener(async_update_entry))
def init_xknx(self) -> None:
"""Initialize XKNX object."""
self.xknx = XKNX(
own_address=self.config[DOMAIN][CONF_KNX_INDIVIDUAL_ADDRESS],
rate_limit=self.config[DOMAIN][ConnectionSchema.CONF_KNX_RATE_LIMIT],
multicast_group=self.config[DOMAIN][ConnectionSchema.CONF_KNX_MCAST_GRP],
multicast_port=self.config[DOMAIN][ConnectionSchema.CONF_KNX_MCAST_PORT],
own_address=self.config[CONF_KNX_INDIVIDUAL_ADDRESS],
rate_limit=self.config[ConnectionSchema.CONF_KNX_RATE_LIMIT],
multicast_group=self.config[ConnectionSchema.CONF_KNX_MCAST_GRP],
multicast_port=self.config[ConnectionSchema.CONF_KNX_MCAST_PORT],
connection_config=self.connection_config(),
state_updater=self.config[DOMAIN][ConnectionSchema.CONF_KNX_STATE_UPDATER],
state_updater=self.config[ConnectionSchema.CONF_KNX_STATE_UPDATER],
)
async def start(self) -> None:
"""Start XKNX object. Connect to tunneling or Routing device."""
await self.xknx.start()
self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self.stop)
async def stop(self, event: Event) -> None:
async def stop(self, event: Event | None = None) -> None:
"""Stop XKNX object. Disconnect from tunneling or Routing device."""
await self.xknx.stop()
def connection_config(self) -> ConnectionConfig:
"""Return the connection_config."""
if CONF_KNX_TUNNELING in self.config[DOMAIN]:
return self.connection_config_tunneling()
if CONF_KNX_ROUTING in self.config[DOMAIN]:
return self.connection_config_routing()
return ConnectionConfig(auto_reconnect=True)
def connection_config_routing(self) -> ConnectionConfig:
"""Return the connection_config if routing is configured."""
local_ip = None
# all configuration values are optional
if self.config[DOMAIN][CONF_KNX_ROUTING] is not None:
local_ip = self.config[DOMAIN][CONF_KNX_ROUTING].get(
ConnectionSchema.CONF_KNX_LOCAL_IP
_conn_type: str = self.config[CONF_KNX_CONNECTION_TYPE]
if _conn_type == CONF_KNX_ROUTING:
return ConnectionConfig(
connection_type=ConnectionType.ROUTING,
auto_reconnect=True,
)
if _conn_type == CONF_KNX_TUNNELING:
return ConnectionConfig(
connection_type=ConnectionType.TUNNELING,
gateway_ip=self.config[CONF_HOST],
gateway_port=self.config[CONF_PORT],
route_back=self.config.get(ConnectionSchema.CONF_KNX_ROUTE_BACK, False),
auto_reconnect=True,
)
return ConnectionConfig(
connection_type=ConnectionType.ROUTING, local_ip=local_ip
)
def connection_config_tunneling(self) -> ConnectionConfig:
"""Return the connection_config if tunneling is configured."""
gateway_ip = self.config[DOMAIN][CONF_KNX_TUNNELING][CONF_HOST]
gateway_port = self.config[DOMAIN][CONF_KNX_TUNNELING][CONF_PORT]
local_ip = self.config[DOMAIN][CONF_KNX_TUNNELING].get(
ConnectionSchema.CONF_KNX_LOCAL_IP
)
route_back = self.config[DOMAIN][CONF_KNX_TUNNELING][
ConnectionSchema.CONF_KNX_ROUTE_BACK
]
return ConnectionConfig(
connection_type=ConnectionType.TUNNELING,
gateway_ip=gateway_ip,
gateway_port=gateway_port,
local_ip=local_ip,
route_back=route_back,
auto_reconnect=True,
)
return ConnectionConfig(auto_reconnect=True)
async def connection_state_changed_cb(self, state: XknxConnectionState) -> None:
"""Call invoked after a KNX connection state change was received."""
@ -409,10 +455,8 @@ class KNXModule:
"""Register callback for knx_event within XKNX TelegramQueue."""
# backwards compatibility for deprecated CONF_KNX_EVENT_FILTER
# use `address_filters = []` when this is not needed anymore
address_filters = list(
map(AddressFilter, self.config[DOMAIN][CONF_KNX_EVENT_FILTER])
)
for filter_set in self.config[DOMAIN][CONF_EVENT]:
address_filters = list(map(AddressFilter, self.config[CONF_KNX_EVENT_FILTER]))
for filter_set in self.config[CONF_EVENT]:
_filters = list(map(AddressFilter, filter_set[KNX_ADDRESS]))
address_filters.extend(_filters)
if (dpt := filter_set.get(CONF_TYPE)) and (

26
homeassistant/components/knx/binary_sensor.py
View File

@ -6,6 +6,7 @@ from typing import Any
from xknx import XKNX
from xknx.devices import BinarySensor as XknxBinarySensor
from homeassistant import config_entries
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.const import (
CONF_DEVICE_CLASS,
@ -18,28 +19,31 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType
from .const import ATTR_COUNTER, ATTR_SOURCE, DOMAIN
from .const import (
ATTR_COUNTER,
ATTR_SOURCE,
DATA_KNX_CONFIG,
DOMAIN,
SupportedPlatforms,
)
from .knx_entity import KnxEntity
from .schema import BinarySensorSchema
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up binary sensor(s) for KNX platform."""
if not discovery_info or not discovery_info["platform_config"]:
return
platform_config = discovery_info["platform_config"]
"""Set up the KNX binary sensor platform."""
xknx: XKNX = hass.data[DOMAIN].xknx
config: ConfigType = hass.data[DATA_KNX_CONFIG]
async_add_entities(
KNXBinarySensor(xknx, entity_config) for entity_config in platform_config
KNXBinarySensor(xknx, entity_config)
for entity_config in config[SupportedPlatforms.BINARY_SENSOR.value]
)

26
homeassistant/components/knx/button.py
View File

@ -4,30 +4,36 @@ from __future__ import annotations
from xknx import XKNX
from xknx.devices import RawValue as XknxRawValue
from homeassistant import config_entries
from homeassistant.components.button import ButtonEntity
from homeassistant.const import CONF_ENTITY_CATEGORY, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType
from .const import CONF_PAYLOAD, CONF_PAYLOAD_LENGTH, DOMAIN, KNX_ADDRESS
from .const import (
CONF_PAYLOAD,
CONF_PAYLOAD_LENGTH,
DATA_KNX_CONFIG,
DOMAIN,
KNX_ADDRESS,
SupportedPlatforms,
)
from .knx_entity import KnxEntity
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up buttons for KNX platform."""
if not discovery_info or not discovery_info["platform_config"]:
return
platform_config = discovery_info["platform_config"]
"""Set up the KNX binary sensor platform."""
xknx: XKNX = hass.data[DOMAIN].xknx
config: ConfigType = hass.data[DATA_KNX_CONFIG]
async_add_entities(
KNXButton(xknx, entity_config) for entity_config in platform_config
KNXButton(xknx, entity_config)
for entity_config in config[SupportedPlatforms.BUTTON.value]
)

30
homeassistant/components/knx/climate.py
View File

@ -8,6 +8,7 @@ from xknx.devices import Climate as XknxClimate, ClimateMode as XknxClimateMode
from xknx.dpt.dpt_hvac_mode import HVACControllerMode, HVACOperationMode
from xknx.telegram.address import parse_device_group_address
from homeassistant import config_entries
from homeassistant.components.climate import ClimateEntity
from homeassistant.components.climate.const import (
CURRENT_HVAC_IDLE,
@ -26,9 +27,16 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType
from .const import CONTROLLER_MODES, CURRENT_HVAC_ACTIONS, DOMAIN, PRESET_MODES
from .const import (
CONTROLLER_MODES,
CURRENT_HVAC_ACTIONS,
DATA_KNX_CONFIG,
DOMAIN,
PRESET_MODES,
SupportedPlatforms,
)
from .knx_entity import KnxEntity
from .schema import ClimateSchema
@ -37,23 +45,19 @@ CONTROLLER_MODES_INV = {value: key for key, value in CONTROLLER_MODES.items()}
PRESET_MODES_INV = {value: key for key, value in PRESET_MODES.items()}
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up climate(s) for KNX platform."""
if not discovery_info or not discovery_info["platform_config"]:
return
platform_config = discovery_info["platform_config"]
xknx: XKNX = hass.data[DOMAIN].xknx
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][
SupportedPlatforms.CLIMATE.value
]
_async_migrate_unique_id(hass, platform_config)
async_add_entities(
KNXClimate(xknx, entity_config) for entity_config in platform_config
)
_async_migrate_unique_id(hass, config)
async_add_entities(KNXClimate(xknx, entity_config) for entity_config in config)
@callback

409
homeassistant/components/knx/config_flow.py Normal file
View File

@ -0,0 +1,409 @@
"""Config flow for KNX."""
from __future__ import annotations
from typing import Any, Final
import voluptuous as vol
from xknx import XKNX
from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
from xknx.io.gateway_scanner import GatewayDescriptor, GatewayScanner
from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry, OptionsFlow
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
import homeassistant.helpers.config_validation as cv
from .const import (
CONF_KNX_AUTOMATIC,
CONF_KNX_CONNECTION_TYPE,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_INITIAL_CONNECTION_TYPES,
CONF_KNX_ROUTING,
CONF_KNX_TUNNELING,
DOMAIN,
)
from .schema import ConnectionSchema
CONF_KNX_GATEWAY: Final = "gateway"
CONF_MAX_RATE_LIMIT: Final = 60
DEFAULT_ENTRY_DATA: Final = {
ConnectionSchema.CONF_KNX_STATE_UPDATER: ConnectionSchema.CONF_KNX_DEFAULT_STATE_UPDATER,
ConnectionSchema.CONF_KNX_RATE_LIMIT: ConnectionSchema.CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
}
class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a KNX config flow."""
VERSION = 1
_tunnels: list
_gateway_ip: str = ""
_gateway_port: int = DEFAULT_MCAST_PORT
@staticmethod
@callback
def async_get_options_flow(config_entry: ConfigEntry) -> KNXOptionsFlowHandler:
"""Get the options flow for this handler."""
return KNXOptionsFlowHandler(config_entry)
async def async_step_user(self, user_input: dict | None = None) -> FlowResult:
"""Handle a flow initialized by the user."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
self._tunnels = []
return await self.async_step_type()
async def async_step_type(self, user_input: dict | None = None) -> FlowResult:
"""Handle connection type configuration."""
errors: dict = {}
supported_connection_types = CONF_KNX_INITIAL_CONNECTION_TYPES.copy()
fields = {}
if user_input is None:
gateways = await scan_for_gateways()
if gateways:
supported_connection_types.insert(0, CONF_KNX_AUTOMATIC)
self._tunnels = [
gateway for gateway in gateways if gateway.supports_tunnelling
]
fields = {
vol.Required(CONF_KNX_CONNECTION_TYPE): vol.In(
supported_connection_types
)
}
if user_input is not None:
connection_type = user_input[CONF_KNX_CONNECTION_TYPE]
if connection_type == CONF_KNX_AUTOMATIC:
return self.async_create_entry(
title=CONF_KNX_AUTOMATIC.capitalize(),
data={**DEFAULT_ENTRY_DATA, **user_input},
)
if connection_type == CONF_KNX_ROUTING:
return await self.async_step_routing()
if connection_type == CONF_KNX_TUNNELING and self._tunnels:
return await self.async_step_tunnel()
return await self.async_step_manual_tunnel()
return self.async_show_form(
step_id="type", data_schema=vol.Schema(fields), errors=errors
)
async def async_step_manual_tunnel(
self, user_input: dict | None = None
) -> FlowResult:
"""General setup."""
errors: dict = {}
if user_input is not None:
return self.async_create_entry(
title=f"{CONF_KNX_TUNNELING.capitalize()} @ {user_input[CONF_HOST]}",
data={
**DEFAULT_ENTRY_DATA,
CONF_HOST: user_input[CONF_HOST],
CONF_PORT: user_input[CONF_PORT],
CONF_KNX_INDIVIDUAL_ADDRESS: user_input[
CONF_KNX_INDIVIDUAL_ADDRESS
],
ConnectionSchema.CONF_KNX_ROUTE_BACK: user_input[
ConnectionSchema.CONF_KNX_ROUTE_BACK
],
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
},
)
fields = {
vol.Required(CONF_HOST, default=self._gateway_ip): str,
vol.Required(CONF_PORT, default=self._gateway_port): vol.Coerce(int),
vol.Required(
CONF_KNX_INDIVIDUAL_ADDRESS, default=XKNX.DEFAULT_ADDRESS
): str,
vol.Required(
ConnectionSchema.CONF_KNX_ROUTE_BACK, default=False
): vol.Coerce(bool),
}
return self.async_show_form(
step_id="manual_tunnel", data_schema=vol.Schema(fields), errors=errors
)
async def async_step_tunnel(self, user_input: dict | None = None) -> FlowResult:
"""Select a tunnel from a list. Will be skipped if the gateway scan was unsuccessful or if only one gateway was found."""
errors: dict = {}
if user_input is not None:
gateway: GatewayDescriptor = next(
gateway
for gateway in self._tunnels
if user_input[CONF_KNX_GATEWAY] == str(gateway)
)
self._gateway_ip = gateway.ip_addr
self._gateway_port = gateway.port
return await self.async_step_manual_tunnel()
tunnel_repr = {
str(tunnel) for tunnel in self._tunnels if tunnel.supports_tunnelling
}
# skip this step if the user has only one unique gateway.
if len(tunnel_repr) == 1:
_gateway: GatewayDescriptor = self._tunnels[0]
self._gateway_ip = _gateway.ip_addr
self._gateway_port = _gateway.port
return await self.async_step_manual_tunnel()
fields = {vol.Required(CONF_KNX_GATEWAY): vol.In(tunnel_repr)}
return self.async_show_form(
step_id="tunnel", data_schema=vol.Schema(fields), errors=errors
)
async def async_step_routing(self, user_input: dict | None = None) -> FlowResult:
"""Routing setup."""
errors: dict = {}
if user_input is not None:
return self.async_create_entry(
title=CONF_KNX_ROUTING.capitalize(),
data={
**DEFAULT_ENTRY_DATA,
ConnectionSchema.CONF_KNX_MCAST_GRP: user_input[
ConnectionSchema.CONF_KNX_MCAST_GRP
],
ConnectionSchema.CONF_KNX_MCAST_PORT: user_input[
ConnectionSchema.CONF_KNX_MCAST_PORT
],
CONF_KNX_INDIVIDUAL_ADDRESS: user_input[
CONF_KNX_INDIVIDUAL_ADDRESS
],
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
},
)
fields = {
vol.Required(
CONF_KNX_INDIVIDUAL_ADDRESS, default=XKNX.DEFAULT_ADDRESS
): str,
vol.Required(
ConnectionSchema.CONF_KNX_MCAST_GRP, default=DEFAULT_MCAST_GRP
): str,
vol.Required(
ConnectionSchema.CONF_KNX_MCAST_PORT, default=DEFAULT_MCAST_PORT
): cv.port,
}
return self.async_show_form(
step_id="routing", data_schema=vol.Schema(fields), errors=errors
)
async def async_step_import(self, config: dict | None = None) -> FlowResult:
"""Import a config entry.
Performs a one time import of the YAML configuration and creates a config entry based on it
if not already done before.
"""
if self._async_current_entries() or not config:
return self.async_abort(reason="single_instance_allowed")
data = {
ConnectionSchema.CONF_KNX_RATE_LIMIT: min(
config[ConnectionSchema.CONF_KNX_RATE_LIMIT], CONF_MAX_RATE_LIMIT
),
ConnectionSchema.CONF_KNX_STATE_UPDATER: config[
ConnectionSchema.CONF_KNX_STATE_UPDATER
],
ConnectionSchema.CONF_KNX_MCAST_GRP: config[
ConnectionSchema.CONF_KNX_MCAST_GRP
],
ConnectionSchema.CONF_KNX_MCAST_PORT: config[
ConnectionSchema.CONF_KNX_MCAST_PORT
],
CONF_KNX_INDIVIDUAL_ADDRESS: config[CONF_KNX_INDIVIDUAL_ADDRESS],
}
if CONF_KNX_TUNNELING in config:
return self.async_create_entry(
title=f"{CONF_KNX_TUNNELING.capitalize()} @ {config[CONF_KNX_TUNNELING][CONF_HOST]}",
data={
**DEFAULT_ENTRY_DATA,
CONF_HOST: config[CONF_KNX_TUNNELING][CONF_HOST],
CONF_PORT: config[CONF_KNX_TUNNELING][CONF_PORT],
ConnectionSchema.CONF_KNX_ROUTE_BACK: config[CONF_KNX_TUNNELING][
ConnectionSchema.CONF_KNX_ROUTE_BACK
],
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
**data,
},
)
if CONF_KNX_ROUTING in config:
return self.async_create_entry(
title=CONF_KNX_ROUTING.capitalize(),
data={
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
**data,
},
)
return self.async_create_entry(
title=CONF_KNX_AUTOMATIC.capitalize(),
data={
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
**data,
},
)
class KNXOptionsFlowHandler(OptionsFlow):
"""Handle KNX options."""
general_settings: dict
current_config: dict
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize KNX options flow."""
self.config_entry = config_entry
async def async_step_tunnel(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage KNX tunneling options."""
if (
self.general_settings.get(CONF_KNX_CONNECTION_TYPE) == CONF_KNX_TUNNELING
and user_input is None
):
return self.async_show_form(
step_id="tunnel",
data_schema=vol.Schema(
{
vol.Required(
CONF_HOST, default=self.current_config.get(CONF_HOST)
): str,
vol.Required(
CONF_PORT, default=self.current_config.get(CONF_PORT, 3671)
): cv.port,
vol.Required(
ConnectionSchema.CONF_KNX_ROUTE_BACK,
default=self.current_config.get(
ConnectionSchema.CONF_KNX_ROUTE_BACK, False
),
): vol.Coerce(bool),
}
),
last_step=True,
)
entry_data = {
**DEFAULT_ENTRY_DATA,
**self.general_settings,
CONF_HOST: self.current_config.get(CONF_HOST, ""),
}
if user_input is not None:
entry_data = {
**entry_data,
**user_input,
}
entry_title = entry_data[CONF_KNX_CONNECTION_TYPE].capitalize()
if entry_data[CONF_KNX_CONNECTION_TYPE] == CONF_KNX_TUNNELING:
entry_title = f"{CONF_KNX_TUNNELING.capitalize()} @ {entry_data[CONF_HOST]}"
self.hass.config_entries.async_update_entry(
self.config_entry,
data=entry_data,
title=entry_title,
)
return self.async_create_entry(title="", data={})
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage KNX options."""
if user_input is not None:
self.general_settings = user_input
return await self.async_step_tunnel()
supported_connection_types = [
CONF_KNX_AUTOMATIC,
CONF_KNX_TUNNELING,
CONF_KNX_ROUTING,
]
self.current_config = self.config_entry.data # type: ignore
data_schema = {
vol.Required(
CONF_KNX_CONNECTION_TYPE,
default=self.current_config.get(CONF_KNX_CONNECTION_TYPE),
): vol.In(supported_connection_types),
vol.Required(
CONF_KNX_INDIVIDUAL_ADDRESS,
default=self.current_config[CONF_KNX_INDIVIDUAL_ADDRESS],
): str,
vol.Required(
ConnectionSchema.CONF_KNX_MCAST_GRP,
default=self.current_config.get(
ConnectionSchema.CONF_KNX_MCAST_GRP, DEFAULT_MCAST_GRP
),
): str,
vol.Required(
ConnectionSchema.CONF_KNX_MCAST_PORT,
default=self.current_config.get(
ConnectionSchema.CONF_KNX_MCAST_PORT, DEFAULT_MCAST_PORT
),
): cv.port,
}
if self.show_advanced_options:
data_schema[
vol.Required(
ConnectionSchema.CONF_KNX_STATE_UPDATER,
default=self.current_config.get(
ConnectionSchema.CONF_KNX_STATE_UPDATER,
ConnectionSchema.CONF_KNX_DEFAULT_STATE_UPDATER,
),
)
] = bool
data_schema[
vol.Required(
ConnectionSchema.CONF_KNX_RATE_LIMIT,
default=self.current_config.get(
ConnectionSchema.CONF_KNX_RATE_LIMIT,
ConnectionSchema.CONF_KNX_DEFAULT_RATE_LIMIT,
),
)
] = vol.All(vol.Coerce(int), vol.Range(min=1, max=CONF_MAX_RATE_LIMIT))
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(data_schema),
last_step=self.current_config.get(CONF_KNX_CONNECTION_TYPE)
!= CONF_KNX_TUNNELING,
)
async def scan_for_gateways(stop_on_found: int = 0) -> list:
"""Scan for gateways within the network."""
xknx = XKNX()
gatewayscanner = GatewayScanner(
xknx, stop_on_found=stop_on_found, timeout_in_seconds=2
)
return await gatewayscanner.scan()

5
homeassistant/components/knx/const.py
View File

@ -29,6 +29,8 @@ KNX_ADDRESS: Final = "address"
CONF_INVERT: Final = "invert"
CONF_KNX_EXPOSE: Final = "expose"
CONF_KNX_INDIVIDUAL_ADDRESS: Final = "individual_address"
CONF_KNX_CONNECTION_TYPE: Final = "connection_type"
CONF_KNX_AUTOMATIC: Final = "automatic"
CONF_KNX_ROUTING: Final = "routing"
CONF_KNX_TUNNELING: Final = "tunneling"
CONF_PAYLOAD: Final = "payload"
@ -37,6 +39,9 @@ CONF_RESET_AFTER: Final = "reset_after"
CONF_RESPOND_TO_READ: Final = "respond_to_read"
CONF_STATE_ADDRESS: Final = "state_address"
CONF_SYNC_STATE: Final = "sync_state"
CONF_KNX_INITIAL_CONNECTION_TYPES: Final = [CONF_KNX_TUNNELING, CONF_KNX_ROUTING]
DATA_KNX_CONFIG: Final = "knx_config"
ATTR_COUNTER: Final = "counter"
ATTR_SOURCE: Final = "source"

22
homeassistant/components/knx/cover.py
View File

@ -9,6 +9,7 @@ from xknx import XKNX
from xknx.devices import Cover as XknxCover, Device as XknxDevice
from xknx.telegram.address import parse_device_group_address
from homeassistant import config_entries
from homeassistant.components.cover import (
ATTR_POSITION,
ATTR_TILT_POSITION,
@ -28,29 +29,26 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_utc_time_change
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN
from .const import DATA_KNX_CONFIG, DOMAIN, SupportedPlatforms
from .knx_entity import KnxEntity
from .schema import CoverSchema
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up cover(s) for KNX platform."""
if not discovery_info or not discovery_info["platform_config"]:
return
platform_config = discovery_info["platform_config"]
xknx: XKNX = hass.data[DOMAIN].xknx
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][
SupportedPlatforms.COVER.value
]
_async_migrate_unique_id(hass, platform_config)
async_add_entities(
KNXCover(xknx, entity_config) for entity_config in platform_config
)
_async_migrate_unique_id(hass, config)
async_add_entities(KNXCover(xknx, entity_config) for entity_config in config)
@callback

18
homeassistant/components/knx/fan.py
View File

@ -7,37 +7,35 @@ from typing import Any, Final
from xknx import XKNX
from xknx.devices import Fan as XknxFan
from homeassistant import config_entries
from homeassistant.components.fan import SUPPORT_OSCILLATE, SUPPORT_SET_SPEED, FanEntity
from homeassistant.const import CONF_ENTITY_CATEGORY, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.percentage import (
int_states_in_range,
percentage_to_ranged_value,
ranged_value_to_percentage,
)
from .const import DOMAIN, KNX_ADDRESS
from .const import DATA_KNX_CONFIG, DOMAIN, KNX_ADDRESS, SupportedPlatforms
from .knx_entity import KnxEntity
from .schema import FanSchema
DEFAULT_PERCENTAGE: Final = 50
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up fans for KNX platform."""
if not discovery_info or not discovery_info["platform_config"]:
return
platform_config = discovery_info["platform_config"]
"""Set up fan(s) for KNX platform."""
xknx: XKNX = hass.data[DOMAIN].xknx
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][SupportedPlatforms.FAN.value]
async_add_entities(KNXFan(xknx, entity_config) for entity_config in platform_config)
async_add_entities(KNXFan(xknx, entity_config) for entity_config in config)
class KNXFan(KnxEntity, FanEntity):

30
homeassistant/components/knx/light.py
View File

@ -7,6 +7,7 @@ from xknx import XKNX
from xknx.devices.light import Light as XknxLight, XYYColor
from xknx.telegram.address import parse_device_group_address
from homeassistant import config_entries
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_COLOR_TEMP,
@ -27,30 +28,33 @@ from homeassistant.const import CONF_ENTITY_CATEGORY, CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType
import homeassistant.util.color as color_util
from .const import DOMAIN, KNX_ADDRESS, ColorTempModes
from .const import (
DATA_KNX_CONFIG,
DOMAIN,
KNX_ADDRESS,
ColorTempModes,
SupportedPlatforms,
)
from .knx_entity import KnxEntity
from .schema import LightSchema
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up lights for KNX platform."""
if not discovery_info or not discovery_info["platform_config"]:
return
platform_config = discovery_info["platform_config"]
"""Set up light(s) for KNX platform."""
xknx: XKNX = hass.data[DOMAIN].xknx
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][
SupportedPlatforms.LIGHT.value
]
_async_migrate_unique_id(hass, platform_config)
async_add_entities(
KNXLight(xknx, entity_config) for entity_config in platform_config
)
_async_migrate_unique_id(hass, config)
async_add_entities(KNXLight(xknx, entity_config) for entity_config in config)
@callback

13
homeassistant/components/knx/manifest.json
View File

@ -1,9 +1,16 @@
{
"domain": "knx",
"name": "KNX",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/knx",
"requirements": ["xknx==0.18.13"],
"codeowners": ["@Julius2342", "@farmio", "@marvin-w"],
"requirements": [
"xknx==0.18.13"
],
"codeowners": [
"@Julius2342",
"@farmio",
"@marvin-w"
],
"quality_scale": "silver",
"iot_class": "local_push"
}
}

4
homeassistant/components/knx/notify.py
View File

@ -20,10 +20,10 @@ async def async_get_service(
discovery_info: DiscoveryInfoType | None = None,
) -> KNXNotificationService | None:
"""Get the KNX notification service."""
if not discovery_info or not discovery_info["platform_config"]:
if not discovery_info:
return None
platform_config = discovery_info["platform_config"]
platform_config: dict = discovery_info
xknx: XKNX = hass.data[DOMAIN].xknx
notification_devices = []

29
homeassistant/components/knx/number.py
View File

@ -6,6 +6,7 @@ from typing import cast
from xknx import XKNX
from xknx.devices import NumericValue
from homeassistant import config_entries
from homeassistant.components.number import NumberEntity
from homeassistant.const import (
CONF_ENTITY_CATEGORY,
@ -18,28 +19,32 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType
from .const import CONF_RESPOND_TO_READ, CONF_STATE_ADDRESS, DOMAIN, KNX_ADDRESS
from .const import (
CONF_RESPOND_TO_READ,
CONF_STATE_ADDRESS,
DATA_KNX_CONFIG,
DOMAIN,
KNX_ADDRESS,
SupportedPlatforms,
)
from .knx_entity import KnxEntity
from .schema import NumberSchema
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up number entities for KNX platform."""
if not discovery_info or not discovery_info["platform_config"]:
return
platform_config = discovery_info["platform_config"]
"""Set up number(s) for KNX platform."""
xknx: XKNX = hass.data[DOMAIN].xknx
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][
SupportedPlatforms.NUMBER.value
]
async_add_entities(
KNXNumber(xknx, entity_config) for entity_config in platform_config
)
async_add_entities(KNXNumber(xknx, entity_config) for entity_config in config)
def _create_numeric_value(xknx: XKNX, config: ConfigType) -> NumericValue:

22
homeassistant/components/knx/scene.py
View File

@ -6,32 +6,30 @@ from typing import Any
from xknx import XKNX
from xknx.devices import Scene as XknxScene
from homeassistant import config_entries
from homeassistant.components.scene import Scene
from homeassistant.const import CONF_ENTITY_CATEGORY, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, KNX_ADDRESS
from .const import DATA_KNX_CONFIG, DOMAIN, KNX_ADDRESS, SupportedPlatforms
from .knx_entity import KnxEntity
from .schema import SceneSchema
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the scenes for KNX platform."""
if not discovery_info or not discovery_info["platform_config"]:
return
platform_config = discovery_info["platform_config"]
"""Set up scene(s) for KNX platform."""
xknx: XKNX = hass.data[DOMAIN].xknx
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][
SupportedPlatforms.SCENE.value
]
async_add_entities(
KNXScene(xknx, entity_config) for entity_config in platform_config
)
async_add_entities(KNXScene(xknx, entity_config) for entity_config in config)
class KNXScene(KnxEntity, Scene):

15
homeassistant/components/knx/schema.py
View File

@ -201,7 +201,11 @@ sync_state_validator = vol.Any(
class ConnectionSchema:
"""Voluptuous schema for KNX connection."""
"""
Voluptuous schema for KNX connection.
DEPRECATED: Migrated to config and options flow. Will be removed in a future version of Home Assistant.
"""
CONF_KNX_LOCAL_IP = "local_ip"
CONF_KNX_MCAST_GRP = "multicast_group"
@ -210,6 +214,9 @@ class ConnectionSchema:
CONF_KNX_ROUTE_BACK = "route_back"
CONF_KNX_STATE_UPDATER = "state_updater"
CONF_KNX_DEFAULT_STATE_UPDATER = True
CONF_KNX_DEFAULT_RATE_LIMIT = 20
TUNNELING_SCHEMA = vol.Schema(
{
vol.Optional(CONF_PORT, default=DEFAULT_MCAST_PORT): cv.port,
@ -229,8 +236,10 @@ class ConnectionSchema:
): ia_validator,
vol.Optional(CONF_KNX_MCAST_GRP, default=DEFAULT_MCAST_GRP): cv.string,
vol.Optional(CONF_KNX_MCAST_PORT, default=DEFAULT_MCAST_PORT): cv.port,
vol.Optional(CONF_KNX_STATE_UPDATER, default=True): cv.boolean,
vol.Optional(CONF_KNX_RATE_LIMIT, default=20): vol.All(
vol.Optional(
CONF_KNX_STATE_UPDATER, default=CONF_KNX_DEFAULT_STATE_UPDATER
): cv.boolean,
vol.Optional(CONF_KNX_RATE_LIMIT, default=CONF_KNX_DEFAULT_RATE_LIMIT): vol.All(
vol.Coerce(int), vol.Range(min=1, max=100)
),
}

22
homeassistant/components/knx/select.py
View File

@ -4,6 +4,7 @@ from __future__ import annotations
from xknx import XKNX
from xknx.devices import Device as XknxDevice, RawValue
from homeassistant import config_entries
from homeassistant.components.select import SelectEntity
from homeassistant.const import (
CONF_ENTITY_CATEGORY,
@ -14,7 +15,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType
from .const import (
CONF_PAYLOAD,
@ -22,28 +23,27 @@ from .const import (
CONF_RESPOND_TO_READ,
CONF_STATE_ADDRESS,
CONF_SYNC_STATE,
DATA_KNX_CONFIG,
DOMAIN,
KNX_ADDRESS,
SupportedPlatforms,
)
from .knx_entity import KnxEntity
from .schema import SelectSchema
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up select entities for KNX platform."""
if not discovery_info or not discovery_info["platform_config"]:
return
platform_config = discovery_info["platform_config"]
"""Set up select(s) for KNX platform."""
xknx: XKNX = hass.data[DOMAIN].xknx
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][
SupportedPlatforms.SELECT.value
]
async_add_entities(
KNXSelect(xknx, entity_config) for entity_config in platform_config
)
async_add_entities(KNXSelect(xknx, entity_config) for entity_config in config)
def _create_raw_value(xknx: XKNX, config: ConfigType) -> RawValue:

20
homeassistant/components/knx/sensor.py
View File

@ -6,6 +6,7 @@ from typing import Any
from xknx import XKNX
from xknx.devices import Sensor as XknxSensor
from homeassistant import config_entries
from homeassistant.components.sensor import (
CONF_STATE_CLASS,
DEVICE_CLASSES,
@ -14,28 +15,25 @@ from homeassistant.components.sensor import (
from homeassistant.const import CONF_ENTITY_CATEGORY, CONF_NAME, CONF_TYPE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, StateType
from homeassistant.helpers.typing import ConfigType, StateType
from .const import ATTR_SOURCE, DOMAIN
from .const import ATTR_SOURCE, DATA_KNX_CONFIG, DOMAIN, SupportedPlatforms
from .knx_entity import KnxEntity
from .schema import SensorSchema
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up sensor(s) for KNX platform."""
if not discovery_info or not discovery_info["platform_config"]:
return
platform_config = discovery_info["platform_config"]
xknx: XKNX = hass.data[DOMAIN].xknx
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][
SupportedPlatforms.SENSOR.value
]
async_add_entities(
KNXSensor(xknx, entity_config) for entity_config in platform_config
)
async_add_entities(KNXSensor(xknx, entity_config) for entity_config in config)
def _create_sensor(xknx: XKNX, config: ConfigType) -> XknxSensor:

3
homeassistant/components/knx/services.yaml
View File

@ -100,6 +100,3 @@ exposure_register:
default: false
selector:
boolean:
reload:
name: "Reload KNX configuration"
description: "Reload the KNX configuration from YAML."

63
homeassistant/components/knx/strings.json Normal file
View File

@ -0,0 +1,63 @@
{
"config": {
"step": {
"type": {
"description": "Please enter the connection type we should use for your KNX connection. \n AUTOMATIC - The integration takes care of the connectivity to your KNX Bus by performing a gateway scan. \n TUNNELING - The integration will connect to your KNX bus via tunneling. \n ROUTING - The integration will connect to your KNX bus via routing.",
"data": {
"connection_type": "KNX Connection Type"
}
},
"tunnel": {
"description": "Please select a gateway from the list.",
"data": {
"gateway": "KNX Tunnel Connection"
}
},
"manual_tunnel": {
"description": "Please enter the connection information of your tunneling device.",
"data": {
"port": "[%key:common::config_flow::data::port%]",
"host": "[%key:common::config_flow::data::host%]",
"individual_address": "Individual address for the connection",
"route_back": "Route Back / NAT Mode"
}
},
"routing": {
"description": "Please configure the routing options.",
"data": {
"individual_address": "Individual address for the routing connection",
"multicast_group": "The multicast group used for routing",
"multicast_port": "The multicast port used for routing"
}
}
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
}
},
"options": {
"step": {
"init": {
"data": {
"connection_type": "KNX Connection Type",
"individual_address": "Default individual address",
"multicast_group": "Multicast group used for routing and discovery",
"multicast_port": "Multicast port used for routing and discovery",
"state_updater": "Globally enable reading states from the KNX Bus",
"rate_limit": "Maximum outgoing telegrams per second"
}
},
"tunnel": {
"data": {
"port": "[%key:common::config_flow::data::port%]",
"host": "[%key:common::config_flow::data::host%]",
"route_back": "Route Back / NAT Mode"
}
}
}
}
}

26
homeassistant/components/knx/switch.py
View File

@ -6,6 +6,7 @@ from typing import Any
from xknx import XKNX
from xknx.devices import Switch as XknxSwitch
from homeassistant import config_entries
from homeassistant.components.switch import SwitchEntity
from homeassistant.const import (
CONF_ENTITY_CATEGORY,
@ -17,28 +18,31 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType
from .const import CONF_RESPOND_TO_READ, DOMAIN, KNX_ADDRESS
from .const import (
CONF_RESPOND_TO_READ,
DATA_KNX_CONFIG,
DOMAIN,
KNX_ADDRESS,
SupportedPlatforms,
)
from .knx_entity import KnxEntity
from .schema import SwitchSchema
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up switch(es) for KNX platform."""
if not discovery_info or not discovery_info["platform_config"]:
return
platform_config = discovery_info["platform_config"]
xknx: XKNX = hass.data[DOMAIN].xknx
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][
SupportedPlatforms.SWITCH.value
]
async_add_entities(
KNXSwitch(xknx, entity_config) for entity_config in platform_config
)
async_add_entities(KNXSwitch(xknx, entity_config) for entity_config in config)
class KNXSwitch(KnxEntity, SwitchEntity, RestoreEntity):

63
homeassistant/components/knx/translations/en.json Normal file
View File

@ -0,0 +1,63 @@
{
"config": {
"step": {
"type": {
"description": "Please enter the connection type we should use for your KNX connection. \n AUTOMATIC - The integration takes care of the connectivity to your KNX Bus by performing a gateway scan. \n TUNNELING - The integration will connect to your KNX bus via tunneling. \n ROUTING - The integration will connect to your KNX bus via routing.",
"data": {
"connection_type": "KNX Connection Type"
}
},
"tunnel": {
"description": "Please select a gateway from the list.",
"data": {
"gateway": "KNX Tunnel Connection"
}
},
"manual_tunnel": {
"description": "Please enter the connection information of your tunneling device.",
"data": {
"host": "IP Address of KNX gateway",
"port": "Port of KNX gateway",
"individual_address": "Individual address for the connection",
"route_back": "Route Back / NAT Mode"
}
},
"routing": {
"description": "Please configure the routing options.",
"data": {
"individual_address": "Individual address for the routing connection",
"multicast_group": "The multicast group used for routing",
"multicast_port": "The multicast port used for routing"
}
}
},
"abort": {
"already_configured": "Service is already configured",
"single_instance_allowed": "Already configured. Only a single configuration possible."
},
"error": {
"cannot_connect": "Failed to connect."
}
},
"options": {
"step": {
"init": {
"data": {
"connection_type": "KNX Connection Type",
"individual_address": "Default individual address",
"multicast_group": "Multicast group used for routing and discovery",
"multicast_port": "Multicast port used for routing and discovery",
"state_updater": "Globally enable reading states from the KNX Bus",
"rate_limit": "Maximum outgoing telegrams per second"
}
},
"tunnel": {
"data": {
"port": "Port of KNX gateway",
"host": "IP Address of KNX gateway",
"route_back": "Route Back / NAT Mode"
}
}
}
}
}

22
homeassistant/components/knx/weather.py
View File

@ -4,32 +4,30 @@ from __future__ import annotations
from xknx import XKNX
from xknx.devices import Weather as XknxWeather
from homeassistant import config_entries
from homeassistant.components.weather import WeatherEntity
from homeassistant.const import CONF_ENTITY_CATEGORY, CONF_NAME, TEMP_CELSIUS
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN
from .const import DATA_KNX_CONFIG, DOMAIN, SupportedPlatforms
from .knx_entity import KnxEntity
from .schema import WeatherSchema
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
config_entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up weather entities for KNX platform."""
if not discovery_info or not discovery_info["platform_config"]:
return
platform_config = discovery_info["platform_config"]
"""Set up switch(es) for KNX platform."""
xknx: XKNX = hass.data[DOMAIN].xknx
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][
SupportedPlatforms.WEATHER.value
]
async_add_entities(
KNXWeather(xknx, entity_config) for entity_config in platform_config
)
async_add_entities(KNXWeather(xknx, entity_config) for entity_config in config)
def _create_weather(xknx: XKNX, config: ConfigType) -> XknxWeather:

1
homeassistant/generated/config_flows.py
View File

@ -152,6 +152,7 @@ FLOWS = [
"juicenet",
"keenetic_ndms2",
"kmtronic",
"knx",
"kodi",
"konnected",
"kostal_plenticore",

34
tests/components/knx/conftest.py
View File

@ -8,23 +8,33 @@ import pytest
from xknx import XKNX
from xknx.core import XknxConnectionState
from xknx.dpt import DPTArray, DPTBinary
from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
from xknx.telegram import Telegram, TelegramDirection
from xknx.telegram.address import GroupAddress, IndividualAddress
from xknx.telegram.apci import APCI, GroupValueRead, GroupValueResponse, GroupValueWrite
from homeassistant.components.knx.const import DOMAIN as KNX_DOMAIN
from homeassistant.components.knx import ConnectionSchema
from homeassistant.components.knx.const import (
CONF_KNX_AUTOMATIC,
CONF_KNX_CONNECTION_TYPE,
CONF_KNX_INDIVIDUAL_ADDRESS,
DOMAIN as KNX_DOMAIN,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
class KNXTestKit:
"""Test helper for the KNX integration."""
INDIVIDUAL_ADDRESS = "1.2.3"
def __init__(self, hass: HomeAssistant):
def __init__(self, hass: HomeAssistant, mock_config_entry: MockConfigEntry):
"""Init KNX test helper class."""
self.hass: HomeAssistant = hass
self.mock_config_entry: MockConfigEntry = mock_config_entry
self.xknx: XKNX
# outgoing telegrams will be put in the Queue instead of sent to the interface
# telegrams to an InternalGroupAddress won't be queued here
@ -60,6 +70,7 @@ class KNXTestKit:
return_value=knx_ip_interface_mock(),
side_effect=fish_xknx,
):
self.mock_config_entry.add_to_hass(self.hass)
await async_setup_component(self.hass, KNX_DOMAIN, {KNX_DOMAIN: config})
await self.xknx.connection_manager.connection_state_changed(
XknxConnectionState.CONNECTED
@ -191,8 +202,23 @@ class KNXTestKit:
@pytest.fixture
async def knx(request, hass):
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="KNX",
domain=KNX_DOMAIN,
data={
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
},
)
@pytest.fixture
async def knx(request, hass, mock_config_entry: MockConfigEntry):
"""Create a KNX TestKit instance."""
knx_test_kit = KNXTestKit(hass)
knx_test_kit = KNXTestKit(hass, mock_config_entry)
yield knx_test_kit
await knx_test_kit.assert_no_telegram()

573
tests/components/knx/test_config_flow.py Normal file
View File

@ -0,0 +1,573 @@
"""Test the KNX config flow."""
from unittest.mock import patch
from xknx import XKNX
from xknx.io import DEFAULT_MCAST_GRP
from xknx.io.gateway_scanner import GatewayDescriptor
from homeassistant import config_entries
from homeassistant.components.knx import ConnectionSchema
from homeassistant.components.knx.config_flow import (
CONF_KNX_GATEWAY,
DEFAULT_ENTRY_DATA,
)
from homeassistant.components.knx.const import (
CONF_KNX_AUTOMATIC,
CONF_KNX_CONNECTION_TYPE,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_ROUTING,
CONF_KNX_TUNNELING,
DOMAIN,
)
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY, RESULT_TYPE_FORM
from tests.common import MockConfigEntry
def _gateway_descriptor(ip: str, port: int) -> GatewayDescriptor:
"""Get mock gw descriptor."""
return GatewayDescriptor("Test", ip, port, "eth0", "127.0.0.1", True)
async def test_user_single_instance(hass):
"""Test we only allow a single config flow."""
MockConfigEntry(domain=DOMAIN).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "abort"
assert result["reason"] == "single_instance_allowed"
async def test_routing_setup(hass: HomeAssistant) -> None:
"""Test routing setup."""
with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways:
gateways.return_value = []
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_FORM
assert result2["step_id"] == "routing"
assert not result2["errors"]
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
},
)
await hass.async_block_till_done()
assert result3["type"] == RESULT_TYPE_CREATE_ENTRY
assert result3["title"] == CONF_KNX_ROUTING.capitalize()
assert result3["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_tunneling_setup(hass: HomeAssistant) -> None:
"""Test tunneling if only one gateway is found."""
gateway = _gateway_descriptor("192.168.0.1", 3675)
with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways:
gateways.return_value = [gateway]
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_FORM
assert result2["step_id"] == "manual_tunnel"
assert not result2["errors"]
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
},
)
await hass.async_block_till_done()
assert result3["type"] == RESULT_TYPE_CREATE_ENTRY
assert result3["title"] == "Tunneling @ 192.168.0.1"
assert result3["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_tunneling_setup_for_multiple_found_gateways(hass: HomeAssistant) -> None:
"""Test tunneling if only one gateway is found."""
gateway = _gateway_descriptor("192.168.0.1", 3675)
gateway2 = _gateway_descriptor("192.168.1.100", 3675)
with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways:
gateways.return_value = [gateway, gateway2]
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert not result["errors"]
tunnel_flow = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
},
)
await hass.async_block_till_done()
assert tunnel_flow["type"] == RESULT_TYPE_FORM
assert tunnel_flow["step_id"] == "tunnel"
assert not tunnel_flow["errors"]
manual_tunnel = await hass.config_entries.flow.async_configure(
tunnel_flow["flow_id"],
{CONF_KNX_GATEWAY: str(gateway)},
)
await hass.async_block_till_done()
assert manual_tunnel["type"] == RESULT_TYPE_FORM
assert manual_tunnel["step_id"] == "manual_tunnel"
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
manual_tunnel_flow = await hass.config_entries.flow.async_configure(
manual_tunnel["flow_id"],
{
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
},
)
await hass.async_block_till_done()
assert manual_tunnel_flow["type"] == RESULT_TYPE_CREATE_ENTRY
assert manual_tunnel_flow["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: False,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_manual_tunnel_step_when_no_gateway(hass: HomeAssistant) -> None:
"""Test manual tunnel if no gateway is found and tunneling is selected."""
with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways:
gateways.return_value = []
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert not result["errors"]
tunnel_flow = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
},
)
await hass.async_block_till_done()
assert tunnel_flow["type"] == RESULT_TYPE_FORM
assert tunnel_flow["step_id"] == "manual_tunnel"
assert not tunnel_flow["errors"]
async def test_form_with_automatic_connection_handling(hass: HomeAssistant) -> None:
"""Test we get the form."""
with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways:
gateways.return_value = [_gateway_descriptor("192.168.0.1", 3675)]
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert not result["errors"]
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == CONF_KNX_AUTOMATIC.capitalize()
assert result2["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
}
assert len(mock_setup_entry.mock_calls) == 1
##
# Import Tests
##
async def test_import_config_tunneling(hass: HomeAssistant) -> None:
"""Test tunneling import from config.yaml."""
config = {
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, # has a default in the original config
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20, # has a default in the original config
ConnectionSchema.CONF_KNX_STATE_UPDATER: True, # has a default in the original config
CONF_KNX_TUNNELING: {
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
},
}
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Tunneling @ 192.168.1.1"
assert result["data"] == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_config_routing(hass: HomeAssistant) -> None:
"""Test routing import from config.yaml."""
config = {
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, # has a default in the original config
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20, # has a default in the original config
ConnectionSchema.CONF_KNX_STATE_UPDATER: True, # has a default in the original config
CONF_KNX_ROUTING: {}, # is required when using routing
}
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == CONF_KNX_ROUTING.capitalize()
assert result["data"] == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_config_automatic(hass: HomeAssistant) -> None:
"""Test automatic import from config.yaml."""
config = {
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, # has a default in the original config
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20, # has a default in the original config
ConnectionSchema.CONF_KNX_STATE_UPDATER: True, # has a default in the original config
}
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == CONF_KNX_AUTOMATIC.capitalize()
assert result["data"] == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_rate_limit_out_of_range(hass: HomeAssistant) -> None:
"""Test automatic import from config.yaml."""
config = {
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, # has a default in the original config
ConnectionSchema.CONF_KNX_RATE_LIMIT: 80,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True, # has a default in the original config
}
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == CONF_KNX_AUTOMATIC.capitalize()
assert result["data"] == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 60,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_options(hass: HomeAssistant) -> None:
"""Test import from config.yaml with options."""
config = {
CONF_KNX_INDIVIDUAL_ADDRESS: XKNX.DEFAULT_ADDRESS, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, # has a default in the original config
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675, # has a default in the original config
ConnectionSchema.CONF_KNX_STATE_UPDATER: False,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 30,
}
with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == CONF_KNX_AUTOMATIC.capitalize()
assert result["data"] == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_STATE_UPDATER: False,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 30,
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_abort_if_entry_exists_already(hass: HomeAssistant) -> None:
"""Test routing import from config.yaml."""
MockConfigEntry(domain=DOMAIN).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}
)
assert result["type"] == "abort"
assert result["reason"] == "single_instance_allowed"
async def test_options_flow(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test options config flow."""
mock_config_entry.add_to_hass(hass)
gateway = _gateway_descriptor("192.168.0.1", 3675)
with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways:
gateways.return_value = [gateway]
result = await hass.config_entries.options.async_init(
mock_config_entry.entry_id
)
assert result.get("type") == RESULT_TYPE_FORM
assert result.get("step_id") == "init"
assert "flow_id" in result
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
},
)
await hass.async_block_till_done()
assert result2.get("type") == RESULT_TYPE_CREATE_ENTRY
assert not result2.get("data")
assert mock_config_entry.data == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
CONF_HOST: "",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
}
async def test_tunneling_options_flow(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test options flow for tunneling."""
mock_config_entry.add_to_hass(hass)
gateway = _gateway_descriptor("192.168.0.1", 3675)
with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways:
gateways.return_value = [gateway]
result = await hass.config_entries.options.async_init(
mock_config_entry.entry_id
)
assert result.get("type") == RESULT_TYPE_FORM
assert result.get("step_id") == "init"
assert "flow_id" in result
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
},
)
assert result2.get("type") == RESULT_TYPE_FORM
assert not result2.get("data")
assert "flow_id" in result2
result3 = await hass.config_entries.options.async_configure(
result2["flow_id"],
user_input={
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
},
)
await hass.async_block_till_done()
assert result3.get("type") == RESULT_TYPE_CREATE_ENTRY
assert not result3.get("data")
assert mock_config_entry.data == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.255",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 20,
ConnectionSchema.CONF_KNX_STATE_UPDATER: True,
CONF_HOST: "192.168.1.1",
CONF_PORT: 3675,
ConnectionSchema.CONF_KNX_ROUTE_BACK: True,
}
async def test_advanced_options(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test options config flow."""
mock_config_entry.add_to_hass(hass)
gateway = _gateway_descriptor("192.168.0.1", 3675)
with patch("xknx.io.gateway_scanner.GatewayScanner.scan") as gateways:
gateways.return_value = [gateway]
result = await hass.config_entries.options.async_init(
mock_config_entry.entry_id, context={"show_advanced_options": True}
)
assert result.get("type") == RESULT_TYPE_FORM
assert result.get("step_id") == "init"
assert "flow_id" in result
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 25,
ConnectionSchema.CONF_KNX_STATE_UPDATER: False,
},
)
await hass.async_block_till_done()
assert result2.get("type") == RESULT_TYPE_CREATE_ENTRY
assert not result2.get("data")
assert mock_config_entry.data == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_INDIVIDUAL_ADDRESS: "15.15.250",
CONF_HOST: "",
ConnectionSchema.CONF_KNX_MCAST_PORT: 3675,
ConnectionSchema.CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
ConnectionSchema.CONF_KNX_RATE_LIMIT: 25,
ConnectionSchema.CONF_KNX_STATE_UPDATER: False,
}