Handle LCN entity instances only in corresponding platform (#124589)

* Handle switch entity instance only in switch platform

* Add other platforms
This commit is contained in:
Andre Lengwenus 2024-08-26 10:17:51 +02:00 committed by GitHub
parent 302ffe5e56
commit 7ddd755acc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 281 additions and 139 deletions

View File

@ -1,6 +1,7 @@
"""Support for LCN binary sensors.""" """Support for LCN binary sensors."""
from __future__ import annotations from collections.abc import Iterable
from functools import partial
import pypck import pypck
@ -25,22 +26,37 @@ from .const import (
from .helpers import DeviceConnectionType, InputType, get_device_connection from .helpers import DeviceConnectionType, InputType, get_device_connection
def create_lcn_binary_sensor_entity( def add_lcn_entities(
hass: HomeAssistant, entity_config: ConfigType, config_entry: ConfigEntry hass: HomeAssistant,
) -> LcnEntity: config_entry: ConfigEntry,
"""Set up an entity for this domain.""" async_add_entities: AddEntitiesCallback,
device_connection = get_device_connection( entity_configs: Iterable[ConfigType],
hass, entity_config[CONF_ADDRESS], config_entry ) -> None:
) """Add entities for this domain."""
entities: list[LcnRegulatorLockSensor | LcnBinarySensor | LcnLockKeysSensor] = []
if entity_config[CONF_DOMAIN_DATA][CONF_SOURCE] in SETPOINTS: for entity_config in entity_configs:
return LcnRegulatorLockSensor( device_connection = get_device_connection(
entity_config, config_entry.entry_id, device_connection hass, entity_config[CONF_ADDRESS], config_entry
) )
if entity_config[CONF_DOMAIN_DATA][CONF_SOURCE] in BINSENSOR_PORTS:
return LcnBinarySensor(entity_config, config_entry.entry_id, device_connection) if entity_config[CONF_DOMAIN_DATA][CONF_SOURCE] in SETPOINTS:
# in KEY entities.append(
return LcnLockKeysSensor(entity_config, config_entry.entry_id, device_connection) LcnRegulatorLockSensor(
entity_config, config_entry.entry_id, device_connection
)
)
elif entity_config[CONF_DOMAIN_DATA][CONF_SOURCE] in BINSENSOR_PORTS:
entities.append(
LcnBinarySensor(entity_config, config_entry.entry_id, device_connection)
)
else: # in KEY
entities.append(
LcnLockKeysSensor(
entity_config, config_entry.entry_id, device_connection
)
)
async_add_entities(entities)
async def async_setup_entry( async def async_setup_entry(
@ -49,14 +65,23 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up LCN switch entities from a config entry.""" """Set up LCN switch entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update( add_entities = partial(
{DOMAIN_BINARY_SENSOR: (async_add_entities, create_lcn_binary_sensor_entity)} add_lcn_entities,
hass,
config_entry,
async_add_entities,
) )
async_add_entities( hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
create_lcn_binary_sensor_entity(hass, entity_config, config_entry) {DOMAIN_BINARY_SENSOR: add_entities}
for entity_config in config_entry.data[CONF_ENTITIES] )
if entity_config[CONF_DOMAIN] == DOMAIN_BINARY_SENSOR
add_entities(
(
entity_config
for entity_config in config_entry.data[CONF_ENTITIES]
if entity_config[CONF_DOMAIN] == DOMAIN_BINARY_SENSOR
),
) )

View File

@ -1,7 +1,7 @@
"""Support for LCN climate control.""" """Support for LCN climate control."""
from __future__ import annotations from collections.abc import Iterable
from functools import partial
from typing import Any, cast from typing import Any, cast
import pypck import pypck
@ -41,15 +41,24 @@ from .helpers import DeviceConnectionType, InputType, get_device_connection
PARALLEL_UPDATES = 0 PARALLEL_UPDATES = 0
def create_lcn_climate_entity( def add_lcn_entities(
hass: HomeAssistant, entity_config: ConfigType, config_entry: ConfigEntry hass: HomeAssistant,
) -> LcnEntity: config_entry: ConfigEntry,
"""Set up an entity for this domain.""" async_add_entities: AddEntitiesCallback,
device_connection = get_device_connection( entity_configs: Iterable[ConfigType],
hass, entity_config[CONF_ADDRESS], config_entry ) -> None:
) """Add entities for this domain."""
entities: list[LcnClimate] = []
for entity_config in entity_configs:
device_connection = get_device_connection(
hass, entity_config[CONF_ADDRESS], config_entry
)
return LcnClimate(entity_config, config_entry.entry_id, device_connection) entities.append(
LcnClimate(entity_config, config_entry.entry_id, device_connection)
)
async_add_entities(entities)
async def async_setup_entry( async def async_setup_entry(
@ -58,14 +67,23 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up LCN switch entities from a config entry.""" """Set up LCN switch entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update( add_entities = partial(
{DOMAIN_CLIMATE: (async_add_entities, create_lcn_climate_entity)} add_lcn_entities,
hass,
config_entry,
async_add_entities,
) )
async_add_entities( hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
create_lcn_climate_entity(hass, entity_config, config_entry) {DOMAIN_CLIMATE: add_entities}
for entity_config in config_entry.data[CONF_ENTITIES] )
if entity_config[CONF_DOMAIN] == DOMAIN_CLIMATE
add_entities(
(
entity_config
for entity_config in config_entry.data[CONF_ENTITIES]
if entity_config[CONF_DOMAIN] == DOMAIN_CLIMATE
),
) )

View File

@ -1,7 +1,7 @@
"""Support for LCN covers.""" """Support for LCN covers."""
from __future__ import annotations from collections.abc import Iterable
from functools import partial
from typing import Any from typing import Any
import pypck import pypck
@ -26,18 +26,29 @@ from .helpers import DeviceConnectionType, InputType, get_device_connection
PARALLEL_UPDATES = 0 PARALLEL_UPDATES = 0
def create_lcn_cover_entity( def add_lcn_entities(
hass: HomeAssistant, entity_config: ConfigType, config_entry: ConfigEntry hass: HomeAssistant,
) -> LcnEntity: config_entry: ConfigEntry,
"""Set up an entity for this domain.""" async_add_entities: AddEntitiesCallback,
device_connection = get_device_connection( entity_configs: Iterable[ConfigType],
hass, entity_config[CONF_ADDRESS], config_entry ) -> None:
) """Add entities for this domain."""
entities: list[LcnOutputsCover | LcnRelayCover] = []
for entity_config in entity_configs:
device_connection = get_device_connection(
hass, entity_config[CONF_ADDRESS], config_entry
)
if entity_config[CONF_DOMAIN_DATA][CONF_MOTOR] in "OUTPUTS": if entity_config[CONF_DOMAIN_DATA][CONF_MOTOR] in "OUTPUTS":
return LcnOutputsCover(entity_config, config_entry.entry_id, device_connection) entities.append(
# in RELAYS LcnOutputsCover(entity_config, config_entry.entry_id, device_connection)
return LcnRelayCover(entity_config, config_entry.entry_id, device_connection) )
else: # in RELAYS
entities.append(
LcnRelayCover(entity_config, config_entry.entry_id, device_connection)
)
async_add_entities(entities)
async def async_setup_entry( async def async_setup_entry(
@ -46,14 +57,23 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up LCN cover entities from a config entry.""" """Set up LCN cover entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update( add_entities = partial(
{DOMAIN_COVER: (async_add_entities, create_lcn_cover_entity)} add_lcn_entities,
hass,
config_entry,
async_add_entities,
) )
async_add_entities( hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
create_lcn_cover_entity(hass, entity_config, config_entry) {DOMAIN_COVER: add_entities}
for entity_config in config_entry.data[CONF_ENTITIES] )
if entity_config[CONF_DOMAIN] == DOMAIN_COVER
add_entities(
(
entity_config
for entity_config in config_entry.data[CONF_ENTITIES]
if entity_config[CONF_DOMAIN] == DOMAIN_COVER
),
) )

View File

@ -1,7 +1,7 @@
"""Support for LCN lights.""" """Support for LCN lights."""
from __future__ import annotations from collections.abc import Iterable
from functools import partial
from typing import Any from typing import Any
import pypck import pypck
@ -35,18 +35,29 @@ from .helpers import DeviceConnectionType, InputType, get_device_connection
PARALLEL_UPDATES = 0 PARALLEL_UPDATES = 0
def create_lcn_light_entity( def add_lcn_entities(
hass: HomeAssistant, entity_config: ConfigType, config_entry: ConfigEntry hass: HomeAssistant,
) -> LcnEntity: config_entry: ConfigEntry,
"""Set up an entity for this domain.""" async_add_entities: AddEntitiesCallback,
device_connection = get_device_connection( entity_configs: Iterable[ConfigType],
hass, entity_config[CONF_ADDRESS], config_entry ) -> None:
) """Add entities for this domain."""
entities: list[LcnOutputLight | LcnRelayLight] = []
for entity_config in entity_configs:
device_connection = get_device_connection(
hass, entity_config[CONF_ADDRESS], config_entry
)
if entity_config[CONF_DOMAIN_DATA][CONF_OUTPUT] in OUTPUT_PORTS: if entity_config[CONF_DOMAIN_DATA][CONF_OUTPUT] in OUTPUT_PORTS:
return LcnOutputLight(entity_config, config_entry.entry_id, device_connection) entities.append(
# in RELAY_PORTS LcnOutputLight(entity_config, config_entry.entry_id, device_connection)
return LcnRelayLight(entity_config, config_entry.entry_id, device_connection) )
else: # in RELAY_PORTS
entities.append(
LcnRelayLight(entity_config, config_entry.entry_id, device_connection)
)
async_add_entities(entities)
async def async_setup_entry( async def async_setup_entry(
@ -55,14 +66,23 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up LCN light entities from a config entry.""" """Set up LCN light entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update( add_entities = partial(
{DOMAIN_LIGHT: (async_add_entities, create_lcn_light_entity)} add_lcn_entities,
hass,
config_entry,
async_add_entities,
) )
async_add_entities( hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
create_lcn_light_entity(hass, entity_config, config_entry) {DOMAIN_LIGHT: add_entities}
for entity_config in config_entry.data[CONF_ENTITIES] )
if entity_config[CONF_DOMAIN] == DOMAIN_LIGHT
add_entities(
(
entity_config
for entity_config in config_entry.data[CONF_ENTITIES]
if entity_config[CONF_DOMAIN] == DOMAIN_LIGHT
),
) )

View File

@ -1,7 +1,7 @@
"""Support for LCN scenes.""" """Support for LCN scenes."""
from __future__ import annotations from collections.abc import Iterable
from functools import partial
from typing import Any from typing import Any
import pypck import pypck
@ -28,15 +28,24 @@ from .helpers import DeviceConnectionType, get_device_connection
PARALLEL_UPDATES = 0 PARALLEL_UPDATES = 0
def create_lcn_scene_entity( def add_lcn_entities(
hass: HomeAssistant, entity_config: ConfigType, config_entry: ConfigEntry hass: HomeAssistant,
) -> LcnEntity: config_entry: ConfigEntry,
"""Set up an entity for this domain.""" async_add_entities: AddEntitiesCallback,
device_connection = get_device_connection( entity_configs: Iterable[ConfigType],
hass, entity_config[CONF_ADDRESS], config_entry ) -> None:
) """Add entities for this domain."""
entities: list[LcnScene] = []
for entity_config in entity_configs:
device_connection = get_device_connection(
hass, entity_config[CONF_ADDRESS], config_entry
)
return LcnScene(entity_config, config_entry.entry_id, device_connection) entities.append(
LcnScene(entity_config, config_entry.entry_id, device_connection)
)
async_add_entities(entities)
async def async_setup_entry( async def async_setup_entry(
@ -45,14 +54,23 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up LCN switch entities from a config entry.""" """Set up LCN switch entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update( add_entities = partial(
{DOMAIN_SCENE: (async_add_entities, create_lcn_scene_entity)} add_lcn_entities,
hass,
config_entry,
async_add_entities,
) )
async_add_entities( hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
create_lcn_scene_entity(hass, entity_config, config_entry) {DOMAIN_SCENE: add_entities}
for entity_config in config_entry.data[CONF_ENTITIES] )
if entity_config[CONF_DOMAIN] == DOMAIN_SCENE
add_entities(
(
entity_config
for entity_config in config_entry.data[CONF_ENTITIES]
if entity_config[CONF_DOMAIN] == DOMAIN_SCENE
),
) )

View File

@ -1,7 +1,7 @@
"""Support for LCN sensors.""" """Support for LCN sensors."""
from __future__ import annotations from collections.abc import Iterable
from functools import partial
from itertools import chain from itertools import chain
from typing import cast from typing import cast
@ -34,22 +34,35 @@ from .const import (
from .helpers import DeviceConnectionType, InputType, get_device_connection from .helpers import DeviceConnectionType, InputType, get_device_connection
def create_lcn_sensor_entity( def add_lcn_entities(
hass: HomeAssistant, entity_config: ConfigType, config_entry: ConfigEntry hass: HomeAssistant,
) -> LcnEntity: config_entry: ConfigEntry,
"""Set up an entity for this domain.""" async_add_entities: AddEntitiesCallback,
device_connection = get_device_connection( entity_configs: Iterable[ConfigType],
hass, entity_config[CONF_ADDRESS], config_entry ) -> None:
) """Add entities for this domain."""
entities: list[LcnVariableSensor | LcnLedLogicSensor] = []
if entity_config[CONF_DOMAIN_DATA][CONF_SOURCE] in chain( for entity_config in entity_configs:
VARIABLES, SETPOINTS, THRESHOLDS, S0_INPUTS device_connection = get_device_connection(
): hass, entity_config[CONF_ADDRESS], config_entry
return LcnVariableSensor(
entity_config, config_entry.entry_id, device_connection
) )
# in LED_PORTS + LOGICOP_PORTS
return LcnLedLogicSensor(entity_config, config_entry.entry_id, device_connection) if entity_config[CONF_DOMAIN_DATA][CONF_SOURCE] in chain(
VARIABLES, SETPOINTS, THRESHOLDS, S0_INPUTS
):
entities.append(
LcnVariableSensor(
entity_config, config_entry.entry_id, device_connection
)
)
else: # in LED_PORTS + LOGICOP_PORTS
entities.append(
LcnLedLogicSensor(
entity_config, config_entry.entry_id, device_connection
)
)
async_add_entities(entities)
async def async_setup_entry( async def async_setup_entry(
@ -58,14 +71,23 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up LCN switch entities from a config entry.""" """Set up LCN switch entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update( add_entities = partial(
{DOMAIN_SENSOR: (async_add_entities, create_lcn_sensor_entity)} add_lcn_entities,
hass,
config_entry,
async_add_entities,
) )
async_add_entities( hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
create_lcn_sensor_entity(hass, entity_config, config_entry) {DOMAIN_SENSOR: add_entities}
for entity_config in config_entry.data[CONF_ENTITIES] )
if entity_config[CONF_DOMAIN] == DOMAIN_SENSOR
add_entities(
(
entity_config
for entity_config in config_entry.data[CONF_ENTITIES]
if entity_config[CONF_DOMAIN] == DOMAIN_SENSOR
),
) )

View File

@ -1,7 +1,7 @@
"""Support for LCN switches.""" """Support for LCN switches."""
from __future__ import annotations from collections.abc import Iterable
from functools import partial
from typing import Any from typing import Any
import pypck import pypck
@ -26,18 +26,29 @@ from .helpers import DeviceConnectionType, InputType, get_device_connection
PARALLEL_UPDATES = 0 PARALLEL_UPDATES = 0
def create_lcn_switch_entity( def add_lcn_switch_entities(
hass: HomeAssistant, entity_config: ConfigType, config_entry: ConfigEntry hass: HomeAssistant,
) -> LcnEntity: config_entry: ConfigEntry,
"""Set up an entity for this domain.""" async_add_entities: AddEntitiesCallback,
device_connection = get_device_connection( entity_configs: Iterable[ConfigType],
hass, entity_config[CONF_ADDRESS], config_entry ) -> None:
) """Add entities for this domain."""
entities: list[LcnOutputSwitch | LcnRelaySwitch] = []
for entity_config in entity_configs:
device_connection = get_device_connection(
hass, entity_config[CONF_ADDRESS], config_entry
)
if entity_config[CONF_DOMAIN_DATA][CONF_OUTPUT] in OUTPUT_PORTS: if entity_config[CONF_DOMAIN_DATA][CONF_OUTPUT] in OUTPUT_PORTS:
return LcnOutputSwitch(entity_config, config_entry.entry_id, device_connection) entities.append(
# in RELAY_PORTS LcnOutputSwitch(entity_config, config_entry.entry_id, device_connection)
return LcnRelaySwitch(entity_config, config_entry.entry_id, device_connection) )
else: # in RELAY_PORTS
entities.append(
LcnRelaySwitch(entity_config, config_entry.entry_id, device_connection)
)
async_add_entities(entities)
async def async_setup_entry( async def async_setup_entry(
@ -46,14 +57,23 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up LCN switch entities from a config entry.""" """Set up LCN switch entities from a config entry."""
hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update( add_entities = partial(
{DOMAIN_SWITCH: (async_add_entities, create_lcn_switch_entity)} add_lcn_switch_entities,
hass,
config_entry,
async_add_entities,
) )
async_add_entities( hass.data[DOMAIN][config_entry.entry_id][ADD_ENTITIES_CALLBACKS].update(
create_lcn_switch_entity(hass, entity_config, config_entry) {DOMAIN_SWITCH: add_entities}
for entity_config in config_entry.data[CONF_ENTITIES] )
if entity_config[CONF_DOMAIN] == DOMAIN_SWITCH
add_entities(
(
entity_config
for entity_config in config_entry.data[CONF_ENTITIES]
if entity_config[CONF_DOMAIN] == DOMAIN_SWITCH
),
) )

View File

@ -341,11 +341,10 @@ async def websocket_add_entity(
} }
# Create new entity and add to corresponding component # Create new entity and add to corresponding component
callbacks = hass.data[DOMAIN][msg["entry_id"]][ADD_ENTITIES_CALLBACKS] add_entities = hass.data[DOMAIN][msg["entry_id"]][ADD_ENTITIES_CALLBACKS][
async_add_entities, create_lcn_entity = callbacks[msg[CONF_DOMAIN]] msg[CONF_DOMAIN]
]
entity = create_lcn_entity(hass, entity_config, config_entry) add_entities([entity_config])
async_add_entities([entity])
# Add entity config to config_entry # Add entity config to config_entry
entity_configs = [*config_entry.data[CONF_ENTITIES], entity_config] entity_configs = [*config_entry.data[CONF_ENTITIES], entity_config]