KNX: move some Schema to schema.py (#51307)

* create platform schema node from schema class

* move connection schema to schema.py

* rename SCHEMA to ENTITY_SCHEMA

* Final module level constants
pull/51324/head
Matthias Alphart 2021-06-01 08:59:23 +02:00 committed by GitHub
parent fb682665e2
commit 164e45f0a7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 143 additions and 118 deletions

View File

@ -3,18 +3,14 @@ from __future__ import annotations
import asyncio
import logging
from typing import Final
import voluptuous as vol
from xknx import XKNX
from xknx.core.telegram_queue import TelegramQueue
from xknx.dpt import DPTArray, DPTBase, DPTBinary
from xknx.exceptions import XKNXException
from xknx.io import (
DEFAULT_MCAST_GRP,
DEFAULT_MCAST_PORT,
ConnectionConfig,
ConnectionType,
)
from xknx.io import ConnectionConfig, ConnectionType
from xknx.telegram import AddressFilter, Telegram
from xknx.telegram.address import parse_device_group_address
from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite
@ -34,7 +30,15 @@ from homeassistant.helpers.reload import async_integration_yaml_config
from homeassistant.helpers.service import async_register_admin_service
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, KNX_ADDRESS, SupportedPlatforms
from .const import (
CONF_KNX_EXPOSE,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_ROUTING,
CONF_KNX_TUNNELING,
DOMAIN,
KNX_ADDRESS,
SupportedPlatforms,
)
from .expose import KNXExposeSensor, KNXExposeTime, create_knx_exposure
from .schema import (
BinarySensorSchema,
@ -50,30 +54,22 @@ from .schema import (
SwitchSchema,
WeatherSchema,
ga_validator,
ia_validator,
sensor_type_validator,
)
_LOGGER = logging.getLogger(__name__)
CONF_KNX_ROUTING = "routing"
CONF_KNX_TUNNELING = "tunneling"
CONF_KNX_FIRE_EVENT = "fire_event"
CONF_KNX_EVENT_FILTER = "event_filter"
CONF_KNX_INDIVIDUAL_ADDRESS = "individual_address"
CONF_KNX_MCAST_GRP = "multicast_group"
CONF_KNX_MCAST_PORT = "multicast_port"
CONF_KNX_STATE_UPDATER = "state_updater"
CONF_KNX_RATE_LIMIT = "rate_limit"
CONF_KNX_EXPOSE = "expose"
SERVICE_KNX_SEND = "send"
SERVICE_KNX_ATTR_PAYLOAD = "payload"
SERVICE_KNX_ATTR_TYPE = "type"
SERVICE_KNX_ATTR_REMOVE = "remove"
SERVICE_KNX_EVENT_REGISTER = "event_register"
SERVICE_KNX_EXPOSURE_REGISTER = "exposure_register"
SERVICE_KNX_READ = "read"
CONF_KNX_FIRE_EVENT: Final = "fire_event"
CONF_KNX_EVENT_FILTER: Final = "event_filter"
SERVICE_KNX_SEND: Final = "send"
SERVICE_KNX_ATTR_PAYLOAD: Final = "payload"
SERVICE_KNX_ATTR_TYPE: Final = "type"
SERVICE_KNX_ATTR_REMOVE: Final = "remove"
SERVICE_KNX_EVENT_REGISTER: Final = "event_register"
SERVICE_KNX_EXPOSURE_REGISTER: Final = "exposure_register"
SERVICE_KNX_READ: Final = "read"
CONFIG_SCHEMA = vol.Schema(
{
@ -85,62 +81,22 @@ CONFIG_SCHEMA = vol.Schema(
cv.deprecated("fire_event_filter", replacement_key=CONF_KNX_EVENT_FILTER),
vol.Schema(
{
vol.Exclusive(
CONF_KNX_ROUTING, "connection_type"
): ConnectionSchema.ROUTING_SCHEMA,
vol.Exclusive(
CONF_KNX_TUNNELING, "connection_type"
): ConnectionSchema.TUNNELING_SCHEMA,
**ConnectionSchema.SCHEMA,
vol.Optional(CONF_KNX_FIRE_EVENT): cv.boolean,
vol.Optional(CONF_KNX_EVENT_FILTER, default=[]): vol.All(
cv.ensure_list, [cv.string]
),
vol.Optional(
CONF_KNX_INDIVIDUAL_ADDRESS, default=XKNX.DEFAULT_ADDRESS
): ia_validator,
vol.Optional(
CONF_KNX_MCAST_GRP, default=DEFAULT_MCAST_GRP
): cv.string,
vol.Optional(
CONF_KNX_MCAST_PORT, default=DEFAULT_MCAST_PORT
): cv.port,
vol.Optional(CONF_KNX_STATE_UPDATER, default=True): cv.boolean,
vol.Optional(CONF_KNX_RATE_LIMIT, default=20): vol.All(
vol.Coerce(int), vol.Range(min=1, max=100)
),
vol.Optional(CONF_KNX_EXPOSE): vol.All(
cv.ensure_list, [ExposeSchema.SCHEMA]
),
vol.Optional(SupportedPlatforms.COVER.value): vol.All(
cv.ensure_list, [CoverSchema.SCHEMA]
),
vol.Optional(SupportedPlatforms.BINARY_SENSOR.value): vol.All(
cv.ensure_list, [BinarySensorSchema.SCHEMA]
),
vol.Optional(SupportedPlatforms.LIGHT.value): vol.All(
cv.ensure_list, [LightSchema.SCHEMA]
),
vol.Optional(SupportedPlatforms.CLIMATE.value): vol.All(
cv.ensure_list, [ClimateSchema.SCHEMA]
),
vol.Optional(SupportedPlatforms.NOTIFY.value): vol.All(
cv.ensure_list, [NotifySchema.SCHEMA]
),
vol.Optional(SupportedPlatforms.SWITCH.value): vol.All(
cv.ensure_list, [SwitchSchema.SCHEMA]
),
vol.Optional(SupportedPlatforms.SENSOR.value): vol.All(
cv.ensure_list, [SensorSchema.SCHEMA]
),
vol.Optional(SupportedPlatforms.SCENE.value): vol.All(
cv.ensure_list, [SceneSchema.SCHEMA]
),
vol.Optional(SupportedPlatforms.WEATHER.value): vol.All(
cv.ensure_list, [WeatherSchema.SCHEMA]
),
vol.Optional(SupportedPlatforms.FAN.value): vol.All(
cv.ensure_list, [FanSchema.SCHEMA]
),
**ExposeSchema.platform_node(),
**BinarySensorSchema.platform_node(),
**ClimateSchema.platform_node(),
**CoverSchema.platform_node(),
**FanSchema.platform_node(),
**LightSchema.platform_node(),
**NotifySchema.platform_node(),
**SceneSchema.platform_node(),
**SensorSchema.platform_node(),
**SwitchSchema.platform_node(),
**WeatherSchema.platform_node(),
}
),
)
@ -315,11 +271,11 @@ class KNXModule:
"""Initialize XKNX object."""
self.xknx = XKNX(
own_address=self.config[DOMAIN][CONF_KNX_INDIVIDUAL_ADDRESS],
rate_limit=self.config[DOMAIN][CONF_KNX_RATE_LIMIT],
multicast_group=self.config[DOMAIN][CONF_KNX_MCAST_GRP],
multicast_port=self.config[DOMAIN][CONF_KNX_MCAST_PORT],
rate_limit=self.config[DOMAIN][ConnectionSchema.CONF_KNX_RATE_LIMIT],
multicast_group=self.config[DOMAIN][ConnectionSchema.CONF_KNX_MCAST_GRP],
multicast_port=self.config[DOMAIN][ConnectionSchema.CONF_KNX_MCAST_PORT],
connection_config=self.connection_config(),
state_updater=self.config[DOMAIN][CONF_KNX_STATE_UPDATER],
state_updater=self.config[DOMAIN][ConnectionSchema.CONF_KNX_STATE_UPDATER],
)
async def start(self) -> None:

View File

@ -1,5 +1,6 @@
"""Constants for the KNX integration."""
from enum import Enum
from typing import Final
from homeassistant.components.climate.const import (
HVAC_MODE_AUTO,
@ -15,19 +16,23 @@ from homeassistant.components.climate.const import (
PRESET_SLEEP,
)
DOMAIN = "knx"
DOMAIN: Final = "knx"
# Address is used for configuration and services by the same functions so the key has to match
KNX_ADDRESS = "address"
KNX_ADDRESS: Final = "address"
CONF_INVERT = "invert"
CONF_STATE_ADDRESS = "state_address"
CONF_SYNC_STATE = "sync_state"
CONF_RESET_AFTER = "reset_after"
CONF_KNX_ROUTING: Final = "routing"
CONF_KNX_TUNNELING: Final = "tunneling"
CONF_KNX_INDIVIDUAL_ADDRESS: Final = "individual_address"
CONF_INVERT: Final = "invert"
CONF_KNX_EXPOSE: Final = "expose"
CONF_STATE_ADDRESS: Final = "state_address"
CONF_SYNC_STATE: Final = "sync_state"
CONF_RESET_AFTER: Final = "reset_after"
ATTR_COUNTER = "counter"
ATTR_SOURCE = "source"
ATTR_LAST_KNX_UPDATE = "last_knx_update"
ATTR_COUNTER: Final = "counter"
ATTR_SOURCE: Final = "source"
ATTR_LAST_KNX_UPDATE: Final = "last_knx_update"
class ColorTempModes(Enum):
@ -53,7 +58,7 @@ class SupportedPlatforms(Enum):
# Map KNX controller modes to HA modes. This list might not be complete.
CONTROLLER_MODES = {
CONTROLLER_MODES: Final = {
# Map DPT 20.105 HVAC control modes
"Auto": HVAC_MODE_AUTO,
"Heat": HVAC_MODE_HEAT,
@ -63,7 +68,7 @@ CONTROLLER_MODES = {
"Dry": HVAC_MODE_DRY,
}
PRESET_MODES = {
PRESET_MODES: Final = {
# Map DPT 20.102 HVAC operating modes to HA presets
"Auto": PRESET_NONE,
"Frost Protection": PRESET_ECO,

View File

@ -2,7 +2,7 @@
from __future__ import annotations
import math
from typing import Any
from typing import Any, Final
from xknx import XKNX
from xknx.devices import Fan as XknxFan
@ -22,7 +22,7 @@ from .const import DOMAIN, KNX_ADDRESS
from .knx_entity import KnxEntity
from .schema import FanSchema
DEFAULT_PERCENTAGE = 50
DEFAULT_PERCENTAGE: Final = 50
async def async_setup_platform(

View File

@ -1,13 +1,15 @@
"""Voluptuous schemas for the KNX integration."""
from __future__ import annotations
from typing import Any
from abc import ABC
from typing import Any, ClassVar
import voluptuous as vol
from xknx import XKNX
from xknx.devices.climate import SetpointShiftMode
from xknx.dpt import DPTBase
from xknx.exceptions import CouldNotParseAddress
from xknx.io import DEFAULT_MCAST_PORT
from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
from xknx.telegram.address import IndividualAddress, parse_device_group_address
from homeassistant.components.binary_sensor import (
@ -26,6 +28,10 @@ import homeassistant.helpers.config_validation as cv
from .const import (
CONF_INVERT,
CONF_KNX_EXPOSE,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_ROUTING,
CONF_KNX_TUNNELING,
CONF_RESET_AFTER,
CONF_STATE_ADDRESS,
CONF_SYNC_STATE,
@ -33,6 +39,7 @@ from .const import (
KNX_ADDRESS,
PRESET_MODES,
ColorTempModes,
SupportedPlatforms,
)
##################
@ -76,6 +83,7 @@ sync_state_validator = vol.Any(
cv.matches_regex(r"^(init|expire|every)( \d*)?$"),
)
##############
# CONNECTION
##############
@ -85,7 +93,11 @@ class ConnectionSchema:
"""Voluptuous schema for KNX connection."""
CONF_KNX_LOCAL_IP = "local_ip"
CONF_KNX_MCAST_GRP = "multicast_group"
CONF_KNX_MCAST_PORT = "multicast_port"
CONF_KNX_RATE_LIMIT = "rate_limit"
CONF_KNX_ROUTE_BACK = "route_back"
CONF_KNX_STATE_UPDATER = "state_updater"
TUNNELING_SCHEMA = vol.Schema(
{
@ -98,15 +110,47 @@ class ConnectionSchema:
ROUTING_SCHEMA = vol.Maybe(vol.Schema({vol.Optional(CONF_KNX_LOCAL_IP): cv.string}))
SCHEMA = {
vol.Exclusive(CONF_KNX_ROUTING, "connection_type"): ROUTING_SCHEMA,
vol.Exclusive(CONF_KNX_TUNNELING, "connection_type"): TUNNELING_SCHEMA,
vol.Optional(
CONF_KNX_INDIVIDUAL_ADDRESS, default=XKNX.DEFAULT_ADDRESS
): ia_validator,
vol.Optional(CONF_KNX_MCAST_GRP, default=DEFAULT_MCAST_GRP): cv.string,
vol.Optional(CONF_KNX_MCAST_PORT, default=DEFAULT_MCAST_PORT): cv.port,
vol.Optional(CONF_KNX_STATE_UPDATER, default=True): cv.boolean,
vol.Optional(CONF_KNX_RATE_LIMIT, default=20): vol.All(
vol.Coerce(int), vol.Range(min=1, max=100)
),
}
#############
# PLATFORMS
#############
class BinarySensorSchema:
class KNXPlatformSchema(ABC):
"""Voluptuous schema for KNX platform entity configuration."""
PLATFORM_NAME: ClassVar[str]
ENTITY_SCHEMA: ClassVar[vol.Schema]
@classmethod
def platform_node(cls) -> dict[vol.Optional, vol.All]:
"""Return a schema node for the platform."""
return {
vol.Optional(cls.PLATFORM_NAME): vol.All(
cv.ensure_list, [cls.ENTITY_SCHEMA]
)
}
class BinarySensorSchema(KNXPlatformSchema):
"""Voluptuous schema for KNX binary sensors."""
PLATFORM_NAME = SupportedPlatforms.BINARY_SENSOR.value
CONF_STATE_ADDRESS = CONF_STATE_ADDRESS
CONF_SYNC_STATE = CONF_SYNC_STATE
CONF_INVERT = CONF_INVERT
@ -116,7 +160,7 @@ class BinarySensorSchema:
DEFAULT_NAME = "KNX Binary Sensor"
SCHEMA = vol.All(
ENTITY_SCHEMA = vol.All(
# deprecated since September 2020
cv.deprecated("significant_bit"),
cv.deprecated("automation"),
@ -137,9 +181,11 @@ class BinarySensorSchema:
)
class ClimateSchema:
class ClimateSchema(KNXPlatformSchema):
"""Voluptuous schema for KNX climate devices."""
PLATFORM_NAME = SupportedPlatforms.CLIMATE.value
CONF_SETPOINT_SHIFT_ADDRESS = "setpoint_shift_address"
CONF_SETPOINT_SHIFT_STATE_ADDRESS = "setpoint_shift_state_address"
CONF_SETPOINT_SHIFT_MODE = "setpoint_shift_mode"
@ -178,7 +224,7 @@ class ClimateSchema:
DEFAULT_TEMPERATURE_STEP = 0.1
DEFAULT_ON_OFF_INVERT = False
SCHEMA = vol.All(
ENTITY_SCHEMA = vol.All(
# deprecated since September 2020
cv.deprecated("setpoint_shift_step", replacement_key=CONF_TEMPERATURE_STEP),
# deprecated since 2021.6
@ -245,9 +291,11 @@ class ClimateSchema:
)
class CoverSchema:
class CoverSchema(KNXPlatformSchema):
"""Voluptuous schema for KNX covers."""
PLATFORM_NAME = SupportedPlatforms.COVER.value
CONF_MOVE_LONG_ADDRESS = "move_long_address"
CONF_MOVE_SHORT_ADDRESS = "move_short_address"
CONF_STOP_ADDRESS = "stop_address"
@ -263,7 +311,7 @@ class CoverSchema:
DEFAULT_TRAVEL_TIME = 25
DEFAULT_NAME = "KNX Cover"
SCHEMA = vol.All(
ENTITY_SCHEMA = vol.All(
vol.Schema(
{
vol.Required(
@ -297,9 +345,11 @@ class CoverSchema:
)
class ExposeSchema:
class ExposeSchema(KNXPlatformSchema):
"""Voluptuous schema for KNX exposures."""
PLATFORM_NAME = CONF_KNX_EXPOSE
CONF_KNX_EXPOSE_TYPE = CONF_TYPE
CONF_KNX_EXPOSE_ATTRIBUTE = "attribute"
CONF_KNX_EXPOSE_BINARY = "binary"
@ -329,12 +379,14 @@ class ExposeSchema:
vol.Optional(CONF_KNX_EXPOSE_DEFAULT): cv.match_all,
}
)
SCHEMA = vol.Any(EXPOSE_SENSOR_SCHEMA, EXPOSE_TIME_SCHEMA)
ENTITY_SCHEMA = vol.Any(EXPOSE_SENSOR_SCHEMA, EXPOSE_TIME_SCHEMA)
class FanSchema:
class FanSchema(KNXPlatformSchema):
"""Voluptuous schema for KNX fans."""
PLATFORM_NAME = SupportedPlatforms.FAN.value
CONF_STATE_ADDRESS = CONF_STATE_ADDRESS
CONF_OSCILLATION_ADDRESS = "oscillation_address"
CONF_OSCILLATION_STATE_ADDRESS = "oscillation_state_address"
@ -342,7 +394,7 @@ class FanSchema:
DEFAULT_NAME = "KNX Fan"
SCHEMA = vol.Schema(
ENTITY_SCHEMA = vol.Schema(
{
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(KNX_ADDRESS): ga_list_validator,
@ -354,9 +406,11 @@ class FanSchema:
)
class LightSchema:
class LightSchema(KNXPlatformSchema):
"""Voluptuous schema for KNX lights."""
PLATFORM_NAME = SupportedPlatforms.LIGHT.value
CONF_STATE_ADDRESS = CONF_STATE_ADDRESS
CONF_BRIGHTNESS_ADDRESS = "brightness_address"
CONF_BRIGHTNESS_STATE_ADDRESS = "brightness_state_address"
@ -390,7 +444,7 @@ class LightSchema:
}
)
SCHEMA = vol.All(
ENTITY_SCHEMA = vol.All(
vol.Schema(
{
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
@ -452,12 +506,14 @@ class LightSchema:
)
class NotifySchema:
class NotifySchema(KNXPlatformSchema):
"""Voluptuous schema for KNX notifications."""
PLATFORM_NAME = SupportedPlatforms.NOTIFY.value
DEFAULT_NAME = "KNX Notify"
SCHEMA = vol.Schema(
ENTITY_SCHEMA = vol.Schema(
{
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(KNX_ADDRESS): ga_validator,
@ -465,13 +521,15 @@ class NotifySchema:
)
class SceneSchema:
class SceneSchema(KNXPlatformSchema):
"""Voluptuous schema for KNX scenes."""
PLATFORM_NAME = SupportedPlatforms.SCENE.value
CONF_SCENE_NUMBER = "scene_number"
DEFAULT_NAME = "KNX SCENE"
SCHEMA = vol.Schema(
ENTITY_SCHEMA = vol.Schema(
{
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(KNX_ADDRESS): ga_list_validator,
@ -482,15 +540,17 @@ class SceneSchema:
)
class SensorSchema:
class SensorSchema(KNXPlatformSchema):
"""Voluptuous schema for KNX sensors."""
PLATFORM_NAME = SupportedPlatforms.SENSOR.value
CONF_ALWAYS_CALLBACK = "always_callback"
CONF_STATE_ADDRESS = CONF_STATE_ADDRESS
CONF_SYNC_STATE = CONF_SYNC_STATE
DEFAULT_NAME = "KNX Sensor"
SCHEMA = vol.Schema(
ENTITY_SCHEMA = vol.Schema(
{
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_SYNC_STATE, default=True): sync_state_validator,
@ -501,14 +561,16 @@ class SensorSchema:
)
class SwitchSchema:
class SwitchSchema(KNXPlatformSchema):
"""Voluptuous schema for KNX switches."""
PLATFORM_NAME = SupportedPlatforms.SWITCH.value
CONF_INVERT = CONF_INVERT
CONF_STATE_ADDRESS = CONF_STATE_ADDRESS
DEFAULT_NAME = "KNX Switch"
SCHEMA = vol.Schema(
ENTITY_SCHEMA = vol.Schema(
{
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_INVERT, default=False): cv.boolean,
@ -518,9 +580,11 @@ class SwitchSchema:
)
class WeatherSchema:
class WeatherSchema(KNXPlatformSchema):
"""Voluptuous schema for KNX weather station."""
PLATFORM_NAME = SupportedPlatforms.WEATHER.value
CONF_SYNC_STATE = CONF_SYNC_STATE
CONF_KNX_TEMPERATURE_ADDRESS = "address_temperature"
CONF_KNX_BRIGHTNESS_SOUTH_ADDRESS = "address_brightness_south"
@ -538,7 +602,7 @@ class WeatherSchema:
DEFAULT_NAME = "KNX Weather Station"
SCHEMA = vol.All(
ENTITY_SCHEMA = vol.All(
# deprecated since 2021.6
cv.deprecated("create_sensors"),
vol.Schema(