Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add config flow step user to dsmr (#50318) 

Co-authored-by: Franck Nijhof <git@frenck.dev>
pull/52143/head
Rob Bierbooms 2021-06-24 10:16:08 +02:00 committed by GitHub
parent 0714ee68eb
commit aa56a21b45
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 490 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

148
homeassistant/components/dsmr/config_flow.py
View File

@ -4,16 +4,18 @@ from __future__ import annotations
import asyncio
from functools import partial
import logging
import os
from typing import Any
from async_timeout import timeout
from dsmr_parser import obis_references as obis_ref
from dsmr_parser.clients.protocol import create_dsmr_reader, create_tcp_dsmr_reader
import serial
import serial.tools.list_ports
import voluptuous as vol
from homeassistant import config_entries, core, exceptions
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TYPE
from homeassistant.core import callback
from .const import (
@ -27,6 +29,8 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
CONF_MANUAL_PATH = "Enter Manually"
class DSMRConnection:
"""Test the connection to DSMR and receive telegram to read serial ids."""
@ -124,6 +128,10 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
def __init__(self):
"""Initialize flow instance."""
self._dsmr_version = None
@staticmethod
@callback
def async_get_options_flow(config_entry):
@ -160,6 +168,132 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return None
async def async_step_user(self, user_input=None):
"""Step when user initializes a integration."""
errors = {}
if user_input is not None:
user_selection = user_input[CONF_TYPE]
if user_selection == "Serial":
return await self.async_step_setup_serial()
return await self.async_step_setup_network()
list_of_types = ["Serial", "Network"]
schema = vol.Schema({vol.Required(CONF_TYPE): vol.In(list_of_types)})
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
async def async_step_setup_network(self, user_input=None):
"""Step when setting up network configuration."""
errors = {}
if user_input is not None:
data = await self.async_validate_dsmr(user_input, errors)
if not errors:
return self.async_create_entry(
title=f"{data[CONF_HOST]}:{data[CONF_PORT]}", data=data
)
schema = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_PORT): int,
vol.Required(CONF_DSMR_VERSION): vol.In(["2.2", "4", "5", "5B", "5L"]),
}
)
return self.async_show_form(
step_id="setup_network",
data_schema=schema,
errors=errors,
)
async def async_step_setup_serial(self, user_input=None):
"""Step when setting up serial configuration."""
errors = {}
if user_input is not None:
user_selection = user_input[CONF_PORT]
if user_selection == CONF_MANUAL_PATH:
self._dsmr_version = user_input[CONF_DSMR_VERSION]
return await self.async_step_setup_serial_manual_path()
dev_path = await self.hass.async_add_executor_job(
get_serial_by_id, user_selection
)
validate_data = {
CONF_PORT: dev_path,
CONF_DSMR_VERSION: user_input[CONF_DSMR_VERSION],
}
data = await self.async_validate_dsmr(validate_data, errors)
if not errors:
return self.async_create_entry(title=data[CONF_PORT], data=data)
ports = await self.hass.async_add_executor_job(serial.tools.list_ports.comports)
list_of_ports = {}
for port in ports:
list_of_ports[
port.device
] = f"{port}, s/n: {port.serial_number or 'n/a'}" + (
f" - {port.manufacturer}" if port.manufacturer else ""
)
list_of_ports[CONF_MANUAL_PATH] = CONF_MANUAL_PATH
schema = vol.Schema(
{
vol.Required(CONF_PORT): vol.In(list_of_ports),
vol.Required(CONF_DSMR_VERSION): vol.In(["2.2", "4", "5", "5B", "5L"]),
}
)
return self.async_show_form(
step_id="setup_serial",
data_schema=schema,
errors=errors,
)
async def async_step_setup_serial_manual_path(self, user_input=None):
"""Select path manually."""
errors = {}
if user_input is not None:
validate_data = {
CONF_PORT: user_input[CONF_PORT],
CONF_DSMR_VERSION: self._dsmr_version,
}
data = await self.async_validate_dsmr(validate_data, errors)
if not errors:
return self.async_create_entry(title=data[CONF_PORT], data=data)
schema = vol.Schema({vol.Required(CONF_PORT): str})
return self.async_show_form(
step_id="setup_serial_manual_path",
data_schema=schema,
errors=errors,
)
async def async_validate_dsmr(self, input_data, errors):
"""Validate dsmr connection and create data."""
data = input_data
try:
info = await _validate_dsmr_connection(self.hass, data)
data = {**data, **info}
await self.async_set_unique_id(info[CONF_SERIAL_ID])
self._abort_if_unique_id_configured()
except CannotConnect:
errors["base"] = "cannot_connect"
except CannotCommunicate:
errors["base"] = "cannot_communicate"
return data
async def async_step_import(self, import_config=None):
"""Handle the initial step."""
host = import_config.get(CONF_HOST)
@ -216,6 +350,18 @@ class DSMROptionFlowHandler(config_entries.OptionsFlow):
)
def get_serial_by_id(dev_path: str) -> str:
"""Return a /dev/serial/by-id match for given device if available."""
by_id = "/dev/serial/by-id"
if not os.path.isdir(by_id):
return dev_path
for path in (entry.path for entry in os.scandir(by_id) if entry.is_symlink()):
if os.path.realpath(path) == dev_path:
return path
return dev_path
class CannotConnect(exceptions.HomeAssistantError):
"""Error to indicate we cannot connect."""

2
homeassistant/components/dsmr/manifest.json
View File

@ -4,6 +4,6 @@
"documentation": "https://www.home-assistant.io/integrations/dsmr",
"requirements": ["dsmr_parser==0.29"],
"codeowners": ["@Robbie1221"],
"config_flow": false,
"config_flow": true,
"iot_class": "local_push"
}

4
homeassistant/components/dsmr/sensor.py
View File

@ -344,7 +344,9 @@ class DSMREntity(SensorEntity):
return self.translate_tariff(value, self._config[CONF_DSMR_VERSION])
with suppress(TypeError):
value = round(float(value), self._config[CONF_PRECISION])
value = round(
float(value), self._config.get(CONF_PRECISION, DEFAULT_PRECISION)
)
if value is not None:
return value

40
homeassistant/components/dsmr/strings.json
View File

@ -1,9 +1,43 @@
{
"config": {
"step": {},
"error": {},
"step": {
"user": {
"data": {
"type": "Connection type"
},
"title": "Select connection type"
},
"setup_network": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]",
"dsmr_version": "Select DSMR version"
},
"title": "Select connection address"
},
"setup_serial": {
"data": {
"port": "Select device",
"dsmr_version": "Select DSMR version"
},
"title": "Device"
},
"setup_serial_manual_path": {
"data": {
"port": "[%key:common::config_flow::data::usb_path%]"
},
"title": "Path"
}
},
"error": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"cannot_communicate": "Failed to communicate"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"cannot_communicate": "Failed to communicate"
}
},
"options": {

38
homeassistant/components/dsmr/translations/en.json
View File

@ -1,7 +1,43 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
"already_configured": "Device is already configured",
"cannot_communicate": "Failed to communicate",
"cannot_connect": "Failed to connect"
},
"error": {
"already_configured": "Device is already configured",
"cannot_communicate": "Failed to communicate",
"cannot_connect": "Failed to connect"
},
"step": {
"setup_network": {
"data": {
"dsmr_version": "Select DSMR version",
"host": "Host",
"port": "Port"
},
"title": "Select connection address"
},
"setup_serial": {
"data": {
"dsmr_version": "Select DSMR version",
"port": "Select device"
},
"title": "Device"
},
"setup_serial_manual_path": {
"data": {
"port": "USB Device Path"
},
"title": "Path"
},
"user": {
"data": {
"type": "Connection type"
},
"title": "Select connection type"
}
}
},
"options": {

1
homeassistant/generated/config_flows.py
View File

@ -56,6 +56,7 @@ FLOWS = [
"dialogflow",
"directv",
"doorbird",
"dsmr",
"dunehd",
"dynalite",
"eafm",

266
tests/components/dsmr/test_config_flow.py
View File

@ -1,18 +1,233 @@
"""Test the DSMR config flow."""
import asyncio
from itertools import chain, repeat
from unittest.mock import DEFAULT, AsyncMock, patch
import os
from unittest.mock import DEFAULT, AsyncMock, MagicMock, patch, sentinel
import serial
import serial.tools.list_ports
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.dsmr import DOMAIN
from homeassistant.components.dsmr import DOMAIN, config_flow
from tests.common import MockConfigEntry
SERIAL_DATA = {"serial_id": "12345678", "serial_id_gas": "123456789"}
def com_port():
"""Mock of a serial port."""
port = serial.tools.list_ports_common.ListPortInfo("/dev/ttyUSB1234")
port.serial_number = "1234"
port.manufacturer = "Virtual serial port"
port.device = "/dev/ttyUSB1234"
port.description = "Some serial port"
return port
async def test_setup_network(hass, dsmr_connection_send_validate_fixture):
"""Test we can setup network."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"type": "Network"},
)
assert result["type"] == "form"
assert result["step_id"] == "setup_network"
assert result["errors"] == {}
with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"host": "10.10.0.1", "port": 1234, "dsmr_version": "2.2"},
)
entry_data = {
"host": "10.10.0.1",
"port": 1234,
"dsmr_version": "2.2",
}
assert result["type"] == "create_entry"
assert result["title"] == "10.10.0.1:1234"
assert result["data"] == {**entry_data, **SERIAL_DATA}
@patch("serial.tools.list_ports.comports", return_value=[com_port()])
async def test_setup_serial(com_mock, hass, dsmr_connection_send_validate_fixture):
"""Test we can setup serial."""
port = com_port()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"type": "Serial"},
)
assert result["type"] == "form"
assert result["step_id"] == "setup_serial"
assert result["errors"] == {}
with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"port": port.device, "dsmr_version": "2.2"}
)
entry_data = {
"port": port.device,
"dsmr_version": "2.2",
}
assert result["type"] == "create_entry"
assert result["title"] == port.device
assert result["data"] == {**entry_data, **SERIAL_DATA}
@patch("serial.tools.list_ports.comports", return_value=[com_port()])
async def test_setup_serial_manual(
com_mock, hass, dsmr_connection_send_validate_fixture
):
"""Test we can setup serial with manual entry."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"type": "Serial"},
)
assert result["type"] == "form"
assert result["step_id"] == "setup_serial"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"port": "Enter Manually", "dsmr_version": "2.2"}
)
assert result["type"] == "form"
assert result["step_id"] == "setup_serial_manual_path"
assert result["errors"] == {}
with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"port": "/dev/ttyUSB0"}
)
entry_data = {
"port": "/dev/ttyUSB0",
"dsmr_version": "2.2",
}
assert result["type"] == "create_entry"
assert result["title"] == "/dev/ttyUSB0"
assert result["data"] == {**entry_data, **SERIAL_DATA}
@patch("serial.tools.list_ports.comports", return_value=[com_port()])
async def test_setup_serial_fail(com_mock, hass, dsmr_connection_send_validate_fixture):
"""Test failed serial connection."""
(connection_factory, transport, protocol) = dsmr_connection_send_validate_fixture
await setup.async_setup_component(hass, "persistent_notification", {})
port = com_port()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
# override the mock to have it fail the first time and succeed after
first_fail_connection_factory = AsyncMock(
return_value=(transport, protocol),
side_effect=chain([serial.serialutil.SerialException], repeat(DEFAULT)),
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"type": "Serial"},
)
assert result["type"] == "form"
assert result["step_id"] == "setup_serial"
assert result["errors"] == {}
with patch(
"homeassistant.components.dsmr.config_flow.create_dsmr_reader",
first_fail_connection_factory,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"port": port.device, "dsmr_version": "2.2"}
)
assert result["type"] == "form"
assert result["step_id"] == "setup_serial"
assert result["errors"] == {"base": "cannot_connect"}
@patch("serial.tools.list_ports.comports", return_value=[com_port()])
async def test_setup_serial_wrong_telegram(
com_mock, hass, dsmr_connection_send_validate_fixture
):
"""Test failed telegram data."""
(connection_factory, transport, protocol) = dsmr_connection_send_validate_fixture
await setup.async_setup_component(hass, "persistent_notification", {})
port = com_port()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
protocol.telegram = {}
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"type": "Serial"},
)
assert result["type"] == "form"
assert result["step_id"] == "setup_serial"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"port": port.device, "dsmr_version": "2.2"}
)
assert result["type"] == "form"
assert result["step_id"] == "setup_serial"
assert result["errors"] == {"base": "cannot_communicate"}
async def test_import_usb(hass, dsmr_connection_send_validate_fixture):
"""Test we can import."""
await setup.async_setup_component(hass, "persistent_notification", {})
@ -265,3 +480,50 @@ async def test_import_luxembourg(hass, dsmr_connection_send_validate_fixture):
assert result["type"] == "create_entry"
assert result["title"] == "/dev/ttyUSB0"
assert result["data"] == {**entry_data, **SERIAL_DATA}
def test_get_serial_by_id_no_dir():
"""Test serial by id conversion if there's no /dev/serial/by-id."""
p1 = patch("os.path.isdir", MagicMock(return_value=False))
p2 = patch("os.scandir")
with p1 as is_dir_mock, p2 as scan_mock:
res = config_flow.get_serial_by_id(sentinel.path)
assert res is sentinel.path
assert is_dir_mock.call_count == 1
assert scan_mock.call_count == 0
def test_get_serial_by_id():
"""Test serial by id conversion."""
p1 = patch("os.path.isdir", MagicMock(return_value=True))
p2 = patch("os.scandir")
def _realpath(path):
if path is sentinel.matched_link:
return sentinel.path
return sentinel.serial_link_path
p3 = patch("os.path.realpath", side_effect=_realpath)
with p1 as is_dir_mock, p2 as scan_mock, p3:
res = config_flow.get_serial_by_id(sentinel.path)
assert res is sentinel.path
assert is_dir_mock.call_count == 1
assert scan_mock.call_count == 1
entry1 = MagicMock(spec_set=os.DirEntry)
entry1.is_symlink.return_value = True
entry1.path = sentinel.some_path
entry2 = MagicMock(spec_set=os.DirEntry)
entry2.is_symlink.return_value = False
entry2.path = sentinel.other_path
entry3 = MagicMock(spec_set=os.DirEntry)
entry3.is_symlink.return_value = True
entry3.path = sentinel.matched_link
scan_mock.return_value = [entry1, entry2, entry3]
res = config_flow.get_serial_by_id(sentinel.path)
assert res is sentinel.matched_link
assert is_dir_mock.call_count == 2
assert scan_mock.call_count == 2