Add Config Flow to AlarmDecoder (#37998)

pull/40034/head
AJ Schmidt 2020-09-13 13:29:25 -04:00 committed by GitHub
parent 17efa1bda5
commit c32f698671
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 1169 additions and 210 deletions

View File

@ -33,7 +33,11 @@ omit =
homeassistant/components/airvisual/air_quality.py
homeassistant/components/airvisual/sensor.py
homeassistant/components/aladdin_connect/cover.py
homeassistant/components/alarmdecoder/*
homeassistant/components/alarmdecoder/__init__.py
homeassistant/components/alarmdecoder/alarm_control_panel.py
homeassistant/components/alarmdecoder/binary_sensor.py
homeassistant/components/alarmdecoder/const.py
homeassistant/components/alarmdecoder/sensor.py
homeassistant/components/alpha_vantage/sensor.py
homeassistant/components/amazon_polly/tts.py
homeassistant/components/ambiclimate/climate.py

View File

@ -1,167 +1,82 @@
"""Support for AlarmDecoder devices."""
import asyncio
from datetime import timedelta
import logging
from adext import AdExt
from alarmdecoder.devices import SerialDevice, SocketDevice, USBDevice
from alarmdecoder.devices import SerialDevice, SocketDevice
from alarmdecoder.util import NoDeviceError
import voluptuous as vol
from homeassistant.components.binary_sensor import DEVICE_CLASSES_SCHEMA
from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STOP
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.discovery import load_platform
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_PORT,
CONF_PROTOCOL,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util import dt as dt_util
from .const import (
CONF_DEVICE_BAUD,
CONF_DEVICE_PATH,
DATA_AD,
DATA_REMOVE_STOP_LISTENER,
DATA_REMOVE_UPDATE_LISTENER,
DATA_RESTART,
DOMAIN,
PROTOCOL_SERIAL,
PROTOCOL_SOCKET,
SIGNAL_PANEL_MESSAGE,
SIGNAL_REL_MESSAGE,
SIGNAL_RFX_MESSAGE,
SIGNAL_ZONE_FAULT,
SIGNAL_ZONE_RESTORE,
)
_LOGGER = logging.getLogger(__name__)
DOMAIN = "alarmdecoder"
DATA_AD = "alarmdecoder"
CONF_DEVICE = "device"
CONF_DEVICE_BAUD = "baudrate"
CONF_DEVICE_PATH = "path"
CONF_DEVICE_PORT = "port"
CONF_DEVICE_TYPE = "type"
CONF_AUTO_BYPASS = "autobypass"
CONF_PANEL_DISPLAY = "panel_display"
CONF_ZONE_NAME = "name"
CONF_ZONE_TYPE = "type"
CONF_ZONE_LOOP = "loop"
CONF_ZONE_RFID = "rfid"
CONF_ZONES = "zones"
CONF_RELAY_ADDR = "relayaddr"
CONF_RELAY_CHAN = "relaychan"
CONF_CODE_ARM_REQUIRED = "code_arm_required"
DEFAULT_DEVICE_TYPE = "socket"
DEFAULT_DEVICE_HOST = "localhost"
DEFAULT_DEVICE_PORT = 10000
DEFAULT_DEVICE_PATH = "/dev/ttyUSB0"
DEFAULT_DEVICE_BAUD = 115200
DEFAULT_AUTO_BYPASS = False
DEFAULT_PANEL_DISPLAY = False
DEFAULT_CODE_ARM_REQUIRED = True
DEFAULT_ZONE_TYPE = "opening"
SIGNAL_PANEL_MESSAGE = "alarmdecoder.panel_message"
SIGNAL_PANEL_ARM_AWAY = "alarmdecoder.panel_arm_away"
SIGNAL_PANEL_ARM_HOME = "alarmdecoder.panel_arm_home"
SIGNAL_PANEL_DISARM = "alarmdecoder.panel_disarm"
SIGNAL_ZONE_FAULT = "alarmdecoder.zone_fault"
SIGNAL_ZONE_RESTORE = "alarmdecoder.zone_restore"
SIGNAL_RFX_MESSAGE = "alarmdecoder.rfx_message"
SIGNAL_REL_MESSAGE = "alarmdecoder.rel_message"
DEVICE_SOCKET_SCHEMA = vol.Schema(
{
vol.Required(CONF_DEVICE_TYPE): "socket",
vol.Optional(CONF_HOST, default=DEFAULT_DEVICE_HOST): cv.string,
vol.Optional(CONF_DEVICE_PORT, default=DEFAULT_DEVICE_PORT): cv.port,
}
)
DEVICE_SERIAL_SCHEMA = vol.Schema(
{
vol.Required(CONF_DEVICE_TYPE): "serial",
vol.Optional(CONF_DEVICE_PATH, default=DEFAULT_DEVICE_PATH): cv.string,
vol.Optional(CONF_DEVICE_BAUD, default=DEFAULT_DEVICE_BAUD): cv.string,
}
)
DEVICE_USB_SCHEMA = vol.Schema({vol.Required(CONF_DEVICE_TYPE): "usb"})
ZONE_SCHEMA = vol.Schema(
{
vol.Required(CONF_ZONE_NAME): cv.string,
vol.Optional(CONF_ZONE_TYPE, default=DEFAULT_ZONE_TYPE): vol.Any(
DEVICE_CLASSES_SCHEMA
),
vol.Optional(CONF_ZONE_RFID): cv.string,
vol.Optional(CONF_ZONE_LOOP): vol.All(vol.Coerce(int), vol.Range(min=1, max=4)),
vol.Inclusive(
CONF_RELAY_ADDR,
"relaylocation",
"Relay address and channel must exist together",
): cv.byte,
vol.Inclusive(
CONF_RELAY_CHAN,
"relaylocation",
"Relay address and channel must exist together",
): cv.byte,
}
)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_DEVICE): vol.Any(
DEVICE_SOCKET_SCHEMA, DEVICE_SERIAL_SCHEMA, DEVICE_USB_SCHEMA
),
vol.Optional(
CONF_PANEL_DISPLAY, default=DEFAULT_PANEL_DISPLAY
): cv.boolean,
vol.Optional(CONF_AUTO_BYPASS, default=DEFAULT_AUTO_BYPASS): cv.boolean,
vol.Optional(
CONF_CODE_ARM_REQUIRED, default=DEFAULT_CODE_ARM_REQUIRED
): cv.boolean,
vol.Optional(CONF_ZONES): {vol.Coerce(int): ZONE_SCHEMA},
}
)
},
extra=vol.ALLOW_EXTRA,
)
PLATFORMS = ["alarm_control_panel", "sensor", "binary_sensor"]
def setup(hass, config):
async def async_setup(hass, config):
"""Set up for the AlarmDecoder devices."""
conf = config.get(DOMAIN)
return True
restart = False
device = conf[CONF_DEVICE]
display = conf[CONF_PANEL_DISPLAY]
auto_bypass = conf[CONF_AUTO_BYPASS]
code_arm_required = conf[CONF_CODE_ARM_REQUIRED]
zones = conf.get(CONF_ZONES)
device_type = device[CONF_DEVICE_TYPE]
host = DEFAULT_DEVICE_HOST
port = DEFAULT_DEVICE_PORT
path = DEFAULT_DEVICE_PATH
baud = DEFAULT_DEVICE_BAUD
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
"""Set up AlarmDecoder config flow."""
undo_listener = entry.add_update_listener(_update_listener)
ad_connection = entry.data
protocol = ad_connection[CONF_PROTOCOL]
def stop_alarmdecoder(event):
"""Handle the shutdown of AlarmDecoder."""
if not hass.data.get(DOMAIN):
return
_LOGGER.debug("Shutting down alarmdecoder")
nonlocal restart
restart = False
hass.data[DOMAIN][entry.entry_id][DATA_RESTART] = False
controller.close()
def open_connection(now=None):
"""Open a connection to AlarmDecoder."""
nonlocal restart
try:
controller.open(baud)
except NoDeviceError:
_LOGGER.debug("Failed to connect. Retrying in 5 seconds")
_LOGGER.debug("Failed to connect. Retrying in 5 seconds")
hass.helpers.event.track_point_in_time(
open_connection, dt_util.utcnow() + timedelta(seconds=5)
)
return
_LOGGER.debug("Established a connection with the alarmdecoder")
restart = True
hass.data[DOMAIN][entry.entry_id][DATA_RESTART] = True
def handle_closed_connection(event):
"""Restart after unexpected loss of connection."""
nonlocal restart
if not restart:
if not hass.data[DOMAIN][entry.entry_id][DATA_RESTART]:
return
restart = False
hass.data[DOMAIN][entry.entry_id][DATA_RESTART] = False
_LOGGER.warning("AlarmDecoder unexpectedly lost connection")
hass.add_job(open_connection)
@ -186,17 +101,14 @@ def setup(hass, config):
hass.helpers.dispatcher.dispatcher_send(SIGNAL_REL_MESSAGE, message)
controller = False
if device_type == "socket":
host = device[CONF_HOST]
port = device[CONF_DEVICE_PORT]
baud = ad_connection[CONF_DEVICE_BAUD]
if protocol == PROTOCOL_SOCKET:
host = ad_connection[CONF_HOST]
port = ad_connection[CONF_PORT]
controller = AdExt(SocketDevice(interface=(host, port)))
elif device_type == "serial":
path = device[CONF_DEVICE_PATH]
baud = device[CONF_DEVICE_BAUD]
if protocol == PROTOCOL_SERIAL:
path = ad_connection[CONF_DEVICE_PATH]
controller = AdExt(SerialDevice(interface=path))
elif device_type == "usb":
AdExt(USBDevice.find())
return False
controller.on_message += handle_message
controller.on_rfx_message += handle_rfx_message
@ -205,24 +117,56 @@ def setup(hass, config):
controller.on_close += handle_closed_connection
controller.on_expander_message += handle_rel_message
hass.data[DATA_AD] = controller
remove_stop_listener = hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STOP, stop_alarmdecoder
)
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = {
DATA_AD: controller,
DATA_REMOVE_UPDATE_LISTENER: undo_listener,
DATA_REMOVE_STOP_LISTENER: remove_stop_listener,
DATA_RESTART: False,
}
open_connection()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_alarmdecoder)
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
load_platform(
hass,
"alarm_control_panel",
DOMAIN,
{CONF_AUTO_BYPASS: auto_bypass, CONF_CODE_ARM_REQUIRED: code_arm_required},
config,
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry):
"""Unload a AlarmDecoder entry."""
hass.data[DOMAIN][entry.entry_id][DATA_RESTART] = False
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in PLATFORMS
]
)
)
if zones:
load_platform(hass, "binary_sensor", DOMAIN, {CONF_ZONES: zones}, config)
if not unload_ok:
return False
if display:
load_platform(hass, "sensor", DOMAIN, conf, config)
hass.data[DOMAIN][entry.entry_id][DATA_REMOVE_UPDATE_LISTENER]()
hass.data[DOMAIN][entry.entry_id][DATA_REMOVE_STOP_LISTENER]()
hass.data[DOMAIN][entry.entry_id][DATA_AD].close()
if hass.data[DOMAIN][entry.entry_id]:
hass.data[DOMAIN].pop(entry.entry_id)
if not hass.data[DOMAIN]:
hass.data.pop(DOMAIN)
return True
async def _update_listener(hass: HomeAssistantType, entry: ConfigEntry):
"""Handle options update."""
_LOGGER.debug("AlarmDecoder options updated: %s", entry.as_dict()["options"])
await hass.config_entries.async_reload(entry.entry_id)

View File

@ -12,74 +12,90 @@ from homeassistant.components.alarm_control_panel.const import (
SUPPORT_ALARM_ARM_HOME,
SUPPORT_ALARM_ARM_NIGHT,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_CODE,
ATTR_ENTITY_ID,
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMED_HOME,
STATE_ALARM_ARMED_NIGHT,
STATE_ALARM_DISARMED,
STATE_ALARM_TRIGGERED,
)
from homeassistant.helpers import entity_platform
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import HomeAssistantType
from . import (
from .const import (
CONF_ALT_NIGHT_MODE,
CONF_AUTO_BYPASS,
CONF_CODE_ARM_REQUIRED,
DATA_AD,
DEFAULT_ARM_OPTIONS,
DOMAIN,
OPTIONS_ARM,
SIGNAL_PANEL_MESSAGE,
)
_LOGGER = logging.getLogger(__name__)
SERVICE_ALARM_TOGGLE_CHIME = "alarm_toggle_chime"
ALARM_TOGGLE_CHIME_SCHEMA = vol.Schema({vol.Required(ATTR_CODE): cv.string})
ALARM_TOGGLE_CHIME_SCHEMA = vol.Schema(
{
vol.Required(ATTR_ENTITY_ID, default=[]): cv.entity_ids,
vol.Required(ATTR_CODE): cv.string,
}
)
SERVICE_ALARM_KEYPRESS = "alarm_keypress"
ATTR_KEYPRESS = "keypress"
ALARM_KEYPRESS_SCHEMA = vol.Schema({vol.Required(ATTR_KEYPRESS): cv.string})
ALARM_KEYPRESS_SCHEMA = vol.Schema(
{
vol.Required(ATTR_ENTITY_ID, default=[]): cv.entity_ids,
vol.Required(ATTR_KEYPRESS): cv.string,
}
)
def setup_platform(hass, config, add_entities, discovery_info=None):
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
):
"""Set up for AlarmDecoder alarm panels."""
if discovery_info is None:
return
options = entry.options
arm_options = options.get(OPTIONS_ARM, DEFAULT_ARM_OPTIONS)
client = hass.data[DOMAIN][entry.entry_id][DATA_AD]
auto_bypass = discovery_info[CONF_AUTO_BYPASS]
code_arm_required = discovery_info[CONF_CODE_ARM_REQUIRED]
entity = AlarmDecoderAlarmPanel(auto_bypass, code_arm_required)
add_entities([entity])
entity = AlarmDecoderAlarmPanel(
client=client,
auto_bypass=arm_options[CONF_AUTO_BYPASS],
code_arm_required=arm_options[CONF_CODE_ARM_REQUIRED],
alt_night_mode=arm_options[CONF_ALT_NIGHT_MODE],
)
async_add_entities([entity])
def alarm_toggle_chime_handler(service):
"""Register toggle chime handler."""
code = service.data.get(ATTR_CODE)
entity.alarm_toggle_chime(code)
platform = entity_platform.current_platform.get()
hass.services.register(
DOMAIN,
platform.async_register_entity_service(
SERVICE_ALARM_TOGGLE_CHIME,
alarm_toggle_chime_handler,
schema=ALARM_TOGGLE_CHIME_SCHEMA,
ALARM_TOGGLE_CHIME_SCHEMA,
"alarm_toggle_chime",
)
def alarm_keypress_handler(service):
"""Register keypress handler."""
keypress = service.data[ATTR_KEYPRESS]
entity.alarm_keypress(keypress)
hass.services.register(
DOMAIN,
platform.async_register_entity_service(
SERVICE_ALARM_KEYPRESS,
alarm_keypress_handler,
schema=ALARM_KEYPRESS_SCHEMA,
ALARM_KEYPRESS_SCHEMA,
"alarm_keypress",
)
return True
class AlarmDecoderAlarmPanel(AlarmControlPanelEntity):
"""Representation of an AlarmDecoder-based alarm panel."""
def __init__(self, auto_bypass, code_arm_required):
def __init__(self, client, auto_bypass, code_arm_required, alt_night_mode):
"""Initialize the alarm panel."""
self._client = client
self._display = ""
self._name = "Alarm Panel"
self._state = None
@ -95,6 +111,7 @@ class AlarmDecoderAlarmPanel(AlarmControlPanelEntity):
self._zone_bypassed = None
self._auto_bypass = auto_bypass
self._code_arm_required = code_arm_required
self._alt_night_mode = alt_night_mode
async def async_added_to_hass(self):
"""Register callbacks."""
@ -180,11 +197,11 @@ class AlarmDecoderAlarmPanel(AlarmControlPanelEntity):
def alarm_disarm(self, code=None):
"""Send disarm command."""
if code:
self.hass.data[DATA_AD].send(f"{code!s}1")
self._client.send(f"{code!s}1")
def alarm_arm_away(self, code=None):
"""Send arm away command."""
self.hass.data[DATA_AD].arm_away(
self._client.arm_away(
code=code,
code_arm_required=self._code_arm_required,
auto_bypass=self._auto_bypass,
@ -192,7 +209,7 @@ class AlarmDecoderAlarmPanel(AlarmControlPanelEntity):
def alarm_arm_home(self, code=None):
"""Send arm home command."""
self.hass.data[DATA_AD].arm_home(
self._client.arm_home(
code=code,
code_arm_required=self._code_arm_required,
auto_bypass=self._auto_bypass,
@ -200,18 +217,19 @@ class AlarmDecoderAlarmPanel(AlarmControlPanelEntity):
def alarm_arm_night(self, code=None):
"""Send arm night command."""
self.hass.data[DATA_AD].arm_night(
self._client.arm_night(
code=code,
code_arm_required=self._code_arm_required,
alt_night_mode=self._alt_night_mode,
auto_bypass=self._auto_bypass,
)
def alarm_toggle_chime(self, code=None):
"""Send toggle chime command."""
if code:
self.hass.data[DATA_AD].send(f"{code!s}9")
self._client.send(f"{code!s}9")
def alarm_keypress(self, keypress):
"""Send custom keypresses."""
if keypress:
self.hass.data[DATA_AD].send(keypress)
self._client.send(keypress)

View File

@ -2,20 +2,23 @@
import logging
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.typing import HomeAssistantType
from . import (
from .const import (
CONF_RELAY_ADDR,
CONF_RELAY_CHAN,
CONF_ZONE_LOOP,
CONF_ZONE_NAME,
CONF_ZONE_NUMBER,
CONF_ZONE_RFID,
CONF_ZONE_TYPE,
CONF_ZONES,
DEFAULT_ZONE_OPTIONS,
OPTIONS_ZONES,
SIGNAL_REL_MESSAGE,
SIGNAL_RFX_MESSAGE,
SIGNAL_ZONE_FAULT,
SIGNAL_ZONE_RESTORE,
ZONE_SCHEMA,
)
_LOGGER = logging.getLogger(__name__)
@ -30,26 +33,28 @@ ATTR_RF_LOOP4 = "rf_loop4"
ATTR_RF_LOOP1 = "rf_loop1"
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the AlarmDecoder binary sensor devices."""
configured_zones = discovery_info[CONF_ZONES]
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
):
"""Set up for AlarmDecoder sensor."""
zones = entry.options.get(OPTIONS_ZONES, DEFAULT_ZONE_OPTIONS)
devices = []
for zone_num in configured_zones:
device_config_data = ZONE_SCHEMA(configured_zones[zone_num])
zone_type = device_config_data[CONF_ZONE_TYPE]
zone_name = device_config_data[CONF_ZONE_NAME]
zone_rfid = device_config_data.get(CONF_ZONE_RFID)
zone_loop = device_config_data.get(CONF_ZONE_LOOP)
relay_addr = device_config_data.get(CONF_RELAY_ADDR)
relay_chan = device_config_data.get(CONF_RELAY_CHAN)
for zone_num in zones:
zone_info = zones[zone_num]
zone_type = zone_info[CONF_ZONE_TYPE]
zone_name = zone_info[CONF_ZONE_NAME]
zone_rfid = zone_info.get(CONF_ZONE_RFID)
zone_loop = zone_info.get(CONF_ZONE_LOOP)
relay_addr = zone_info.get(CONF_RELAY_ADDR)
relay_chan = zone_info.get(CONF_RELAY_CHAN)
device = AlarmDecoderBinarySensor(
zone_num, zone_name, zone_type, zone_rfid, zone_loop, relay_addr, relay_chan
)
devices.append(device)
add_entities(devices)
async_add_entities(devices)
return True
@ -67,7 +72,7 @@ class AlarmDecoderBinarySensor(BinarySensorEntity):
relay_chan,
):
"""Initialize the binary_sensor."""
self._zone_number = zone_number
self._zone_number = int(zone_number)
self._zone_type = zone_type
self._state = None
self._name = zone_name
@ -117,6 +122,7 @@ class AlarmDecoderBinarySensor(BinarySensorEntity):
def device_state_attributes(self):
"""Return the state attributes."""
attr = {}
attr[CONF_ZONE_NUMBER] = self._zone_number
if self._rfid and self._rfstate is not None:
attr[ATTR_RF_BIT0] = bool(self._rfstate & 0x01)
attr[ATTR_RF_LOW_BAT] = bool(self._rfstate & 0x02)

View File

@ -0,0 +1,356 @@
"""Config flow for AlarmDecoder."""
import logging
from adext import AdExt
from alarmdecoder.devices import SerialDevice, SocketDevice
from alarmdecoder.util import NoDeviceError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.binary_sensor import DEVICE_CLASSES
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_PROTOCOL
from homeassistant.core import callback
from .const import ( # pylint: disable=unused-import
CONF_ALT_NIGHT_MODE,
CONF_AUTO_BYPASS,
CONF_CODE_ARM_REQUIRED,
CONF_DEVICE_BAUD,
CONF_DEVICE_PATH,
CONF_RELAY_ADDR,
CONF_RELAY_CHAN,
CONF_ZONE_LOOP,
CONF_ZONE_NAME,
CONF_ZONE_NUMBER,
CONF_ZONE_RFID,
CONF_ZONE_TYPE,
DEFAULT_ARM_OPTIONS,
DEFAULT_DEVICE_BAUD,
DEFAULT_DEVICE_HOST,
DEFAULT_DEVICE_PATH,
DEFAULT_DEVICE_PORT,
DEFAULT_ZONE_OPTIONS,
DEFAULT_ZONE_TYPE,
DOMAIN,
OPTIONS_ARM,
OPTIONS_ZONES,
PROTOCOL_SERIAL,
PROTOCOL_SOCKET,
)
EDIT_KEY = "edit_selection"
EDIT_ZONES = "Zones"
EDIT_SETTINGS = "Arming Settings"
_LOGGER = logging.getLogger(__name__)
class AlarmDecoderFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a AlarmDecoder config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
def __init__(self):
"""Initialize AlarmDecoder ConfigFlow."""
self.protocol = None
@staticmethod
@callback
def async_get_options_flow(config_entry):
"""Get the options flow for AlarmDecoder."""
return AlarmDecoderOptionsFlowHandler(config_entry)
async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user."""
if user_input is not None:
self.protocol = user_input[CONF_PROTOCOL]
return await self.async_step_protocol()
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_PROTOCOL): vol.In(
[PROTOCOL_SOCKET, PROTOCOL_SERIAL]
),
}
),
)
async def async_step_protocol(self, user_input=None):
"""Handle AlarmDecoder protocol setup."""
errors = {}
if user_input is not None:
if _device_already_added(
self._async_current_entries(), user_input, self.protocol
):
return self.async_abort(reason="already_configured")
connection = {}
if self.protocol == PROTOCOL_SOCKET:
baud = connection[CONF_DEVICE_BAUD] = None
host = connection[CONF_HOST] = user_input[CONF_HOST]
port = connection[CONF_PORT] = user_input[CONF_PORT]
title = f"{host}:{port}"
device = SocketDevice(interface=(host, port))
if self.protocol == PROTOCOL_SERIAL:
path = connection[CONF_DEVICE_PATH] = user_input[CONF_DEVICE_PATH]
baud = connection[CONF_DEVICE_BAUD] = user_input[CONF_DEVICE_BAUD]
title = path
device = SerialDevice(interface=path)
controller = AdExt(device)
try:
with controller:
controller.open(baudrate=baud)
return self.async_create_entry(
title=title, data={CONF_PROTOCOL: self.protocol, **connection}
)
except NoDeviceError:
errors["base"] = "service_unavailable"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception during AlarmDecoder setup")
errors["base"] = "unknown"
if self.protocol == PROTOCOL_SOCKET:
schema = vol.Schema(
{
vol.Required(CONF_HOST, default=DEFAULT_DEVICE_HOST): str,
vol.Required(CONF_PORT, default=DEFAULT_DEVICE_PORT): int,
}
)
if self.protocol == PROTOCOL_SERIAL:
schema = vol.Schema(
{
vol.Required(CONF_DEVICE_PATH, default=DEFAULT_DEVICE_PATH): str,
vol.Required(CONF_DEVICE_BAUD, default=DEFAULT_DEVICE_BAUD): int,
}
)
return self.async_show_form(
step_id="protocol",
data_schema=schema,
errors=errors,
)
class AlarmDecoderOptionsFlowHandler(config_entries.OptionsFlow):
"""Handle AlarmDecoder options."""
def __init__(self, config_entry: config_entries.ConfigEntry):
"""Initialize AlarmDecoder options flow."""
self.arm_options = config_entry.options.get(OPTIONS_ARM, DEFAULT_ARM_OPTIONS)
self.zone_options = config_entry.options.get(
OPTIONS_ZONES, DEFAULT_ZONE_OPTIONS
)
self.selected_zone = None
async def async_step_init(self, user_input=None):
"""Manage the options."""
if user_input is not None:
if user_input[EDIT_KEY] == EDIT_SETTINGS:
return await self.async_step_arm_settings()
if user_input[EDIT_KEY] == EDIT_ZONES:
return await self.async_step_zone_select()
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Required(EDIT_KEY, default=EDIT_SETTINGS): vol.In(
[EDIT_SETTINGS, EDIT_ZONES]
)
},
),
)
async def async_step_arm_settings(self, user_input=None):
"""Arming options form."""
if user_input is not None:
return self.async_create_entry(
title="",
data={OPTIONS_ARM: user_input, OPTIONS_ZONES: self.zone_options},
)
return self.async_show_form(
step_id="arm_settings",
data_schema=vol.Schema(
{
vol.Optional(
CONF_ALT_NIGHT_MODE,
default=self.arm_options[CONF_ALT_NIGHT_MODE],
): bool,
vol.Optional(
CONF_AUTO_BYPASS, default=self.arm_options[CONF_AUTO_BYPASS]
): bool,
vol.Optional(
CONF_CODE_ARM_REQUIRED,
default=self.arm_options[CONF_CODE_ARM_REQUIRED],
): bool,
},
),
)
async def async_step_zone_select(self, user_input=None):
"""Zone selection form."""
errors = _validate_zone_input(user_input)
if user_input is not None and not errors:
self.selected_zone = str(
int(user_input[CONF_ZONE_NUMBER])
) # remove leading zeros
return await self.async_step_zone_details()
return self.async_show_form(
step_id="zone_select",
data_schema=vol.Schema({vol.Required(CONF_ZONE_NUMBER): str}),
errors=errors,
)
async def async_step_zone_details(self, user_input=None):
"""Zone details form."""
errors = _validate_zone_input(user_input)
if user_input is not None and not errors:
zone_options = self.zone_options.copy()
zone_id = self.selected_zone
zone_options[zone_id] = _fix_input_types(user_input)
# Delete zone entry if zone_name is omitted
if CONF_ZONE_NAME not in zone_options[zone_id]:
zone_options.pop(zone_id)
return self.async_create_entry(
title="",
data={OPTIONS_ARM: self.arm_options, OPTIONS_ZONES: zone_options},
)
existing_zone_settings = self.zone_options.get(self.selected_zone, {})
return self.async_show_form(
step_id="zone_details",
description_placeholders={CONF_ZONE_NUMBER: self.selected_zone},
data_schema=vol.Schema(
{
vol.Optional(
CONF_ZONE_NAME,
description={
"suggested_value": existing_zone_settings.get(
CONF_ZONE_NAME
)
},
): str,
vol.Optional(
CONF_ZONE_TYPE,
default=existing_zone_settings.get(
CONF_ZONE_TYPE, DEFAULT_ZONE_TYPE
),
): vol.In(DEVICE_CLASSES),
vol.Optional(
CONF_ZONE_RFID,
description={
"suggested_value": existing_zone_settings.get(
CONF_ZONE_RFID
)
},
): str,
vol.Optional(
CONF_ZONE_LOOP,
description={
"suggested_value": existing_zone_settings.get(
CONF_ZONE_LOOP
)
},
): str,
vol.Optional(
CONF_RELAY_ADDR,
description={
"suggested_value": existing_zone_settings.get(
CONF_RELAY_ADDR
)
},
): str,
vol.Optional(
CONF_RELAY_CHAN,
description={
"suggested_value": existing_zone_settings.get(
CONF_RELAY_CHAN
)
},
): str,
}
),
errors=errors,
)
def _validate_zone_input(zone_input):
if not zone_input:
return {}
errors = {}
# CONF_RELAY_ADDR & CONF_RELAY_CHAN are inclusive
if (CONF_RELAY_ADDR in zone_input and CONF_RELAY_CHAN not in zone_input) or (
CONF_RELAY_ADDR not in zone_input and CONF_RELAY_CHAN in zone_input
):
errors["base"] = "relay_inclusive"
# The following keys must be int
for key in [CONF_ZONE_NUMBER, CONF_ZONE_LOOP, CONF_RELAY_ADDR, CONF_RELAY_CHAN]:
if key in zone_input:
try:
int(zone_input[key])
except ValueError:
errors[key] = "int"
# CONF_ZONE_LOOP depends on CONF_ZONE_RFID
if CONF_ZONE_LOOP in zone_input and CONF_ZONE_RFID not in zone_input:
errors[CONF_ZONE_LOOP] = "loop_rfid"
# CONF_ZONE_LOOP must be 1-4
if (
CONF_ZONE_LOOP in zone_input
and zone_input[CONF_ZONE_LOOP].isdigit()
and int(zone_input[CONF_ZONE_LOOP]) not in list(range(1, 5))
):
errors[CONF_ZONE_LOOP] = "loop_range"
return errors
def _fix_input_types(zone_input):
"""Convert necessary keys to int.
Since ConfigFlow inputs of type int cannot default to an empty string, we collect the values below as
strings and then convert them to ints.
"""
for key in [CONF_ZONE_LOOP, CONF_RELAY_ADDR, CONF_RELAY_CHAN]:
if key in zone_input:
zone_input[key] = int(zone_input[key])
return zone_input
def _device_already_added(current_entries, user_input, protocol):
"""Determine if entry has already been added to HA."""
user_host = user_input.get(CONF_HOST)
user_port = user_input.get(CONF_PORT)
user_path = user_input.get(CONF_DEVICE_PATH)
user_baud = user_input.get(CONF_DEVICE_BAUD)
for entry in current_entries:
entry_host = entry.data.get(CONF_HOST)
entry_port = entry.data.get(CONF_PORT)
entry_path = entry.data.get(CONF_DEVICE_PATH)
entry_baud = entry.data.get(CONF_DEVICE_BAUD)
if protocol == PROTOCOL_SOCKET:
if user_host == entry_host and user_port == entry_port:
return True
if protocol == PROTOCOL_SERIAL:
if user_baud == entry_baud and user_path == entry_path:
return True
return False

View File

@ -0,0 +1,49 @@
"""Constants for the AlarmDecoder component."""
CONF_ALT_NIGHT_MODE = "alt_night_mode"
CONF_AUTO_BYPASS = "auto_bypass"
CONF_CODE_ARM_REQUIRED = "code_arm_required"
CONF_DEVICE_BAUD = "device_baudrate"
CONF_DEVICE_PATH = "device_path"
CONF_RELAY_ADDR = "zone_relayaddr"
CONF_RELAY_CHAN = "zone_relaychan"
CONF_ZONE_LOOP = "zone_loop"
CONF_ZONE_NAME = "zone_name"
CONF_ZONE_NUMBER = "zone_number"
CONF_ZONE_RFID = "zone_rfid"
CONF_ZONE_TYPE = "zone_type"
DATA_AD = "alarmdecoder"
DATA_REMOVE_STOP_LISTENER = "rm_stop_listener"
DATA_REMOVE_UPDATE_LISTENER = "rm_update_listener"
DATA_RESTART = "restart"
DEFAULT_ALT_NIGHT_MODE = False
DEFAULT_AUTO_BYPASS = False
DEFAULT_CODE_ARM_REQUIRED = True
DEFAULT_DEVICE_BAUD = 115200
DEFAULT_DEVICE_HOST = "alarmdecoder"
DEFAULT_DEVICE_PATH = "/dev/ttyUSB0"
DEFAULT_DEVICE_PORT = 10000
DEFAULT_ZONE_TYPE = "window"
DEFAULT_ARM_OPTIONS = {
CONF_ALT_NIGHT_MODE: DEFAULT_ALT_NIGHT_MODE,
CONF_AUTO_BYPASS: DEFAULT_AUTO_BYPASS,
CONF_CODE_ARM_REQUIRED: DEFAULT_CODE_ARM_REQUIRED,
}
DEFAULT_ZONE_OPTIONS = {}
DOMAIN = "alarmdecoder"
OPTIONS_ARM = "arm_options"
OPTIONS_ZONES = "zone_options"
PROTOCOL_SERIAL = "serial"
PROTOCOL_SOCKET = "socket"
SIGNAL_PANEL_MESSAGE = "alarmdecoder.panel_message"
SIGNAL_REL_MESSAGE = "alarmdecoder.rel_message"
SIGNAL_RFX_MESSAGE = "alarmdecoder.rfx_message"
SIGNAL_ZONE_FAULT = "alarmdecoder.zone_fault"
SIGNAL_ZONE_RESTORE = "alarmdecoder.zone_restore"

View File

@ -3,5 +3,6 @@
"name": "AlarmDecoder",
"documentation": "https://www.home-assistant.io/integrations/alarmdecoder",
"requirements": ["adext==0.3"],
"codeowners": ["@ajschmidt8"]
"codeowners": ["@ajschmidt8"],
"config_flow": true
}

View File

@ -1,26 +1,29 @@
"""Support for AlarmDecoder sensors (Shows Panel Display)."""
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import HomeAssistantType
from . import SIGNAL_PANEL_MESSAGE
from .const import SIGNAL_PANEL_MESSAGE
_LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up for AlarmDecoder sensor devices."""
_LOGGER.debug("AlarmDecoderSensor: setup_platform")
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
):
"""Set up for AlarmDecoder sensor."""
device = AlarmDecoderSensor(hass)
add_entities([device])
entity = AlarmDecoderSensor()
async_add_entities([entity])
return True
class AlarmDecoderSensor(Entity):
"""Representation of an AlarmDecoder keypad."""
def __init__(self, hass):
def __init__(self):
"""Initialize the alarm panel."""
self._display = ""
self._state = None

View File

@ -1,6 +1,9 @@
alarm_keypress:
description: Send custom keypresses to the alarm.
fields:
entity_id:
description: Name of alarm control panel to deliver keypress.
example: "alarm_control_panel.main"
keypress:
description: "String to send to the alarm panel."
example: "*71"
@ -8,6 +11,9 @@ alarm_keypress:
alarm_toggle_chime:
description: Send the alarm the toggle chime command.
fields:
entity_id:
description: Name of alarm control panel to toggle chime.
example: "alarm_control_panel.main"
code:
description: A required code to toggle the alarm control panel chime with.
example: 1234

View File

@ -0,0 +1,72 @@
{
"config": {
"step": {
"user": {
"title": "Choose AlarmDecoder Protocol",
"data": {
"protocol": "Protocol"
}
},
"protocol": {
"title": "Configure connection settings",
"data": {
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]",
"device_baudrate": "Device Baud Rate",
"device_path": "Device Path"
}
}
},
"error": {
"service_unavailable": "[%key:common::config_flow::error::cannot_connect%]"
},
"create_entry": { "default": "Successfully connected to AlarmDecoder." },
"abort": {
"already_configured": "AlarmDecoder device is already configured."
}
},
"options": {
"step": {
"init": {
"title": "Configure AlarmDecoder",
"description": "What would you like to edit?",
"data": {
"edit_select": "Edit"
}
},
"arm_settings": {
"title": "Configure AlarmDecoder",
"data": {
"auto_bypass": "Auto Bypass on Arm",
"code_arm_required": "Code Required for Arming",
"alt_night_mode": "Alternative Night Mode"
}
},
"zone_select": {
"title": "Configure AlarmDecoder",
"description": "Enter the zone number you'd like to to add, edit, or remove.",
"data": {
"zone_number": "Zone Number"
}
},
"zone_details": {
"title": "Configure AlarmDecoder",
"description": "Enter details for zone {zone_number}. To delete zone {zone_number}, leave Zone Name blank.",
"data": {
"zone_name": "Zone Name",
"zone_type": "Zone Type",
"zone_rfid": "RF Serial",
"zone_loop": "RF Loop",
"zone_relayaddr": "Relay Address",
"zone_relaychan": "Relay Channel"
}
}
},
"error": {
"relay_inclusive": "Relay Address and Relay Channel are codependent and must be included together.",
"int": "The field below must be an integer.",
"loop_rfid": "RF Loop cannot be used without RF Serial.",
"loop_range": "RF Loop must be an integer between 1 and 4."
}
}
}

View File

@ -0,0 +1,72 @@
{
"config": {
"step": {
"user": {
"title": "Choose AlarmDecoder Protocol",
"data": {
"protocol": "Protocol"
}
},
"protocol": {
"title": "Configure connection settings",
"data": {
"host": "Host",
"port": "Port",
"device_baudrate": "Device Baud Rate",
"device_path": "Device Path"
}
}
},
"error": {
"service_unavailable": "Failed to connect"
},
"create_entry": { "default": "Successfully connected to AlarmDecoder." },
"abort": {
"already_configured": "AlarmDecoder device is already configured."
}
},
"options": {
"step": {
"init": {
"title": "Configure AlarmDecoder",
"description": "What would you like to edit?",
"data": {
"edit_select": "Edit"
}
},
"arm_settings": {
"title": "Configure AlarmDecoder",
"data": {
"auto_bypass": "Auto Bypass on Arm",
"code_arm_required": "Code Required for Arming",
"alt_night_mode": "Alternative Night Mode"
}
},
"zone_select": {
"title": "Configure AlarmDecoder",
"description": "Enter the zone number you'd like to to add, edit, or remove.",
"data": {
"zone_number": "Zone Number"
}
},
"zone_details": {
"title": "Configure AlarmDecoder",
"description": "Enter details for zone {zone_number}. To delete zone {zone_number}, leave Zone Name blank.",
"data": {
"zone_name": "Zone Name",
"zone_type": "Zone Type",
"zone_rfid": "RF Serial",
"zone_loop": "RF Loop",
"zone_relayaddr": "Relay Address",
"zone_relaychan": "Relay Channel"
}
}
},
"error": {
"relay_inclusive": "Relay Address and Relay Channel are codependent and must be included together.",
"int": "The field below must be an integer.",
"loop_rfid": "RF Loop cannot be used without RF Serial.",
"loop_range": "RF Loop must be an integer between 1 and 4."
}
}
}

View File

@ -13,6 +13,7 @@ FLOWS = [
"agent_dvr",
"airly",
"airvisual",
"alarmdecoder",
"almond",
"ambiclimate",
"ambient_station",

View File

@ -50,6 +50,9 @@ accuweather==0.0.10
# homeassistant.components.androidtv
adb-shell[async]==0.2.1
# homeassistant.components.alarmdecoder
adext==0.3
# homeassistant.components.adguard
adguardhome==0.4.2

View File

@ -0,0 +1,424 @@
"""Test the AlarmDecoder config flow."""
from alarmdecoder.util import NoDeviceError
import pytest
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.alarmdecoder import config_flow
from homeassistant.components.alarmdecoder.const import (
CONF_ALT_NIGHT_MODE,
CONF_AUTO_BYPASS,
CONF_CODE_ARM_REQUIRED,
CONF_DEVICE_BAUD,
CONF_DEVICE_PATH,
CONF_RELAY_ADDR,
CONF_RELAY_CHAN,
CONF_ZONE_LOOP,
CONF_ZONE_NAME,
CONF_ZONE_NUMBER,
CONF_ZONE_RFID,
CONF_ZONE_TYPE,
DEFAULT_ARM_OPTIONS,
DEFAULT_ZONE_OPTIONS,
DOMAIN,
OPTIONS_ARM,
OPTIONS_ZONES,
PROTOCOL_SERIAL,
PROTOCOL_SOCKET,
)
from homeassistant.components.binary_sensor import DEVICE_CLASS_WINDOW
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_PROTOCOL
from homeassistant.core import HomeAssistant
from tests.async_mock import patch
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
"protocol,connection,baud,title",
[
(
PROTOCOL_SOCKET,
{
CONF_HOST: "alarmdecoder123",
CONF_PORT: 10001,
},
None,
"alarmdecoder123:10001",
),
(
PROTOCOL_SERIAL,
{
CONF_DEVICE_PATH: "/dev/ttyUSB123",
CONF_DEVICE_BAUD: 115000,
},
115000,
"/dev/ttyUSB123",
),
],
)
async def test_setups(hass: HomeAssistant, protocol, connection, baud, title):
"""Test flow for setting up the available AlarmDecoder protocols."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PROTOCOL: protocol},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "protocol"
with patch("homeassistant.components.alarmdecoder.config_flow.AdExt.open"), patch(
"homeassistant.components.alarmdecoder.config_flow.AdExt.close"
), patch(
"homeassistant.components.alarmdecoder.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.alarmdecoder.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"], connection
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == title
assert result["data"] == {
**connection,
CONF_PROTOCOL: protocol,
CONF_DEVICE_BAUD: baud,
}
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_setup_connection_error(hass: HomeAssistant):
"""Test flow for setup with a connection error."""
port = 1001
host = "alarmdecoder"
protocol = PROTOCOL_SOCKET
connection_settings = {CONF_HOST: host, CONF_PORT: port}
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PROTOCOL: protocol},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "protocol"
with patch(
"homeassistant.components.alarmdecoder.config_flow.AdExt.open",
side_effect=NoDeviceError,
), patch("homeassistant.components.alarmdecoder.config_flow.AdExt.close"):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], connection_settings
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "service_unavailable"}
async def test_options_arm_flow(hass: HomeAssistant):
"""Test arm options flow."""
user_input = {
CONF_ALT_NIGHT_MODE: True,
CONF_AUTO_BYPASS: True,
CONF_CODE_ARM_REQUIRED: True,
}
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"edit_selection": "Arming Settings"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "arm_settings"
with patch(
"homeassistant.components.alarmdecoder.async_setup_entry", return_value=True
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input=user_input,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert entry.options == {
OPTIONS_ARM: user_input,
OPTIONS_ZONES: DEFAULT_ZONE_OPTIONS,
}
async def test_options_zone_flow(hass: HomeAssistant):
"""Test options flow for adding/deleting zones."""
zone_number = "2"
zone_settings = {CONF_ZONE_NAME: "Front Entry", CONF_ZONE_TYPE: DEVICE_CLASS_WINDOW}
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"edit_selection": "Zones"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "zone_select"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_ZONE_NUMBER: zone_number},
)
with patch(
"homeassistant.components.alarmdecoder.async_setup_entry", return_value=True
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input=zone_settings,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert entry.options == {
OPTIONS_ARM: DEFAULT_ARM_OPTIONS,
OPTIONS_ZONES: {zone_number: zone_settings},
}
# Make sure zone can be removed...
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"edit_selection": "Zones"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "zone_select"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_ZONE_NUMBER: zone_number},
)
with patch(
"homeassistant.components.alarmdecoder.async_setup_entry", return_value=True
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert entry.options == {
OPTIONS_ARM: DEFAULT_ARM_OPTIONS,
OPTIONS_ZONES: {},
}
async def test_options_zone_flow_validation(hass: HomeAssistant):
"""Test input validation for zone options flow."""
zone_number = "2"
zone_settings = {CONF_ZONE_NAME: "Front Entry", CONF_ZONE_TYPE: DEVICE_CLASS_WINDOW}
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"edit_selection": "Zones"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "zone_select"
# Zone Number must be int
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_ZONE_NUMBER: "asd"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "zone_select"
assert result["errors"] == {CONF_ZONE_NUMBER: "int"}
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_ZONE_NUMBER: zone_number},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "zone_details"
# CONF_RELAY_ADDR & CONF_RELAY_CHAN are inclusive
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={**zone_settings, CONF_RELAY_ADDR: "1"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "zone_details"
assert result["errors"] == {"base": "relay_inclusive"}
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={**zone_settings, CONF_RELAY_CHAN: "1"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "zone_details"
assert result["errors"] == {"base": "relay_inclusive"}
# CONF_RELAY_ADDR, CONF_RELAY_CHAN must be int
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={**zone_settings, CONF_RELAY_ADDR: "abc", CONF_RELAY_CHAN: "abc"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "zone_details"
assert result["errors"] == {
CONF_RELAY_ADDR: "int",
CONF_RELAY_CHAN: "int",
}
# CONF_ZONE_LOOP depends on CONF_ZONE_RFID
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={**zone_settings, CONF_ZONE_LOOP: "1"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "zone_details"
assert result["errors"] == {CONF_ZONE_LOOP: "loop_rfid"}
# CONF_ZONE_LOOP must be int
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={**zone_settings, CONF_ZONE_RFID: "rfid123", CONF_ZONE_LOOP: "ab"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "zone_details"
assert result["errors"] == {CONF_ZONE_LOOP: "int"}
# CONF_ZONE_LOOP must be between [1,4]
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={**zone_settings, CONF_ZONE_RFID: "rfid123", CONF_ZONE_LOOP: "5"},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "zone_details"
assert result["errors"] == {CONF_ZONE_LOOP: "loop_range"}
# All valid settings
with patch(
"homeassistant.components.alarmdecoder.async_setup_entry", return_value=True
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
**zone_settings,
CONF_ZONE_RFID: "rfid123",
CONF_ZONE_LOOP: "2",
CONF_RELAY_ADDR: "12",
CONF_RELAY_CHAN: "1",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert entry.options == {
OPTIONS_ARM: DEFAULT_ARM_OPTIONS,
OPTIONS_ZONES: {
zone_number: {
**zone_settings,
CONF_ZONE_RFID: "rfid123",
CONF_ZONE_LOOP: 2,
CONF_RELAY_ADDR: 12,
CONF_RELAY_CHAN: 1,
}
},
}
@pytest.mark.parametrize(
"protocol,connection",
[
(
PROTOCOL_SOCKET,
{
CONF_HOST: "alarmdecoder123",
CONF_PORT: 10001,
},
),
(
PROTOCOL_SERIAL,
{
CONF_DEVICE_PATH: "/dev/ttyUSB123",
CONF_DEVICE_BAUD: 115000,
},
),
],
)
async def test_one_device_allowed(hass, protocol, connection):
"""Test that only one AlarmDecoder device is allowed."""
flow = config_flow.AlarmDecoderFlowHandler()
flow.hass = hass
MockConfigEntry(
domain=DOMAIN,
data=connection,
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PROTOCOL: protocol},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "protocol"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], connection
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"