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

Add switch port support

Browse Source
This commit is contained in:
René Bachmann
2026-02-13 13:25:49 +01:00
parent 74f5133bc1
commit d6ec48e2e6
7 changed files with 150 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
custom_components/ha_knx_bridge/bridge.py
View File

@@ -12,6 +12,7 @@ from homeassistant.helpers import event as event_helper
from .const import (
CONF_ANGLE_ADDRESS,
CONF_ANGLE_STATE_ADDRESS,
CONF_COMMAND_ADDRESS,
CONF_MOVE_LONG_ADDRESS,
CONF_MOVE_SHORT_ADDRESS,
CONF_POSITION_ADDRESS,
@@ -43,6 +44,13 @@ class CoverPort:
angle_state_address: str | None
@dataclass(frozen=True)
class SwitchPort:
entity_id: str
command_address: str | None
state_address: str | None
class BridgeManager:
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
self.hass = hass
@@ -56,9 +64,9 @@ class BridgeManager:
if not self.hass.services.has_service(KNX_DOMAIN, "send"):
raise ConfigEntryNotReady("KNX integration services not available")
binary_ports, cover_ports = self._load_ports()
self._register_outgoing(binary_ports, cover_ports)
await self._register_incoming(cover_ports)
binary_ports, switch_ports, cover_ports = self._load_ports()
self._register_outgoing(binary_ports, switch_ports, cover_ports)
await self._register_incoming(switch_ports, cover_ports)
async def async_unload(self) -> None:
for unsub in self._unsub_listeners:
@@ -71,8 +79,11 @@ class BridgeManager:
await self._unregister_knx_events()
def _load_ports(self) -> tuple[list[BinarySensorPort], list[CoverPort]]:
def _load_ports(
self,
) -> tuple[list[BinarySensorPort], list[SwitchPort], list[CoverPort]]:
binary_ports: list[BinarySensorPort] = []
switch_ports: list[SwitchPort] = []
cover_ports: list[CoverPort] = []
subentries = getattr(self.entry, "subentries", [])
@@ -85,6 +96,16 @@ class BridgeManager:
state_address=_clean_address(data.get(CONF_STATE_ADDRESS)),
)
)
elif subentry.type == "switch":
switch_ports.append(
SwitchPort(
entity_id=data["entity_id"],
command_address=_clean_address(
data.get(CONF_COMMAND_ADDRESS)
),
state_address=_clean_address(data.get(CONF_STATE_ADDRESS)),
)
)
elif subentry.type == "cover":
cover_ports.append(
CoverPort(
@@ -103,10 +124,13 @@ class BridgeManager:
)
)
return binary_ports, cover_ports
return binary_ports, switch_ports, cover_ports
def _register_outgoing(
self, binary_ports: list[BinarySensorPort], cover_ports: list[CoverPort]
self,
binary_ports: list[BinarySensorPort],
switch_ports: list[SwitchPort],
cover_ports: list[CoverPort],
) -> None:
for port in binary_ports:
if not port.state_address:
@@ -117,6 +141,15 @@ class BridgeManager:
)
)
for port in switch_ports:
if not port.state_address:
continue
self._unsub_listeners.append(
event_helper.async_track_state_change_event(
self.hass, [port.entity_id], self._switch_changed(port)
)
)
for port in cover_ports:
if not (port.position_state_address or port.angle_state_address):
continue
@@ -126,7 +159,12 @@ class BridgeManager:
)
)
async def _register_incoming(self, cover_ports: list[CoverPort]) -> None:
async def _register_incoming(
self, switch_ports: list[SwitchPort], cover_ports: list[CoverPort]
) -> None:
for port in switch_ports:
self._register_knx_switch_address(port.command_address, port)
for port in cover_ports:
self._register_knx_address(port.move_long_address, None, port, "move_long")
self._register_knx_address(port.move_short_address, None, port, "move_short")
@@ -160,6 +198,18 @@ class BridgeManager:
)
self._registered_addresses.append((address, value_type))
def _register_knx_switch_address(
self,
address: str | None,
port: SwitchPort,
) -> None:
if not address:
return
self._address_handlers[address] = lambda event: self._handle_switch_action(
port, event
)
self._registered_addresses.append((address, None))
async def _register_knx_events(self) -> None:
for address, value_type in self._registered_addresses:
data: dict[str, Any] = {"address": address}
@@ -191,6 +241,18 @@ class BridgeManager:
return _handler
def _switch_changed(self, port: SwitchPort) -> callable[[Event], Any]:
async def _handler(event: Event) -> None:
new_state: State | None = event.data.get("new_state")
if new_state is None:
return
if new_state.state not in ("on", "off"):
return
payload = 1 if new_state.state == "on" else 0
await self._knx_send_raw(port.state_address, payload)
return _handler
def _cover_changed(self, port: CoverPort) -> callable[[Event], Any]:
async def _handler(event: Event) -> None:
new_state: State | None = event.data.get("new_state")
@@ -250,6 +312,15 @@ class BridgeManager:
port.entity_id, "set_cover_tilt_position", {"tilt_position": value}
)
async def _handle_switch_action(self, port: SwitchPort, event: Event) -> None:
value = _extract_event_value(event)
if value is None:
return
if value == 0:
await self._call_switch_service(port.entity_id, "turn_off")
elif value == 1:
await self._call_switch_service(port.entity_id, "turn_on")
async def _call_cover_service(
self, entity_id: str, service: str, service_data: dict[str, Any] | None = None
) -> None:
@@ -260,6 +331,16 @@ class BridgeManager:
"cover", service, data, blocking=False
)
async def _call_switch_service(
self, entity_id: str, service: str, service_data: dict[str, Any] | None = None
) -> None:
data = {"entity_id": entity_id}
if service_data:
data.update(service_data)
await self.hass.services.async_call(
"switch", service, data, blocking=False
)
async def _knx_send_raw(self, address: str | None, payload: int) -> None:
if not address:
return

38
custom_components/ha_knx_bridge/config_flow.py
View File

@@ -12,6 +12,7 @@ from homeassistant.helpers import selector
from .const import (
CONF_ANGLE_ADDRESS,
CONF_ANGLE_STATE_ADDRESS,
CONF_COMMAND_ADDRESS,
CONF_KNX_ENTRY_ID,
CONF_MOVE_LONG_ADDRESS,
CONF_MOVE_SHORT_ADDRESS,
@@ -66,6 +67,9 @@ class HAKnxBridgeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"binary_sensor": config_entries.SubentryType(
name="Binary Sensor Port", flow_class=BinarySensorPortSubentryFlow
),
"switch": config_entries.SubentryType(
name="Switch Port", flow_class=SwitchPortSubentryFlow
),
"cover": config_entries.SubentryType(
name="Cover Port", flow_class=CoverPortSubentryFlow
),
@@ -102,6 +106,24 @@ class BinarySensorPortSubentryFlow(config_entries.ConfigSubentryFlow):
return self.async_show_form(step_id="user", data_schema=_binary_sensor_schema())
class SwitchPortSubentryFlow(config_entries.ConfigSubentryFlow):
VERSION = 1
async def async_step_user(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="user", data_schema=_switch_schema(), errors=errors
)
title = _entity_title(self.hass, user_input[CONF_ENTITY_ID])
return self.async_create_entry(title=title, data=user_input)
return self.async_show_form(step_id="user", data_schema=_switch_schema())
class CoverPortSubentryFlow(config_entries.ConfigSubentryFlow):
VERSION = 1
@@ -180,6 +202,22 @@ def _cover_schema() -> vol.Schema:
)
def _switch_schema() -> vol.Schema:
return vol.Schema(
{
vol.Required(CONF_ENTITY_ID): selector.EntitySelector(
selector.EntitySelectorConfig(domain=["switch"])
),
vol.Optional(CONF_COMMAND_ADDRESS): selector.TextSelector(
selector.TextSelectorConfig(type="text")
),
vol.Optional(CONF_STATE_ADDRESS): selector.TextSelector(
selector.TextSelectorConfig(type="text")
),
}
)
def _validate_knx_addresses(
user_input: dict, keys: list[str]
) -> tuple[dict, dict[str, str]]:

1
custom_components/ha_knx_bridge/const.py
View File

@@ -3,6 +3,7 @@ DOMAIN = "ha_knx_bridge"
CONF_KNX_ENTRY_ID = "knx_entry_id"
CONF_STATE_ADDRESS = "state_address"
CONF_COMMAND_ADDRESS = "command_address"
CONF_MOVE_LONG_ADDRESS = "move_long_address"
CONF_MOVE_SHORT_ADDRESS = "move_short_address"

2
custom_components/ha_knx_bridge/manifest.json
View File

@@ -1,7 +1,7 @@
{
"domain": "ha_knx_bridge",
"name": "HA KNX Bridge",
"version": "0.0.2",
"version": "0.0.3",
"config_flow": true,
"documentation": "https://github.com/bahmcloud/HA-KNX-Bridge",
"issue_tracker": "https://github.com/bahmcloud/HA-KNX-Bridge/issues",

12
custom_components/ha_knx_bridge/strings.json
View File

@@ -39,6 +39,18 @@
}
}
},
"switch": {
"step": {
"user": {
"title": "Switch Port",
"data": {
"entity_id": "Switch entity",
"command_address": "Command group address (DPT 1)",
"state_address": "State group address (DPT 1)"
}
}
}
},
"cover": {
"step": {
"user": {