evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Migrate LCN configuration to ConfigEntry (Part 1) (#44090)

Browse source
This commit is contained in:
Andre Lengwenus 2021-03-15 13:45:13 +01:00 • committed by GitHub
parent fb34d1a56e
commit 50b7b1cc7a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
17 changed files with 885 additions and 350 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
.coveragerc
View file

@ -505,7 +505,18 @@ omit =
homeassistant/components/lastfm/sensor.py
homeassistant/components/launch_library/const.py
homeassistant/components/launch_library/sensor.py
homeassistant/components/lcn/*
homeassistant/components/lcn/__init__.py
homeassistant/components/lcn/binary_sensor.py
homeassistant/components/lcn/climate.py
homeassistant/components/lcn/const.py
homeassistant/components/lcn/cover.py
homeassistant/components/lcn/helpers.py
homeassistant/components/lcn/light.py
homeassistant/components/lcn/scene.py
homeassistant/components/lcn/schemas.py
homeassistant/components/lcn/sensor.py
homeassistant/components/lcn/services.py
homeassistant/components/lcn/switch.py
homeassistant/components/lg_netcast/media_player.py
homeassistant/components/lg_soundbar/media_player.py
homeassistant/components/life360/*

232
homeassistant/components/lcn/__init__.py
View file

@ -1,137 +1,162 @@
"""Support for LCN devices."""
import asyncio
import logging
import pypck
from homeassistant import config_entries
from homeassistant.const import (
CONF_BINARY_SENSORS,
CONF_COVERS,
CONF_HOST,
CONF_LIGHTS,
CONF_IP_ADDRESS,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_SENSORS,
CONF_SWITCHES,
CONF_RESOURCE,
CONF_USERNAME,
)
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity import Entity
from .const import (
CONF_CLIMATES,
CONF_CONNECTIONS,
CONF_DIM_MODE,
CONF_SCENES,
CONF_SK_NUM_TRIES,
DATA_LCN,
DOMAIN,
)
from .const import CONF_DIM_MODE, CONF_SK_NUM_TRIES, CONNECTION, DOMAIN
from .helpers import generate_unique_id, import_lcn_config
from .schemas import CONFIG_SCHEMA # noqa: F401
from .services import (
DynText,
Led,
LockKeys,
LockRegulator,
OutputAbs,
OutputRel,
OutputToggle,
Pck,
Relays,
SendKeys,
VarAbs,
VarRel,
VarReset,
)
from .services import SERVICES
PLATFORMS = ["binary_sensor", "climate", "cover", "light", "scene", "sensor", "switch"]
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass, config):
"""Set up the LCN component."""
hass.data[DATA_LCN] = {}
if DOMAIN not in config:
return True
conf_connections = config[DOMAIN][CONF_CONNECTIONS]
connections = []
for conf_connection in conf_connections:
connection_name = conf_connection.get(CONF_NAME)
# initialize a config_flow for all LCN configurations read from
# configuration.yaml
config_entries_data = import_lcn_config(config[DOMAIN])
settings = {
"SK_NUM_TRIES": conf_connection[CONF_SK_NUM_TRIES],
"DIM_MODE": pypck.lcn_defs.OutputPortDimMode[
conf_connection[CONF_DIM_MODE]
],
}
connection = pypck.connection.PchkConnectionManager(
conf_connection[CONF_HOST],
conf_connection[CONF_PORT],
conf_connection[CONF_USERNAME],
conf_connection[CONF_PASSWORD],
settings=settings,
connection_id=connection_name,
)
try:
# establish connection to PCHK server
await hass.async_create_task(connection.async_connect(timeout=15))
connections.append(connection)
_LOGGER.info('LCN connected to "%s"', connection_name)
except TimeoutError:
_LOGGER.error('Connection to PCHK server "%s" failed', connection_name)
return False
hass.data[DATA_LCN][CONF_CONNECTIONS] = connections
# load platforms
for component, conf_key in (
("binary_sensor", CONF_BINARY_SENSORS),
("climate", CONF_CLIMATES),
("cover", CONF_COVERS),
("light", CONF_LIGHTS),
("scene", CONF_SCENES),
("sensor", CONF_SENSORS),
("switch", CONF_SWITCHES),
):
if conf_key in config[DOMAIN]:
hass.async_create_task(
async_load_platform(
hass, component, DOMAIN, config[DOMAIN][conf_key], config
)
for config_entry_data in config_entries_data:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=config_entry_data,
)
)
return True
async def async_setup_entry(hass, config_entry):
"""Set up a connection to PCHK host from a config entry."""
hass.data.setdefault(DOMAIN, {})
if config_entry.entry_id in hass.data[DOMAIN]:
return False
settings = {
"SK_NUM_TRIES": config_entry.data[CONF_SK_NUM_TRIES],
"DIM_MODE": pypck.lcn_defs.OutputPortDimMode[config_entry.data[CONF_DIM_MODE]],
}
# connect to PCHK
lcn_connection = pypck.connection.PchkConnectionManager(
config_entry.data[CONF_IP_ADDRESS],
config_entry.data[CONF_PORT],
config_entry.data[CONF_USERNAME],
config_entry.data[CONF_PASSWORD],
settings=settings,
connection_id=config_entry.entry_id,
)
try:
# establish connection to PCHK server
await lcn_connection.async_connect(timeout=15)
except pypck.connection.PchkAuthenticationError:
_LOGGER.warning('Authentication on PCHK "%s" failed', config_entry.title)
return False
except pypck.connection.PchkLicenseError:
_LOGGER.warning(
'Maximum number of connections on PCHK "%s" was '
"reached. An additional license key is required",
config_entry.title,
)
return False
except TimeoutError:
_LOGGER.warning('Connection to PCHK "%s" failed', config_entry.title)
return False
_LOGGER.debug('LCN connected to "%s"', config_entry.title)
hass.data[DOMAIN][config_entry.entry_id] = {
CONNECTION: lcn_connection,
}
# remove orphans from entity registry which are in ConfigEntry but were removed
# from configuration.yaml
if config_entry.source == config_entries.SOURCE_IMPORT:
entity_registry = await er.async_get_registry(hass)
entity_registry.async_clear_config_entry(config_entry.entry_id)
# forward config_entry to components
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, component)
)
# register service calls
for service_name, service in (
("output_abs", OutputAbs),
("output_rel", OutputRel),
("output_toggle", OutputToggle),
("relays", Relays),
("var_abs", VarAbs),
("var_reset", VarReset),
("var_rel", VarRel),
("lock_regulator", LockRegulator),
("led", Led),
("send_keys", SendKeys),
("lock_keys", LockKeys),
("dyn_text", DynText),
("pck", Pck),
):
hass.services.async_register(
DOMAIN, service_name, service(hass).async_call_service, service.schema
)
for service_name, service in SERVICES:
if not hass.services.has_service(DOMAIN, service_name):
hass.services.async_register(
DOMAIN, service_name, service(hass).async_call_service, service.schema
)
return True
class LcnEntity(Entity):
"""Parent class for all devices associated with the LCN component."""
async def async_unload_entry(hass, config_entry):
"""Close connection to PCHK host represented by config_entry."""
# forward unloading to platforms
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(config_entry, component)
for component in PLATFORMS
]
)
)
def __init__(self, config, device_connection):
if unload_ok and config_entry.entry_id in hass.data[DOMAIN]:
host = hass.data[DOMAIN].pop(config_entry.entry_id)
await host[CONNECTION].async_close()
# unregister service calls
if unload_ok and not hass.data[DOMAIN]: # check if this is the last entry to unload
for service_name, _ in SERVICES:
hass.services.async_remove(DOMAIN, service_name)
return unload_ok
class LcnEntity(Entity):
"""Parent class for all entities associated with the LCN component."""
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN device."""
self.config = config
self.entry_id = entry_id
self.device_connection = device_connection
self._unregister_for_inputs = None
self._name = config[CONF_NAME]
@property
def unique_id(self):
"""Return a unique ID."""
unique_device_id = generate_unique_id(
(
self.device_connection.seg_id,
self.device_connection.addr_id,
self.device_connection.is_group,
)
)
return f"{self.entry_id}-{unique_device_id}-{self.config[CONF_RESOURCE]}"
@property
def should_poll(self):
"""Lcn device entity pushes its state to HA."""
@ -140,7 +165,14 @@ class LcnEntity(Entity):
async def async_added_to_hass(self):
"""Run when entity about to be added to hass."""
if not self.device_connection.is_group:
self.device_connection.register_for_inputs(self.input_received)
self._unregister_for_inputs = self.device_connection.register_for_inputs(
self.input_received
)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
if self._unregister_for_inputs is not None:
self._unregister_for_inputs()
@property
def name(self):

99
homeassistant/components/lcn/binary_sensor.py
View file

@ -1,49 +1,56 @@
"""Support for LCN binary sensors."""
import pypck
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.const import CONF_ADDRESS, CONF_SOURCE
from homeassistant.components.binary_sensor import (
DOMAIN as DOMAIN_BINARY_SENSOR,
BinarySensorEntity,
)
from homeassistant.const import CONF_ADDRESS, CONF_DOMAIN, CONF_ENTITIES, CONF_SOURCE
from . import LcnEntity
from .const import BINSENSOR_PORTS, CONF_CONNECTIONS, DATA_LCN, SETPOINTS
from .helpers import get_connection
from .const import BINSENSOR_PORTS, CONF_DOMAIN_DATA, SETPOINTS
from .helpers import get_device_connection
async def async_setup_platform(
hass, hass_config, async_add_entities, discovery_info=None
):
"""Set up the LCN binary sensor platform."""
if discovery_info is None:
return
def create_lcn_binary_sensor_entity(hass, entity_config, config_entry):
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
)
devices = []
for config in discovery_info:
address, connection_id = config[CONF_ADDRESS]
addr = pypck.lcn_addr.LcnAddr(*address)
connections = hass.data[DATA_LCN][CONF_CONNECTIONS]
connection = get_connection(connections, connection_id)
address_connection = connection.get_address_conn(addr)
if entity_config[CONF_DOMAIN_DATA][CONF_SOURCE] in SETPOINTS:
return LcnRegulatorLockSensor(
entity_config, config_entry.entry_id, device_connection
)
if entity_config[CONF_DOMAIN_DATA][CONF_SOURCE] in BINSENSOR_PORTS:
return LcnBinarySensor(entity_config, config_entry.entry_id, device_connection)
# in KEY
return LcnLockKeysSensor(entity_config, config_entry.entry_id, device_connection)
if config[CONF_SOURCE] in SETPOINTS:
device = LcnRegulatorLockSensor(config, address_connection)
elif config[CONF_SOURCE] in BINSENSOR_PORTS:
device = LcnBinarySensor(config, address_connection)
else: # in KEYS
device = LcnLockKeysSensor(config, address_connection)
devices.append(device)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up LCN switch entities from a config entry."""
entities = []
async_add_entities(devices)
for entity_config in config_entry.data[CONF_ENTITIES]:
if entity_config[CONF_DOMAIN] == DOMAIN_BINARY_SENSOR:
entities.append(
create_lcn_binary_sensor_entity(hass, entity_config, config_entry)
)
async_add_entities(entities)
class LcnRegulatorLockSensor(LcnEntity, BinarySensorEntity):
"""Representation of a LCN binary sensor for regulator locks."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN binary sensor."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
self.setpoint_variable = pypck.lcn_defs.Var[config[CONF_SOURCE]]
self.setpoint_variable = pypck.lcn_defs.Var[
config[CONF_DOMAIN_DATA][CONF_SOURCE]
]
self._value = None
@ -55,6 +62,14 @@ class LcnRegulatorLockSensor(LcnEntity, BinarySensorEntity):
self.setpoint_variable
)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(
self.setpoint_variable
)
@property
def is_on(self):
"""Return true if the binary sensor is on."""
@ -75,11 +90,13 @@ class LcnRegulatorLockSensor(LcnEntity, BinarySensorEntity):
class LcnBinarySensor(LcnEntity, BinarySensorEntity):
"""Representation of a LCN binary sensor for binary sensor ports."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN binary sensor."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
self.bin_sensor_port = pypck.lcn_defs.BinSensorPort[config[CONF_SOURCE]]
self.bin_sensor_port = pypck.lcn_defs.BinSensorPort[
config[CONF_DOMAIN_DATA][CONF_SOURCE]
]
self._value = None
@ -91,6 +108,14 @@ class LcnBinarySensor(LcnEntity, BinarySensorEntity):
self.bin_sensor_port
)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(
self.bin_sensor_port
)
@property
def is_on(self):
"""Return true if the binary sensor is on."""
@ -108,11 +133,11 @@ class LcnBinarySensor(LcnEntity, BinarySensorEntity):
class LcnLockKeysSensor(LcnEntity, BinarySensorEntity):
"""Representation of a LCN sensor for key locks."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN sensor."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
self.source = pypck.lcn_defs.Key[config[CONF_SOURCE]]
self.source = pypck.lcn_defs.Key[config[CONF_DOMAIN_DATA][CONF_SOURCE]]
self._value = None
async def async_added_to_hass(self):
@ -121,6 +146,12 @@ class LcnLockKeysSensor(LcnEntity, BinarySensorEntity):
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.source)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.source)
@property
def is_on(self):
"""Return true if the binary sensor is on."""

73
homeassistant/components/lcn/climate.py
View file

@ -1,68 +1,76 @@
"""Support for LCN climate control."""
import pypck
from homeassistant.components.climate import ClimateEntity, const
from homeassistant.components.climate import (
DOMAIN as DOMAIN_CLIMATE,
ClimateEntity,
const,
)
from homeassistant.const import (
ATTR_TEMPERATURE,
CONF_ADDRESS,
CONF_DOMAIN,
CONF_ENTITIES,
CONF_SOURCE,
CONF_UNIT_OF_MEASUREMENT,
)
from . import LcnEntity
from .const import (
CONF_CONNECTIONS,
CONF_DOMAIN_DATA,
CONF_LOCKABLE,
CONF_MAX_TEMP,
CONF_MIN_TEMP,
CONF_SETPOINT,
DATA_LCN,
)
from .helpers import get_connection
from .helpers import get_device_connection
PARALLEL_UPDATES = 0
async def async_setup_platform(
hass, hass_config, async_add_entities, discovery_info=None
):
"""Set up the LCN climate platform."""
if discovery_info is None:
return
def create_lcn_climate_entity(hass, entity_config, config_entry):
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
)
devices = []
for config in discovery_info:
address, connection_id = config[CONF_ADDRESS]
addr = pypck.lcn_addr.LcnAddr(*address)
connections = hass.data[DATA_LCN][CONF_CONNECTIONS]
connection = get_connection(connections, connection_id)
address_connection = connection.get_address_conn(addr)
return LcnClimate(entity_config, config_entry.entry_id, device_connection)
devices.append(LcnClimate(config, address_connection))
async_add_entities(devices)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up LCN switch entities from a config entry."""
entities = []
for entity_config in config_entry.data[CONF_ENTITIES]:
if entity_config[CONF_DOMAIN] == DOMAIN_CLIMATE:
entities.append(
create_lcn_climate_entity(hass, entity_config, config_entry)
)
async_add_entities(entities)
class LcnClimate(LcnEntity, ClimateEntity):
"""Representation of a LCN climate device."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize of a LCN climate device."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
self.variable = pypck.lcn_defs.Var[config[CONF_SOURCE]]
self.setpoint = pypck.lcn_defs.Var[config[CONF_SETPOINT]]
self.unit = pypck.lcn_defs.VarUnit.parse(config[CONF_UNIT_OF_MEASUREMENT])
self.variable = pypck.lcn_defs.Var[config[CONF_DOMAIN_DATA][CONF_SOURCE]]
self.setpoint = pypck.lcn_defs.Var[config[CONF_DOMAIN_DATA][CONF_SETPOINT]]
self.unit = pypck.lcn_defs.VarUnit.parse(
config[CONF_DOMAIN_DATA][CONF_UNIT_OF_MEASUREMENT]
)
self.regulator_id = pypck.lcn_defs.Var.to_set_point_id(self.setpoint)
self.is_lockable = config[CONF_LOCKABLE]
self._max_temp = config[CONF_MAX_TEMP]
self._min_temp = config[CONF_MIN_TEMP]
self.is_lockable = config[CONF_DOMAIN_DATA][CONF_LOCKABLE]
self._max_temp = config[CONF_DOMAIN_DATA][CONF_MAX_TEMP]
self._min_temp = config[CONF_DOMAIN_DATA][CONF_MIN_TEMP]
self._current_temperature = None
self._target_temperature = None
self._is_on = None
self._is_on = True
async def async_added_to_hass(self):
"""Run when entity about to be added to hass."""
@ -71,6 +79,13 @@ class LcnClimate(LcnEntity, ClimateEntity):
await self.device_connection.activate_status_request_handler(self.variable)
await self.device_connection.activate_status_request_handler(self.setpoint)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.variable)
await self.device_connection.cancel_status_request_handler(self.setpoint)
@property
def supported_features(self):
"""Return the list of supported features."""

97
homeassistant/components/lcn/config_flow.py Normal file
View file

@ -0,0 +1,97 @@
"""Config flow to configure the LCN integration."""
import logging
import pypck
from homeassistant import config_entries
from homeassistant.const import (
CONF_HOST,
CONF_IP_ADDRESS,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
)
from .const import CONF_DIM_MODE, CONF_SK_NUM_TRIES, DOMAIN
_LOGGER = logging.getLogger(__name__)
def get_config_entry(hass, data):
"""Check config entries for already configured entries based on the ip address/port."""
return next(
(
entry
for entry in hass.config_entries.async_entries(DOMAIN)
if entry.data[CONF_IP_ADDRESS] == data[CONF_IP_ADDRESS]
and entry.data[CONF_PORT] == data[CONF_PORT]
),
None,
)
async def validate_connection(host_name, data):
"""Validate if a connection to LCN can be established."""
host = data[CONF_IP_ADDRESS]
port = data[CONF_PORT]
username = data[CONF_USERNAME]
password = data[CONF_PASSWORD]
sk_num_tries = data[CONF_SK_NUM_TRIES]
dim_mode = data[CONF_DIM_MODE]
settings = {
"SK_NUM_TRIES": sk_num_tries,
"DIM_MODE": pypck.lcn_defs.OutputPortDimMode[dim_mode],
}
_LOGGER.debug("Validating connection parameters to PCHK host '%s'", host_name)
connection = pypck.connection.PchkConnectionManager(
host, port, username, password, settings=settings
)
await connection.async_connect(timeout=5)
_LOGGER.debug("LCN connection validated")
await connection.async_close()
return data
@config_entries.HANDLERS.register(DOMAIN)
class LcnFlowHandler(config_entries.ConfigFlow):
"""Handle a LCN config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
async def async_step_import(self, data):
"""Import existing configuration from LCN."""
host_name = data[CONF_HOST]
# validate the imported connection parameters
try:
await validate_connection(host_name, data)
except pypck.connection.PchkAuthenticationError:
_LOGGER.warning('Authentication on PCHK "%s" failed', host_name)
return self.async_abort(reason="authentication_error")
except pypck.connection.PchkLicenseError:
_LOGGER.warning(
'Maximum number of connections on PCHK "%s" was '
"reached. An additional license key is required",
host_name,
)
return self.async_abort(reason="license_error")
except TimeoutError:
_LOGGER.warning('Connection to PCHK "%s" failed', host_name)
return self.async_abort(reason="connection_timeout")
# check if we already have a host with the same address configured
entry = get_config_entry(self.hass, data)
if entry:
entry.source = config_entries.SOURCE_IMPORT
self.hass.config_entries.async_update_entry(entry, data=data)
return self.async_abort(reason="existing_configuration_updated")
return self.async_create_entry(
title=f"{host_name}",
data=data,
)

7
homeassistant/components/lcn/const.py
View file

@ -14,6 +14,13 @@ DOMAIN = "lcn"
DATA_LCN = "lcn"
DEFAULT_NAME = "pchk"
CONNECTION = "connection"
CONF_HARDWARE_SERIAL = "hardware_serial"
CONF_SOFTWARE_SERIAL = "software_serial"
CONF_HARDWARE_TYPE = "hardware_type"
CONF_RESOURCE = "resource"
CONF_DOMAIN_DATA = "domain_data"
CONF_CONNECTIONS = "connections"
CONF_SK_NUM_TRIES = "sk_num_tries"
CONF_OUTPUT = "output"

89
homeassistant/components/lcn/cover.py
View file

@ -1,53 +1,54 @@
"""Support for LCN covers."""
import pypck
from homeassistant.components.cover import CoverEntity
from homeassistant.const import CONF_ADDRESS
from homeassistant.components.cover import DOMAIN as DOMAIN_COVER, CoverEntity
from homeassistant.const import CONF_ADDRESS, CONF_DOMAIN, CONF_ENTITIES
from . import LcnEntity
from .const import CONF_CONNECTIONS, CONF_MOTOR, CONF_REVERSE_TIME, DATA_LCN
from .helpers import get_connection
from .const import CONF_DOMAIN_DATA, CONF_MOTOR, CONF_REVERSE_TIME
from .helpers import get_device_connection
PARALLEL_UPDATES = 0
async def async_setup_platform(
hass, hass_config, async_add_entities, discovery_info=None
):
"""Setups the LCN cover platform."""
if discovery_info is None:
return
def create_lcn_cover_entity(hass, entity_config, config_entry):
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
)
devices = []
for config in discovery_info:
address, connection_id = config[CONF_ADDRESS]
addr = pypck.lcn_addr.LcnAddr(*address)
connections = hass.data[DATA_LCN][CONF_CONNECTIONS]
connection = get_connection(connections, connection_id)
address_connection = connection.get_address_conn(addr)
if entity_config[CONF_DOMAIN_DATA][CONF_MOTOR] in "OUTPUTS":
return LcnOutputsCover(entity_config, config_entry.entry_id, device_connection)
# in RELAYS
return LcnRelayCover(entity_config, config_entry.entry_id, device_connection)
if config[CONF_MOTOR] == "OUTPUTS":
devices.append(LcnOutputsCover(config, address_connection))
else: # RELAYS
devices.append(LcnRelayCover(config, address_connection))
async_add_entities(devices)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up LCN cover entities from a config entry."""
entities = []
for entity_config in config_entry.data[CONF_ENTITIES]:
if entity_config[CONF_DOMAIN] == DOMAIN_COVER:
entities.append(create_lcn_cover_entity(hass, entity_config, config_entry))
async_add_entities(entities)
class LcnOutputsCover(LcnEntity, CoverEntity):
"""Representation of a LCN cover connected to output ports."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN cover."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
self.output_ids = [
pypck.lcn_defs.OutputPort["OUTPUTUP"].value,
pypck.lcn_defs.OutputPort["OUTPUTDOWN"].value,
]
if CONF_REVERSE_TIME in config:
if CONF_REVERSE_TIME in config[CONF_DOMAIN_DATA]:
self.reverse_time = pypck.lcn_defs.MotorReverseTime[
config[CONF_REVERSE_TIME]
config[CONF_DOMAIN_DATA][CONF_REVERSE_TIME]
]
else:
self.reverse_time = None
@ -59,12 +60,24 @@ class LcnOutputsCover(LcnEntity, CoverEntity):
async def async_added_to_hass(self):
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
await self.device_connection.activate_status_request_handler(
pypck.lcn_defs.OutputPort["OUTPUTUP"]
)
await self.device_connection.activate_status_request_handler(
pypck.lcn_defs.OutputPort["OUTPUTDOWN"]
)
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(
pypck.lcn_defs.OutputPort["OUTPUTUP"]
)
await self.device_connection.activate_status_request_handler(
pypck.lcn_defs.OutputPort["OUTPUTDOWN"]
)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(
pypck.lcn_defs.OutputPort["OUTPUTUP"]
)
await self.device_connection.cancel_status_request_handler(
pypck.lcn_defs.OutputPort["OUTPUTDOWN"]
)
@property
def is_closed(self):
@ -146,11 +159,11 @@ class LcnOutputsCover(LcnEntity, CoverEntity):
class LcnRelayCover(LcnEntity, CoverEntity):
"""Representation of a LCN cover connected to relays."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN cover."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
self.motor = pypck.lcn_defs.MotorPort[config[CONF_MOTOR]]
self.motor = pypck.lcn_defs.MotorPort[config[CONF_DOMAIN_DATA][CONF_MOTOR]]
self.motor_port_onoff = self.motor.value * 2
self.motor_port_updown = self.motor_port_onoff + 1
@ -164,6 +177,12 @@ class LcnRelayCover(LcnEntity, CoverEntity):
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.motor)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.motor)
@property
def is_closed(self):
"""Return if the cover is closed."""

185
homeassistant/components/lcn/helpers.py
View file

@ -1,11 +1,42 @@
"""Helpers for LCN component."""
import re
import pypck
import voluptuous as vol
from homeassistant.const import CONF_NAME
from homeassistant.const import (
CONF_ADDRESS,
CONF_BINARY_SENSORS,
CONF_COVERS,
CONF_DEVICES,
CONF_DOMAIN,
CONF_ENTITIES,
CONF_HOST,
CONF_IP_ADDRESS,
CONF_LIGHTS,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_SENSORS,
CONF_SWITCHES,
CONF_USERNAME,
)
from .const import DEFAULT_NAME
from .const import (
CONF_CLIMATES,
CONF_CONNECTIONS,
CONF_DIM_MODE,
CONF_DOMAIN_DATA,
CONF_HARDWARE_SERIAL,
CONF_HARDWARE_TYPE,
CONF_RESOURCE,
CONF_SCENES,
CONF_SK_NUM_TRIES,
CONF_SOFTWARE_SERIAL,
CONNECTION,
DEFAULT_NAME,
DOMAIN,
)
# Regex for address validation
PATTERN_ADDRESS = re.compile(
@ -13,17 +44,145 @@ PATTERN_ADDRESS = re.compile(
)
def get_connection(connections, connection_id=None):
"""Return the connection object from list."""
if connection_id is None:
connection = connections[0]
else:
for connection in connections:
if connection.connection_id == connection_id:
break
else:
raise ValueError("Unknown connection_id.")
return connection
DOMAIN_LOOKUP = {
CONF_BINARY_SENSORS: "binary_sensor",
CONF_CLIMATES: "climate",
CONF_COVERS: "cover",
CONF_LIGHTS: "light",
CONF_SCENES: "scene",
CONF_SENSORS: "sensor",
CONF_SWITCHES: "switch",
}
def get_device_connection(hass, address, config_entry):
"""Return a lcn device_connection."""
host_connection = hass.data[DOMAIN][config_entry.entry_id][CONNECTION]
addr = pypck.lcn_addr.LcnAddr(*address)
return host_connection.get_address_conn(addr)
def get_resource(domain_name, domain_data):
"""Return the resource for the specified domain_data."""
if domain_name in ["switch", "light"]:
return domain_data["output"]
if domain_name in ["binary_sensor", "sensor"]:
return domain_data["source"]
if domain_name == "cover":
return domain_data["motor"]
if domain_name == "climate":
return f'{domain_data["source"]}.{domain_data["setpoint"]}'
if domain_name == "scene":
return f'{domain_data["register"]}.{domain_data["scene"]}'
raise ValueError("Unknown domain")
def generate_unique_id(address):
"""Generate a unique_id from the given parameters."""
is_group = "g" if address[2] else "m"
return f"{is_group}{address[0]:03d}{address[1]:03d}"
def import_lcn_config(lcn_config):
"""Convert lcn settings from configuration.yaml to config_entries data.
Create a list of config_entry data structures like:
"data": {
"host": "pchk",
"ip_address": "192.168.2.41",
"port": 4114,
"username": "lcn",
"password": "lcn,
"sk_num_tries: 0,
"dim_mode: "STEPS200",
"devices": [
{
"address": (0, 7, False)
"name": "",
"hardware_serial": -1,
"software_serial": -1,
"hardware_type": -1
}, ...
],
"entities": [
{
"address": (0, 7, False)
"name": "Light_Output1",
"resource": "output1",
"domain": "light",
"domain_data": {
"output": "OUTPUT1",
"dimmable": True,
"transition": 5000.0
}
}, ...
]
}
"""
data = {}
for connection in lcn_config[CONF_CONNECTIONS]:
host = {
CONF_HOST: connection[CONF_NAME],
CONF_IP_ADDRESS: connection[CONF_HOST],
CONF_PORT: connection[CONF_PORT],
CONF_USERNAME: connection[CONF_USERNAME],
CONF_PASSWORD: connection[CONF_PASSWORD],
CONF_SK_NUM_TRIES: connection[CONF_SK_NUM_TRIES],
CONF_DIM_MODE: connection[CONF_DIM_MODE],
CONF_DEVICES: [],
CONF_ENTITIES: [],
}
data[connection[CONF_NAME]] = host
for confkey, domain_config in lcn_config.items():
if confkey == CONF_CONNECTIONS:
continue
domain = DOMAIN_LOOKUP[confkey]
# loop over entities in configuration.yaml
for domain_data in domain_config:
# remove name and address from domain_data
entity_name = domain_data.pop(CONF_NAME)
address, host_name = domain_data.pop(CONF_ADDRESS)
if host_name is None:
host_name = DEFAULT_NAME
# check if we have a new device config
for device_config in data[host_name][CONF_DEVICES]:
if address == device_config[CONF_ADDRESS]:
break
else: # create new device_config
device_config = {
CONF_ADDRESS: address,
CONF_NAME: "",
CONF_HARDWARE_SERIAL: -1,
CONF_SOFTWARE_SERIAL: -1,
CONF_HARDWARE_TYPE: -1,
}
data[host_name][CONF_DEVICES].append(device_config)
# insert entity config
resource = get_resource(domain, domain_data).lower()
for entity_config in data[host_name][CONF_ENTITIES]:
if (
address == entity_config[CONF_ADDRESS]
and resource == entity_config[CONF_RESOURCE]
and domain == entity_config[CONF_DOMAIN]
):
break
else: # create new entity_config
entity_config = {
CONF_ADDRESS: address,
CONF_NAME: entity_name,
CONF_RESOURCE: resource,
CONF_DOMAIN: domain,
CONF_DOMAIN_DATA: domain_data.copy(),
}
data[host_name][CONF_ENTITIES].append(entity_config)
return list(data.values())
def has_unique_host_names(hosts):

79
homeassistant/components/lcn/light.py
View file

@ -1,68 +1,69 @@
"""Support for LCN lights."""
import pypck
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_TRANSITION,
DOMAIN as DOMAIN_LIGHT,
SUPPORT_BRIGHTNESS,
SUPPORT_TRANSITION,
LightEntity,
)
from homeassistant.const import CONF_ADDRESS
from homeassistant.const import CONF_ADDRESS, CONF_DOMAIN, CONF_ENTITIES
from . import LcnEntity
from .const import (
CONF_CONNECTIONS,
CONF_DIMMABLE,
CONF_DOMAIN_DATA,
CONF_OUTPUT,
CONF_TRANSITION,
DATA_LCN,
OUTPUT_PORTS,
)
from .helpers import get_connection
from .helpers import get_device_connection
PARALLEL_UPDATES = 0
async def async_setup_platform(
hass, hass_config, async_add_entities, discovery_info=None
):
"""Set up the LCN light platform."""
if discovery_info is None:
return
def create_lcn_light_entity(hass, entity_config, config_entry):
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
)
devices = []
for config in discovery_info:
address, connection_id = config[CONF_ADDRESS]
addr = pypck.lcn_addr.LcnAddr(*address)
connections = hass.data[DATA_LCN][CONF_CONNECTIONS]
connection = get_connection(connections, connection_id)
address_connection = connection.get_address_conn(addr)
if entity_config[CONF_DOMAIN_DATA][CONF_OUTPUT] in OUTPUT_PORTS:
return LcnOutputLight(entity_config, config_entry.entry_id, device_connection)
# in RELAY_PORTS
return LcnRelayLight(entity_config, config_entry.entry_id, device_connection)
if config[CONF_OUTPUT] in OUTPUT_PORTS:
device = LcnOutputLight(config, address_connection)
else: # in RELAY_PORTS
device = LcnRelayLight(config, address_connection)
devices.append(device)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up LCN light entities from a config entry."""
entities = []
async_add_entities(devices)
for entity_config in config_entry.data[CONF_ENTITIES]:
if entity_config[CONF_DOMAIN] == DOMAIN_LIGHT:
entities.append(create_lcn_light_entity(hass, entity_config, config_entry))
async_add_entities(entities)
class LcnOutputLight(LcnEntity, LightEntity):
"""Representation of a LCN light for output ports."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN light."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
self.output = pypck.lcn_defs.OutputPort[config[CONF_OUTPUT]]
self.output = pypck.lcn_defs.OutputPort[config[CONF_DOMAIN_DATA][CONF_OUTPUT]]
self._transition = pypck.lcn_defs.time_to_ramp_value(config[CONF_TRANSITION])
self.dimmable = config[CONF_DIMMABLE]
self._transition = pypck.lcn_defs.time_to_ramp_value(
config[CONF_DOMAIN_DATA][CONF_TRANSITION]
)
self.dimmable = config[CONF_DOMAIN_DATA][CONF_DIMMABLE]
self._brightness = 255
self._is_on = None
self._is_on = False
self._is_dimming_to_zero = False
async def async_added_to_hass(self):
@ -71,6 +72,12 @@ class LcnOutputLight(LcnEntity, LightEntity):
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.output)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.output)
@property
def supported_features(self):
"""Flag supported features."""
@ -145,13 +152,13 @@ class LcnOutputLight(LcnEntity, LightEntity):
class LcnRelayLight(LcnEntity, LightEntity):
"""Representation of a LCN light for relay ports."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN light."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
self.output = pypck.lcn_defs.RelayPort[config[CONF_OUTPUT]]
self.output = pypck.lcn_defs.RelayPort[config[CONF_DOMAIN_DATA][CONF_OUTPUT]]
self._is_on = None
self._is_on = False
async def async_added_to_hass(self):
"""Run when entity about to be added to hass."""
@ -159,6 +166,12 @@ class LcnRelayLight(LcnEntity, LightEntity):
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.output)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.output)
@property
def is_on(self):
"""Return True if entity is on."""

1
homeassistant/components/lcn/manifest.json
View file

@ -1,6 +1,7 @@
{
"domain": "lcn",
"name": "LCN",
"config_flow": false,
"documentation": "https://www.home-assistant.io/integrations/lcn",
"requirements": [
"pypck==0.7.9"

61
homeassistant/components/lcn/scene.py
View file

@ -1,72 +1,69 @@
"""Support for LCN scenes."""
from typing import Any
import pypck
from homeassistant.components.scene import Scene
from homeassistant.const import CONF_ADDRESS, CONF_SCENE
from homeassistant.components.scene import DOMAIN as DOMAIN_SCENE, Scene
from homeassistant.const import CONF_ADDRESS, CONF_DOMAIN, CONF_ENTITIES, CONF_SCENE
from . import LcnEntity
from .const import (
CONF_CONNECTIONS,
CONF_DOMAIN_DATA,
CONF_OUTPUTS,
CONF_REGISTER,
CONF_TRANSITION,
DATA_LCN,
OUTPUT_PORTS,
)
from .helpers import get_connection
from .helpers import get_device_connection
PARALLEL_UPDATES = 0
async def async_setup_platform(
hass, hass_config, async_add_entities, discovery_info=None
):
"""Set up the LCN scene platform."""
if discovery_info is None:
return
def create_lcn_scene_entity(hass, entity_config, config_entry):
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
)
devices = []
for config in discovery_info:
address, connection_id = config[CONF_ADDRESS]
addr = pypck.lcn_addr.LcnAddr(*address)
connections = hass.data[DATA_LCN][CONF_CONNECTIONS]
connection = get_connection(connections, connection_id)
address_connection = connection.get_address_conn(addr)
return LcnScene(entity_config, config_entry.entry_id, device_connection)
devices.append(LcnScene(config, address_connection))
async_add_entities(devices)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up LCN switch entities from a config entry."""
entities = []
for entity_config in config_entry.data[CONF_ENTITIES]:
if entity_config[CONF_DOMAIN] == DOMAIN_SCENE:
entities.append(create_lcn_scene_entity(hass, entity_config, config_entry))
async_add_entities(entities)
class LcnScene(LcnEntity, Scene):
"""Representation of a LCN scene."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN scene."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
self.register_id = config[CONF_REGISTER]
self.scene_id = config[CONF_SCENE]
self.register_id = config[CONF_DOMAIN_DATA][CONF_REGISTER]
self.scene_id = config[CONF_DOMAIN_DATA][CONF_SCENE]
self.output_ports = []
self.relay_ports = []
for port in config[CONF_OUTPUTS]:
for port in config[CONF_DOMAIN_DATA][CONF_OUTPUTS]:
if port in OUTPUT_PORTS:
self.output_ports.append(pypck.lcn_defs.OutputPort[port])
else: # in RELEAY_PORTS
self.relay_ports.append(pypck.lcn_defs.RelayPort[port])
if config[CONF_TRANSITION] is None:
if config[CONF_DOMAIN_DATA][CONF_TRANSITION] is None:
self.transition = None
else:
self.transition = pypck.lcn_defs.time_to_ramp_value(config[CONF_TRANSITION])
self.transition = pypck.lcn_defs.time_to_ramp_value(
config[CONF_DOMAIN_DATA][CONF_TRANSITION]
)
async def async_added_to_hass(self):
"""Run when entity about to be added to hass."""
async def async_activate(self, **kwargs: Any) -> None:
async def async_activate(self, **kwargs):
"""Activate scene."""
await self.device_connection.activate_scene(
self.register_id,

90
homeassistant/components/lcn/sensor.py
View file

@ -1,55 +1,67 @@
"""Support for LCN sensors."""
import pypck
from homeassistant.const import CONF_ADDRESS, CONF_SOURCE, CONF_UNIT_OF_MEASUREMENT
from homeassistant.components.sensor import DOMAIN as DOMAIN_SENSOR
from homeassistant.const import (
CONF_ADDRESS,
CONF_DOMAIN,
CONF_ENTITIES,
CONF_SOURCE,
CONF_UNIT_OF_MEASUREMENT,
)
from . import LcnEntity
from .const import (
CONF_CONNECTIONS,
DATA_LCN,
CONF_DOMAIN_DATA,
LED_PORTS,
S0_INPUTS,
SETPOINTS,
THRESHOLDS,
VARIABLES,
)
from .helpers import get_connection
from .helpers import get_device_connection
async def async_setup_platform(
hass, hass_config, async_add_entities, discovery_info=None
):
"""Set up the LCN sensor platform."""
if discovery_info is None:
return
def create_lcn_sensor_entity(hass, entity_config, config_entry):
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
)
devices = []
for config in discovery_info:
address, connection_id = config[CONF_ADDRESS]
addr = pypck.lcn_addr.LcnAddr(*address)
connections = hass.data[DATA_LCN][CONF_CONNECTIONS]
connection = get_connection(connections, connection_id)
device_connection = connection.get_address_conn(addr)
if (
entity_config[CONF_DOMAIN_DATA][CONF_SOURCE]
in VARIABLES + SETPOINTS + THRESHOLDS + S0_INPUTS
):
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 config[CONF_SOURCE] in VARIABLES + SETPOINTS + THRESHOLDS + S0_INPUTS:
device = LcnVariableSensor(config, device_connection)
else: # in LED_PORTS + LOGICOP_PORTS
device = LcnLedLogicSensor(config, device_connection)
devices.append(device)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up LCN switch entities from a config entry."""
entities = []
async_add_entities(devices)
for entity_config in config_entry.data[CONF_ENTITIES]:
if entity_config[CONF_DOMAIN] == DOMAIN_SENSOR:
entities.append(create_lcn_sensor_entity(hass, entity_config, config_entry))
async_add_entities(entities)
class LcnVariableSensor(LcnEntity):
"""Representation of a LCN sensor for variables."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN sensor."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
self.variable = pypck.lcn_defs.Var[config[CONF_SOURCE]]
self.unit = pypck.lcn_defs.VarUnit.parse(config[CONF_UNIT_OF_MEASUREMENT])
self.variable = pypck.lcn_defs.Var[config[CONF_DOMAIN_DATA][CONF_SOURCE]]
self.unit = pypck.lcn_defs.VarUnit.parse(
config[CONF_DOMAIN_DATA][CONF_UNIT_OF_MEASUREMENT]
)
self._value = None
@ -59,6 +71,12 @@ class LcnVariableSensor(LcnEntity):
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.variable)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.variable)
@property
def state(self):
"""Return the state of the entity."""
@ -84,14 +102,16 @@ class LcnVariableSensor(LcnEntity):
class LcnLedLogicSensor(LcnEntity):
"""Representation of a LCN sensor for leds and logicops."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN sensor."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
if config[CONF_SOURCE] in LED_PORTS:
self.source = pypck.lcn_defs.LedPort[config[CONF_SOURCE]]
if config[CONF_DOMAIN_DATA][CONF_SOURCE] in LED_PORTS:
self.source = pypck.lcn_defs.LedPort[config[CONF_DOMAIN_DATA][CONF_SOURCE]]
else:
self.source = pypck.lcn_defs.LogicOpPort[config[CONF_SOURCE]]
self.source = pypck.lcn_defs.LogicOpPort[
config[CONF_DOMAIN_DATA][CONF_SOURCE]
]
self._value = None
@ -101,6 +121,12 @@ class LcnLedLogicSensor(LcnEntity):
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.source)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.source)
@property
def state(self):
"""Return the state of the entity."""

43
homeassistant/components/lcn/services.py
View file

@ -6,6 +6,7 @@ import voluptuous as vol
from homeassistant.const import (
CONF_ADDRESS,
CONF_BRIGHTNESS,
CONF_HOST,
CONF_STATE,
CONF_UNIT_OF_MEASUREMENT,
TIME_SECONDS,
@ -13,7 +14,6 @@ from homeassistant.const import (
import homeassistant.helpers.config_validation as cv
from .const import (
CONF_CONNECTIONS,
CONF_KEYS,
CONF_LED,
CONF_OUTPUT,
@ -28,7 +28,7 @@ from .const import (
CONF_TRANSITION,
CONF_VALUE,
CONF_VARIABLE,
DATA_LCN,
DOMAIN,
LED_PORTS,
LED_STATUS,
OUTPUT_PORTS,
@ -41,7 +41,7 @@ from .const import (
VARIABLES,
)
from .helpers import (
get_connection,
get_device_connection,
is_address,
is_key_lock_states_string,
is_relays_states_string,
@ -56,18 +56,20 @@ class LcnServiceCall:
def __init__(self, hass):
"""Initialize service call."""
self.hass = hass
self.connections = hass.data[DATA_LCN][CONF_CONNECTIONS]
def get_device_connection(self, service):
"""Get device connection object."""
addr, connection_id = service.data[CONF_ADDRESS]
addr = pypck.lcn_addr.LcnAddr(*addr)
if connection_id is None:
connection = self.connections[0]
else:
connection = get_connection(self.connections, connection_id)
"""Get address connection object."""
address, host_name = service.data[CONF_ADDRESS]
return connection.get_address_conn(addr)
for config_entry in self.hass.config_entries.async_entries(DOMAIN):
if config_entry.data[CONF_HOST] == host_name:
device_connection = get_device_connection(
self.hass, address, config_entry
)
if device_connection is None:
raise ValueError("Wrong address.")
return device_connection
raise ValueError("Invalid host name.")
async def async_call_service(self, service):
"""Execute service call."""
@ -392,3 +394,20 @@ class Pck(LcnServiceCall):
pck = service.data[CONF_PCK]
device_connection = self.get_device_connection(service)
await device_connection.pck(pck)
SERVICES = (
("output_abs", OutputAbs),
("output_rel", OutputRel),
("output_toggle", OutputToggle),
("relays", Relays),
("var_abs", VarAbs),
("var_reset", VarReset),
("var_rel", VarRel),
("lock_regulator", LockRegulator),
("led", Led),
("send_keys", SendKeys),
("lock_keys", LockKeys),
("dyn_text", DynText),
("pck", Pck),
)

70
homeassistant/components/lcn/switch.py
View file

@ -1,49 +1,49 @@
"""Support for LCN switches."""
import pypck
from homeassistant.components.switch import SwitchEntity
from homeassistant.const import CONF_ADDRESS
from homeassistant.components.switch import DOMAIN as DOMAIN_SWITCH, SwitchEntity
from homeassistant.const import CONF_ADDRESS, CONF_DOMAIN, CONF_ENTITIES
from . import LcnEntity
from .const import CONF_CONNECTIONS, CONF_OUTPUT, DATA_LCN, OUTPUT_PORTS
from .helpers import get_connection
from .const import CONF_DOMAIN_DATA, CONF_OUTPUT, OUTPUT_PORTS
from .helpers import get_device_connection
PARALLEL_UPDATES = 0
async def async_setup_platform(
hass, hass_config, async_add_entities, discovery_info=None
):
"""Set up the LCN switch platform."""
if discovery_info is None:
return
def create_lcn_switch_entity(hass, entity_config, config_entry):
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
)
devices = []
for config in discovery_info:
address, connection_id = config[CONF_ADDRESS]
addr = pypck.lcn_addr.LcnAddr(*address)
connections = hass.data[DATA_LCN][CONF_CONNECTIONS]
connection = get_connection(connections, connection_id)
address_connection = connection.get_address_conn(addr)
if entity_config[CONF_DOMAIN_DATA][CONF_OUTPUT] in OUTPUT_PORTS:
return LcnOutputSwitch(entity_config, config_entry.entry_id, device_connection)
# in RELAY_PORTS
return LcnRelaySwitch(entity_config, config_entry.entry_id, device_connection)
if config[CONF_OUTPUT] in OUTPUT_PORTS:
device = LcnOutputSwitch(config, address_connection)
else: # in RELAY_PORTS
device = LcnRelaySwitch(config, address_connection)
devices.append(device)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up LCN switch entities from a config entry."""
async_add_entities(devices)
entities = []
for entity_config in config_entry.data[CONF_ENTITIES]:
if entity_config[CONF_DOMAIN] == DOMAIN_SWITCH:
entities.append(create_lcn_switch_entity(hass, entity_config, config_entry))
async_add_entities(entities)
class LcnOutputSwitch(LcnEntity, SwitchEntity):
"""Representation of a LCN switch for output ports."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN switch."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
self.output = pypck.lcn_defs.OutputPort[config[CONF_OUTPUT]]
self.output = pypck.lcn_defs.OutputPort[config[CONF_DOMAIN_DATA][CONF_OUTPUT]]
self._is_on = None
@ -53,6 +53,12 @@ class LcnOutputSwitch(LcnEntity, SwitchEntity):
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.output)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.output)
@property
def is_on(self):
"""Return True if entity is on."""
@ -87,11 +93,11 @@ class LcnOutputSwitch(LcnEntity, SwitchEntity):
class LcnRelaySwitch(LcnEntity, SwitchEntity):
"""Representation of a LCN switch for relay ports."""
def __init__(self, config, device_connection):
def __init__(self, config, entry_id, device_connection):
"""Initialize the LCN switch."""
super().__init__(config, device_connection)
super().__init__(config, entry_id, device_connection)
self.output = pypck.lcn_defs.RelayPort[config[CONF_OUTPUT]]
self.output = pypck.lcn_defs.RelayPort[config[CONF_DOMAIN_DATA][CONF_OUTPUT]]
self._is_on = None
@ -101,6 +107,12 @@ class LcnRelaySwitch(LcnEntity, SwitchEntity):
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.output)
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.output)
@property
def is_on(self):
"""Return True if entity is on."""

3
requirements_test_all.txt
View file

@ -863,6 +863,9 @@ pyowm==3.2.0
# homeassistant.components.onewire
pyownet==0.10.0.post1
# homeassistant.components.lcn
pypck==0.7.9
# homeassistant.components.plaato
pyplaato==0.0.15

1
tests/components/lcn/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Tests for LCN."""

92
tests/components/lcn/test_config_flow.py Normal file
View file

@ -0,0 +1,92 @@
"""Tests for the LCN config flow."""
from unittest.mock import patch
from pypck.connection import PchkAuthenticationError, PchkLicenseError
import pytest
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.lcn.const import CONF_DIM_MODE, CONF_SK_NUM_TRIES, DOMAIN
from homeassistant.const import (
CONF_HOST,
CONF_IP_ADDRESS,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
)
from tests.common import MockConfigEntry
IMPORT_DATA = {
CONF_HOST: "pchk",
CONF_IP_ADDRESS: "127.0.0.1",
CONF_PORT: 4114,
CONF_USERNAME: "lcn",
CONF_PASSWORD: "lcn",
CONF_SK_NUM_TRIES: 0,
CONF_DIM_MODE: "STEPS200",
}
async def test_step_import(hass):
"""Test for import step."""
await setup.async_setup_component(hass, "persistent_notification", {})
with patch("pypck.connection.PchkConnectionManager.async_connect"), patch(
"homeassistant.components.lcn.async_setup", return_value=True
), patch("homeassistant.components.lcn.async_setup_entry", return_value=True):
data = IMPORT_DATA.copy()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=data
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "pchk"
assert result["data"] == IMPORT_DATA
async def test_step_import_existing_host(hass):
"""Test for update of config_entry if imported host already exists."""
await setup.async_setup_component(hass, "persistent_notification", {})
# Create config entry and add it to hass
mock_data = IMPORT_DATA.copy()
mock_data.update({CONF_SK_NUM_TRIES: 3, CONF_DIM_MODE: 50})
mock_entry = MockConfigEntry(domain=DOMAIN, data=mock_data)
mock_entry.add_to_hass(hass)
# Inititalize a config flow with different data but same host address
with patch("pypck.connection.PchkConnectionManager.async_connect"):
imported_data = IMPORT_DATA.copy()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=imported_data
)
await hass.async_block_till_done()
# Check if config entry was updated
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "existing_configuration_updated"
assert mock_entry.source == config_entries.SOURCE_IMPORT
assert mock_entry.data == IMPORT_DATA
@pytest.mark.parametrize(
"error,reason",
[
(PchkAuthenticationError, "authentication_error"),
(PchkLicenseError, "license_error"),
(TimeoutError, "connection_timeout"),
],
)
async def test_step_import_error(hass, error, reason):
"""Test for authentication error is handled correctly."""
await setup.async_setup_component(hass, "persistent_notification", {})
with patch(
"pypck.connection.PchkConnectionManager.async_connect", side_effect=error
):
data = IMPORT_DATA.copy()
data.update({CONF_HOST: "pchk"})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=data
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == reason