bahmcloud/HA-KNX-Bridge
1
0
Fork 0
mirror of https://github.com/bahmcloud/HA-KNX-Bridge.git synced 2026-04-06 16:51:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add options flow fallback for ports

Browse Source
This commit is contained in:
René Bachmann
2026-02-13 19:33:55 +01:00
parent 8b20cf4744
commit 502a33dbac
7 changed files with 202 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
custom_components/ha_knx_bridge/config_flow.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import logging
import uuid
import voluptuous as vol
@@ -18,6 +19,8 @@ from .const import (
CONF_KNX_ENTRY_ID,
CONF_MOVE_LONG_ADDRESS,
CONF_MOVE_SHORT_ADDRESS,
CONF_PORTS,
CONF_PORT_ID,
CONF_POSITION_ADDRESS,
CONF_POSITION_STATE_ADDRESS,
CONF_STATE_ADDRESS,
@@ -89,11 +92,114 @@ class HAKnxBridgeOptionsFlow(config_entries.OptionsFlow):
self._config_entry = config_entry
async def async_step_init(self, user_input: dict | None = None):
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
return self.async_show_menu(
step_id="init",
menu_options=[
"add_binary_sensor",
"add_switch",
"add_cover",
"remove_port",
],
)
schema = vol.Schema({})
return self.async_show_form(step_id="init", data_schema=schema)
async def async_step_add_binary_sensor(self, user_input: dict | None = None):
if user_input is not None:
user_input, errors = _validate_knx_addresses(
user_input,
[
CONF_STATE_ADDRESS,
],
)
if errors:
return self.async_show_form(
step_id="add_binary_sensor",
data_schema=_binary_sensor_schema(),
errors=errors,
)
return await self._async_store_port("binary_sensor", user_input)
return self.async_show_form(
step_id="add_binary_sensor", data_schema=_binary_sensor_schema()
)
async def async_step_add_switch(self, user_input: dict | None = None):
if user_input is not None:
user_input, errors = _validate_knx_addresses(
user_input, [CONF_COMMAND_ADDRESS, CONF_STATE_ADDRESS]
)
if errors:
return self.async_show_form(
step_id="add_switch",
data_schema=_switch_schema(),
errors=errors,
)
return await self._async_store_port("switch", user_input)
return self.async_show_form(
step_id="add_switch", data_schema=_switch_schema()
)
async def async_step_add_cover(self, user_input: dict | None = None):
if user_input is not None:
user_input, errors = _validate_knx_addresses(
user_input,
[
CONF_MOVE_LONG_ADDRESS,
CONF_MOVE_SHORT_ADDRESS,
CONF_STOP_ADDRESS,
CONF_POSITION_ADDRESS,
CONF_POSITION_STATE_ADDRESS,
CONF_ANGLE_ADDRESS,
CONF_ANGLE_STATE_ADDRESS,
],
)
if errors:
return self.async_show_form(
step_id="add_cover", data_schema=_cover_schema(), errors=errors
)
return await self._async_store_port("cover", user_input)
return self.async_show_form(step_id="add_cover", data_schema=_cover_schema())
async def async_step_remove_port(self, user_input: dict | None = None):
ports = list(self._config_entry.options.get(CONF_PORTS, []))
if not ports:
return self.async_abort(reason="no_ports_to_remove")
if user_input is not None:
port_id = user_input[CONF_PORT_ID]
ports = [port for port in ports if port.get("id") != port_id]
return self.async_create_entry(title="", data={CONF_PORTS: ports})
options = [
{
"value": port.get("id"),
"label": port.get("title", port.get("id")),
}
for port in ports
if port.get("id")
]
schema = vol.Schema(
{
vol.Required(CONF_PORT_ID): selector.SelectSelector(
selector.SelectSelectorConfig(options=options, mode="dropdown")
)
}
)
return self.async_show_form(step_id="remove_port", data_schema=schema)
async def _async_store_port(self, port_type: str, user_input: dict):
ports = list(self._config_entry.options.get(CONF_PORTS, []))
title = _entity_title(self.hass, user_input[CONF_ENTITY_ID])
ports.append(
{
"id": uuid.uuid4().hex,
"type": port_type,
"title": title,
"data": user_input,
}
)
return self.async_create_entry(title="", data={CONF_PORTS: ports})
class BinarySensorPortSubentryFlow(config_entries.ConfigSubentryFlow):