mirror of
https://github.com/home-assistant/core.git
synced 2025-07-19 03:07:37 +00:00
Cleanup KNX integration (#52168)
Co-authored-by: Franck Nijhof <frenck@frenck.nl>
This commit is contained in:
parent
b7c15d4474
commit
5687ced7b3
@ -298,7 +298,6 @@ class KNXModule:
|
|||||||
return self.connection_config_tunneling()
|
return self.connection_config_tunneling()
|
||||||
if CONF_KNX_ROUTING in self.config[DOMAIN]:
|
if CONF_KNX_ROUTING in self.config[DOMAIN]:
|
||||||
return self.connection_config_routing()
|
return self.connection_config_routing()
|
||||||
# config from xknx.yaml always has priority later on
|
|
||||||
return ConnectionConfig(auto_reconnect=True)
|
return ConnectionConfig(auto_reconnect=True)
|
||||||
|
|
||||||
def connection_config_routing(self) -> ConnectionConfig:
|
def connection_config_routing(self) -> ConnectionConfig:
|
||||||
@ -379,14 +378,16 @@ class KNXModule:
|
|||||||
"Service event_register could not remove event for '%s'",
|
"Service event_register could not remove event for '%s'",
|
||||||
str(group_address),
|
str(group_address),
|
||||||
)
|
)
|
||||||
else:
|
return
|
||||||
for group_address in group_addresses:
|
|
||||||
if group_address not in self._knx_event_callback.group_addresses:
|
for group_address in group_addresses:
|
||||||
self._knx_event_callback.group_addresses.append(group_address)
|
if group_address in self._knx_event_callback.group_addresses:
|
||||||
_LOGGER.debug(
|
continue
|
||||||
"Service event_register registered event for '%s'",
|
self._knx_event_callback.group_addresses.append(group_address)
|
||||||
str(group_address),
|
_LOGGER.debug(
|
||||||
)
|
"Service event_register registered event for '%s'",
|
||||||
|
str(group_address),
|
||||||
|
)
|
||||||
|
|
||||||
async def service_exposure_register_modify(self, call: ServiceCall) -> None:
|
async def service_exposure_register_modify(self, call: ServiceCall) -> None:
|
||||||
"""Service for adding or removing an exposure to KNX bus."""
|
"""Service for adding or removing an exposure to KNX bus."""
|
||||||
@ -405,7 +406,6 @@ class KNXModule:
|
|||||||
|
|
||||||
if group_address in self.service_exposures:
|
if group_address in self.service_exposures:
|
||||||
replaced_exposure = self.service_exposures.pop(group_address)
|
replaced_exposure = self.service_exposures.pop(group_address)
|
||||||
assert replaced_exposure.device is not None
|
|
||||||
_LOGGER.warning(
|
_LOGGER.warning(
|
||||||
"Service exposure_register replacing already registered exposure for '%s' - %s",
|
"Service exposure_register replacing already registered exposure for '%s' - %s",
|
||||||
group_address,
|
group_address,
|
||||||
|
@ -31,19 +31,18 @@ async def async_setup_platform(
|
|||||||
platform_config = discovery_info["platform_config"]
|
platform_config = discovery_info["platform_config"]
|
||||||
xknx: XKNX = hass.data[DOMAIN].xknx
|
xknx: XKNX = hass.data[DOMAIN].xknx
|
||||||
|
|
||||||
entities = []
|
async_add_entities(
|
||||||
for entity_config in platform_config:
|
KNXBinarySensor(xknx, entity_config) for entity_config in platform_config
|
||||||
entities.append(KNXBinarySensor(xknx, entity_config))
|
)
|
||||||
|
|
||||||
async_add_entities(entities)
|
|
||||||
|
|
||||||
|
|
||||||
class KNXBinarySensor(KnxEntity, BinarySensorEntity):
|
class KNXBinarySensor(KnxEntity, BinarySensorEntity):
|
||||||
"""Representation of a KNX binary sensor."""
|
"""Representation of a KNX binary sensor."""
|
||||||
|
|
||||||
|
_device: XknxBinarySensor
|
||||||
|
|
||||||
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
||||||
"""Initialize of KNX binary sensor."""
|
"""Initialize of KNX binary sensor."""
|
||||||
self._device: XknxBinarySensor
|
|
||||||
super().__init__(
|
super().__init__(
|
||||||
device=XknxBinarySensor(
|
device=XknxBinarySensor(
|
||||||
xknx,
|
xknx,
|
||||||
@ -58,13 +57,9 @@ class KNXBinarySensor(KnxEntity, BinarySensorEntity):
|
|||||||
reset_after=config.get(BinarySensorSchema.CONF_RESET_AFTER),
|
reset_after=config.get(BinarySensorSchema.CONF_RESET_AFTER),
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
self._device_class: str | None = config.get(CONF_DEVICE_CLASS)
|
self._attr_device_class = config.get(CONF_DEVICE_CLASS)
|
||||||
self._unique_id = f"{self._device.remote_value.group_address_state}"
|
self._attr_force_update = self._device.ignore_internal_state
|
||||||
|
self._attr_unique_id = str(self._device.remote_value.group_address_state)
|
||||||
@property
|
|
||||||
def device_class(self) -> str | None:
|
|
||||||
"""Return the class of this sensor."""
|
|
||||||
return self._device_class
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def is_on(self) -> bool:
|
def is_on(self) -> bool:
|
||||||
@ -84,13 +79,3 @@ class KNXBinarySensor(KnxEntity, BinarySensorEntity):
|
|||||||
dt.as_utc(self._device.last_telegram.timestamp)
|
dt.as_utc(self._device.last_telegram.timestamp)
|
||||||
)
|
)
|
||||||
return attr
|
return attr
|
||||||
|
|
||||||
@property
|
|
||||||
def force_update(self) -> bool:
|
|
||||||
"""
|
|
||||||
Return True if state updates should be forced.
|
|
||||||
|
|
||||||
If True, a state change will be triggered anytime the state property is
|
|
||||||
updated, not just when the value changes.
|
|
||||||
"""
|
|
||||||
return self._device.ignore_internal_state
|
|
||||||
|
@ -46,11 +46,9 @@ async def async_setup_platform(
|
|||||||
xknx: XKNX = hass.data[DOMAIN].xknx
|
xknx: XKNX = hass.data[DOMAIN].xknx
|
||||||
|
|
||||||
_async_migrate_unique_id(hass, platform_config)
|
_async_migrate_unique_id(hass, platform_config)
|
||||||
entities = []
|
async_add_entities(
|
||||||
for entity_config in platform_config:
|
KNXClimate(xknx, entity_config) for entity_config in platform_config
|
||||||
entities.append(KNXClimate(xknx, entity_config))
|
)
|
||||||
|
|
||||||
async_add_entities(entities)
|
|
||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
@ -168,22 +166,20 @@ def _create_climate(xknx: XKNX, config: ConfigType) -> XknxClimate:
|
|||||||
class KNXClimate(KnxEntity, ClimateEntity):
|
class KNXClimate(KnxEntity, ClimateEntity):
|
||||||
"""Representation of a KNX climate device."""
|
"""Representation of a KNX climate device."""
|
||||||
|
|
||||||
|
_device: XknxClimate
|
||||||
|
_attr_supported_features = SUPPORT_TARGET_TEMPERATURE | SUPPORT_PRESET_MODE
|
||||||
|
_attr_temperature_unit = TEMP_CELSIUS
|
||||||
|
|
||||||
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
||||||
"""Initialize of a KNX climate device."""
|
"""Initialize of a KNX climate device."""
|
||||||
self._device: XknxClimate
|
|
||||||
super().__init__(_create_climate(xknx, config))
|
super().__init__(_create_climate(xknx, config))
|
||||||
self._unique_id = (
|
self._attr_target_temperature_step = self._device.temperature_step
|
||||||
|
self._attr_unique_id = (
|
||||||
f"{self._device.temperature.group_address_state}_"
|
f"{self._device.temperature.group_address_state}_"
|
||||||
f"{self._device.target_temperature.group_address_state}_"
|
f"{self._device.target_temperature.group_address_state}_"
|
||||||
f"{self._device.target_temperature.group_address}_"
|
f"{self._device.target_temperature.group_address}_"
|
||||||
f"{self._device._setpoint_shift.group_address}" # pylint: disable=protected-access
|
f"{self._device._setpoint_shift.group_address}" # pylint: disable=protected-access
|
||||||
)
|
)
|
||||||
self._unit_of_measurement = TEMP_CELSIUS
|
|
||||||
|
|
||||||
@property
|
|
||||||
def supported_features(self) -> int:
|
|
||||||
"""Return the list of supported features."""
|
|
||||||
return SUPPORT_TARGET_TEMPERATURE | SUPPORT_PRESET_MODE
|
|
||||||
|
|
||||||
async def async_update(self) -> None:
|
async def async_update(self) -> None:
|
||||||
"""Request a state update from KNX bus."""
|
"""Request a state update from KNX bus."""
|
||||||
@ -191,21 +187,11 @@ class KNXClimate(KnxEntity, ClimateEntity):
|
|||||||
if self._device.mode is not None:
|
if self._device.mode is not None:
|
||||||
await self._device.mode.sync()
|
await self._device.mode.sync()
|
||||||
|
|
||||||
@property
|
|
||||||
def temperature_unit(self) -> str:
|
|
||||||
"""Return the unit of measurement."""
|
|
||||||
return self._unit_of_measurement
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def current_temperature(self) -> float | None:
|
def current_temperature(self) -> float | None:
|
||||||
"""Return the current temperature."""
|
"""Return the current temperature."""
|
||||||
return self._device.temperature.value
|
return self._device.temperature.value
|
||||||
|
|
||||||
@property
|
|
||||||
def target_temperature_step(self) -> float:
|
|
||||||
"""Return the supported step of target temperature."""
|
|
||||||
return self._device.temperature_step
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def target_temperature(self) -> float | None:
|
def target_temperature(self) -> float | None:
|
||||||
"""Return the temperature we try to reach."""
|
"""Return the temperature we try to reach."""
|
||||||
|
@ -37,8 +37,8 @@ CONF_STATE_ADDRESS: Final = "state_address"
|
|||||||
CONF_SYNC_STATE: Final = "sync_state"
|
CONF_SYNC_STATE: Final = "sync_state"
|
||||||
|
|
||||||
ATTR_COUNTER: Final = "counter"
|
ATTR_COUNTER: Final = "counter"
|
||||||
ATTR_SOURCE: Final = "source"
|
|
||||||
ATTR_LAST_KNX_UPDATE: Final = "last_knx_update"
|
ATTR_LAST_KNX_UPDATE: Final = "last_knx_update"
|
||||||
|
ATTR_SOURCE: Final = "source"
|
||||||
|
|
||||||
|
|
||||||
class ColorTempModes(Enum):
|
class ColorTempModes(Enum):
|
||||||
|
@ -44,15 +44,12 @@ async def async_setup_platform(
|
|||||||
if not discovery_info or not discovery_info["platform_config"]:
|
if not discovery_info or not discovery_info["platform_config"]:
|
||||||
return
|
return
|
||||||
platform_config = discovery_info["platform_config"]
|
platform_config = discovery_info["platform_config"]
|
||||||
_async_migrate_unique_id(hass, platform_config)
|
|
||||||
|
|
||||||
xknx: XKNX = hass.data[DOMAIN].xknx
|
xknx: XKNX = hass.data[DOMAIN].xknx
|
||||||
|
|
||||||
entities = []
|
_async_migrate_unique_id(hass, platform_config)
|
||||||
for entity_config in platform_config:
|
async_add_entities(
|
||||||
entities.append(KNXCover(xknx, entity_config))
|
KNXCover(xknx, entity_config) for entity_config in platform_config
|
||||||
|
)
|
||||||
async_add_entities(entities)
|
|
||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
@ -85,9 +82,10 @@ def _async_migrate_unique_id(
|
|||||||
class KNXCover(KnxEntity, CoverEntity):
|
class KNXCover(KnxEntity, CoverEntity):
|
||||||
"""Representation of a KNX cover."""
|
"""Representation of a KNX cover."""
|
||||||
|
|
||||||
|
_device: XknxCover
|
||||||
|
|
||||||
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
||||||
"""Initialize the cover."""
|
"""Initialize the cover."""
|
||||||
self._device: XknxCover
|
|
||||||
super().__init__(
|
super().__init__(
|
||||||
device=XknxCover(
|
device=XknxCover(
|
||||||
xknx,
|
xknx,
|
||||||
@ -109,12 +107,15 @@ class KNXCover(KnxEntity, CoverEntity):
|
|||||||
invert_angle=config[CoverSchema.CONF_INVERT_ANGLE],
|
invert_angle=config[CoverSchema.CONF_INVERT_ANGLE],
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
self._device_class: str | None = config.get(CONF_DEVICE_CLASS)
|
self._unsubscribe_auto_updater: Callable[[], None] | None = None
|
||||||
self._unique_id = (
|
|
||||||
|
self._attr_device_class = config.get(CONF_DEVICE_CLASS) or (
|
||||||
|
DEVICE_CLASS_BLIND if self._device.supports_angle else None
|
||||||
|
)
|
||||||
|
self._attr_unique_id = (
|
||||||
f"{self._device.updown.group_address}_"
|
f"{self._device.updown.group_address}_"
|
||||||
f"{self._device.position_target.group_address}"
|
f"{self._device.position_target.group_address}"
|
||||||
)
|
)
|
||||||
self._unsubscribe_auto_updater: Callable[[], None] | None = None
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
async def after_update_callback(self, device: XknxDevice) -> None:
|
async def after_update_callback(self, device: XknxDevice) -> None:
|
||||||
@ -123,15 +124,6 @@ class KNXCover(KnxEntity, CoverEntity):
|
|||||||
if self._device.is_traveling():
|
if self._device.is_traveling():
|
||||||
self.start_auto_updater()
|
self.start_auto_updater()
|
||||||
|
|
||||||
@property
|
|
||||||
def device_class(self) -> str | None:
|
|
||||||
"""Return the class of this device, from component DEVICE_CLASSES."""
|
|
||||||
if self._device_class:
|
|
||||||
return self._device_class
|
|
||||||
if self._device.supports_angle:
|
|
||||||
return DEVICE_CLASS_BLIND
|
|
||||||
return None
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def supported_features(self) -> int:
|
def supported_features(self) -> int:
|
||||||
"""Flag supported features."""
|
"""Flag supported features."""
|
||||||
|
@ -34,23 +34,19 @@ async def async_setup_platform(
|
|||||||
"""Set up fans for KNX platform."""
|
"""Set up fans for KNX platform."""
|
||||||
if not discovery_info or not discovery_info["platform_config"]:
|
if not discovery_info or not discovery_info["platform_config"]:
|
||||||
return
|
return
|
||||||
|
|
||||||
platform_config = discovery_info["platform_config"]
|
platform_config = discovery_info["platform_config"]
|
||||||
xknx: XKNX = hass.data[DOMAIN].xknx
|
xknx: XKNX = hass.data[DOMAIN].xknx
|
||||||
|
|
||||||
entities = []
|
async_add_entities(KNXFan(xknx, entity_config) for entity_config in platform_config)
|
||||||
for entity_config in platform_config:
|
|
||||||
entities.append(KNXFan(xknx, entity_config))
|
|
||||||
|
|
||||||
async_add_entities(entities)
|
|
||||||
|
|
||||||
|
|
||||||
class KNXFan(KnxEntity, FanEntity):
|
class KNXFan(KnxEntity, FanEntity):
|
||||||
"""Representation of a KNX fan."""
|
"""Representation of a KNX fan."""
|
||||||
|
|
||||||
|
_device: XknxFan
|
||||||
|
|
||||||
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
||||||
"""Initialize of KNX fan."""
|
"""Initialize of KNX fan."""
|
||||||
self._device: XknxFan
|
|
||||||
max_step = config.get(FanSchema.CONF_MAX_STEP)
|
max_step = config.get(FanSchema.CONF_MAX_STEP)
|
||||||
super().__init__(
|
super().__init__(
|
||||||
device=XknxFan(
|
device=XknxFan(
|
||||||
@ -67,10 +63,16 @@ class KNXFan(KnxEntity, FanEntity):
|
|||||||
max_step=max_step,
|
max_step=max_step,
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
self._unique_id = f"{self._device.speed.group_address}"
|
|
||||||
# FanSpeedMode.STEP if max_step is set
|
# FanSpeedMode.STEP if max_step is set
|
||||||
self._step_range: tuple[int, int] | None = (1, max_step) if max_step else None
|
self._step_range: tuple[int, int] | None = (1, max_step) if max_step else None
|
||||||
|
|
||||||
|
self._attr_supported_features = (
|
||||||
|
SUPPORT_SET_SPEED | SUPPORT_OSCILLATE
|
||||||
|
if self._device.supports_oscillation
|
||||||
|
else SUPPORT_SET_SPEED
|
||||||
|
)
|
||||||
|
self._attr_unique_id = str(self._device.speed.group_address)
|
||||||
|
|
||||||
async def async_set_percentage(self, percentage: int) -> None:
|
async def async_set_percentage(self, percentage: int) -> None:
|
||||||
"""Set the speed of the fan, as a percentage."""
|
"""Set the speed of the fan, as a percentage."""
|
||||||
if self._step_range:
|
if self._step_range:
|
||||||
@ -79,16 +81,6 @@ class KNXFan(KnxEntity, FanEntity):
|
|||||||
else:
|
else:
|
||||||
await self._device.set_speed(percentage)
|
await self._device.set_speed(percentage)
|
||||||
|
|
||||||
@property
|
|
||||||
def supported_features(self) -> int:
|
|
||||||
"""Flag supported features."""
|
|
||||||
flags = SUPPORT_SET_SPEED
|
|
||||||
|
|
||||||
if self._device.supports_oscillation:
|
|
||||||
flags |= SUPPORT_OSCILLATE
|
|
||||||
|
|
||||||
return flags
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def percentage(self) -> int | None:
|
def percentage(self) -> int | None:
|
||||||
"""Return the current speed as a percentage."""
|
"""Return the current speed as a percentage."""
|
||||||
|
@ -14,10 +14,11 @@ from .const import DOMAIN
|
|||||||
class KnxEntity(Entity):
|
class KnxEntity(Entity):
|
||||||
"""Representation of a KNX entity."""
|
"""Representation of a KNX entity."""
|
||||||
|
|
||||||
|
_attr_should_poll = False
|
||||||
|
|
||||||
def __init__(self, device: XknxDevice) -> None:
|
def __init__(self, device: XknxDevice) -> None:
|
||||||
"""Set up device."""
|
"""Set up device."""
|
||||||
self._device = device
|
self._device = device
|
||||||
self._unique_id: str | None = None
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def name(self) -> str:
|
def name(self) -> str:
|
||||||
@ -30,16 +31,6 @@ class KnxEntity(Entity):
|
|||||||
knx_module = cast(KNXModule, self.hass.data[DOMAIN])
|
knx_module = cast(KNXModule, self.hass.data[DOMAIN])
|
||||||
return knx_module.connected
|
return knx_module.connected
|
||||||
|
|
||||||
@property
|
|
||||||
def should_poll(self) -> bool:
|
|
||||||
"""No polling needed within KNX."""
|
|
||||||
return False
|
|
||||||
|
|
||||||
@property
|
|
||||||
def unique_id(self) -> str | None:
|
|
||||||
"""Return the unique id of the device."""
|
|
||||||
return self._unique_id
|
|
||||||
|
|
||||||
async def async_update(self) -> None:
|
async def async_update(self) -> None:
|
||||||
"""Request a state update from KNX bus."""
|
"""Request a state update from KNX bus."""
|
||||||
await self._device.sync()
|
await self._device.sync()
|
||||||
|
@ -43,15 +43,12 @@ async def async_setup_platform(
|
|||||||
if not discovery_info or not discovery_info["platform_config"]:
|
if not discovery_info or not discovery_info["platform_config"]:
|
||||||
return
|
return
|
||||||
platform_config = discovery_info["platform_config"]
|
platform_config = discovery_info["platform_config"]
|
||||||
_async_migrate_unique_id(hass, platform_config)
|
|
||||||
|
|
||||||
xknx: XKNX = hass.data[DOMAIN].xknx
|
xknx: XKNX = hass.data[DOMAIN].xknx
|
||||||
|
|
||||||
entities = []
|
_async_migrate_unique_id(hass, platform_config)
|
||||||
for entity_config in platform_config:
|
async_add_entities(
|
||||||
entities.append(KNXLight(xknx, entity_config))
|
KNXLight(xknx, entity_config) for entity_config in platform_config
|
||||||
|
)
|
||||||
async_add_entities(entities)
|
|
||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
@ -223,19 +220,21 @@ def _create_light(xknx: XKNX, config: ConfigType) -> XknxLight:
|
|||||||
class KNXLight(KnxEntity, LightEntity):
|
class KNXLight(KnxEntity, LightEntity):
|
||||||
"""Representation of a KNX light."""
|
"""Representation of a KNX light."""
|
||||||
|
|
||||||
|
_device: XknxLight
|
||||||
|
|
||||||
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
||||||
"""Initialize of KNX light."""
|
"""Initialize of KNX light."""
|
||||||
self._device: XknxLight
|
|
||||||
super().__init__(_create_light(xknx, config))
|
super().__init__(_create_light(xknx, config))
|
||||||
self._unique_id = self._device_unique_id()
|
|
||||||
self._min_kelvin: int = config[LightSchema.CONF_MIN_KELVIN]
|
|
||||||
self._max_kelvin: int = config[LightSchema.CONF_MAX_KELVIN]
|
self._max_kelvin: int = config[LightSchema.CONF_MAX_KELVIN]
|
||||||
self._min_mireds = color_util.color_temperature_kelvin_to_mired(
|
self._min_kelvin: int = config[LightSchema.CONF_MIN_KELVIN]
|
||||||
self._max_kelvin
|
|
||||||
)
|
self._attr_max_mireds = color_util.color_temperature_kelvin_to_mired(
|
||||||
self._max_mireds = color_util.color_temperature_kelvin_to_mired(
|
|
||||||
self._min_kelvin
|
self._min_kelvin
|
||||||
)
|
)
|
||||||
|
self._attr_min_mireds = color_util.color_temperature_kelvin_to_mired(
|
||||||
|
self._max_kelvin
|
||||||
|
)
|
||||||
|
self._attr_unique_id = self._device_unique_id()
|
||||||
|
|
||||||
def _device_unique_id(self) -> str:
|
def _device_unique_id(self) -> str:
|
||||||
"""Return unique id for this device."""
|
"""Return unique id for this device."""
|
||||||
@ -311,16 +310,6 @@ class KNXLight(KnxEntity, LightEntity):
|
|||||||
)
|
)
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@property
|
|
||||||
def min_mireds(self) -> int:
|
|
||||||
"""Return the coldest color temp this light supports in mireds."""
|
|
||||||
return self._min_mireds
|
|
||||||
|
|
||||||
@property
|
|
||||||
def max_mireds(self) -> int:
|
|
||||||
"""Return the warmest color temp this light supports in mireds."""
|
|
||||||
return self._max_mireds
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def color_mode(self) -> str | None:
|
def color_mode(self) -> str | None:
|
||||||
"""Return the color mode of the light."""
|
"""Return the color mode of the light."""
|
||||||
|
@ -27,7 +27,6 @@ async def async_setup_platform(
|
|||||||
"""Set up number entities for KNX platform."""
|
"""Set up number entities for KNX platform."""
|
||||||
if not discovery_info or not discovery_info["platform_config"]:
|
if not discovery_info or not discovery_info["platform_config"]:
|
||||||
return
|
return
|
||||||
|
|
||||||
platform_config = discovery_info["platform_config"]
|
platform_config = discovery_info["platform_config"]
|
||||||
xknx: XKNX = hass.data[DOMAIN].xknx
|
xknx: XKNX = hass.data[DOMAIN].xknx
|
||||||
|
|
||||||
@ -51,25 +50,25 @@ def _create_numeric_value(xknx: XKNX, config: ConfigType) -> NumericValue:
|
|||||||
class KNXNumber(KnxEntity, NumberEntity, RestoreEntity):
|
class KNXNumber(KnxEntity, NumberEntity, RestoreEntity):
|
||||||
"""Representation of a KNX number."""
|
"""Representation of a KNX number."""
|
||||||
|
|
||||||
|
_device: NumericValue
|
||||||
|
|
||||||
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
||||||
"""Initialize a KNX number."""
|
"""Initialize a KNX number."""
|
||||||
self._device: NumericValue
|
|
||||||
super().__init__(_create_numeric_value(xknx, config))
|
super().__init__(_create_numeric_value(xknx, config))
|
||||||
self._unique_id = f"{self._device.sensor_value.group_address}"
|
|
||||||
|
|
||||||
self._attr_min_value = config.get(
|
|
||||||
NumberSchema.CONF_MIN,
|
|
||||||
self._device.sensor_value.dpt_class.value_min,
|
|
||||||
)
|
|
||||||
self._attr_max_value = config.get(
|
self._attr_max_value = config.get(
|
||||||
NumberSchema.CONF_MAX,
|
NumberSchema.CONF_MAX,
|
||||||
self._device.sensor_value.dpt_class.value_max,
|
self._device.sensor_value.dpt_class.value_max,
|
||||||
)
|
)
|
||||||
|
self._attr_min_value = config.get(
|
||||||
|
NumberSchema.CONF_MIN,
|
||||||
|
self._device.sensor_value.dpt_class.value_min,
|
||||||
|
)
|
||||||
self._attr_step = config.get(
|
self._attr_step = config.get(
|
||||||
NumberSchema.CONF_STEP,
|
NumberSchema.CONF_STEP,
|
||||||
self._device.sensor_value.dpt_class.resolution,
|
self._device.sensor_value.dpt_class.resolution,
|
||||||
)
|
)
|
||||||
self._device.sensor_value.value = max(0, self.min_value)
|
self._attr_unique_id = str(self._device.sensor_value.group_address)
|
||||||
|
self._device.sensor_value.value = max(0, self._attr_min_value)
|
||||||
|
|
||||||
async def async_added_to_hass(self) -> None:
|
async def async_added_to_hass(self) -> None:
|
||||||
"""Restore last state."""
|
"""Restore last state."""
|
||||||
|
@ -26,23 +26,21 @@ async def async_setup_platform(
|
|||||||
"""Set up the scenes for KNX platform."""
|
"""Set up the scenes for KNX platform."""
|
||||||
if not discovery_info or not discovery_info["platform_config"]:
|
if not discovery_info or not discovery_info["platform_config"]:
|
||||||
return
|
return
|
||||||
|
|
||||||
platform_config = discovery_info["platform_config"]
|
platform_config = discovery_info["platform_config"]
|
||||||
xknx: XKNX = hass.data[DOMAIN].xknx
|
xknx: XKNX = hass.data[DOMAIN].xknx
|
||||||
|
|
||||||
entities = []
|
async_add_entities(
|
||||||
for entity_config in platform_config:
|
KNXScene(xknx, entity_config) for entity_config in platform_config
|
||||||
entities.append(KNXScene(xknx, entity_config))
|
)
|
||||||
|
|
||||||
async_add_entities(entities)
|
|
||||||
|
|
||||||
|
|
||||||
class KNXScene(KnxEntity, Scene):
|
class KNXScene(KnxEntity, Scene):
|
||||||
"""Representation of a KNX scene."""
|
"""Representation of a KNX scene."""
|
||||||
|
|
||||||
|
_device: XknxScene
|
||||||
|
|
||||||
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
||||||
"""Init KNX scene."""
|
"""Init KNX scene."""
|
||||||
self._device: XknxScene
|
|
||||||
super().__init__(
|
super().__init__(
|
||||||
device=XknxScene(
|
device=XknxScene(
|
||||||
xknx,
|
xknx,
|
||||||
@ -51,7 +49,7 @@ class KNXScene(KnxEntity, Scene):
|
|||||||
scene_number=config[SceneSchema.CONF_SCENE_NUMBER],
|
scene_number=config[SceneSchema.CONF_SCENE_NUMBER],
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
self._unique_id = (
|
self._attr_unique_id = (
|
||||||
f"{self._device.scene_value.group_address}_{self._device.scene_number}"
|
f"{self._device.scene_value.group_address}_{self._device.scene_number}"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -28,10 +28,9 @@ async def async_setup_platform(
|
|||||||
async_add_entities: AddEntitiesCallback,
|
async_add_entities: AddEntitiesCallback,
|
||||||
discovery_info: DiscoveryInfoType | None = None,
|
discovery_info: DiscoveryInfoType | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Set up number entities for KNX platform."""
|
"""Set up select entities for KNX platform."""
|
||||||
if not discovery_info or not discovery_info["platform_config"]:
|
if not discovery_info or not discovery_info["platform_config"]:
|
||||||
return
|
return
|
||||||
|
|
||||||
platform_config = discovery_info["platform_config"]
|
platform_config = discovery_info["platform_config"]
|
||||||
xknx: XKNX = hass.data[DOMAIN].xknx
|
xknx: XKNX = hass.data[DOMAIN].xknx
|
||||||
|
|
||||||
@ -54,20 +53,20 @@ def _create_raw_value(xknx: XKNX, config: ConfigType) -> RawValue:
|
|||||||
|
|
||||||
|
|
||||||
class KNXSelect(KnxEntity, SelectEntity, RestoreEntity):
|
class KNXSelect(KnxEntity, SelectEntity, RestoreEntity):
|
||||||
"""Representation of a KNX number."""
|
"""Representation of a KNX select."""
|
||||||
|
|
||||||
|
_device: RawValue
|
||||||
|
|
||||||
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
||||||
"""Initialize a KNX number."""
|
"""Initialize a KNX select."""
|
||||||
self._device: RawValue
|
|
||||||
super().__init__(_create_raw_value(xknx, config))
|
super().__init__(_create_raw_value(xknx, config))
|
||||||
self._unique_id = f"{self._device.remote_value.group_address}"
|
|
||||||
|
|
||||||
self._option_payloads: dict[str, int] = {
|
self._option_payloads: dict[str, int] = {
|
||||||
option[SelectSchema.CONF_OPTION]: option[SelectSchema.CONF_PAYLOAD]
|
option[SelectSchema.CONF_OPTION]: option[SelectSchema.CONF_PAYLOAD]
|
||||||
for option in config[SelectSchema.CONF_OPTIONS]
|
for option in config[SelectSchema.CONF_OPTIONS]
|
||||||
}
|
}
|
||||||
self._attr_options = list(self._option_payloads)
|
self._attr_options = list(self._option_payloads)
|
||||||
self._attr_current_option = None
|
self._attr_current_option = None
|
||||||
|
self._attr_unique_id = str(self._device.remote_value.group_address)
|
||||||
|
|
||||||
async def async_added_to_hass(self) -> None:
|
async def async_added_to_hass(self) -> None:
|
||||||
"""Restore last state."""
|
"""Restore last state."""
|
||||||
@ -98,7 +97,7 @@ class KNXSelect(KnxEntity, SelectEntity, RestoreEntity):
|
|||||||
|
|
||||||
async def async_select_option(self, option: str) -> None:
|
async def async_select_option(self, option: str) -> None:
|
||||||
"""Change the selected option."""
|
"""Change the selected option."""
|
||||||
if payload := self._option_payloads.get(option):
|
payload = self._option_payloads.get(option)
|
||||||
await self._device.set(payload)
|
if payload is None:
|
||||||
return
|
raise ValueError(f"Invalid option for {self.entity_id}: {option}")
|
||||||
raise ValueError(f"Invalid option for {self.entity_id}: {option}")
|
await self._device.set(payload)
|
||||||
|
@ -27,15 +27,12 @@ async def async_setup_platform(
|
|||||||
"""Set up sensor(s) for KNX platform."""
|
"""Set up sensor(s) for KNX platform."""
|
||||||
if not discovery_info or not discovery_info["platform_config"]:
|
if not discovery_info or not discovery_info["platform_config"]:
|
||||||
return
|
return
|
||||||
|
|
||||||
platform_config = discovery_info["platform_config"]
|
platform_config = discovery_info["platform_config"]
|
||||||
xknx: XKNX = hass.data[DOMAIN].xknx
|
xknx: XKNX = hass.data[DOMAIN].xknx
|
||||||
|
|
||||||
entities = []
|
async_add_entities(
|
||||||
for entity_config in platform_config:
|
KNXSensor(xknx, entity_config) for entity_config in platform_config
|
||||||
entities.append(KNXSensor(xknx, entity_config))
|
)
|
||||||
|
|
||||||
async_add_entities(entities)
|
|
||||||
|
|
||||||
|
|
||||||
def _create_sensor(xknx: XKNX, config: ConfigType) -> XknxSensor:
|
def _create_sensor(xknx: XKNX, config: ConfigType) -> XknxSensor:
|
||||||
@ -53,30 +50,25 @@ def _create_sensor(xknx: XKNX, config: ConfigType) -> XknxSensor:
|
|||||||
class KNXSensor(KnxEntity, SensorEntity):
|
class KNXSensor(KnxEntity, SensorEntity):
|
||||||
"""Representation of a KNX sensor."""
|
"""Representation of a KNX sensor."""
|
||||||
|
|
||||||
|
_device: XknxSensor
|
||||||
|
|
||||||
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
||||||
"""Initialize of a KNX sensor."""
|
"""Initialize of a KNX sensor."""
|
||||||
self._device: XknxSensor
|
|
||||||
super().__init__(_create_sensor(xknx, config))
|
super().__init__(_create_sensor(xknx, config))
|
||||||
self._unique_id = f"{self._device.sensor_value.group_address_state}"
|
self._attr_device_class = (
|
||||||
|
self._device.ha_device_class()
|
||||||
|
if self._device.ha_device_class() in DEVICE_CLASSES
|
||||||
|
else None
|
||||||
|
)
|
||||||
|
self._attr_force_update = self._device.always_callback
|
||||||
|
self._attr_unique_id = str(self._device.sensor_value.group_address_state)
|
||||||
|
self._attr_unit_of_measurement = self._device.unit_of_measurement()
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def state(self) -> StateType:
|
def state(self) -> StateType:
|
||||||
"""Return the state of the sensor."""
|
"""Return the state of the sensor."""
|
||||||
return self._device.resolve_state()
|
return self._device.resolve_state()
|
||||||
|
|
||||||
@property
|
|
||||||
def unit_of_measurement(self) -> str | None:
|
|
||||||
"""Return the unit this state is expressed in."""
|
|
||||||
return self._device.unit_of_measurement()
|
|
||||||
|
|
||||||
@property
|
|
||||||
def device_class(self) -> str | None:
|
|
||||||
"""Return the device class of the sensor."""
|
|
||||||
device_class = self._device.ha_device_class()
|
|
||||||
if device_class in DEVICE_CLASSES:
|
|
||||||
return device_class
|
|
||||||
return None
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def extra_state_attributes(self) -> dict[str, Any] | None:
|
def extra_state_attributes(self) -> dict[str, Any] | None:
|
||||||
"""Return device specific state attributes."""
|
"""Return device specific state attributes."""
|
||||||
@ -88,13 +80,3 @@ class KNXSensor(KnxEntity, SensorEntity):
|
|||||||
dt.as_utc(self._device.last_telegram.timestamp)
|
dt.as_utc(self._device.last_telegram.timestamp)
|
||||||
)
|
)
|
||||||
return attr
|
return attr
|
||||||
|
|
||||||
@property
|
|
||||||
def force_update(self) -> bool:
|
|
||||||
"""
|
|
||||||
Return True if state updates should be forced.
|
|
||||||
|
|
||||||
If True, a state change will be triggered anytime the state property is
|
|
||||||
updated, not just when the value changes.
|
|
||||||
"""
|
|
||||||
return self._device.always_callback
|
|
||||||
|
@ -27,23 +27,21 @@ async def async_setup_platform(
|
|||||||
"""Set up switch(es) for KNX platform."""
|
"""Set up switch(es) for KNX platform."""
|
||||||
if not discovery_info or not discovery_info["platform_config"]:
|
if not discovery_info or not discovery_info["platform_config"]:
|
||||||
return
|
return
|
||||||
|
|
||||||
platform_config = discovery_info["platform_config"]
|
platform_config = discovery_info["platform_config"]
|
||||||
xknx: XKNX = hass.data[DOMAIN].xknx
|
xknx: XKNX = hass.data[DOMAIN].xknx
|
||||||
|
|
||||||
entities = []
|
async_add_entities(
|
||||||
for entity_config in platform_config:
|
KNXSwitch(xknx, entity_config) for entity_config in platform_config
|
||||||
entities.append(KNXSwitch(xknx, entity_config))
|
)
|
||||||
|
|
||||||
async_add_entities(entities)
|
|
||||||
|
|
||||||
|
|
||||||
class KNXSwitch(KnxEntity, SwitchEntity, RestoreEntity):
|
class KNXSwitch(KnxEntity, SwitchEntity, RestoreEntity):
|
||||||
"""Representation of a KNX switch."""
|
"""Representation of a KNX switch."""
|
||||||
|
|
||||||
|
_device: XknxSwitch
|
||||||
|
|
||||||
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
||||||
"""Initialize of KNX switch."""
|
"""Initialize of KNX switch."""
|
||||||
self._device: XknxSwitch
|
|
||||||
super().__init__(
|
super().__init__(
|
||||||
device=XknxSwitch(
|
device=XknxSwitch(
|
||||||
xknx,
|
xknx,
|
||||||
@ -53,7 +51,7 @@ class KNXSwitch(KnxEntity, SwitchEntity, RestoreEntity):
|
|||||||
invert=config[SwitchSchema.CONF_INVERT],
|
invert=config[SwitchSchema.CONF_INVERT],
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
self._unique_id = f"{self._device.switch.group_address}"
|
self._attr_unique_id = str(self._device.switch.group_address)
|
||||||
|
|
||||||
async def async_added_to_hass(self) -> None:
|
async def async_added_to_hass(self) -> None:
|
||||||
"""Restore last state."""
|
"""Restore last state."""
|
||||||
|
@ -24,15 +24,12 @@ async def async_setup_platform(
|
|||||||
"""Set up weather entities for KNX platform."""
|
"""Set up weather entities for KNX platform."""
|
||||||
if not discovery_info or not discovery_info["platform_config"]:
|
if not discovery_info or not discovery_info["platform_config"]:
|
||||||
return
|
return
|
||||||
|
|
||||||
platform_config = discovery_info["platform_config"]
|
platform_config = discovery_info["platform_config"]
|
||||||
xknx: XKNX = hass.data[DOMAIN].xknx
|
xknx: XKNX = hass.data[DOMAIN].xknx
|
||||||
|
|
||||||
entities = []
|
async_add_entities(
|
||||||
for entity_config in platform_config:
|
KNXWeather(xknx, entity_config) for entity_config in platform_config
|
||||||
entities.append(KNXWeather(xknx, entity_config))
|
)
|
||||||
|
|
||||||
async_add_entities(entities)
|
|
||||||
|
|
||||||
|
|
||||||
def _create_weather(xknx: XKNX, config: ConfigType) -> XknxWeather:
|
def _create_weather(xknx: XKNX, config: ConfigType) -> XknxWeather:
|
||||||
@ -74,22 +71,19 @@ def _create_weather(xknx: XKNX, config: ConfigType) -> XknxWeather:
|
|||||||
class KNXWeather(KnxEntity, WeatherEntity):
|
class KNXWeather(KnxEntity, WeatherEntity):
|
||||||
"""Representation of a KNX weather device."""
|
"""Representation of a KNX weather device."""
|
||||||
|
|
||||||
|
_device: XknxWeather
|
||||||
|
_attr_temperature_unit = TEMP_CELSIUS
|
||||||
|
|
||||||
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
|
||||||
"""Initialize of a KNX sensor."""
|
"""Initialize of a KNX sensor."""
|
||||||
self._device: XknxWeather
|
|
||||||
super().__init__(_create_weather(xknx, config))
|
super().__init__(_create_weather(xknx, config))
|
||||||
self._unique_id = f"{self._device._temperature.group_address_state}"
|
self._attr_unique_id = str(self._device._temperature.group_address_state)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def temperature(self) -> float | None:
|
def temperature(self) -> float | None:
|
||||||
"""Return current temperature."""
|
"""Return current temperature."""
|
||||||
return self._device.temperature
|
return self._device.temperature
|
||||||
|
|
||||||
@property
|
|
||||||
def temperature_unit(self) -> str:
|
|
||||||
"""Return temperature unit."""
|
|
||||||
return TEMP_CELSIUS
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def pressure(self) -> float | None:
|
def pressure(self) -> float | None:
|
||||||
"""Return current air pressure."""
|
"""Return current air pressure."""
|
||||||
|
Loading…
x
Reference in New Issue
Block a user