mirror of https://github.com/home-assistant/core
249 lines
7.8 KiB
Python
249 lines
7.8 KiB
Python
"""Support for LCN devices."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from functools import partial
|
|
import logging
|
|
|
|
import pypck
|
|
from pypck.connection import PchkConnectionManager
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.const import (
|
|
CONF_DEVICE_ID,
|
|
CONF_DOMAIN,
|
|
CONF_ENTITIES,
|
|
CONF_IP_ADDRESS,
|
|
CONF_PASSWORD,
|
|
CONF_PORT,
|
|
CONF_USERNAME,
|
|
Platform,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers import config_validation as cv, device_registry as dr
|
|
from homeassistant.helpers.typing import ConfigType
|
|
|
|
from .const import (
|
|
ADD_ENTITIES_CALLBACKS,
|
|
CONF_ACKNOWLEDGE,
|
|
CONF_DIM_MODE,
|
|
CONF_DOMAIN_DATA,
|
|
CONF_SK_NUM_TRIES,
|
|
CONF_TRANSITION,
|
|
CONNECTION,
|
|
DOMAIN,
|
|
PLATFORMS,
|
|
)
|
|
from .helpers import (
|
|
AddressType,
|
|
InputType,
|
|
async_update_config_entry,
|
|
generate_unique_id,
|
|
register_lcn_address_devices,
|
|
register_lcn_host_device,
|
|
)
|
|
from .services import register_services
|
|
from .websocket import register_panel_and_ws_api
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
|
|
|
|
|
|
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
|
"""Set up the LCN component."""
|
|
hass.data.setdefault(DOMAIN, {})
|
|
|
|
await register_services(hass)
|
|
await register_panel_and_ws_api(hass)
|
|
|
|
return True
|
|
|
|
|
|
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
|
|
"""Set up a connection to PCHK host from a config entry."""
|
|
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]],
|
|
"ACKNOWLEDGE": config_entry.data[CONF_ACKNOWLEDGE],
|
|
}
|
|
|
|
# connect to PCHK
|
|
lcn_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,
|
|
ADD_ENTITIES_CALLBACKS: {},
|
|
}
|
|
# Update config_entry with LCN device serials
|
|
await async_update_config_entry(hass, config_entry)
|
|
|
|
# register/update devices for host, modules and groups in device registry
|
|
register_lcn_host_device(hass, config_entry)
|
|
register_lcn_address_devices(hass, config_entry)
|
|
|
|
# forward config_entry to components
|
|
await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
|
|
|
|
# register for LCN bus messages
|
|
device_registry = dr.async_get(hass)
|
|
input_received = partial(
|
|
async_host_input_received, hass, config_entry, device_registry
|
|
)
|
|
lcn_connection.register_for_inputs(input_received)
|
|
|
|
return True
|
|
|
|
|
|
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
|
|
"""Migrate old entry."""
|
|
_LOGGER.debug(
|
|
"Migrating configuration from version %s.%s",
|
|
config_entry.version,
|
|
config_entry.minor_version,
|
|
)
|
|
|
|
new_data = {**config_entry.data}
|
|
|
|
if config_entry.version == 1:
|
|
# update to 1.2 (add acknowledge flag)
|
|
if config_entry.minor_version < 2:
|
|
new_data[CONF_ACKNOWLEDGE] = False
|
|
|
|
# update to 2.1 (fix transitions for lights and switches)
|
|
new_entities_data = [*new_data[CONF_ENTITIES]]
|
|
for entity in new_entities_data:
|
|
if entity[CONF_DOMAIN] in [Platform.LIGHT, Platform.SCENE]:
|
|
if entity[CONF_DOMAIN_DATA][CONF_TRANSITION] is None:
|
|
entity[CONF_DOMAIN_DATA][CONF_TRANSITION] = 0
|
|
entity[CONF_DOMAIN_DATA][CONF_TRANSITION] /= 1000.0
|
|
new_data[CONF_ENTITIES] = new_entities_data
|
|
|
|
hass.config_entries.async_update_entry(
|
|
config_entry, data=new_data, minor_version=1, version=2
|
|
)
|
|
|
|
_LOGGER.debug(
|
|
"Migration to configuration version %s.%s successful",
|
|
config_entry.version,
|
|
config_entry.minor_version,
|
|
)
|
|
return True
|
|
|
|
|
|
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
|
|
"""Close connection to PCHK host represented by config_entry."""
|
|
# forward unloading to platforms
|
|
unload_ok = await hass.config_entries.async_unload_platforms(
|
|
config_entry, PLATFORMS
|
|
)
|
|
|
|
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()
|
|
|
|
return unload_ok
|
|
|
|
|
|
def async_host_input_received(
|
|
hass: HomeAssistant,
|
|
config_entry: ConfigEntry,
|
|
device_registry: dr.DeviceRegistry,
|
|
inp: pypck.inputs.Input,
|
|
) -> None:
|
|
"""Process received input object (command) from LCN bus."""
|
|
if not isinstance(inp, pypck.inputs.ModInput):
|
|
return
|
|
|
|
lcn_connection = hass.data[DOMAIN][config_entry.entry_id][CONNECTION]
|
|
logical_address = lcn_connection.physical_to_logical(inp.physical_source_addr)
|
|
address = (
|
|
logical_address.seg_id,
|
|
logical_address.addr_id,
|
|
logical_address.is_group,
|
|
)
|
|
identifiers = {(DOMAIN, generate_unique_id(config_entry.entry_id, address))}
|
|
device = device_registry.async_get_device(identifiers=identifiers)
|
|
if device is None:
|
|
return
|
|
|
|
if isinstance(inp, pypck.inputs.ModStatusAccessControl):
|
|
_async_fire_access_control_event(hass, device, address, inp)
|
|
elif isinstance(inp, pypck.inputs.ModSendKeysHost):
|
|
_async_fire_send_keys_event(hass, device, address, inp)
|
|
|
|
|
|
def _async_fire_access_control_event(
|
|
hass: HomeAssistant, device: dr.DeviceEntry, address: AddressType, inp: InputType
|
|
) -> None:
|
|
"""Fire access control event (transponder, transmitter, fingerprint, codelock)."""
|
|
event_data = {
|
|
"segment_id": address[0],
|
|
"module_id": address[1],
|
|
"code": inp.code,
|
|
}
|
|
|
|
if device is not None:
|
|
event_data.update({CONF_DEVICE_ID: device.id})
|
|
|
|
if inp.periphery == pypck.lcn_defs.AccessControlPeriphery.TRANSMITTER:
|
|
event_data.update(
|
|
{"level": inp.level, "key": inp.key, "action": inp.action.value}
|
|
)
|
|
|
|
event_name = f"lcn_{inp.periphery.value.lower()}"
|
|
hass.bus.async_fire(event_name, event_data)
|
|
|
|
|
|
def _async_fire_send_keys_event(
|
|
hass: HomeAssistant, device: dr.DeviceEntry, address: AddressType, inp: InputType
|
|
) -> None:
|
|
"""Fire send_keys event."""
|
|
for table, action in enumerate(inp.actions):
|
|
if action == pypck.lcn_defs.SendKeyCommand.DONTSEND:
|
|
continue
|
|
|
|
for key, selected in enumerate(inp.keys):
|
|
if not selected:
|
|
continue
|
|
event_data = {
|
|
"segment_id": address[0],
|
|
"module_id": address[1],
|
|
"key": pypck.lcn_defs.Key(table * 8 + key).name.lower(),
|
|
"action": action.name.lower(),
|
|
}
|
|
|
|
if device is not None:
|
|
event_data.update({CONF_DEVICE_ID: device.id})
|
|
|
|
hass.bus.async_fire("lcn_send_keys", event_data)
|