Compare commits

...

2 Commits

Author SHA1 Message Date
farmio
a166da552b fix type 2026-01-13 22:31:56 +01:00
farmio
31ae6951db KNX Expose: Add support for sending value periodically 2026-01-13 22:16:57 +01:00
3 changed files with 47 additions and 2 deletions

View File

@@ -116,6 +116,7 @@ class KnxExposeOptions:
dpt: type[DPTBase]
respond_to_read: bool
cooldown: float
periodic_send: float
default: Any | None
value_template: Template | None
@@ -130,12 +131,17 @@ def _yaml_config_to_expose_options(config: ConfigType) -> KnxExposeOptions:
else:
dpt = DPTBase.parse_transcoder(config[ExposeSchema.CONF_KNX_EXPOSE_TYPE]) # type: ignore[assignment] # checked by schema validation
ga = parse_device_group_address(config[KNX_ADDRESS])
cooldown_seconds = config[ExposeSchema.CONF_KNX_EXPOSE_COOLDOWN].total_seconds()
periodic_send_seconds = config[
ExposeSchema.CONF_KNX_EXPOSE_PERIODIC_SEND
].total_seconds()
return KnxExposeOptions(
attribute=config.get(ExposeSchema.CONF_KNX_EXPOSE_ATTRIBUTE),
group_address=ga,
dpt=dpt,
respond_to_read=config[CONF_RESPOND_TO_READ],
cooldown=config[ExposeSchema.CONF_KNX_EXPOSE_COOLDOWN],
cooldown=cooldown_seconds,
periodic_send=periodic_send_seconds,
default=config.get(ExposeSchema.CONF_KNX_EXPOSE_DEFAULT),
value_template=config.get(CONF_VALUE_TEMPLATE),
)
@@ -167,6 +173,7 @@ class KnxExposeEntity:
respond_to_read=option.respond_to_read,
value_type=option.dpt,
cooldown=option.cooldown,
periodic_send=option.periodic_send,
),
)
for option in options

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
from abc import ABC
from collections import OrderedDict
from datetime import timedelta
import math
from typing import ClassVar, Final
@@ -538,6 +539,7 @@ class ExposeSchema(KNXPlatformSchema):
CONF_KNX_EXPOSE_ATTRIBUTE = "attribute"
CONF_KNX_EXPOSE_BINARY = "binary"
CONF_KNX_EXPOSE_COOLDOWN = "cooldown"
CONF_KNX_EXPOSE_PERIODIC_SEND = "periodic_send"
CONF_KNX_EXPOSE_DEFAULT = "default"
CONF_TIME = "time"
CONF_DATE = "date"
@@ -554,7 +556,12 @@ class ExposeSchema(KNXPlatformSchema):
)
EXPOSE_SENSOR_SCHEMA = vol.Schema(
{
vol.Optional(CONF_KNX_EXPOSE_COOLDOWN, default=0): cv.positive_float,
vol.Optional(
CONF_KNX_EXPOSE_COOLDOWN, default=timedelta(0)
): cv.positive_time_period,
vol.Optional(
CONF_KNX_EXPOSE_PERIODIC_SEND, default=timedelta(0)
): cv.positive_time_period,
vol.Optional(CONF_RESPOND_TO_READ, default=True): cv.boolean,
vol.Required(CONF_KNX_EXPOSE_TYPE): vol.Any(
CONF_KNX_EXPOSE_BINARY, sensor_type_validator

View File

@@ -78,6 +78,11 @@ async def test_expose_attribute(hass: HomeAssistant, knx: KNXTestKit) -> None:
await hass.async_block_till_done()
await knx.assert_write("1/1/8", (1,))
# Change attribute below resolution of DPT; expect no telegram
hass.states.async_set(entity_id, "on", {attribute: 1.2})
await hass.async_block_till_done()
await knx.assert_no_telegram()
# Read in between
await knx.receive_read("1/1/8")
await knx.assert_response("1/1/8", (1,))
@@ -251,6 +256,32 @@ async def test_expose_cooldown(
await knx.assert_write("1/1/8", (3,))
async def test_expose_periodic_send(
hass: HomeAssistant, knx: KNXTestKit, freezer: FrozenDateTimeFactory
) -> None:
"""Test an expose with periodic send."""
entity_id = "fake.entity"
await knx.setup_integration(
{
CONF_KNX_EXPOSE: {
CONF_TYPE: "percentU8",
KNX_ADDRESS: "1/1/8",
CONF_ENTITY_ID: entity_id,
ExposeSchema.CONF_KNX_EXPOSE_PERIODIC_SEND: {"minutes": 1},
}
},
)
# Initialize state
hass.states.async_set(entity_id, "15", {})
await hass.async_block_till_done()
await knx.assert_write("1/1/8", (15,))
# Wait for time to pass
freezer.tick(timedelta(seconds=60))
async_fire_time_changed(hass)
await hass.async_block_till_done()
await knx.assert_write("1/1/8", (15,))
async def test_expose_value_template(
hass: HomeAssistant, knx: KNXTestKit, caplog: pytest.LogCaptureFixture
) -> None: