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 per-address invert toggles

Browse Source
This commit is contained in:
René Bachmann
2026-02-13 13:44:06 +01:00
parent d6ec48e2e6
commit d91d3edc5a
7 changed files with 353 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

251
custom_components/ha_knx_bridge/bridge.py
View File

@@ -13,6 +13,8 @@ from .const import (
CONF_ANGLE_ADDRESS,
CONF_ANGLE_STATE_ADDRESS,
CONF_COMMAND_ADDRESS,
CONF_INVERT_INCOMING,
CONF_INVERT_OUTGOING,
CONF_MOVE_LONG_ADDRESS,
CONF_MOVE_SHORT_ADDRESS,
CONF_POSITION_ADDRESS,
@@ -30,18 +32,34 @@ KNX_DOMAIN = "knx"
class BinarySensorPort:
entity_id: str
state_address: str | None
state_invert_incoming: bool
state_invert_outgoing: bool
@dataclass(frozen=True)
class CoverPort:
entity_id: str
move_long_address: str | None
move_long_invert_incoming: bool
move_long_invert_outgoing: bool
move_short_address: str | None
move_short_invert_incoming: bool
move_short_invert_outgoing: bool
stop_address: str | None
stop_invert_incoming: bool
stop_invert_outgoing: bool
position_address: str | None
position_invert_incoming: bool
position_invert_outgoing: bool
position_state_address: str | None
position_state_invert_incoming: bool
position_state_invert_outgoing: bool
angle_address: str | None
angle_invert_incoming: bool
angle_invert_outgoing: bool
angle_state_address: str | None
angle_state_invert_incoming: bool
angle_state_invert_outgoing: bool
@dataclass(frozen=True)
@@ -49,6 +67,17 @@ class SwitchPort:
entity_id: str
command_address: str | None
state_address: str | None
command_invert_incoming: bool
command_invert_outgoing: bool
state_invert_incoming: bool
state_invert_outgoing: bool
@dataclass(frozen=True)
class AddressOptions:
value_type: str | None
invert_incoming: bool
invert_outgoing: bool
class BridgeManager:
@@ -59,6 +88,7 @@ class BridgeManager:
self._knx_event_unsub: callable | None = None
self._address_handlers: dict[str, callable[[Event], Any]] = {}
self._registered_addresses: list[tuple[str, str | None]] = []
self._address_options: dict[str, AddressOptions] = {}
async def async_setup(self) -> None:
if not self.hass.services.has_service(KNX_DOMAIN, "send"):
@@ -94,6 +124,12 @@ class BridgeManager:
BinarySensorPort(
entity_id=data["entity_id"],
state_address=_clean_address(data.get(CONF_STATE_ADDRESS)),
state_invert_incoming=_clean_bool(
data.get(_invert_in_key(CONF_STATE_ADDRESS))
),
state_invert_outgoing=_clean_bool(
data.get(_invert_out_key(CONF_STATE_ADDRESS))
),
)
)
elif subentry.type == "switch":
@@ -104,6 +140,18 @@ class BridgeManager:
data.get(CONF_COMMAND_ADDRESS)
),
state_address=_clean_address(data.get(CONF_STATE_ADDRESS)),
command_invert_incoming=_clean_bool(
data.get(_invert_in_key(CONF_COMMAND_ADDRESS))
),
command_invert_outgoing=_clean_bool(
data.get(_invert_out_key(CONF_COMMAND_ADDRESS))
),
state_invert_incoming=_clean_bool(
data.get(_invert_in_key(CONF_STATE_ADDRESS))
),
state_invert_outgoing=_clean_bool(
data.get(_invert_out_key(CONF_STATE_ADDRESS))
),
)
)
elif subentry.type == "cover":
@@ -111,16 +159,58 @@ class BridgeManager:
CoverPort(
entity_id=data["entity_id"],
move_long_address=_clean_address(data.get(CONF_MOVE_LONG_ADDRESS)),
move_long_invert_incoming=_clean_bool(
data.get(_invert_in_key(CONF_MOVE_LONG_ADDRESS))
),
move_long_invert_outgoing=_clean_bool(
data.get(_invert_out_key(CONF_MOVE_LONG_ADDRESS))
),
move_short_address=_clean_address(data.get(CONF_MOVE_SHORT_ADDRESS)),
move_short_invert_incoming=_clean_bool(
data.get(_invert_in_key(CONF_MOVE_SHORT_ADDRESS))
),
move_short_invert_outgoing=_clean_bool(
data.get(_invert_out_key(CONF_MOVE_SHORT_ADDRESS))
),
stop_address=_clean_address(data.get(CONF_STOP_ADDRESS)),
stop_invert_incoming=_clean_bool(
data.get(_invert_in_key(CONF_STOP_ADDRESS))
),
stop_invert_outgoing=_clean_bool(
data.get(_invert_out_key(CONF_STOP_ADDRESS))
),
position_address=_clean_address(data.get(CONF_POSITION_ADDRESS)),
position_invert_incoming=_clean_bool(
data.get(_invert_in_key(CONF_POSITION_ADDRESS))
),
position_invert_outgoing=_clean_bool(
data.get(_invert_out_key(CONF_POSITION_ADDRESS))
),
position_state_address=_clean_address(
data.get(CONF_POSITION_STATE_ADDRESS)
),
position_state_invert_incoming=_clean_bool(
data.get(_invert_in_key(CONF_POSITION_STATE_ADDRESS))
),
position_state_invert_outgoing=_clean_bool(
data.get(_invert_out_key(CONF_POSITION_STATE_ADDRESS))
),
angle_address=_clean_address(data.get(CONF_ANGLE_ADDRESS)),
angle_invert_incoming=_clean_bool(
data.get(_invert_in_key(CONF_ANGLE_ADDRESS))
),
angle_invert_outgoing=_clean_bool(
data.get(_invert_out_key(CONF_ANGLE_ADDRESS))
),
angle_state_address=_clean_address(
data.get(CONF_ANGLE_STATE_ADDRESS)
),
angle_state_invert_incoming=_clean_bool(
data.get(_invert_in_key(CONF_ANGLE_STATE_ADDRESS))
),
angle_state_invert_outgoing=_clean_bool(
data.get(_invert_out_key(CONF_ANGLE_STATE_ADDRESS))
),
)
)
@@ -135,6 +225,12 @@ class BridgeManager:
for port in binary_ports:
if not port.state_address:
continue
self._remember_address_options(
port.state_address,
None,
port.state_invert_incoming,
port.state_invert_outgoing,
)
self._unsub_listeners.append(
event_helper.async_track_state_change_event(
self.hass, [port.entity_id], self._binary_sensor_changed(port)
@@ -144,6 +240,12 @@ class BridgeManager:
for port in switch_ports:
if not port.state_address:
continue
self._remember_address_options(
port.state_address,
None,
port.state_invert_incoming,
port.state_invert_outgoing,
)
self._unsub_listeners.append(
event_helper.async_track_state_change_event(
self.hass, [port.entity_id], self._switch_changed(port)
@@ -153,6 +255,20 @@ class BridgeManager:
for port in cover_ports:
if not (port.position_state_address or port.angle_state_address):
continue
if port.position_state_address:
self._remember_address_options(
port.position_state_address,
"percent",
port.position_state_invert_incoming,
port.position_state_invert_outgoing,
)
if port.angle_state_address:
self._remember_address_options(
port.angle_state_address,
"percent",
port.angle_state_invert_incoming,
port.angle_state_invert_outgoing,
)
self._unsub_listeners.append(
event_helper.async_track_state_change_event(
self.hass, [port.entity_id], self._cover_changed(port)
@@ -163,17 +279,53 @@ class BridgeManager:
self, switch_ports: list[SwitchPort], cover_ports: list[CoverPort]
) -> None:
for port in switch_ports:
self._register_knx_switch_address(port.command_address, port)
self._register_knx_switch_address(
port.command_address,
port.command_invert_incoming,
port.command_invert_outgoing,
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")
self._register_knx_address(port.stop_address, None, port, "stop")
self._register_knx_address(
port.position_address, "percent", port, "position"
port.move_long_address,
None,
port.move_long_invert_incoming,
port.move_long_invert_outgoing,
port,
"move_long",
)
self._register_knx_address(
port.angle_address, "percent", port, "angle"
port.move_short_address,
None,
port.move_short_invert_incoming,
port.move_short_invert_outgoing,
port,
"move_short",
)
self._register_knx_address(
port.stop_address,
None,
port.stop_invert_incoming,
port.stop_invert_outgoing,
port,
"stop",
)
self._register_knx_address(
port.position_address,
"percent",
port.position_invert_incoming,
port.position_invert_outgoing,
port,
"position",
)
self._register_knx_address(
port.angle_address,
"percent",
port.angle_invert_incoming,
port.angle_invert_outgoing,
port,
"angle",
)
if self._address_handlers:
@@ -187,6 +339,8 @@ class BridgeManager:
self,
address: str | None,
value_type: str | None,
invert_incoming: bool,
invert_outgoing: bool,
port: CoverPort,
action: str,
) -> None:
@@ -196,11 +350,16 @@ class BridgeManager:
self._address_handlers[address] = lambda event: self._handle_cover_action(
port, action, event
)
self._remember_address_options(
address, value_type, invert_incoming, invert_outgoing
)
self._registered_addresses.append((address, value_type))
def _register_knx_switch_address(
self,
address: str | None,
invert_incoming: bool,
invert_outgoing: bool,
port: SwitchPort,
) -> None:
if not address:
@@ -208,8 +367,24 @@ class BridgeManager:
self._address_handlers[address] = lambda event: self._handle_switch_action(
port, event
)
self._remember_address_options(
address, None, invert_incoming, invert_outgoing
)
self._registered_addresses.append((address, None))
def _remember_address_options(
self,
address: str,
value_type: str | None,
invert_incoming: bool,
invert_outgoing: bool,
) -> None:
self._address_options[address] = AddressOptions(
value_type=value_type,
invert_incoming=invert_incoming,
invert_outgoing=invert_outgoing,
)
async def _register_knx_events(self) -> None:
for address, value_type in self._registered_addresses:
data: dict[str, Any] = {"address": address}
@@ -237,6 +412,9 @@ class BridgeManager:
if new_state.state not in ("on", "off"):
return
payload = 1 if new_state.state == "on" else 0
payload = _invert_value(
payload, port.state_address, self._address_options, "outgoing"
)
await self._knx_send_raw(port.state_address, payload)
return _handler
@@ -249,6 +427,9 @@ class BridgeManager:
if new_state.state not in ("on", "off"):
return
payload = 1 if new_state.state == "on" else 0
payload = _invert_value(
payload, port.state_address, self._address_options, "outgoing"
)
await self._knx_send_raw(port.state_address, payload)
return _handler
@@ -262,11 +443,25 @@ class BridgeManager:
if port.position_state_address is not None:
position = new_state.attributes.get("current_position")
if position is not None:
await self._knx_send_percent(port.position_state_address, position)
position = _invert_value(
position,
port.position_state_address,
self._address_options,
"outgoing",
)
await self._knx_send_percent(
port.position_state_address, position
)
if port.angle_state_address is not None:
angle = new_state.attributes.get("current_tilt_position")
if angle is not None:
angle = _invert_value(
angle,
port.angle_state_address,
self._address_options,
"outgoing",
)
await self._knx_send_percent(port.angle_state_address, angle)
return _handler
@@ -291,6 +486,9 @@ class BridgeManager:
value = _extract_event_value(event)
if value is None:
return
value = _invert_value(
value, event.data.get("destination"), self._address_options, "incoming"
)
if action == "move_long":
if value == 0:
@@ -316,6 +514,9 @@ class BridgeManager:
value = _extract_event_value(event)
if value is None:
return
value = _invert_value(
value, event.data.get("destination"), self._address_options, "incoming"
)
if value == 0:
await self._call_switch_service(port.entity_id, "turn_off")
elif value == 1:
@@ -386,3 +587,39 @@ def _clean_address(address: Any) -> str | None:
stripped = address.strip()
return stripped or None
return None
def _clean_bool(value: Any) -> bool:
return bool(value)
def _invert_value(
value: int,
address: str | None,
address_options: dict[str, AddressOptions],
direction: str,
) -> int:
if address is None:
return value
options = address_options.get(address)
if options is None:
return value
if direction == "incoming":
if not options.invert_incoming:
return value
else:
if not options.invert_outgoing:
return value
if options.value_type == "percent":
return 100 - value
if value not in (0, 1):
return value
return 0 if value == 1 else 1
def _invert_in_key(address_key: str) -> str:
return f"{address_key}_{CONF_INVERT_INCOMING}"
def _invert_out_key(address_key: str) -> str:
return f"{address_key}_{CONF_INVERT_OUTGOING}"