jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-21 12:17:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support Konnected Pro alarm panel, embrace async, leverage latest HA features/architecture (#30894)

Browse Source 
* fix unique_id computation for switches

* update konnected component to use async, config entries, registries. Pro board support and tests

* clean up formatting comments from PR

* use standard interfaces in tests

* migrate config flow to use options

* address latest pr feedback

* format for import as part of config schema validation

* address pr feedback

* lint fix

* simplify check based on pr feedback

* clarify default schema validation

* fix other schema checks

* fix translations

Co-authored-by: Nate Clark <nate@nateclark.com>
This commit is contained in:
Kit Klein 2020-02-11 16:04:42 -05:00 committed by GitHub
parent 51c35ab9a8
commit 3435281bd1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 3692 additions and 436 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS
View File

@ -186,7 +186,7 @@ homeassistant/components/kef/* @basnijholt
homeassistant/components/keyboard_remote/* @bendavid
homeassistant/components/knx/* @Julius2342
homeassistant/components/kodi/* @armills
homeassistant/components/konnected/* @heythisisnate
homeassistant/components/konnected/* @heythisisnate @kit-klein
homeassistant/components/lametric/* @robbiet480
homeassistant/components/launch_library/* @ludeeus
homeassistant/components/lcn/* @alengwenus

101
homeassistant/components/konnected/.translations/en.json Normal file
View File

@ -0,0 +1,101 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured",
"already_in_progress": "Config flow for device is already in progress.",
"not_konn_panel": "Not a recognized Konnected.io device",
"unknown": "Unknown error occurred"
},
"error": {
"cannot_connect": "Unable to connect to a Konnected Panel at {host}:{port}"
},
"step": {
"confirm": {
"description": "Model: {model}\nHost: {host}\nPort: {port}\n\nYou can configure the IO and panel behavior in the Konnected Alarm Panel settings.",
"title": "Konnected Device Ready"
},
"user": {
"data": {
"host": "Konnected device IP address",
"port": "Konnected device port"
},
"description": "Please enter the host information for your Konnected Panel.",
"title": "Discover Konnected Device"
}
},
"title": "Konnected.io"
},
"options": {
"abort": {
"not_konn_panel": "Not a recognized Konnected.io device"
},
"error": {},
"step": {
"options_binary": {
"data": {
"inverse": "Invert the open/close state",
"name": "Name (optional)",
"type": "Binary Sensor Type"
},
"description": "Please select the options for the binary sensor attached to {zone}",
"title": "Configure Binary Sensor"
},
"options_digital": {
"data": {
"name": "Name (optional)",
"poll_interval": "Poll Interval (minutes) (optional)",
"type": "Sensor Type"
},
"description": "Please select the options for the digital sensor attached to {zone}",
"title": "Configure Digital Sensor"
},
"options_io": {
"data": {
"1": "Zone 1",
"2": "Zone 2",
"3": "Zone 3",
"4": "Zone 4",
"5": "Zone 5",
"6": "Zone 6",
"7": "Zone 7",
"out": "OUT"
},
"description": "Discovered a {model} at {host}. Select the base configuration of each I/O below - depending on the I/O it may allow for binary sensors (open/close contacts), digital sensors (dht and ds18b20), or switchable outputs. You'll be able to configure detailed options in the next steps.",
"title": "Configure I/O"
},
"options_io_ext": {
"data": {
"10": "Zone 10",
"11": "Zone 11",
"12": "Zone 12",
"8": "Zone 8",
"9": "Zone 9",
"alarm1": "ALARM1",
"alarm2_out2": "OUT2/ALARM2",
"out1": "OUT1"
},
"description": "Select the configuration of the remaining I/O below. You'll be able to configure detailed options in the next steps.",
"title": "Configure Extended I/O"
},
"options_misc": {
"data": {
"blink": "Blink panel LED on when sending state change"
},
"description": "Please select the desired behavior for your panel",
"title": "Configure Misc"
},
"options_switch": {
"data": {
"activation": "Output when on",
"momentary": "Pulse duration (ms) (optional)",
"name": "Name (optional)",
"pause": "Pause between pulses (ms) (optional)",
"repeat": "Times to repeat (-1=infinite) (optional)"
},
"description": "Please select the output options for {zone}",
"title": "Configure Switchable Output"
}
},
"title": "Konnected Alarm Panel Options"
}
}

630
homeassistant/components/konnected/__init__.py
View File

@ -1,20 +1,20 @@
"""Support for Konnected devices."""
import asyncio
import copy
import hmac
import json
import logging
from aiohttp.hdrs import AUTHORIZATION
from aiohttp.web import Request, Response
import konnected
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.binary_sensor import DEVICE_CLASSES_SCHEMA
from homeassistant.components.discovery import SERVICE_KONNECTED
from homeassistant.components.http import HomeAssistantView
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_STATE,
CONF_ACCESS_TOKEN,
CONF_BINARY_SENSORS,
CONF_DEVICES,
@ -27,45 +27,106 @@ from homeassistant.const import (
CONF_SWITCHES,
CONF_TYPE,
CONF_ZONE,
EVENT_HOMEASSISTANT_START,
HTTP_BAD_REQUEST,
HTTP_NOT_FOUND,
HTTP_UNAUTHORIZED,
STATE_OFF,
STATE_ON,
)
from homeassistant.helpers import config_validation as cv, discovery
from homeassistant.helpers.dispatcher import dispatcher_send
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from .config_flow import ( # Loading the config flow file will register the flow
CONF_DEFAULT_OPTIONS,
CONF_IO,
CONF_IO_BIN,
CONF_IO_DIG,
CONF_IO_SWI,
OPTIONS_SCHEMA,
)
from .const import (
CONF_ACTIVATION,
CONF_API_HOST,
CONF_BLINK,
CONF_DHT_SENSORS,
CONF_DISCOVERY,
CONF_DS18B20_SENSORS,
CONF_INVERSE,
CONF_MOMENTARY,
CONF_PAUSE,
CONF_POLL_INTERVAL,
CONF_REPEAT,
DOMAIN,
ENDPOINT_ROOT,
PIN_TO_ZONE,
SIGNAL_SENSOR_UPDATE,
STATE_HIGH,
STATE_LOW,
UPDATE_ENDPOINT,
ZONE_TO_PIN,
ZONES,
)
from .errors import CannotConnect
from .handlers import HANDLERS
from .panel import AlarmPanel
_LOGGER = logging.getLogger(__name__)
_BINARY_SENSOR_SCHEMA = vol.All(
def ensure_pin(value):
"""Check if valid pin and coerce to string."""
if value is None:
raise vol.Invalid("pin value is None")
if PIN_TO_ZONE.get(str(value)) is None:
raise vol.Invalid("pin not valid")
return str(value)
def ensure_zone(value):
"""Check if valid zone and coerce to string."""
if value is None:
raise vol.Invalid("zone value is None")
if str(value) not in ZONES is None:
raise vol.Invalid("zone not valid")
return str(value)
def import_validator(config):
"""Validate zones and reformat for import."""
config = copy.deepcopy(config)
io_cfgs = {}
# Replace pins with zones
for conf_platform, conf_io in (
(CONF_BINARY_SENSORS, CONF_IO_BIN),
(CONF_SENSORS, CONF_IO_DIG),
(CONF_SWITCHES, CONF_IO_SWI),
):
for zone in config.get(conf_platform, []):
if zone.get(CONF_PIN):
zone[CONF_ZONE] = PIN_TO_ZONE[zone[CONF_PIN]]
del zone[CONF_PIN]
io_cfgs[zone[CONF_ZONE]] = conf_io
# Migrate config_entry data into default_options structure
config[CONF_IO] = io_cfgs
config[CONF_DEFAULT_OPTIONS] = OPTIONS_SCHEMA(config)
# clean up fields migrated to options
config.pop(CONF_BINARY_SENSORS, None)
config.pop(CONF_SENSORS, None)
config.pop(CONF_SWITCHES, None)
config.pop(CONF_BLINK, None)
config.pop(CONF_DISCOVERY, None)
config.pop(CONF_IO, None)
return config
# configuration.yaml schemas (legacy)
BINARY_SENSOR_SCHEMA_YAML = vol.All(
vol.Schema(
{
vol.Exclusive(CONF_PIN, "s_pin"): vol.Any(*PIN_TO_ZONE),
vol.Exclusive(CONF_ZONE, "s_pin"): vol.Any(*ZONE_TO_PIN),
vol.Exclusive(CONF_ZONE, "s_io"): ensure_zone,
vol.Exclusive(CONF_PIN, "s_io"): ensure_pin,
vol.Required(CONF_TYPE): DEVICE_CLASSES_SCHEMA,
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_INVERSE, default=False): cv.boolean,
@ -74,14 +135,14 @@ _BINARY_SENSOR_SCHEMA = vol.All(
cv.has_at_least_one_key(CONF_PIN, CONF_ZONE),
)
_SENSOR_SCHEMA = vol.All(
SENSOR_SCHEMA_YAML = vol.All(
vol.Schema(
{
vol.Exclusive(CONF_PIN, "s_pin"): vol.Any(*PIN_TO_ZONE),
vol.Exclusive(CONF_ZONE, "s_pin"): vol.Any(*ZONE_TO_PIN),
vol.Exclusive(CONF_ZONE, "s_io"): ensure_zone,
vol.Exclusive(CONF_PIN, "s_io"): ensure_pin,
vol.Required(CONF_TYPE): vol.All(vol.Lower, vol.In(["dht", "ds18b20"])),
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_POLL_INTERVAL): vol.All(
vol.Optional(CONF_POLL_INTERVAL, default=3): vol.All(
vol.Coerce(int), vol.Range(min=1)
),
}
@ -89,11 +150,11 @@ _SENSOR_SCHEMA = vol.All(
cv.has_at_least_one_key(CONF_PIN, CONF_ZONE),
)
_SWITCH_SCHEMA = vol.All(
SWITCH_SCHEMA_YAML = vol.All(
vol.Schema(
{
vol.Exclusive(CONF_PIN, "a_pin"): vol.Any(*PIN_TO_ZONE),
vol.Exclusive(CONF_ZONE, "a_pin"): vol.Any(*ZONE_TO_PIN),
vol.Exclusive(CONF_ZONE, "s_io"): ensure_zone,
vol.Exclusive(CONF_PIN, "s_io"): ensure_pin,
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_ACTIVATION, default=STATE_HIGH): vol.All(
vol.Lower, vol.Any(STATE_HIGH, STATE_LOW)
@ -106,6 +167,24 @@ _SWITCH_SCHEMA = vol.All(
cv.has_at_least_one_key(CONF_PIN, CONF_ZONE),
)
DEVICE_SCHEMA_YAML = vol.All(
vol.Schema(
{
vol.Required(CONF_ID): cv.matches_regex("[0-9a-f]{12}"),
vol.Optional(CONF_BINARY_SENSORS): vol.All(
cv.ensure_list, [BINARY_SENSOR_SCHEMA_YAML]
),
vol.Optional(CONF_SENSORS): vol.All(cv.ensure_list, [SENSOR_SCHEMA_YAML]),
vol.Optional(CONF_SWITCHES): vol.All(cv.ensure_list, [SWITCH_SCHEMA_YAML]),
vol.Inclusive(CONF_HOST, "host_info"): cv.string,
vol.Inclusive(CONF_PORT, "host_info"): cv.port,
vol.Optional(CONF_BLINK, default=True): cv.boolean,
vol.Optional(CONF_DISCOVERY, default=True): cv.boolean,
}
),
import_validator,
)
# pylint: disable=no-value-for-parameter
CONFIG_SCHEMA = vol.Schema(
{
@ -113,352 +192,88 @@ CONFIG_SCHEMA = vol.Schema(
{
vol.Required(CONF_ACCESS_TOKEN): cv.string,
vol.Optional(CONF_API_HOST): vol.Url(),
vol.Required(CONF_DEVICES): [
{
vol.Required(CONF_ID): cv.matches_regex("[0-9a-f]{12}"),
vol.Optional(CONF_BINARY_SENSORS): vol.All(
cv.ensure_list, [_BINARY_SENSOR_SCHEMA]
),
vol.Optional(CONF_SENSORS): vol.All(
cv.ensure_list, [_SENSOR_SCHEMA]
),
vol.Optional(CONF_SWITCHES): vol.All(
cv.ensure_list, [_SWITCH_SCHEMA]
),
vol.Optional(CONF_HOST): cv.string,
vol.Optional(CONF_PORT): cv.port,
vol.Optional(CONF_BLINK, default=True): cv.boolean,
vol.Optional(CONF_DISCOVERY, default=True): cv.boolean,
}
],
vol.Optional(CONF_DEVICES): vol.All(
cv.ensure_list, [DEVICE_SCHEMA_YAML]
),
}
)
},
extra=vol.ALLOW_EXTRA,
)
YAML_CONFIGS = "yaml_configs"
PLATFORMS = ["binary_sensor", "sensor", "switch"]
async def async_setup(hass, config):
async def async_setup(hass: HomeAssistant, config: dict):
"""Set up the Konnected platform."""
cfg = config.get(DOMAIN)
if cfg is None:
cfg = {}
access_token = cfg.get(CONF_ACCESS_TOKEN)
if DOMAIN not in hass.data:
hass.data[DOMAIN] = {
CONF_ACCESS_TOKEN: access_token,
CONF_ACCESS_TOKEN: cfg.get(CONF_ACCESS_TOKEN),
CONF_API_HOST: cfg.get(CONF_API_HOST),
CONF_DEVICES: {},
}
def setup_device(host, port):
"""Set up a Konnected device at `host` listening on `port`."""
discovered = DiscoveredDevice(hass, host, port)
if discovered.is_configured:
discovered.setup()
else:
_LOGGER.warning(
"Konnected device %s was discovered on the network"
" but not specified in configuration.yaml",
discovered.device_id,
hass.http.register_view(KonnectedView)
# Check if they have yaml configured devices
if CONF_DEVICES not in cfg:
return True
for device in cfg.get(CONF_DEVICES, []):
# Attempt to importing the cfg. Use
# hass.async_add_job to avoid a deadlock.
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=device,
)
def device_discovered(service, info):
"""Call when a Konnected device has been discovered."""
host = info.get(CONF_HOST)
port = info.get(CONF_PORT)
setup_device(host, port)
async def manual_discovery(event):
"""Init devices on the network with manually assigned addresses."""
specified = [
dev
for dev in cfg.get(CONF_DEVICES)
if dev.get(CONF_HOST) and dev.get(CONF_PORT)
]
while specified:
for dev in specified:
_LOGGER.debug(
"Discovering Konnected device %s at %s:%s",
dev.get(CONF_ID),
dev.get(CONF_HOST),
dev.get(CONF_PORT),
)
try:
await hass.async_add_executor_job(
setup_device, dev.get(CONF_HOST), dev.get(CONF_PORT)
)
specified.remove(dev)
except konnected.Client.ClientError as err:
_LOGGER.error(err)
await asyncio.sleep(10) # try again in 10 seconds
# Initialize devices specified in the configuration on boot
for device in cfg.get(CONF_DEVICES):
ConfiguredDevice(hass, device, config).save_data()
discovery.async_listen(hass, SERVICE_KONNECTED, device_discovered)
hass.http.register_view(KonnectedView(access_token))
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_START, manual_discovery)
)
return True
class ConfiguredDevice:
"""A representation of a configured Konnected device."""
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Set up panel from a config entry."""
client = AlarmPanel(hass, entry)
# create a data store in hass.data[DOMAIN][CONF_DEVICES]
await client.async_save_data()
def __init__(self, hass, config, hass_config):
"""Initialize the Konnected device."""
self.hass = hass
self.config = config
self.hass_config = hass_config
try:
await client.async_connect()
except CannotConnect:
# this will trigger a retry in the future
raise config_entries.ConfigEntryNotReady
@property
def device_id(self):
"""Device id is the MAC address as string with punctuation removed."""
return self.config.get(CONF_ID)
def save_data(self):
"""Save the device configuration to `hass.data`."""
binary_sensors = {}
for entity in self.config.get(CONF_BINARY_SENSORS) or []:
if CONF_ZONE in entity:
pin = ZONE_TO_PIN[entity[CONF_ZONE]]
else:
pin = entity[CONF_PIN]
binary_sensors[pin] = {
CONF_TYPE: entity[CONF_TYPE],
CONF_NAME: entity.get(
CONF_NAME,
"Konnected {} Zone {}".format(self.device_id[6:], PIN_TO_ZONE[pin]),
),
CONF_INVERSE: entity.get(CONF_INVERSE),
ATTR_STATE: None,
}
_LOGGER.debug(
"Set up binary_sensor %s (initial state: %s)",
binary_sensors[pin].get("name"),
binary_sensors[pin].get(ATTR_STATE),
)
actuators = []
for entity in self.config.get(CONF_SWITCHES) or []:
if CONF_ZONE in entity:
pin = ZONE_TO_PIN[entity[CONF_ZONE]]
else:
pin = entity[CONF_PIN]
act = {
CONF_PIN: pin,
CONF_NAME: entity.get(
CONF_NAME,
"Konnected {} Actuator {}".format(
self.device_id[6:], PIN_TO_ZONE[pin]
),
),
ATTR_STATE: None,
CONF_ACTIVATION: entity[CONF_ACTIVATION],
CONF_MOMENTARY: entity.get(CONF_MOMENTARY),
CONF_PAUSE: entity.get(CONF_PAUSE),
CONF_REPEAT: entity.get(CONF_REPEAT),
}
actuators.append(act)
_LOGGER.debug("Set up switch %s", act)
sensors = []
for entity in self.config.get(CONF_SENSORS) or []:
if CONF_ZONE in entity:
pin = ZONE_TO_PIN[entity[CONF_ZONE]]
else:
pin = entity[CONF_PIN]
sensor = {
CONF_PIN: pin,
CONF_NAME: entity.get(
CONF_NAME,
"Konnected {} Sensor {}".format(
self.device_id[6:], PIN_TO_ZONE[pin]
),
),
CONF_TYPE: entity[CONF_TYPE],
CONF_POLL_INTERVAL: entity.get(CONF_POLL_INTERVAL),
}
sensors.append(sensor)
_LOGGER.debug(
"Set up %s sensor %s (initial state: %s)",
sensor.get(CONF_TYPE),
sensor.get(CONF_NAME),
sensor.get(ATTR_STATE),
)
device_data = {
CONF_BINARY_SENSORS: binary_sensors,
CONF_SENSORS: sensors,
CONF_SWITCHES: actuators,
CONF_BLINK: self.config.get(CONF_BLINK),
CONF_DISCOVERY: self.config.get(CONF_DISCOVERY),
}
if CONF_DEVICES not in self.hass.data[DOMAIN]:
self.hass.data[DOMAIN][CONF_DEVICES] = {}
_LOGGER.debug(
"Storing data in hass.data[%s][%s][%s]: %s",
DOMAIN,
CONF_DEVICES,
self.device_id,
device_data,
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
self.hass.data[DOMAIN][CONF_DEVICES][self.device_id] = device_data
for platform in ["binary_sensor", "sensor", "switch"]:
discovery.load_platform(
self.hass,
platform,
DOMAIN,
{"device_id": self.device_id},
self.hass_config,
)
entry.add_update_listener(async_entry_updated)
return True
class DiscoveredDevice:
"""A representation of a discovered Konnected device."""
def __init__(self, hass, host, port):
"""Initialize the Konnected device."""
self.hass = hass
self.host = host
self.port = port
self.client = konnected.Client(host, str(port))
self.status = self.client.get_status()
def setup(self):
"""Set up a newly discovered Konnected device."""
_LOGGER.info(
"Discovered Konnected device %s. Open http://%s:%s in a "
"web browser to view device status.",
self.device_id,
self.host,
self.port,
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in PLATFORMS
]
)
self.save_data()
self.update_initial_states()
self.sync_device_config()
)
if unload_ok:
hass.data[DOMAIN][CONF_DEVICES].pop(entry.data[CONF_ID])
def save_data(self):
"""Save the discovery information to `hass.data`."""
self.stored_configuration["client"] = self.client
self.stored_configuration["host"] = self.host
self.stored_configuration["port"] = self.port
return unload_ok
@property
def device_id(self):
"""Device id is the MAC address as string with punctuation removed."""
return self.status["mac"].replace(":", "")
@property
def is_configured(self):
"""Return true if device_id is specified in the configuration."""
return bool(self.hass.data[DOMAIN][CONF_DEVICES].get(self.device_id))
@property
def stored_configuration(self):
"""Return the configuration stored in `hass.data` for this device."""
return self.hass.data[DOMAIN][CONF_DEVICES].get(self.device_id)
def binary_sensor_configuration(self):
"""Return the configuration map for syncing binary sensors."""
return [{"pin": p} for p in self.stored_configuration[CONF_BINARY_SENSORS]]
def actuator_configuration(self):
"""Return the configuration map for syncing actuators."""
return [
{
"pin": data.get(CONF_PIN),
"trigger": (0 if data.get(CONF_ACTIVATION) in [0, STATE_LOW] else 1),
}
for data in self.stored_configuration[CONF_SWITCHES]
]
def dht_sensor_configuration(self):
"""Return the configuration map for syncing DHT sensors."""
return [
{CONF_PIN: sensor[CONF_PIN], CONF_POLL_INTERVAL: sensor[CONF_POLL_INTERVAL]}
for sensor in self.stored_configuration[CONF_SENSORS]
if sensor[CONF_TYPE] == "dht"
]
def ds18b20_sensor_configuration(self):
"""Return the configuration map for syncing DS18B20 sensors."""
return [
{"pin": sensor[CONF_PIN]}
for sensor in self.stored_configuration[CONF_SENSORS]
if sensor[CONF_TYPE] == "ds18b20"
]
def update_initial_states(self):
"""Update the initial state of each sensor from status poll."""
for sensor_data in self.status.get("sensors"):
sensor_config = self.stored_configuration[CONF_BINARY_SENSORS].get(
sensor_data.get(CONF_PIN), {}
)
entity_id = sensor_config.get(ATTR_ENTITY_ID)
state = bool(sensor_data.get(ATTR_STATE))
if sensor_config.get(CONF_INVERSE):
state = not state
dispatcher_send(self.hass, SIGNAL_SENSOR_UPDATE.format(entity_id), state)
def desired_settings_payload(self):
"""Return a dict representing the desired device configuration."""
desired_api_host = (
self.hass.data[DOMAIN].get(CONF_API_HOST) or self.hass.config.api.base_url
)
desired_api_endpoint = desired_api_host + ENDPOINT_ROOT
return {
"sensors": self.binary_sensor_configuration(),
"actuators": self.actuator_configuration(),
"dht_sensors": self.dht_sensor_configuration(),
"ds18b20_sensors": self.ds18b20_sensor_configuration(),
"auth_token": self.hass.data[DOMAIN].get(CONF_ACCESS_TOKEN),
"endpoint": desired_api_endpoint,
"blink": self.stored_configuration.get(CONF_BLINK),
"discovery": self.stored_configuration.get(CONF_DISCOVERY),
}
def current_settings_payload(self):
"""Return a dict of configuration currently stored on the device."""
settings = self.status["settings"]
if not settings:
settings = {}
return {
"sensors": [{"pin": s[CONF_PIN]} for s in self.status.get("sensors")],
"actuators": self.status.get("actuators"),
"dht_sensors": self.status.get(CONF_DHT_SENSORS),
"ds18b20_sensors": self.status.get(CONF_DS18B20_SENSORS),
"auth_token": settings.get("token"),
"endpoint": settings.get("apiUrl"),
"blink": settings.get(CONF_BLINK),
"discovery": settings.get(CONF_DISCOVERY),
}
def sync_device_config(self):
"""Sync the new pin configuration to the Konnected device if needed."""
_LOGGER.debug(
"Device %s settings payload: %s",
self.device_id,
self.desired_settings_payload(),
)
if self.desired_settings_payload() != self.current_settings_payload():
_LOGGER.info("pushing settings to device %s", self.device_id)
self.client.put_settings(**self.desired_settings_payload())
async def async_entry_updated(hass: HomeAssistant, entry: ConfigEntry):
"""Reload the config entry when options change."""
await hass.config_entries.async_reload(entry.entry_id)
class KonnectedView(HomeAssistantView):
@ -468,9 +283,8 @@ class KonnectedView(HomeAssistantView):
name = "api:konnected"
requires_auth = False # Uses access token from configuration
def __init__(self, auth_token):
def __init__(self):
"""Initialize the view."""
self.auth_token = auth_token
@staticmethod
def binary_value(state, activation):
@ -479,50 +293,29 @@ class KonnectedView(HomeAssistantView):
return 1 if state == STATE_ON else 0
return 0 if state == STATE_ON else 1
async def get(self, request: Request, device_id) -> Response:
"""Return the current binary state of a switch."""
async def update_sensor(self, request: Request, device_id) -> Response:
"""Process a put or post."""
hass = request.app["hass"]
pin_num = int(request.query.get("pin"))
data = hass.data[DOMAIN]
device = data[CONF_DEVICES][device_id]
if not device:
return self.json_message(
f"Device {device_id} not configured", status_code=HTTP_NOT_FOUND
)
try:
pin = next(
filter(
lambda switch: switch[CONF_PIN] == pin_num, device[CONF_SWITCHES]
)
)
except StopIteration:
pin = None
if not pin:
return self.json_message(
format("Switch on pin {} not configured", pin_num),
status_code=HTTP_NOT_FOUND,
)
return self.json(
{
"pin": pin_num,
"state": self.binary_value(
hass.states.get(pin[ATTR_ENTITY_ID]).state, pin[CONF_ACTIVATION]
),
}
auth = request.headers.get(AUTHORIZATION, None)
tokens = []
if hass.data[DOMAIN].get(CONF_ACCESS_TOKEN):
tokens.extend([hass.data[DOMAIN][CONF_ACCESS_TOKEN]])
tokens.extend(
[
entry.data[CONF_ACCESS_TOKEN]
for entry in hass.config_entries.async_entries(DOMAIN)
]
)
async def put(self, request: Request, device_id) -> Response:
"""Receive a sensor update via PUT request and async set state."""
hass = request.app["hass"]
data = hass.data[DOMAIN]
if auth is None or not next(
(True for token in tokens if hmac.compare_digest(f"Bearer {token}", auth)),
False,
):
return self.json_message("unauthorized", status_code=HTTP_UNAUTHORIZED)
try: # Konnected 2.2.0 and above supports JSON payloads
payload = await request.json()
pin_num = payload["pin"]
except json.decoder.JSONDecodeError:
_LOGGER.error(
(
@ -532,30 +325,97 @@ class KonnectedView(HomeAssistantView):
)
)
auth = request.headers.get(AUTHORIZATION, None)
if not hmac.compare_digest(f"Bearer {self.auth_token}", auth):
return self.json_message("unauthorized", status_code=HTTP_UNAUTHORIZED)
pin_num = int(pin_num)
device = data[CONF_DEVICES].get(device_id)
if device is None:
return self.json_message(
"unregistered device", status_code=HTTP_BAD_REQUEST
)
pin_data = device[CONF_BINARY_SENSORS].get(pin_num) or next(
(s for s in device[CONF_SENSORS] if s[CONF_PIN] == pin_num), None
)
if pin_data is None:
try:
zone_num = str(payload.get(CONF_ZONE) or PIN_TO_ZONE[payload[CONF_PIN]])
zone_data = device[CONF_BINARY_SENSORS].get(zone_num) or next(
(s for s in device[CONF_SENSORS] if s[CONF_ZONE] == zone_num), None
)
except KeyError:
zone_data = None
if zone_data is None:
return self.json_message(
"unregistered sensor/actuator", status_code=HTTP_BAD_REQUEST
)
pin_data["device_id"] = device_id
zone_data["device_id"] = device_id
for attr in ["state", "temp", "humi", "addr"]:
value = payload.get(attr)
handler = HANDLERS.get(attr)
if value is not None and handler:
hass.async_create_task(handler(hass, pin_data, payload))
hass.async_create_task(handler(hass, zone_data, payload))
return self.json_message("ok")
async def get(self, request: Request, device_id) -> Response:
"""Return the current binary state of a switch."""
hass = request.app["hass"]
data = hass.data[DOMAIN]
device = data[CONF_DEVICES].get(device_id)
if not device:
return self.json_message(
f"Device {device_id} not configured", status_code=HTTP_NOT_FOUND
)
# Our data model is based on zone ids but we convert from/to pin ids
# based on whether they are specified in the request
try:
zone_num = str(
request.query.get(CONF_ZONE) or PIN_TO_ZONE[request.query[CONF_PIN]]
)
zone = next(
(
switch
for switch in device[CONF_SWITCHES]
if switch[CONF_ZONE] == zone_num
)
)
except StopIteration:
zone = None
except KeyError:
zone = None
zone_num = None
if not zone:
target = request.query.get(
CONF_ZONE, request.query.get(CONF_PIN, "unknown")
)
return self.json_message(
f"Switch on zone or pin {target} not configured",
status_code=HTTP_NOT_FOUND,
)
resp = {}
if request.query.get(CONF_ZONE):
resp[CONF_ZONE] = zone_num
else:
resp[CONF_PIN] = ZONE_TO_PIN[zone_num]
# Make sure entity is setup
zone_entity_id = zone.get(ATTR_ENTITY_ID)
if zone_entity_id:
resp["state"] = self.binary_value(
hass.states.get(zone_entity_id).state, zone[CONF_ACTIVATION],
)
return self.json(resp)
_LOGGER.warning("Konnected entity not yet setup, returning default")
resp["state"] = self.binary_value(STATE_OFF, zone[CONF_ACTIVATION])
return self.json(resp)
async def put(self, request: Request, device_id) -> Response:
"""Receive a sensor update via PUT request and async set state."""
return await self.update_sensor(request, device_id)
async def post(self, request: Request, device_id) -> Response:
"""Receive a sensor update via POST request and async set state."""
return await self.update_sensor(request, device_id)

24
homeassistant/components/konnected/binary_sensor.py
View File

@ -13,18 +13,15 @@ from homeassistant.const import (
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from . import DOMAIN as KONNECTED_DOMAIN, PIN_TO_ZONE, SIGNAL_SENSOR_UPDATE
from .const import DOMAIN as KONNECTED_DOMAIN, SIGNAL_SENSOR_UPDATE
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up binary sensors attached to a Konnected device."""
if discovery_info is None:
return
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up binary sensors attached to a Konnected device from a config entry."""
data = hass.data[KONNECTED_DOMAIN]
device_id = discovery_info["device_id"]
device_id = config_entry.data["id"]
sensors = [
KonnectedBinarySensor(device_id, pin_num, pin_data)
for pin_num, pin_data in data[CONF_DEVICES][device_id][
@ -37,14 +34,14 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
class KonnectedBinarySensor(BinarySensorDevice):
"""Representation of a Konnected binary sensor."""
def __init__(self, device_id, pin_num, data):
def __init__(self, device_id, zone_num, data):
"""Initialize the Konnected binary sensor."""
self._data = data
self._device_id = device_id
self._pin_num = pin_num
self._zone_num = zone_num
self._state = self._data.get(ATTR_STATE)
self._device_class = self._data.get(CONF_TYPE)
self._unique_id = "{}-{}".format(device_id, PIN_TO_ZONE[pin_num])
self._unique_id = f"{device_id}-{zone_num}"
self._name = self._data.get(CONF_NAME)
@property
@ -72,6 +69,13 @@ class KonnectedBinarySensor(BinarySensorDevice):
"""Return the device class."""
return self._device_class
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(KONNECTED_DOMAIN, self._device_id)},
}
async def async_added_to_hass(self):
"""Store entity_id and register state change callback."""
self._data[ATTR_ENTITY_ID] = self.entity_id

739
homeassistant/components/konnected/config_flow.py Normal file
View File

@ -0,0 +1,739 @@
"""Config flow for konnected.io integration."""
import asyncio
import copy
import logging
import random
import string
from urllib.parse import urlparse
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.binary_sensor import (
DEVICE_CLASS_DOOR,
DEVICE_CLASSES_SCHEMA,
)
from homeassistant.components.ssdp import ATTR_UPNP_MANUFACTURER, ATTR_UPNP_MODEL_NAME
from homeassistant.const import (
CONF_ACCESS_TOKEN,
CONF_BINARY_SENSORS,
CONF_HOST,
CONF_ID,
CONF_NAME,
CONF_PORT,
CONF_SENSORS,
CONF_SWITCHES,
CONF_TYPE,
CONF_ZONE,
)
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv
from .const import (
CONF_ACTIVATION,
CONF_BLINK,
CONF_DISCOVERY,
CONF_INVERSE,
CONF_MODEL,
CONF_MOMENTARY,
CONF_PAUSE,
CONF_POLL_INTERVAL,
CONF_REPEAT,
DOMAIN,
STATE_HIGH,
STATE_LOW,
ZONES,
)
from .errors import CannotConnect
from .panel import KONN_MODEL, KONN_MODEL_PRO, get_status
_LOGGER = logging.getLogger(__name__)
ATTR_KONN_UPNP_MODEL_NAME = "model_name" # standard upnp is modelName
CONF_IO = "io"
CONF_IO_DIS = "Disabled"
CONF_IO_BIN = "Binary Sensor"
CONF_IO_DIG = "Digital Sensor"
CONF_IO_SWI = "Switchable Output"
KONN_MANUFACTURER = "konnected.io"
KONN_PANEL_MODEL_NAMES = {
KONN_MODEL: "Konnected Alarm Panel",
KONN_MODEL_PRO: "Konnected Alarm Panel Pro",
}
OPTIONS_IO_ANY = vol.In([CONF_IO_DIS, CONF_IO_BIN, CONF_IO_DIG, CONF_IO_SWI])
OPTIONS_IO_INPUT_ONLY = vol.In([CONF_IO_DIS, CONF_IO_BIN, CONF_IO_DIG])
OPTIONS_IO_OUTPUT_ONLY = vol.In([CONF_IO_DIS, CONF_IO_SWI])
# Config entry schemas
IO_SCHEMA = vol.Schema(
{
vol.Optional("1", default=CONF_IO_DIS): OPTIONS_IO_ANY,
vol.Optional("2", default=CONF_IO_DIS): OPTIONS_IO_ANY,
vol.Optional("3", default=CONF_IO_DIS): OPTIONS_IO_ANY,
vol.Optional("4", default=CONF_IO_DIS): OPTIONS_IO_ANY,
vol.Optional("5", default=CONF_IO_DIS): OPTIONS_IO_ANY,
vol.Optional("6", default=CONF_IO_DIS): OPTIONS_IO_ANY,
vol.Optional("7", default=CONF_IO_DIS): OPTIONS_IO_ANY,
vol.Optional("8", default=CONF_IO_DIS): OPTIONS_IO_ANY,
vol.Optional("9", default=CONF_IO_DIS): OPTIONS_IO_INPUT_ONLY,
vol.Optional("10", default=CONF_IO_DIS): OPTIONS_IO_INPUT_ONLY,
vol.Optional("11", default=CONF_IO_DIS): OPTIONS_IO_INPUT_ONLY,
vol.Optional("12", default=CONF_IO_DIS): OPTIONS_IO_INPUT_ONLY,
vol.Optional("out", default=CONF_IO_DIS): OPTIONS_IO_OUTPUT_ONLY,
vol.Optional("alarm1", default=CONF_IO_DIS): OPTIONS_IO_OUTPUT_ONLY,
vol.Optional("out1", default=CONF_IO_DIS): OPTIONS_IO_OUTPUT_ONLY,
vol.Optional("alarm2_out2", default=CONF_IO_DIS): OPTIONS_IO_OUTPUT_ONLY,
}
)
BINARY_SENSOR_SCHEMA = vol.Schema(
{
vol.Required(CONF_ZONE): vol.In(ZONES),
vol.Required(CONF_TYPE, default=DEVICE_CLASS_DOOR): DEVICE_CLASSES_SCHEMA,
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_INVERSE, default=False): cv.boolean,
}
)
SENSOR_SCHEMA = vol.Schema(
{
vol.Required(CONF_ZONE): vol.In(ZONES),
vol.Required(CONF_TYPE, default="dht"): vol.All(
vol.Lower, vol.In(["dht", "ds18b20"])
),
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_POLL_INTERVAL, default=3): vol.All(
vol.Coerce(int), vol.Range(min=1)
),
}
)
SWITCH_SCHEMA = vol.Schema(
{
vol.Required(CONF_ZONE): vol.In(ZONES),
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_ACTIVATION, default=STATE_HIGH): vol.All(
vol.Lower, vol.Any(STATE_HIGH, STATE_LOW)
),
vol.Optional(CONF_MOMENTARY): vol.All(vol.Coerce(int), vol.Range(min=10)),
vol.Optional(CONF_PAUSE): vol.All(vol.Coerce(int), vol.Range(min=10)),
vol.Optional(CONF_REPEAT): vol.All(vol.Coerce(int), vol.Range(min=-1)),
}
)
OPTIONS_SCHEMA = vol.Schema(
{
vol.Required(CONF_IO): IO_SCHEMA,
vol.Optional(CONF_BINARY_SENSORS): vol.All(
cv.ensure_list, [BINARY_SENSOR_SCHEMA]
),
vol.Optional(CONF_SENSORS): vol.All(cv.ensure_list, [SENSOR_SCHEMA]),
vol.Optional(CONF_SWITCHES): vol.All(cv.ensure_list, [SWITCH_SCHEMA]),
vol.Optional(CONF_BLINK, default=True): cv.boolean,
vol.Optional(CONF_DISCOVERY, default=True): cv.boolean,
},
extra=vol.REMOVE_EXTRA,
)
CONF_DEFAULT_OPTIONS = "default_options"
CONFIG_ENTRY_SCHEMA = vol.Schema(
{
vol.Required(CONF_ID): cv.matches_regex("[0-9a-f]{12}"),
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_PORT): cv.port,
vol.Required(CONF_MODEL): vol.Any(*KONN_PANEL_MODEL_NAMES),
vol.Required(CONF_ACCESS_TOKEN): cv.matches_regex("[a-zA-Z0-9]+"),
vol.Required(CONF_DEFAULT_OPTIONS): OPTIONS_SCHEMA,
},
extra=vol.REMOVE_EXTRA,
)
class KonnectedFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for NEW_NAME."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
def __init__(self):
"""Initialize the Konnected flow."""
self.data = {}
self.options = OPTIONS_SCHEMA({CONF_IO: {}})
async def async_gen_config(self, host, port):
"""Populate self.data based on panel status.
This will raise CannotConnect if an error occurs
"""
self.data[CONF_HOST] = host
self.data[CONF_PORT] = port
try:
status = await get_status(self.hass, host, port)
self.data[CONF_ID] = status["mac"].replace(":", "")
except (CannotConnect, KeyError):
raise CannotConnect
else:
self.data[CONF_MODEL] = status.get("name", KONN_MODEL)
self.data[CONF_ACCESS_TOKEN] = "".join(
random.choices(f"{string.ascii_uppercase}{string.digits}", k=20)
)
async def async_step_import(self, device_config):
"""Import a configuration.yaml config.
This flow is triggered by `async_setup` for configured panels.
"""
_LOGGER.debug(device_config)
# save the data and confirm connection via user step
await self.async_set_unique_id(device_config["id"])
self.options = device_config[CONF_DEFAULT_OPTIONS]
# config schema ensures we have port if we have host
if device_config.get(CONF_HOST):
return await self.async_step_user(
user_input={
CONF_HOST: device_config[CONF_HOST],
CONF_PORT: device_config[CONF_PORT],
}
)
# if we have no host info wait for it or abort if previously configured
self._abort_if_unique_id_configured()
return await self.async_step_user()
async def async_step_ssdp(self, discovery_info):
"""Handle a discovered konnected panel.
This flow is triggered by the SSDP component. It will check if the
device is already configured and attempt to finish the config if not.
"""
_LOGGER.debug(discovery_info)
try:
if discovery_info[ATTR_UPNP_MANUFACTURER] != KONN_MANUFACTURER:
return self.async_abort(reason="not_konn_panel")
if not any(
name in discovery_info[ATTR_UPNP_MODEL_NAME]
for name in KONN_PANEL_MODEL_NAMES
):
_LOGGER.warning(
"Discovered unrecognized Konnected device %s",
discovery_info.get(ATTR_UPNP_MODEL_NAME, "Unknown"),
)
return self.async_abort(reason="not_konn_panel")
# If MAC is missing it is a bug in the device fw but we'll guard
# against it since the field is so vital
except KeyError:
_LOGGER.error("Malformed Konnected SSDP info")
else:
# extract host/port from ssdp_location
netloc = urlparse(discovery_info["ssdp_location"]).netloc.split(":")
return await self.async_step_user(
user_input={CONF_HOST: netloc[0], CONF_PORT: int(netloc[1])}
)
return self.async_abort(reason="unknown")
async def async_step_user(self, user_input=None):
"""Connect to panel and get config."""
errors = {}
if user_input:
# build config info and wait for user confirmation
self.data[CONF_HOST] = user_input[CONF_HOST]
self.data[CONF_PORT] = user_input[CONF_PORT]
self.data[CONF_ACCESS_TOKEN] = self.hass.data.get(DOMAIN, {}).get(
CONF_ACCESS_TOKEN
) or "".join(
random.choices(f"{string.ascii_uppercase}{string.digits}", k=20)
)
# brief delay to allow processing of recent status req
await asyncio.sleep(0.1)
try:
status = await get_status(
self.hass, self.data[CONF_HOST], self.data[CONF_PORT]
)
except CannotConnect:
errors["base"] = "cannot_connect"
else:
self.data[CONF_ID] = status["mac"].replace(":", "")
self.data[CONF_MODEL] = status.get("name", KONN_MODEL)
return await self.async_step_confirm()
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST, default=self.data.get(CONF_HOST)): str,
vol.Required(CONF_PORT, default=self.data.get(CONF_PORT)): int,
}
),
errors=errors,
)
async def async_step_confirm(self, user_input=None):
"""Attempt to link with the Konnected panel.
Given a configured host, will ask the user to confirm and finalize
the connection.
"""
if user_input is None:
# update an existing config entry if host info changes
entry = await self.async_set_unique_id(
self.data[CONF_ID], raise_on_progress=False
)
if entry and (
entry.data[CONF_HOST] != self.data[CONF_HOST]
or entry.data[CONF_PORT] != self.data[CONF_PORT]
):
entry_data = copy.deepcopy(entry.data)
entry_data.update(self.data)
self.hass.config_entries.async_update_entry(entry, data=entry_data)
self._abort_if_unique_id_configured()
return self.async_show_form(
step_id="confirm",
description_placeholders={
"model": KONN_PANEL_MODEL_NAMES[self.data[CONF_MODEL]],
"host": self.data[CONF_HOST],
"port": self.data[CONF_PORT],
},
)
# Attach default options and create entry
self.data[CONF_DEFAULT_OPTIONS] = self.options
return self.async_create_entry(
title=KONN_PANEL_MODEL_NAMES[self.data[CONF_MODEL]], data=self.data,
)
@staticmethod
@callback
def async_get_options_flow(config_entry):
"""Return the Options Flow."""
return OptionsFlowHandler(config_entry)
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle a option flow for a Konnected Panel."""
def __init__(self, config_entry: config_entries.ConfigEntry):
"""Initialize options flow."""
self.entry = config_entry
self.model = self.entry.data[CONF_MODEL]
self.current_opt = self.entry.options or self.entry.data[CONF_DEFAULT_OPTIONS]
# as config proceeds we'll build up new options and then replace what's in the config entry
self.new_opt = {CONF_IO: {}}
self.active_cfg = None
self.io_cfg = {}
@callback
def get_current_cfg(self, io_type, zone):
"""Get the current zone config."""
return next(
(
cfg
for cfg in self.current_opt.get(io_type, [])
if cfg[CONF_ZONE] == zone
),
{},
)
async def async_step_init(self, user_input=None):
"""Handle options flow."""
return await self.async_step_options_io()
async def async_step_options_io(self, user_input=None):
"""Configure legacy panel IO or first half of pro IO."""
errors = {}
current_io = self.current_opt.get(CONF_IO, {})
if user_input is not None:
# strip out disabled io and save for options cfg
for key, value in user_input.items():
if value != CONF_IO_DIS:
self.new_opt[CONF_IO][key] = value
return await self.async_step_options_io_ext()
if self.model == KONN_MODEL:
return self.async_show_form(
step_id="options_io",
data_schema=vol.Schema(
{
vol.Required(
"1", default=current_io.get("1", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"2", default=current_io.get("2", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"3", default=current_io.get("3", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"4", default=current_io.get("4", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"5", default=current_io.get("5", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"6", default=current_io.get("6", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"out", default=current_io.get("out", CONF_IO_DIS)
): OPTIONS_IO_OUTPUT_ONLY,
}
),
description_placeholders={
"model": KONN_PANEL_MODEL_NAMES[self.model],
"host": self.entry.data[CONF_HOST],
},
errors=errors,
)
# configure the first half of the pro board io
if self.model == KONN_MODEL_PRO:
return self.async_show_form(
step_id="options_io",
data_schema=vol.Schema(
{
vol.Required(
"1", default=current_io.get("1", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"2", default=current_io.get("2", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"3", default=current_io.get("3", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"4", default=current_io.get("4", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"5", default=current_io.get("5", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"6", default=current_io.get("6", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"7", default=current_io.get("7", CONF_IO_DIS)
): OPTIONS_IO_ANY,
}
),
description_placeholders={
"model": KONN_PANEL_MODEL_NAMES[self.model],
"host": self.entry.data[CONF_HOST],
},
errors=errors,
)
return self.async_abort(reason="not_konn_panel")
async def async_step_options_io_ext(self, user_input=None):
"""Allow the user to configure the extended IO for pro."""
errors = {}
current_io = self.current_opt.get(CONF_IO, {})
if user_input is not None:
# strip out disabled io and save for options cfg
for key, value in user_input.items():
if value != CONF_IO_DIS:
self.new_opt[CONF_IO].update({key: value})
self.io_cfg = copy.deepcopy(self.new_opt[CONF_IO])
return await self.async_step_options_binary()
if self.model == KONN_MODEL:
self.io_cfg = copy.deepcopy(self.new_opt[CONF_IO])
return await self.async_step_options_binary()
if self.model == KONN_MODEL_PRO:
return self.async_show_form(
step_id="options_io_ext",
data_schema=vol.Schema(
{
vol.Required(
"8", default=current_io.get("8", CONF_IO_DIS)
): OPTIONS_IO_ANY,
vol.Required(
"9", default=current_io.get("9", CONF_IO_DIS)
): OPTIONS_IO_INPUT_ONLY,
vol.Required(
"10", default=current_io.get("10", CONF_IO_DIS)
): OPTIONS_IO_INPUT_ONLY,
vol.Required(
"11", default=current_io.get("11", CONF_IO_DIS)
): OPTIONS_IO_INPUT_ONLY,
vol.Required(
"12", default=current_io.get("12", CONF_IO_DIS)
): OPTIONS_IO_INPUT_ONLY,
vol.Required(
"alarm1", default=current_io.get("alarm1", CONF_IO_DIS)
): OPTIONS_IO_OUTPUT_ONLY,
vol.Required(
"out1", default=current_io.get("out1", CONF_IO_DIS)
): OPTIONS_IO_OUTPUT_ONLY,
vol.Required(
"alarm2_out2",
default=current_io.get("alarm2_out2", CONF_IO_DIS),
): OPTIONS_IO_OUTPUT_ONLY,
}
),
description_placeholders={
"model": KONN_PANEL_MODEL_NAMES[self.model],
"host": self.entry.data[CONF_HOST],
},
errors=errors,
)
return self.async_abort(reason="not_konn_panel")
async def async_step_options_binary(self, user_input=None):
"""Allow the user to configure the IO options for binary sensors."""
errors = {}
if user_input is not None:
zone = {"zone": self.active_cfg}
zone.update(user_input)
self.new_opt[CONF_BINARY_SENSORS] = self.new_opt.get(
CONF_BINARY_SENSORS, []
) + [zone]
self.io_cfg.pop(self.active_cfg)
self.active_cfg = None
if self.active_cfg:
current_cfg = self.get_current_cfg(CONF_BINARY_SENSORS, self.active_cfg)
return self.async_show_form(
step_id="options_binary",
data_schema=vol.Schema(
{
vol.Required(
CONF_TYPE,
default=current_cfg.get(CONF_TYPE, DEVICE_CLASS_DOOR),
): DEVICE_CLASSES_SCHEMA,
vol.Optional(
CONF_NAME, default=current_cfg.get(CONF_NAME, vol.UNDEFINED)
): str,
vol.Optional(
CONF_INVERSE, default=current_cfg.get(CONF_INVERSE, False)
): bool,
}
),
description_placeholders={
"zone": f"Zone {self.active_cfg}"
if len(self.active_cfg) < 3
else self.active_cfg.upper
},
errors=errors,
)
# find the next unconfigured binary sensor
for key, value in self.io_cfg.items():
if value == CONF_IO_BIN:
self.active_cfg = key
current_cfg = self.get_current_cfg(CONF_BINARY_SENSORS, self.active_cfg)
return self.async_show_form(
step_id="options_binary",
data_schema=vol.Schema(
{
vol.Required(
CONF_TYPE,
default=current_cfg.get(CONF_TYPE, DEVICE_CLASS_DOOR),
): DEVICE_CLASSES_SCHEMA,
vol.Optional(
CONF_NAME,
default=current_cfg.get(CONF_NAME, vol.UNDEFINED),
): str,
vol.Optional(
CONF_INVERSE,
default=current_cfg.get(CONF_INVERSE, False),
): bool,
}
),
description_placeholders={
"zone": "Zone {self.active_cfg}"
if len(self.active_cfg) < 3
else self.active_cfg.upper
},
errors=errors,
)
return await self.async_step_options_digital()
async def async_step_options_digital(self, user_input=None):
"""Allow the user to configure the IO options for digital sensors."""
errors = {}
if user_input is not None:
zone = {"zone": self.active_cfg}
zone.update(user_input)
self.new_opt[CONF_SENSORS] = self.new_opt.get(CONF_SENSORS, []) + [zone]
self.io_cfg.pop(self.active_cfg)
self.active_cfg = None
if self.active_cfg:
current_cfg = self.get_current_cfg(CONF_SENSORS, self.active_cfg)
return self.async_show_form(
step_id="options_digital",
data_schema=vol.Schema(
{
vol.Required(
CONF_TYPE, default=current_cfg.get(CONF_TYPE, "dht")
): vol.All(vol.Lower, vol.In(["dht", "ds18b20"])),
vol.Optional(
CONF_NAME, default=current_cfg.get(CONF_NAME, vol.UNDEFINED)
): str,
vol.Optional(
CONF_POLL_INTERVAL,
default=current_cfg.get(CONF_POLL_INTERVAL, 3),
): vol.All(vol.Coerce(int), vol.Range(min=1)),
}
),
description_placeholders={
"zone": "Zone {self.active_cfg}"
if len(self.active_cfg) < 3
else self.active_cfg.upper()
},
errors=errors,
)
# find the next unconfigured digital sensor
for key, value in self.io_cfg.items():
if value == CONF_IO_DIG:
self.active_cfg = key
current_cfg = self.get_current_cfg(CONF_SENSORS, self.active_cfg)
return self.async_show_form(
step_id="options_digital",
data_schema=vol.Schema(
{
vol.Required(
CONF_TYPE, default=current_cfg.get(CONF_TYPE, "dht")
): vol.All(vol.Lower, vol.In(["dht", "ds18b20"])),
vol.Optional(
CONF_NAME,
default=current_cfg.get(CONF_NAME, vol.UNDEFINED),
): str,
vol.Optional(
CONF_POLL_INTERVAL,
default=current_cfg.get(CONF_POLL_INTERVAL, 3),
): vol.All(vol.Coerce(int), vol.Range(min=1)),
}
),
description_placeholders={
"zone": "Zone {self.active_cfg}"
if len(self.active_cfg) < 3
else self.active_cfg.upper()
},
errors=errors,
)
return await self.async_step_options_switch()
async def async_step_options_switch(self, user_input=None):
"""Allow the user to configure the IO options for switches."""
errors = {}
if user_input is not None:
zone = {"zone": self.active_cfg}
zone.update(user_input)
self.new_opt[CONF_SWITCHES] = self.new_opt.get(CONF_SWITCHES, []) + [zone]
self.io_cfg.pop(self.active_cfg)
self.active_cfg = None
if self.active_cfg:
current_cfg = self.get_current_cfg(CONF_SWITCHES, self.active_cfg)
return self.async_show_form(
step_id="options_switch",
data_schema=vol.Schema(
{
vol.Optional(
CONF_NAME, default=current_cfg.get(CONF_NAME, vol.UNDEFINED)
): str,
vol.Optional(
CONF_ACTIVATION,
default=current_cfg.get(CONF_ACTIVATION, STATE_HIGH),
): vol.All(vol.Lower, vol.Any(STATE_HIGH, STATE_LOW)),
vol.Optional(
CONF_MOMENTARY,
default=current_cfg.get(CONF_MOMENTARY, vol.UNDEFINED),
): vol.All(vol.Coerce(int), vol.Range(min=10)),
vol.Optional(
CONF_PAUSE,
default=current_cfg.get(CONF_PAUSE, vol.UNDEFINED),
): vol.All(vol.Coerce(int), vol.Range(min=10)),
vol.Optional(
CONF_REPEAT,
default=current_cfg.get(CONF_REPEAT, vol.UNDEFINED),
): vol.All(vol.Coerce(int), vol.Range(min=-1)),
}
),
description_placeholders={
"zone": "Zone {self.active_cfg}"
if len(self.active_cfg) < 3
else self.active_cfg.upper()
},
errors=errors,
)
# find the next unconfigured switch
for key, value in self.io_cfg.items():
if value == CONF_IO_SWI:
self.active_cfg = key
current_cfg = self.get_current_cfg(CONF_SWITCHES, self.active_cfg)
return self.async_show_form(
step_id="options_switch",
data_schema=vol.Schema(
{
vol.Optional(
CONF_NAME,
default=current_cfg.get(CONF_NAME, vol.UNDEFINED),
): str,
vol.Optional(
CONF_ACTIVATION,
default=current_cfg.get(CONF_ACTIVATION, "high"),
): vol.In(["low", "high"]),
vol.Optional(
CONF_MOMENTARY,
default=current_cfg.get(CONF_MOMENTARY, vol.UNDEFINED),
): vol.All(vol.Coerce(int), vol.Range(min=10)),
vol.Optional(
CONF_PAUSE,
default=current_cfg.get(CONF_PAUSE, vol.UNDEFINED),
): vol.All(vol.Coerce(int), vol.Range(min=10)),
vol.Optional(
CONF_REPEAT,
default=current_cfg.get(CONF_REPEAT, vol.UNDEFINED),
): vol.All(vol.Coerce(int), vol.Range(min=-1)),
}
),
description_placeholders={
"zone": "Zone {self.active_cfg}"
if len(self.active_cfg) < 3
else self.active_cfg.upper()
},
errors=errors,
)
return await self.async_step_options_misc()
async def async_step_options_misc(self, user_input=None):
"""Allow the user to configure the LED behavior."""
errors = {}
if user_input is not None:
self.new_opt[CONF_BLINK] = user_input[CONF_BLINK]
return self.async_create_entry(title="", data=self.new_opt)
return self.async_show_form(
step_id="options_misc",
data_schema=vol.Schema(
{
vol.Required(
CONF_BLINK, default=self.current_opt.get(CONF_BLINK, True)
): bool,
}
),
errors=errors,
)

24
homeassistant/components/konnected/const.py
View File

@ -14,11 +14,33 @@ CONF_BLINK = "blink"
CONF_DISCOVERY = "discovery"
CONF_DHT_SENSORS = "dht_sensors"
CONF_DS18B20_SENSORS = "ds18b20_sensors"
CONF_MODEL = "model"
STATE_LOW = "low"
STATE_HIGH = "high"
PIN_TO_ZONE = {1: 1, 2: 2, 5: 3, 6: 4, 7: 5, 8: "out", 9: 6}
ZONES = [
"1",
"2",
"3",
"4",
"5",
"6",
"7",
"8",
"9",
"10",
"11",
"12",
"alarm1",
"out1",
"alarm2_out2",
"out",
]
# alarm panel pro only handles zones,
# alarm panel allows specifying pins via configuration.yaml
PIN_TO_ZONE = {"1": "1", "2": "2", "5": "3", "6": "4", "7": "5", "8": "out", "9": "6"}
ZONE_TO_PIN = {zone: pin for pin, zone in PIN_TO_ZONE.items()}
ENDPOINT_ROOT = "/api/konnected"

10
homeassistant/components/konnected/errors.py Normal file
View File

@ -0,0 +1,10 @@
"""Errors for the Konnected component."""
from homeassistant.exceptions import HomeAssistantError
class KonnectedException(HomeAssistantError):
"""Base class for Konnected exceptions."""
class CannotConnect(KonnectedException):
"""Unable to connect to the panel."""

19
homeassistant/components/konnected/manifest.json
View File

@ -1,8 +1,21 @@
{
"domain": "konnected",
"name": "Konnected",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/konnected",
"requirements": ["konnected==0.1.5"],
"dependencies": ["http"],
"codeowners": ["@heythisisnate"]
"requirements": [
"konnected==1.1.0"
],
"ssdp": [
{
"manufacturer": "konnected.io"
}
],
"dependencies": [
"http"
],
"codeowners": [
"@heythisisnate",
"@kit-klein"
]
}

359
homeassistant/components/konnected/panel.py Normal file
View File

@ -0,0 +1,359 @@
"""Support for Konnected devices."""
import asyncio
import logging
import konnected
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_STATE,
CONF_ACCESS_TOKEN,
CONF_BINARY_SENSORS,
CONF_DEVICES,
CONF_HOST,
CONF_ID,
CONF_NAME,
CONF_PIN,
CONF_PORT,
CONF_SENSORS,
CONF_SWITCHES,
CONF_TYPE,
CONF_ZONE,
)
from homeassistant.core import callback
from homeassistant.helpers import aiohttp_client, device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import (
CONF_ACTIVATION,
CONF_API_HOST,
CONF_BLINK,
CONF_DHT_SENSORS,
CONF_DISCOVERY,
CONF_DS18B20_SENSORS,
CONF_INVERSE,
CONF_MOMENTARY,
CONF_PAUSE,
CONF_POLL_INTERVAL,
CONF_REPEAT,
DOMAIN,
ENDPOINT_ROOT,
SIGNAL_SENSOR_UPDATE,
STATE_LOW,
ZONE_TO_PIN,
)
from .errors import CannotConnect
_LOGGER = logging.getLogger(__name__)
KONN_MODEL = "Konnected"
KONN_MODEL_PRO = "Konnected Pro"
# Indicate how each unit is controlled (pin or zone)
KONN_API_VERSIONS = {
KONN_MODEL: CONF_PIN,
KONN_MODEL_PRO: CONF_ZONE,
}
class AlarmPanel:
"""A representation of a Konnected alarm panel."""
def __init__(self, hass, config_entry):
"""Initialize the Konnected device."""
self.hass = hass
self.config_entry = config_entry
self.config = config_entry.data
self.options = config_entry.options
self.host = self.config.get(CONF_HOST)
self.port = self.config.get(CONF_PORT)
self.client = None
self.status = None
self.api_version = KONN_API_VERSIONS[KONN_MODEL]
@property
def device_id(self):
"""Device id is the MAC address as string with punctuation removed."""
return self.config.get(CONF_ID)
@property
def stored_configuration(self):
"""Return the configuration stored in `hass.data` for this device."""
return self.hass.data[DOMAIN][CONF_DEVICES].get(self.device_id)
def format_zone(self, zone, other_items=None):
"""Get zone or pin based dict based on the client type."""
payload = {
self.api_version: zone
if self.api_version == CONF_ZONE
else ZONE_TO_PIN[zone]
}
payload.update(other_items or {})
return payload
async def async_connect(self):
"""Connect to and setup a Konnected device."""
try:
self.client = konnected.Client(
host=self.host,
port=str(self.port),
websession=aiohttp_client.async_get_clientsession(self.hass),
)
self.status = await self.client.get_status()
self.api_version = KONN_API_VERSIONS.get(
self.status.get("model", KONN_MODEL), KONN_API_VERSIONS[KONN_MODEL]
)
_LOGGER.info(
"Connected to new %s device", self.status.get("model", "Konnected")
)
_LOGGER.debug(self.status)
await self.async_update_initial_states()
# brief delay to allow processing of recent status req
await asyncio.sleep(0.1)
await self.async_sync_device_config()
except self.client.ClientError as err:
_LOGGER.warning("Exception trying to connect to panel: %s", err)
raise CannotConnect
_LOGGER.info(
"Set up Konnected device %s. Open http://%s:%s in a "
"web browser to view device status",
self.device_id,
self.host,
self.port,
)
device_registry = await dr.async_get_registry(self.hass)
device_registry.async_get_or_create(
config_entry_id=self.config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, self.status.get("mac"))},
identifiers={(DOMAIN, self.device_id)},
manufacturer="Konnected.io",
name=self.config_entry.title,
model=self.config_entry.title,
sw_version=self.status.get("swVersion"),
)
async def update_switch(self, zone, state, momentary=None, times=None, pause=None):
"""Update the state of a switchable output."""
try:
if self.client:
if self.api_version == CONF_ZONE:
return await self.client.put_zone(
zone, state, momentary, times, pause,
)
# device endpoint uses pin number instead of zone
return await self.client.put_device(
ZONE_TO_PIN[zone], state, momentary, times, pause,
)
except self.client.ClientError as err:
_LOGGER.warning("Exception trying to update panel: %s", err)
raise CannotConnect
async def async_save_data(self):
"""Save the device configuration to `hass.data`."""
binary_sensors = {}
for entity in self.options.get(CONF_BINARY_SENSORS) or []:
zone = entity[CONF_ZONE]
binary_sensors[zone] = {
CONF_TYPE: entity[CONF_TYPE],
CONF_NAME: entity.get(
CONF_NAME, f"Konnected {self.device_id[6:]} Zone {zone}"
),
CONF_INVERSE: entity.get(CONF_INVERSE),
ATTR_STATE: None,
}
_LOGGER.debug(
"Set up binary_sensor %s (initial state: %s)",
binary_sensors[zone].get("name"),
binary_sensors[zone].get(ATTR_STATE),
)
actuators = []
for entity in self.options.get(CONF_SWITCHES) or []:
zone = entity[CONF_ZONE]
act = {
CONF_ZONE: zone,
CONF_NAME: entity.get(
CONF_NAME, f"Konnected {self.device_id[6:]} Actuator {zone}",
),
ATTR_STATE: None,
CONF_ACTIVATION: entity[CONF_ACTIVATION],
CONF_MOMENTARY: entity.get(CONF_MOMENTARY),
CONF_PAUSE: entity.get(CONF_PAUSE),
CONF_REPEAT: entity.get(CONF_REPEAT),
}
actuators.append(act)
_LOGGER.debug("Set up switch %s", act)
sensors = []
for entity in self.options.get(CONF_SENSORS) or []:
zone = entity[CONF_ZONE]
sensor = {
CONF_ZONE: zone,
CONF_NAME: entity.get(
CONF_NAME, f"Konnected {self.device_id[6:]} Sensor {zone}"
),
CONF_TYPE: entity[CONF_TYPE],
CONF_POLL_INTERVAL: entity.get(CONF_POLL_INTERVAL),
}
sensors.append(sensor)
_LOGGER.debug(
"Set up %s sensor %s (initial state: %s)",
sensor.get(CONF_TYPE),
sensor.get(CONF_NAME),
sensor.get(ATTR_STATE),
)
device_data = {
CONF_BINARY_SENSORS: binary_sensors,
CONF_SENSORS: sensors,
CONF_SWITCHES: actuators,
CONF_BLINK: self.options.get(CONF_BLINK),
CONF_DISCOVERY: self.options.get(CONF_DISCOVERY),
CONF_HOST: self.host,
CONF_PORT: self.port,
"panel": self,
}
if CONF_DEVICES not in self.hass.data[DOMAIN]:
self.hass.data[DOMAIN][CONF_DEVICES] = {}
_LOGGER.debug(
"Storing data in hass.data[%s][%s][%s]: %s",
DOMAIN,
CONF_DEVICES,
self.device_id,
device_data,
)
self.hass.data[DOMAIN][CONF_DEVICES][self.device_id] = device_data
@callback
def async_binary_sensor_configuration(self):
"""Return the configuration map for syncing binary sensors."""
return [
self.format_zone(p) for p in self.stored_configuration[CONF_BINARY_SENSORS]
]
@callback
def async_actuator_configuration(self):
"""Return the configuration map for syncing actuators."""
return [
self.format_zone(
data[CONF_ZONE],
{"trigger": (0 if data.get(CONF_ACTIVATION) in [0, STATE_LOW] else 1)},
)
for data in self.stored_configuration[CONF_SWITCHES]
]
@callback
def async_dht_sensor_configuration(self):
"""Return the configuration map for syncing DHT sensors."""
return [
self.format_zone(
sensor[CONF_ZONE], {CONF_POLL_INTERVAL: sensor[CONF_POLL_INTERVAL]}
)
for sensor in self.stored_configuration[CONF_SENSORS]
if sensor[CONF_TYPE] == "dht"
]
@callback
def async_ds18b20_sensor_configuration(self):
"""Return the configuration map for syncing DS18B20 sensors."""
return [
self.format_zone(sensor[CONF_ZONE])
for sensor in self.stored_configuration[CONF_SENSORS]
if sensor[CONF_TYPE] == "ds18b20"
]
async def async_update_initial_states(self):
"""Update the initial state of each sensor from status poll."""
for sensor_data in self.status.get("sensors"):
sensor_config = self.stored_configuration[CONF_BINARY_SENSORS].get(
sensor_data.get(CONF_ZONE, sensor_data.get(CONF_PIN)), {}
)
entity_id = sensor_config.get(ATTR_ENTITY_ID)
state = bool(sensor_data.get(ATTR_STATE))
if sensor_config.get(CONF_INVERSE):
state = not state
async_dispatcher_send(
self.hass, SIGNAL_SENSOR_UPDATE.format(entity_id), state
)
@callback
def async_desired_settings_payload(self):
"""Return a dict representing the desired device configuration."""
desired_api_host = (
self.hass.data[DOMAIN].get(CONF_API_HOST) or self.hass.config.api.base_url
)
desired_api_endpoint = desired_api_host + ENDPOINT_ROOT
return {
"sensors": self.async_binary_sensor_configuration(),
"actuators": self.async_actuator_configuration(),
"dht_sensors": self.async_dht_sensor_configuration(),
"ds18b20_sensors": self.async_ds18b20_sensor_configuration(),
"auth_token": self.config.get(CONF_ACCESS_TOKEN),
"endpoint": desired_api_endpoint,
"blink": self.options.get(CONF_BLINK, True),
"discovery": self.options.get(CONF_DISCOVERY, True),
}
@callback
def async_current_settings_payload(self):
"""Return a dict of configuration currently stored on the device."""
settings = self.status["settings"]
if not settings:
settings = {}
return {
"sensors": [
{self.api_version: s[self.api_version]}
for s in self.status.get("sensors")
],
"actuators": self.status.get("actuators"),
"dht_sensors": self.status.get(CONF_DHT_SENSORS),
"ds18b20_sensors": self.status.get(CONF_DS18B20_SENSORS),
"auth_token": settings.get("token"),
"endpoint": settings.get("endpoint"),
"blink": settings.get(CONF_BLINK),
"discovery": settings.get(CONF_DISCOVERY),
}
async def async_sync_device_config(self):
"""Sync the new zone configuration to the Konnected device if needed."""
_LOGGER.debug(
"Device %s settings payload: %s",
self.device_id,
self.async_desired_settings_payload(),
)
if (
self.async_desired_settings_payload()
!= self.async_current_settings_payload()
):
_LOGGER.info("pushing settings to device %s", self.device_id)
await self.client.put_settings(**self.async_desired_settings_payload())
async def get_status(hass, host, port):
"""Get the status of a Konnected Panel."""
client = konnected.Client(
host, str(port), aiohttp_client.async_get_clientsession(hass)
)
try:
return await client.get_status()
except client.ClientError as err:
_LOGGER.error("Exception trying to get panel status: %s", err)
raise CannotConnect

26
homeassistant/components/konnected/sensor.py
View File

@ -4,9 +4,9 @@ import logging
from homeassistant.const import (
CONF_DEVICES,
CONF_NAME,
CONF_PIN,
CONF_SENSORS,
CONF_TYPE,
CONF_ZONE,
DEVICE_CLASS_HUMIDITY,
DEVICE_CLASS_TEMPERATURE,
TEMP_CELSIUS,
@ -25,13 +25,10 @@ SENSOR_TYPES = {
}
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up sensors attached to a Konnected device."""
if discovery_info is None:
return
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up sensors attached to a Konnected device from a config entry."""
data = hass.data[KONNECTED_DOMAIN]
device_id = discovery_info["device_id"]
device_id = config_entry.data["id"]
sensors = []
# Initialize all DHT sensors.
@ -53,7 +50,7 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
(
s
for s in data[CONF_DEVICES][device_id][CONF_SENSORS]
if s[CONF_TYPE] == "ds18b20" and s[CONF_PIN] == attrs.get(CONF_PIN)
if s[CONF_TYPE] == "ds18b20" and s[CONF_ZONE] == attrs.get(CONF_ZONE)
),
None,
)
@ -85,10 +82,10 @@ class KonnectedSensor(Entity):
self._data = data
self._device_id = device_id
self._type = sensor_type
self._pin_num = self._data.get(CONF_PIN)
self._zone_num = self._data.get(CONF_ZONE)
self._unit_of_measurement = SENSOR_TYPES[sensor_type][1]
self._unique_id = addr or "{}-{}-{}".format(
device_id, self._pin_num, sensor_type
device_id, self._zone_num, sensor_type
)
# set initial state if known at initialization
@ -99,7 +96,7 @@ class KonnectedSensor(Entity):
# set entity name if given
self._name = self._data.get(CONF_NAME)
if self._name:
self._name += " " + SENSOR_TYPES[sensor_type][0]
self._name += f" {SENSOR_TYPES[sensor_type][0]}"
@property
def unique_id(self) -> str:
@ -121,6 +118,13 @@ class KonnectedSensor(Entity):
"""Return the unit of measurement."""
return self._unit_of_measurement
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(KONNECTED_DOMAIN, self._device_id)},
}
async def async_added_to_hass(self):
"""Store entity_id and register state change callback."""
entity_id_key = self._addr or self._type

101
homeassistant/components/konnected/strings.json Normal file
View File

@ -0,0 +1,101 @@
{
"config": {
"title": "Konnected.io",
"step": {
"user": {
"title": "Discover Konnected Device",
"description": "Please enter the host information for your Konnected Panel.",
"data": {
"host": "Konnected device IP address",
"port": "Konnected device port"
}
},
"confirm": {
"title": "Konnected Device Ready",
"description": "Model: {model}\nHost: {host}\nPort: {port}\n\nYou can configure the IO and panel behavior in the Konnected Alarm Panel settings."
}
},
"error": {
"cannot_connect": "Unable to connect to a Konnected Panel at {host}:{port}"
},
"abort": {
"unknown": "Unknown error occurred",
"already_configured": "Device is already configured",
"already_in_progress": "Config flow for device is already in progress.",
"not_konn_panel": "Not a recognized Konnected.io device"
}
},
"options": {
"title": "Konnected Alarm Panel Options",
"step": {
"options_io": {
"title": "Configure I/O",
"description": "Discovered a {model} at {host}. Select the base configuration of each I/O below - depending on the I/O it may allow for binary sensors (open/close contacts), digital sensors (dht and ds18b20), or switchable outputs. You'll be able to configure detailed options in the next steps.",
"data": {
"1": "Zone 1",
"2": "Zone 2",
"3": "Zone 3",
"4": "Zone 4",
"5": "Zone 5",
"6": "Zone 6",
"7": "Zone 7",
"out": "OUT"
}
},
"options_io_ext": {
"title": "Configure Extended I/O",
"description": "Select the configuration of the remaining I/O below. You'll be able to configure detailed options in the next steps.",
"data": {
"8": "Zone 8",
"9": "Zone 9",
"10": "Zone 10",
"11": "Zone 11",
"12": "Zone 12",
"out1": "OUT1",
"alarm1": "ALARM1",
"alarm2_out2": "OUT2/ALARM2"
}
},
"options_binary": {
"title": "Configure Binary Sensor",
"description": "Please select the options for the binary sensor attached to {zone}",
"data": {
"type": "Binary Sensor Type",
"name": "Name (optional)",
"inverse": "Invert the open/close state"
}
},
"options_digital": {
"title": "Configure Digital Sensor",
"description": "Please select the options for the digital sensor attached to {zone}",
"data": {
"type": "Sensor Type",
"name": "Name (optional)",
"poll_interval": "Poll Interval (minutes) (optional)"
}
},
"options_switch": {
"title": "Configure Switchable Output",
"description": "Please select the output options for {zone}",
"data": {
"name": "Name (optional)",
"activation": "Output when on",
"momentary": "Pulse duration (ms) (optional)",
"pause": "Pause between pulses (ms) (optional)",
"repeat": "Times to repeat (-1=infinite) (optional)"
}
},
"options_misc": {
"title": "Configure Misc",
"description": "Please select the desired behavior for your panel",
"data": {
"blink": "Blink panel LED on when sending state change"
}
}
},
"error": {},
"abort": {
"not_konn_panel": "Not a recognized Konnected.io device"
}
}
}

53
homeassistant/components/konnected/switch.py
View File

@ -5,12 +5,12 @@ from homeassistant.const import (
ATTR_STATE,
CONF_DEVICES,
CONF_NAME,
CONF_PIN,
CONF_SWITCHES,
CONF_ZONE,
)
from homeassistant.helpers.entity import ToggleEntity
from . import (
from .const import (
CONF_ACTIVATION,
CONF_MOMENTARY,
CONF_PAUSE,
@ -23,16 +23,13 @@ from . import (
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set switches attached to a Konnected device."""
if discovery_info is None:
return
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up switches attached to a Konnected device from a config entry."""
data = hass.data[KONNECTED_DOMAIN]
device_id = discovery_info["device_id"]
device_id = config_entry.data["id"]
switches = [
KonnectedSwitch(device_id, pin_data.get(CONF_PIN), pin_data)
for pin_data in data[CONF_DEVICES][device_id][CONF_SWITCHES]
KonnectedSwitch(device_id, zone_data.get(CONF_ZONE), zone_data)
for zone_data in data[CONF_DEVICES][device_id][CONF_SWITCHES]
]
async_add_entities(switches)
@ -40,11 +37,11 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
class KonnectedSwitch(ToggleEntity):
"""Representation of a Konnected switch."""
def __init__(self, device_id, pin_num, data):
def __init__(self, device_id, zone_num, data):
"""Initialize the Konnected switch."""
self._data = data
self._device_id = device_id
self._pin_num = pin_num
self._zone_num = zone_num
self._activation = self._data.get(CONF_ACTIVATION, STATE_HIGH)
self._momentary = self._data.get(CONF_MOMENTARY)
self._pause = self._data.get(CONF_PAUSE)
@ -52,7 +49,7 @@ class KonnectedSwitch(ToggleEntity):
self._state = self._boolean_state(self._data.get(ATTR_STATE))
self._name = self._data.get(CONF_NAME)
self._unique_id = "{}-{}-{}-{}-{}".format(
device_id, self._pin_num, self._momentary, self._pause, self._repeat
device_id, self._zone_num, self._momentary, self._pause, self._repeat
)
@property
@ -71,16 +68,22 @@ class KonnectedSwitch(ToggleEntity):
return self._state
@property
def client(self):
def panel(self):
"""Return the Konnected HTTP client."""
return self.hass.data[KONNECTED_DOMAIN][CONF_DEVICES][self._device_id].get(
"client"
)
device_data = self.hass.data[KONNECTED_DOMAIN][CONF_DEVICES][self._device_id]
return device_data.get("panel")
def turn_on(self, **kwargs):
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(KONNECTED_DOMAIN, self._device_id)},
}
async def async_turn_on(self, **kwargs):
"""Send a command to turn on the switch."""
resp = self.client.put_device(
self._pin_num,
resp = await self.panel.update_switch(
self._zone_num,
int(self._activation == STATE_HIGH),
self._momentary,
self._repeat,
@ -94,9 +97,11 @@ class KonnectedSwitch(ToggleEntity):
# Immediately set the state back off for momentary switches
self._set_state(False)
def turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs):
"""Send a command to turn off the switch."""
resp = self.client.put_device(self._pin_num, int(self._activation == STATE_LOW))
resp = await self.panel.update_switch(
self._zone_num, int(self._activation == STATE_LOW)
)
if resp.get(ATTR_STATE) is not None:
self._set_state(self._boolean_state(resp.get(ATTR_STATE)))
@ -111,9 +116,9 @@ class KonnectedSwitch(ToggleEntity):
def _set_state(self, state):
self._state = state
self.schedule_update_ha_state()
self.async_schedule_update_ha_state()
_LOGGER.debug(
"Setting status of %s actuator pin %s to %s",
"Setting status of %s actuator zone %s to %s",
self._device_id,
self.name,
state,

1
homeassistant/generated/config_flows.py
View File

@ -47,6 +47,7 @@ FLOWS = [
"ipma",
"iqvia",
"izone",
"konnected",
"life360",
"lifx",
"linky",

5
homeassistant/generated/ssdp.py
View File

@ -36,6 +36,11 @@ SSDP = {
"modelName": "Philips hue bridge 2015"
}
],
"konnected": [
{
"manufacturer": "konnected.io"
}
],
"samsungtv": [
{
"st": "urn:samsung.com:device:RemoteControlReceiver:1"

2
requirements_all.txt
View File

@ -767,7 +767,7 @@ keyrings.alt==3.4.0
kiwiki-client==0.1.1
# homeassistant.components.konnected
konnected==0.1.5
konnected==1.1.0
# homeassistant.components.eufy
lakeside==0.12

3
requirements_test_all.txt
View File

@ -287,6 +287,9 @@ keyring==20.0.0
# homeassistant.scripts.keyring
keyrings.alt==3.4.0
# homeassistant.components.konnected
konnected==1.1.0
# homeassistant.components.dyson
libpurecool==0.6.1

1
tests/components/konnected/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Konnected component."""

1052
tests/components/konnected/test_config_flow.py Normal file
View File

File diff suppressed because it is too large Load Diff

601
tests/components/konnected/test_init.py Normal file
View File

@ -0,0 +1,601 @@
"""Test Konnected setup process."""
from asynctest import patch
import pytest
from homeassistant.components import konnected
from homeassistant.components.konnected import config_flow
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
@pytest.fixture(name="mock_panel")
async def mock_panel_fixture():
"""Mock a Konnected Panel bridge."""
with patch("konnected.Client", autospec=True) as konn_client:
def mock_constructor(host, port, websession):
"""Fake the panel constructor."""
konn_client.host = host
konn_client.port = port
return konn_client
konn_client.side_effect = mock_constructor
konn_client.ClientError = config_flow.CannotConnect
konn_client.get_status.return_value = {
"hwVersion": "2.3.0",
"swVersion": "2.3.1",
"heap": 10000,
"uptime": 12222,
"ip": "192.168.1.90",
"port": 9123,
"sensors": [],
"actuators": [],
"dht_sensors": [],
"ds18b20_sensors": [],
"mac": "11:22:33:44:55:66",
"settings": {},
}
yield konn_client
async def test_config_schema(hass):
"""Test that config schema is imported properly."""
config = {
konnected.DOMAIN: {
konnected.CONF_ACCESS_TOKEN: "abcdefgh",
konnected.CONF_DEVICES: [{konnected.CONF_ID: "aabbccddeeff"}],
}
}
assert konnected.CONFIG_SCHEMA(config) == {
"konnected": {
"access_token": "abcdefgh",
"devices": [
{
"default_options": {
"blink": True,
"discovery": True,
"io": {
"1": "Disabled",
"10": "Disabled",
"11": "Disabled",
"12": "Disabled",
"2": "Disabled",
"3": "Disabled",
"4": "Disabled",
"5": "Disabled",
"6": "Disabled",
"7": "Disabled",
"8": "Disabled",
"9": "Disabled",
"alarm1": "Disabled",
"alarm2_out2": "Disabled",
"out": "Disabled",
"out1": "Disabled",
},
},
"id": "aabbccddeeff",
}
],
}
}
# check with host info
config = {
konnected.DOMAIN: {
konnected.CONF_ACCESS_TOKEN: "abcdefgh",
konnected.CONF_DEVICES: [
{konnected.CONF_ID: "aabbccddeeff", "host": "192.168.1.1", "port": 1234}
],
}
}
assert konnected.CONFIG_SCHEMA(config) == {
"konnected": {
"access_token": "abcdefgh",
"devices": [
{
"default_options": {
"blink": True,
"discovery": True,
"io": {
"1": "Disabled",
"10": "Disabled",
"11": "Disabled",
"12": "Disabled",
"2": "Disabled",
"3": "Disabled",
"4": "Disabled",
"5": "Disabled",
"6": "Disabled",
"7": "Disabled",
"8": "Disabled",
"9": "Disabled",
"alarm1": "Disabled",
"alarm2_out2": "Disabled",
"out": "Disabled",
"out1": "Disabled",
},
},
"id": "aabbccddeeff",
"host": "192.168.1.1",
"port": 1234,
}
],
}
}
# check pin to zone
config = {
konnected.DOMAIN: {
konnected.CONF_ACCESS_TOKEN: "abcdefgh",
konnected.CONF_DEVICES: [
{
konnected.CONF_ID: "aabbccddeeff",
"binary_sensors": [
{"pin": 2, "type": "door"},
{"zone": 1, "type": "door"},
],
}
],
}
}
assert konnected.CONFIG_SCHEMA(config) == {
"konnected": {
"access_token": "abcdefgh",
"devices": [
{
"default_options": {
"blink": True,
"discovery": True,
"io": {
"1": "Binary Sensor",
"10": "Disabled",
"11": "Disabled",
"12": "Disabled",
"2": "Binary Sensor",
"3": "Disabled",
"4": "Disabled",
"5": "Disabled",
"6": "Disabled",
"7": "Disabled",
"8": "Disabled",
"9": "Disabled",
"alarm1": "Disabled",
"alarm2_out2": "Disabled",
"out": "Disabled",
"out1": "Disabled",
},
"binary_sensors": [
{"inverse": False, "type": "door", "zone": "2"},
{"inverse": False, "type": "door", "zone": "1"},
],
},
"id": "aabbccddeeff",
}
],
}
}
async def test_setup_with_no_config(hass):
"""Test that we do not discover anything or try to set up a Konnected panel."""
assert await async_setup_component(hass, konnected.DOMAIN, {})
# No flows started
assert len(hass.config_entries.flow.async_progress()) == 0
# Nothing saved from configuration.yaml
assert hass.data[konnected.DOMAIN][konnected.CONF_ACCESS_TOKEN] is None
assert hass.data[konnected.DOMAIN][konnected.CONF_API_HOST] is None
assert konnected.YAML_CONFIGS not in hass.data[konnected.DOMAIN]
async def test_setup_defined_hosts_known_auth(hass):
"""Test we don't initiate a config entry if configured panel is known."""
MockConfigEntry(
domain="konnected",
unique_id="112233445566",
data={"host": "0.0.0.0", "id": "112233445566"},
).add_to_hass(hass)
MockConfigEntry(
domain="konnected",
unique_id="aabbccddeeff",
data={"host": "1.2.3.4", "id": "aabbccddeeff"},
).add_to_hass(hass)
assert (
await async_setup_component(
hass,
konnected.DOMAIN,
{
konnected.DOMAIN: {
konnected.CONF_ACCESS_TOKEN: "abcdefgh",
konnected.CONF_DEVICES: [
{
config_flow.CONF_ID: "aabbccddeeff",
config_flow.CONF_HOST: "0.0.0.0",
config_flow.CONF_PORT: 1234,
},
],
}
},
)
is True
)
assert hass.data[konnected.DOMAIN][konnected.CONF_ACCESS_TOKEN] == "abcdefgh"
assert konnected.YAML_CONFIGS not in hass.data[konnected.DOMAIN]
# Flow aborted
assert len(hass.config_entries.flow.async_progress()) == 0
async def test_setup_defined_hosts_no_known_auth(hass):
"""Test we initiate config entry if config panel is not known."""
assert (
await async_setup_component(
hass,
konnected.DOMAIN,
{
konnected.DOMAIN: {
konnected.CONF_ACCESS_TOKEN: "abcdefgh",
konnected.CONF_DEVICES: [{konnected.CONF_ID: "aabbccddeeff"}],
}
},
)
is True
)
# Flow started for discovered bridge
assert len(hass.config_entries.flow.async_progress()) == 1
async def test_config_passed_to_config_entry(hass):
"""Test that configured options for a host are loaded via config entry."""
entry = MockConfigEntry(
domain=konnected.DOMAIN,
data={config_flow.CONF_ID: "aabbccddeeff", config_flow.CONF_HOST: "0.0.0.0"},
)
entry.add_to_hass(hass)
with patch.object(konnected, "AlarmPanel", autospec=True) as mock_int:
assert (
await async_setup_component(
hass,
konnected.DOMAIN,
{
konnected.DOMAIN: {
konnected.CONF_ACCESS_TOKEN: "abcdefgh",
konnected.CONF_DEVICES: [{konnected.CONF_ID: "aabbccddeeff"}],
}
},
)
is True
)
assert len(mock_int.mock_calls) == 3
p_hass, p_entry = mock_int.mock_calls[0][1]
assert p_hass is hass
assert p_entry is entry
async def test_unload_entry(hass, mock_panel):
"""Test being able to unload an entry."""
entry = MockConfigEntry(
domain=konnected.DOMAIN, data={konnected.CONF_ID: "aabbccddeeff"}
)
entry.add_to_hass(hass)
assert await async_setup_component(hass, konnected.DOMAIN, {}) is True
assert hass.data[konnected.DOMAIN]["devices"].get("aabbccddeeff") is not None
assert await konnected.async_unload_entry(hass, entry)
assert hass.data[konnected.DOMAIN]["devices"] == {}
async def test_api(hass, aiohttp_client, mock_panel):
"""Test callback view."""
await async_setup_component(hass, "http", {"http": {}})
device_config = config_flow.CONFIG_ENTRY_SCHEMA(
{
"host": "1.2.3.4",
"port": 1234,
"id": "112233445566",
"model": "Konnected Pro",
"access_token": "abcdefgh",
"default_options": config_flow.OPTIONS_SCHEMA({config_flow.CONF_IO: {}}),
}
)
device_options = config_flow.OPTIONS_SCHEMA(
{
"io": {
"1": "Binary Sensor",
"2": "Binary Sensor",
"3": "Binary Sensor",
"4": "Digital Sensor",
"5": "Digital Sensor",
"6": "Switchable Output",
"out": "Switchable Output",
},
"binary_sensors": [
{"zone": "1", "type": "door"},
{"zone": "2", "type": "window", "name": "winder", "inverse": True},
{"zone": "3", "type": "door"},
],
"sensors": [
{"zone": "4", "type": "dht"},
{"zone": "5", "type": "ds18b20", "name": "temper"},
],
"switches": [
{
"zone": "out",
"name": "switcher",
"activation": "low",
"momentary": 50,
"pause": 100,
"repeat": 4,
},
{"zone": "6"},
],
}
)
entry = MockConfigEntry(
domain="konnected",
title="Konnected Alarm Panel",
data=device_config,
options=device_options,
)
entry.add_to_hass(hass)
assert (
await async_setup_component(
hass,
konnected.DOMAIN,
{konnected.DOMAIN: {konnected.CONF_ACCESS_TOKEN: "globaltoken"}},
)
is True
)
client = await aiohttp_client(hass.http.app)
# Test the get endpoint for switch status polling
resp = await client.get("/api/konnected")
assert resp.status == 404 # no device provided
resp = await client.get("/api/konnected/223344556677")
assert resp.status == 404 # unknown device provided
resp = await client.get("/api/konnected/device/112233445566")
assert resp.status == 404 # no zone provided
result = await resp.json()
assert result == {"message": "Switch on zone or pin unknown not configured"}
resp = await client.get("/api/konnected/device/112233445566?zone=8")
assert resp.status == 404 # invalid zone
result = await resp.json()
assert result == {"message": "Switch on zone or pin 8 not configured"}
resp = await client.get("/api/konnected/device/112233445566?pin=12")
assert resp.status == 404 # invalid pin
result = await resp.json()
assert result == {"message": "Switch on zone or pin 12 not configured"}
resp = await client.get("/api/konnected/device/112233445566?zone=out")
assert resp.status == 200
result = await resp.json()
assert result == {"state": 1, "zone": "out"}
resp = await client.get("/api/konnected/device/112233445566?pin=8")
assert resp.status == 200
result = await resp.json()
assert result == {"state": 1, "pin": "8"}
# Test the post endpoint for sensor updates
resp = await client.post("/api/konnected/device", json={"zone": "1", "state": 1})
assert resp.status == 404
resp = await client.post(
"/api/konnected/device/112233445566", json={"zone": "1", "state": 1}
)
assert resp.status == 401
result = await resp.json()
assert result == {"message": "unauthorized"}
resp = await client.post(
"/api/konnected/device/223344556677",
headers={"Authorization": "Bearer abcdefgh"},
json={"zone": "1", "state": 1},
)
assert resp.status == 400
resp = await client.post(
"/api/konnected/device/112233445566",
headers={"Authorization": "Bearer abcdefgh"},
json={"zone": "15", "state": 1},
)
assert resp.status == 400
result = await resp.json()
assert result == {"message": "unregistered sensor/actuator"}
resp = await client.post(
"/api/konnected/device/112233445566",
headers={"Authorization": "Bearer abcdefgh"},
json={"zone": "1", "state": 1},
)
assert resp.status == 200
result = await resp.json()
assert result == {"message": "ok"}
resp = await client.post(
"/api/konnected/device/112233445566",
headers={"Authorization": "Bearer globaltoken"},
json={"zone": "1", "state": 1},
)
assert resp.status == 200
result = await resp.json()
assert result == {"message": "ok"}
resp = await client.post(
"/api/konnected/device/112233445566",
headers={"Authorization": "Bearer abcdefgh"},
json={"zone": "4", "temp": 22, "humi": 20},
)
assert resp.status == 200
result = await resp.json()
assert result == {"message": "ok"}
# Test the put endpoint for sensor updates
resp = await client.post(
"/api/konnected/device/112233445566",
headers={"Authorization": "Bearer abcdefgh"},
json={"zone": "1", "state": 1},
)
assert resp.status == 200
result = await resp.json()
assert result == {"message": "ok"}
async def test_state_updates(hass, aiohttp_client, mock_panel):
"""Test callback view."""
await async_setup_component(hass, "http", {"http": {}})
device_config = config_flow.CONFIG_ENTRY_SCHEMA(
{
"host": "1.2.3.4",
"port": 1234,
"id": "112233445566",
"model": "Konnected Pro",
"access_token": "abcdefgh",
"default_options": config_flow.OPTIONS_SCHEMA({config_flow.CONF_IO: {}}),
}
)
device_options = config_flow.OPTIONS_SCHEMA(
{
"io": {
"1": "Binary Sensor",
"2": "Binary Sensor",
"3": "Binary Sensor",
"4": "Digital Sensor",
"5": "Digital Sensor",
"6": "Switchable Output",
"out": "Switchable Output",
},
"binary_sensors": [
{"zone": "1", "type": "door"},
{"zone": "2", "type": "window", "name": "winder", "inverse": True},
{"zone": "3", "type": "door"},
],
"sensors": [
{"zone": "4", "type": "dht"},
{"zone": "5", "type": "ds18b20", "name": "temper"},
],
"switches": [
{
"zone": "out",
"name": "switcher",
"activation": "low",
"momentary": 50,
"pause": 100,
"repeat": 4,
},
{"zone": "6"},
],
}
)
entry = MockConfigEntry(
domain="konnected",
title="Konnected Alarm Panel",
data=device_config,
options=device_options,
)
entry.add_to_hass(hass)
assert (
await async_setup_component(
hass,
konnected.DOMAIN,
{konnected.DOMAIN: {konnected.CONF_ACCESS_TOKEN: "1122334455"}},
)
is True
)
client = await aiohttp_client(hass.http.app)
# Test updating a binary sensor
resp = await client.post(
"/api/konnected/device/112233445566",
headers={"Authorization": "Bearer abcdefgh"},
json={"zone": "1", "state": 0},
)
assert resp.status == 200
result = await resp.json()
assert result == {"message": "ok"}
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.konnected_445566_zone_1").state == "off"
resp = await client.post(
"/api/konnected/device/112233445566",
headers={"Authorization": "Bearer abcdefgh"},
json={"zone": "1", "state": 1},
)
assert resp.status == 200
result = await resp.json()
assert result == {"message": "ok"}
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.konnected_445566_zone_1").state == "on"
# Test updating sht sensor
resp = await client.post(
"/api/konnected/device/112233445566",
headers={"Authorization": "Bearer abcdefgh"},
json={"zone": "4", "temp": 22, "humi": 20},
)
assert resp.status == 200
result = await resp.json()
assert result == {"message": "ok"}
await hass.async_block_till_done()
assert hass.states.get("sensor.konnected_445566_sensor_4_humidity").state == "20"
assert (
hass.states.get("sensor.konnected_445566_sensor_4_temperature").state == "22.0"
)
resp = await client.post(
"/api/konnected/device/112233445566",
headers={"Authorization": "Bearer abcdefgh"},
json={"zone": "4", "temp": 25, "humi": 23},
)
assert resp.status == 200
result = await resp.json()
assert result == {"message": "ok"}
await hass.async_block_till_done()
assert hass.states.get("sensor.konnected_445566_sensor_4_humidity").state == "23"
assert (
hass.states.get("sensor.konnected_445566_sensor_4_temperature").state == "25.0"
)
# Test updating ds sensor
resp = await client.post(
"/api/konnected/device/112233445566",
headers={"Authorization": "Bearer abcdefgh"},
json={"zone": "5", "temp": 32, "addr": 1},
)
assert resp.status == 200
result = await resp.json()
assert result == {"message": "ok"}
await hass.async_block_till_done()
assert hass.states.get("sensor.temper_temperature").state == "32.0"
resp = await client.post(
"/api/konnected/device/112233445566",
headers={"Authorization": "Bearer abcdefgh"},
json={"zone": "5", "temp": 42, "addr": 1},
)
assert resp.status == 200
result = await resp.json()
assert result == {"message": "ok"}
await hass.async_block_till_done()
assert hass.states.get("sensor.temper_temperature").state == "42.0"

375
tests/components/konnected/test_panel.py Normal file
View File

@ -0,0 +1,375 @@
"""Test Konnected setup process."""
from asynctest import patch
import pytest
from homeassistant.components.konnected import config_flow, panel
from tests.common import MockConfigEntry
@pytest.fixture(name="mock_panel")
async def mock_panel_fixture():
"""Mock a Konnected Panel bridge."""
with patch("konnected.Client", autospec=True) as konn_client:
def mock_constructor(host, port, websession):
"""Fake the panel constructor."""
konn_client.host = host
konn_client.port = port
return konn_client
konn_client.side_effect = mock_constructor
konn_client.ClientError = config_flow.CannotConnect
konn_client.get_status.return_value = {
"hwVersion": "2.3.0",
"swVersion": "2.3.1",
"heap": 10000,
"uptime": 12222,
"ip": "192.168.1.90",
"port": 9123,
"sensors": [],
"actuators": [],
"dht_sensors": [],
"ds18b20_sensors": [],
"mac": "11:22:33:44:55:66",
"model": "Konnected Pro", # `model` field only included in pro
"settings": {},
}
yield konn_client
async def test_create_and_setup(hass, mock_panel):
"""Test that we create a Konnected Panel and save the data."""
device_config = config_flow.CONFIG_ENTRY_SCHEMA(
{
"host": "1.2.3.4",
"port": 1234,
"id": "112233445566",
"model": "Konnected Pro",
"access_token": "11223344556677889900",
"default_options": config_flow.OPTIONS_SCHEMA({config_flow.CONF_IO: {}}),
}
)
device_options = config_flow.OPTIONS_SCHEMA(
{
"io": {
"1": "Binary Sensor",
"2": "Binary Sensor",
"3": "Binary Sensor",
"4": "Digital Sensor",
"5": "Digital Sensor",
"6": "Switchable Output",
"out": "Switchable Output",
},
"binary_sensors": [
{"zone": "1", "type": "door"},
{"zone": "2", "type": "window", "name": "winder", "inverse": True},
{"zone": "3", "type": "door"},
],
"sensors": [
{"zone": "4", "type": "dht"},
{"zone": "5", "type": "ds18b20", "name": "temper"},
],
"switches": [
{
"zone": "out",
"name": "switcher",
"activation": "low",
"momentary": 50,
"pause": 100,
"repeat": 4,
},
{"zone": "6"},
],
}
)
entry = MockConfigEntry(
domain="konnected",
title="Konnected Alarm Panel",
data=device_config,
options=device_options,
)
entry.add_to_hass(hass)
hass.data[panel.DOMAIN] = {
panel.CONF_API_HOST: "192.168.1.1",
}
# override get_status to reflect non-pro board
mock_panel.get_status.return_value = {
"hwVersion": "2.3.0",
"swVersion": "2.3.1",
"heap": 10000,
"uptime": 12222,
"ip": "192.168.1.90",
"port": 9123,
"sensors": [],
"actuators": [],
"dht_sensors": [],
"ds18b20_sensors": [],
"mac": "11:22:33:44:55:66",
"settings": {},
}
device = panel.AlarmPanel(hass, entry)
await device.async_save_data()
await device.async_connect()
await device.update_switch("1", 0)
# confirm the correct api is used
# pylint: disable=no-member
assert device.client.put_device.call_count == 1
assert device.client.put_zone.call_count == 0
# confirm the settings are sent to the panel
# pylint: disable=no-member
assert device.client.put_settings.call_args_list[0][1] == {
"sensors": [{"pin": "1"}, {"pin": "2"}, {"pin": "5"}],
"actuators": [{"trigger": 0, "pin": "8"}, {"trigger": 1, "pin": "9"}],
"dht_sensors": [{"poll_interval": 3, "pin": "6"}],
"ds18b20_sensors": [{"pin": "7"}],
"auth_token": "11223344556677889900",
"blink": True,
"discovery": True,
"endpoint": "192.168.1.1/api/konnected",
}
# confirm the device settings are saved in hass.data
assert hass.data[panel.DOMAIN][panel.CONF_DEVICES] == {
"112233445566": {
"binary_sensors": {
"1": {
"inverse": False,
"name": "Konnected 445566 Zone 1",
"state": None,
"type": "door",
},
"2": {
"inverse": True,
"name": "winder",
"state": None,
"type": "window",
},
"3": {
"inverse": False,
"name": "Konnected 445566 Zone 3",
"state": None,
"type": "door",
},
},
"blink": True,
"panel": device,
"discovery": True,
"host": "1.2.3.4",
"port": 1234,
"sensors": [
{
"name": "Konnected 445566 Sensor 4",
"poll_interval": 3,
"type": "dht",
"zone": "4",
},
{"name": "temper", "poll_interval": 3, "type": "ds18b20", "zone": "5"},
],
"switches": [
{
"activation": "low",
"momentary": 50,
"name": "switcher",
"pause": 100,
"repeat": 4,
"state": None,
"zone": "out",
},
{
"activation": "high",
"momentary": None,
"name": "Konnected 445566 Actuator 6",
"pause": None,
"repeat": None,
"state": None,
"zone": "6",
},
],
}
}
async def test_create_and_setup_pro(hass, mock_panel):
"""Test that we create a Konnected Pro Panel and save the data."""
device_config = config_flow.CONFIG_ENTRY_SCHEMA(
{
"host": "1.2.3.4",
"port": 1234,
"id": "112233445566",
"model": "Konnected Pro",
"access_token": "11223344556677889900",
"default_options": config_flow.OPTIONS_SCHEMA({config_flow.CONF_IO: {}}),
}
)
device_options = config_flow.OPTIONS_SCHEMA(
{
"io": {
"2": "Binary Sensor",
"6": "Binary Sensor",
"10": "Binary Sensor",
"3": "Digital Sensor",
"7": "Digital Sensor",
"11": "Digital Sensor",
"4": "Switchable Output",
"8": "Switchable Output",
"out1": "Switchable Output",
"alarm1": "Switchable Output",
},
"binary_sensors": [
{"zone": "2", "type": "door"},
{"zone": "6", "type": "window", "name": "winder", "inverse": True},
{"zone": "10", "type": "door"},
],
"sensors": [
{"zone": "3", "type": "dht"},
{"zone": "7", "type": "ds18b20", "name": "temper"},
{"zone": "11", "type": "dht", "poll_interval": 5},
],
"switches": [
{"zone": "4"},
{
"zone": "8",
"name": "switcher",
"activation": "low",
"momentary": 50,
"pause": 100,
"repeat": 4,
},
{"zone": "out1"},
{"zone": "alarm1"},
],
}
)
entry = MockConfigEntry(
domain="konnected",
title="Konnected Pro Alarm Panel",
data=device_config,
options=device_options,
)
entry.add_to_hass(hass)
hass.data[panel.DOMAIN] = {
panel.CONF_API_HOST: "192.168.1.1",
}
device = panel.AlarmPanel(hass, entry)
await device.async_save_data()
await device.async_connect()
await device.update_switch("2", 1)
# confirm the correct api is used
# pylint: disable=no-member
assert device.client.put_device.call_count == 0
assert device.client.put_zone.call_count == 1
# confirm the settings are sent to the panel
# pylint: disable=no-member
assert device.client.put_settings.call_args_list[0][1] == {
"sensors": [{"zone": "2"}, {"zone": "6"}, {"zone": "10"}],
"actuators": [
{"trigger": 1, "zone": "4"},
{"trigger": 0, "zone": "8"},
{"trigger": 1, "zone": "out1"},
{"trigger": 1, "zone": "alarm1"},
],
"dht_sensors": [
{"poll_interval": 3, "zone": "3"},
{"poll_interval": 5, "zone": "11"},
],
"ds18b20_sensors": [{"zone": "7"}],
"auth_token": "11223344556677889900",
"blink": True,
"discovery": True,
"endpoint": "192.168.1.1/api/konnected",
}
# confirm the device settings are saved in hass.data
assert hass.data[panel.DOMAIN][panel.CONF_DEVICES] == {
"112233445566": {
"binary_sensors": {
"10": {
"inverse": False,
"name": "Konnected 445566 Zone 10",
"state": None,
"type": "door",
},
"2": {
"inverse": False,
"name": "Konnected 445566 Zone 2",
"state": None,
"type": "door",
},
"6": {
"inverse": True,
"name": "winder",
"state": None,
"type": "window",
},
},
"blink": True,
"panel": device,
"discovery": True,
"host": "1.2.3.4",
"port": 1234,
"sensors": [
{
"name": "Konnected 445566 Sensor 3",
"poll_interval": 3,
"type": "dht",
"zone": "3",
},
{"name": "temper", "poll_interval": 3, "type": "ds18b20", "zone": "7"},
{
"name": "Konnected 445566 Sensor 11",
"poll_interval": 5,
"type": "dht",
"zone": "11",
},
],
"switches": [
{
"activation": "high",
"momentary": None,
"name": "Konnected 445566 Actuator 4",
"pause": None,
"repeat": None,
"state": None,
"zone": "4",
},
{
"activation": "low",
"momentary": 50,
"name": "switcher",
"pause": 100,
"repeat": 4,
"state": None,
"zone": "8",
},
{
"activation": "high",
"momentary": None,
"name": "Konnected 445566 Actuator out1",
"pause": None,
"repeat": None,
"state": None,
"zone": "out1",
},
{
"activation": "high",
"momentary": None,
"name": "Konnected 445566 Actuator alarm1",
"pause": None,
"repeat": None,
"state": None,
"zone": "alarm1",
},
],
}
}