jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-28 07:37:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add cooldown and respond_to_read options for KNX expose (#84613)

Browse Source 
Add cooldown option for KNX expose
This commit is contained in:
Matthias Alphart 2022-12-27 20:36:02 +01:00 committed by GitHub
parent 6c32337c8e
commit acd31d4ae3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 71 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
homeassistant/components/knx/expose.py
View File

@ -21,7 +21,7 @@ from homeassistant.core import Event, HomeAssistant, State, callback
from homeassistant.helpers.event import async_track_state_change_event
from homeassistant.helpers.typing import ConfigType, StateType
from .const import KNX_ADDRESS
from .const import CONF_RESPOND_TO_READ, KNX_ADDRESS
from .schema import ExposeSchema
_LOGGER = logging.getLogger(__name__)
@ -32,27 +32,23 @@ def create_knx_exposure(
hass: HomeAssistant, xknx: XKNX, config: ConfigType
) -> KNXExposeSensor | KNXExposeTime:
"""Create exposures from config."""
address = config[KNX_ADDRESS]
expose_type = config[ExposeSchema.CONF_KNX_EXPOSE_TYPE]
attribute = config.get(ExposeSchema.CONF_KNX_EXPOSE_ATTRIBUTE)
default = config.get(ExposeSchema.CONF_KNX_EXPOSE_DEFAULT)
exposure: KNXExposeSensor | KNXExposeTime
if (
isinstance(expose_type, str)
and expose_type.lower() in ExposeSchema.EXPOSE_TIME_TYPES
):
exposure = KNXExposeTime(xknx, expose_type, address)
exposure = KNXExposeTime(
xknx=xknx,
config=config,
)
else:
entity_id = config[CONF_ENTITY_ID]
exposure = KNXExposeSensor(
hass,
xknx,
expose_type,
entity_id,
attribute,
default,
address,
xknx=xknx,
config=config,
)
return exposure
@ -64,36 +60,37 @@ class KNXExposeSensor:
self,
hass: HomeAssistant,
xknx: XKNX,
expose_type: int | str,
entity_id: str,
attribute: str | None,
default: StateType,
address: str,
config: ConfigType,
) -> None:
"""Initialize of Expose class."""
self.hass = hass
self.xknx = xknx
self.type = expose_type
self.entity_id = entity_id
self.expose_attribute = attribute
self.expose_default = default
self.address = address
self.entity_id: str = config[CONF_ENTITY_ID]
self.expose_attribute: str | None = config.get(
ExposeSchema.CONF_KNX_EXPOSE_ATTRIBUTE
)
self.expose_default = config.get(ExposeSchema.CONF_KNX_EXPOSE_DEFAULT)
self.expose_type: int | str = config[ExposeSchema.CONF_KNX_EXPOSE_TYPE]
self._remove_listener: Callable[[], None] | None = None
self.device: ExposeSensor = self.async_register()
self.device: ExposeSensor = self.async_register(config)
self._init_expose_state()
@callback
def async_register(self) -> ExposeSensor:
def async_register(self, config: ConfigType) -> ExposeSensor:
"""Register listener."""
if self.expose_attribute is not None:
_name = self.entity_id + "__" + self.expose_attribute
else:
_name = self.entity_id
device = ExposeSensor(
self.xknx,
xknx=self.xknx,
name=_name,
group_address=self.address,
value_type=self.type,
group_address=config[KNX_ADDRESS],
respond_to_read=config[CONF_RESPOND_TO_READ],
value_type=self.expose_type,
cooldown=config[ExposeSchema.CONF_KNX_EXPOSE_COOLDOWN],
)
self._remove_listener = async_track_state_change_event(
self.hass, [self.entity_id], self._async_entity_changed
@ -118,7 +115,7 @@ class KNXExposeSensor:
self._remove_listener = None
self.device.shutdown()
def _get_expose_value(self, state: State | None) -> StateType:
def _get_expose_value(self, state: State | None) -> bool | int | float | str | None:
"""Extract value from state."""
if state is None or state.state in (STATE_UNKNOWN, STATE_UNAVAILABLE):
value = self.expose_default
@ -128,7 +125,7 @@ class KNXExposeSensor:
if self.expose_attribute is None
else state.attributes.get(self.expose_attribute, self.expose_default)
)
if self.type == "binary":
if self.expose_type == "binary":
if value in (1, STATE_ON, "True"):
return True
if value in (0, STATE_OFF, "False"):
@ -171,22 +168,21 @@ class KNXExposeSensor:
class KNXExposeTime:
"""Object to Expose Time/Date object to KNX bus."""
def __init__(self, xknx: XKNX, expose_type: str, address: str) -> None:
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
"""Initialize of Expose class."""
self.xknx = xknx
self.expose_type = expose_type
self.address = address
self.device: DateTime = self.async_register()
self.device: DateTime = self.async_register(config)
@callback
def async_register(self) -> DateTime:
def async_register(self, config: ConfigType) -> DateTime:
"""Register listener."""
expose_type = config[ExposeSchema.CONF_KNX_EXPOSE_TYPE]
return DateTime(
self.xknx,
name=self.expose_type.capitalize(),
broadcast_type=self.expose_type.upper(),
name=expose_type.capitalize(),
broadcast_type=expose_type.upper(),
localtime=True,
group_address=self.address,
group_address=config[KNX_ADDRESS],
)
@callback

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

@ -561,6 +561,7 @@ class ExposeSchema(KNXPlatformSchema):
CONF_KNX_EXPOSE_TYPE = CONF_TYPE
CONF_KNX_EXPOSE_ATTRIBUTE = "attribute"
CONF_KNX_EXPOSE_BINARY = "binary"
CONF_KNX_EXPOSE_COOLDOWN = "cooldown"
CONF_KNX_EXPOSE_DEFAULT = "default"
EXPOSE_TIME_TYPES: Final = [
"time",
@ -578,6 +579,8 @@ class ExposeSchema(KNXPlatformSchema):
)
EXPOSE_SENSOR_SCHEMA = vol.Schema(
{
vol.Optional(CONF_KNX_EXPOSE_COOLDOWN, default=0): cv.positive_float,
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
),

35
tests/components/knx/test_expose.py
View File

@ -1,4 +1,5 @@
"""Test KNX expose."""
from datetime import timedelta
import time
from unittest.mock import patch
@ -6,9 +7,12 @@ from homeassistant.components.knx import CONF_KNX_EXPOSE, DOMAIN, KNX_ADDRESS
from homeassistant.components.knx.schema import ExposeSchema
from homeassistant.const import CONF_ATTRIBUTE, CONF_ENTITY_ID, CONF_TYPE
from homeassistant.core import HomeAssistant
from homeassistant.util import dt
from .conftest import KNXTestKit
from tests.common import async_fire_time_changed_exact
async def test_binary_expose(hass: HomeAssistant, knx: KNXTestKit):
"""Test a binary expose to only send telegrams on state change."""
@ -163,6 +167,37 @@ async def test_expose_string(hass: HomeAssistant, knx: KNXTestKit):
)
async def test_expose_cooldown(hass: HomeAssistant, knx: KNXTestKit):
"""Test an expose with cooldown."""
cooldown_time = 2
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_COOLDOWN: cooldown_time,
}
},
)
assert not hass.states.async_all()
# Change state to 1
hass.states.async_set(entity_id, "1", {})
await knx.assert_write("1/1/8", (1,))
# Change state to 2 - skip because of cooldown
hass.states.async_set(entity_id, "2", {})
await knx.assert_no_telegram()
# Change state to 3
hass.states.async_set(entity_id, "3", {})
await knx.assert_no_telegram()
# Wait for cooldown to pass
async_fire_time_changed_exact(hass, dt.utcnow() + timedelta(seconds=cooldown_time))
await hass.async_block_till_done()
await knx.assert_write("1/1/8", (3,))
async def test_expose_conversion_exception(hass: HomeAssistant, knx: KNXTestKit):
"""Test expose throws exception."""