jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

DSMR: Typing cleanup in init & config flow (#52145)

Browse Source
This commit is contained in:
Franck Nijhof 2021-06-24 12:53:16 +02:00 committed by GitHub
parent 17357bf575
commit fbdd6a9d95
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 59 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -21,7 +21,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
task = hass.data[DOMAIN][entry.entry_id][DATA_TASK]
listener = hass.data[DOMAIN][entry.entry_id][DATA_LISTENER]
@ -40,6 +40,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
return unload_ok
async def async_update_options(hass: HomeAssistant, entry: ConfigEntry):
async def async_update_options(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Update options."""
await hass.config_entries.async_reload(entry.entry_id)

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

@ -10,13 +10,17 @@ 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
from dsmr_parser.objects import DSMRObject
import serial
import serial.tools.list_ports
import voluptuous as vol
from homeassistant import config_entries, core, exceptions
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TYPE
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.typing import ConfigType
from .const import (
CONF_DSMR_VERSION,
@ -35,33 +39,34 @@ CONF_MANUAL_PATH = "Enter Manually"
class DSMRConnection:
"""Test the connection to DSMR and receive telegram to read serial ids."""
def __init__(self, host, port, dsmr_version):
def __init__(self, host: str | None, port: int, dsmr_version: str) -> None:
"""Initialize."""
self._host = host
self._port = port
self._dsmr_version = dsmr_version
self._telegram = {}
self._telegram: dict[str, DSMRObject] = {}
self._equipment_identifier = obis_ref.EQUIPMENT_IDENTIFIER
if dsmr_version == "5L":
self._equipment_identifier = obis_ref.LUXEMBOURG_EQUIPMENT_IDENTIFIER
else:
self._equipment_identifier = obis_ref.EQUIPMENT_IDENTIFIER
def equipment_identifier(self):
def equipment_identifier(self) -> str | None:
"""Equipment identifier."""
if self._equipment_identifier in self._telegram:
dsmr_object = self._telegram[self._equipment_identifier]
return getattr(dsmr_object, "value", None)
return None
def equipment_identifier_gas(self):
def equipment_identifier_gas(self) -> str | None:
"""Equipment identifier gas."""
if obis_ref.EQUIPMENT_IDENTIFIER_GAS in self._telegram:
dsmr_object = self._telegram[obis_ref.EQUIPMENT_IDENTIFIER_GAS]
return getattr(dsmr_object, "value", None)
return None
async def validate_connect(self, hass: core.HomeAssistant) -> bool:
"""Test if we can validate connection with the device."""
def update_telegram(telegram):
def update_telegram(telegram: dict[str, DSMRObject]) -> None:
if self._equipment_identifier in telegram:
self._telegram = telegram
transport.close()
@ -101,7 +106,9 @@ class DSMRConnection:
return True
async def _validate_dsmr_connection(hass: core.HomeAssistant, data):
async def _validate_dsmr_connection(
hass: core.HomeAssistant, data: dict[str, Any]
) -> dict[str, str | None]:
"""Validate the user input allows us to connect."""
conn = DSMRConnection(data.get(CONF_HOST), data[CONF_PORT], data[CONF_DSMR_VERSION])
@ -115,26 +122,22 @@ async def _validate_dsmr_connection(hass: core.HomeAssistant, data):
if equipment_identifier is None:
raise CannotCommunicate
info = {
return {
CONF_SERIAL_ID: equipment_identifier,
CONF_SERIAL_ID_GAS: equipment_identifier_gas,
}
return info
class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for DSMR."""
VERSION = 1
def __init__(self):
"""Initialize flow instance."""
self._dsmr_version = None
_dsmr_version: str | None = None
@staticmethod
@callback
def async_get_options_flow(config_entry):
def async_get_options_flow(config_entry: ConfigEntry) -> DSMROptionFlowHandler:
"""Get the options flow for this handler."""
return DSMROptionFlowHandler(config_entry)
@ -144,7 +147,7 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
host: str = None,
updates: dict[Any, Any] | None = None,
reload_on_update: bool = True,
):
) -> FlowResult | None:
"""Test if host and port are already configured."""
for entry in self._async_current_entries():
if entry.data.get(CONF_HOST) == host and entry.data[CONF_PORT] == port:
@ -168,9 +171,10 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return None
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Step when user initializes a integration."""
errors = {}
if user_input is not None:
user_selection = user_input[CONF_TYPE]
if user_selection == "Serial":
@ -181,15 +185,15 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
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)
return self.async_show_form(step_id="user", data_schema=schema)
async def async_step_setup_network(self, user_input=None):
async def async_step_setup_network(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Step when setting up network configuration."""
errors = {}
errors: dict[str, str] = {}
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
@ -208,10 +212,11 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_setup_serial(self, user_input=None):
async def async_step_setup_serial(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Step when setting up serial configuration."""
errors = {}
errors: dict[str, str] = {}
if user_input is not None:
user_selection = user_input[CONF_PORT]
if user_selection == CONF_MANUAL_PATH:
@ -228,18 +233,15 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
}
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 = {
port.device: f"{port}, s/n: {port.serial_number or 'n/a'}"
+ (f" - {port.manufacturer}" if port.manufacturer else "")
for port in ports
}
list_of_ports[CONF_MANUAL_PATH] = CONF_MANUAL_PATH
schema = vol.Schema(
@ -254,18 +256,18 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_setup_serial_manual_path(self, user_input=None):
async def async_step_setup_serial_manual_path(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Select path manually."""
errors = {}
if user_input is not None:
validate_data = {
CONF_PORT: user_input[CONF_PORT],
CONF_DSMR_VERSION: self._dsmr_version,
}
errors: dict[str, str] = {}
data = await self.async_validate_dsmr(validate_data, errors)
if not errors:
return self.async_create_entry(title=data[CONF_PORT], data=data)
@ -273,10 +275,11 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
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):
async def async_validate_dsmr(
self, input_data: dict[str, Any], errors: dict[str, str]
) -> dict[str, Any]:
"""Validate dsmr connection and create data."""
data = input_data
@ -294,7 +297,7 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return data
async def async_step_import(self, import_config=None):
async def async_step_import(self, import_config: ConfigType) -> FlowResult:
"""Handle the initial step."""
host = import_config.get(CONF_HOST)
port = import_config[CONF_PORT]
@ -310,11 +313,7 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
except CannotCommunicate:
return self.async_abort(reason="cannot_communicate")
if host is not None:
name = f"{host}:{port}"
else:
name = port
name = f"{host}:{port}" if host is not None else port
data = {**import_config, **info}
await self.async_set_unique_id(info[CONF_SERIAL_ID])
@ -326,11 +325,13 @@ class DSMRFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
class DSMROptionFlowHandler(config_entries.OptionsFlow):
"""Handle options."""
def __init__(self, config_entry):
def __init__(self, entry: ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
self.entry = entry
async def async_step_init(self, user_input=None):
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
@ -341,7 +342,7 @@ class DSMROptionFlowHandler(config_entries.OptionsFlow):
{
vol.Optional(
CONF_TIME_BETWEEN_UPDATE,
default=self.config_entry.options.get(
default=self.entry.options.get(
CONF_TIME_BETWEEN_UPDATE, DEFAULT_TIME_BETWEEN_UPDATE
),
): vol.All(vol.Coerce(int), vol.Range(min=0)),

3
mypy.ini
View File

@ -1053,9 +1053,6 @@ ignore_errors = true
[mypy-homeassistant.components.doorbird.*]
ignore_errors = true
[mypy-homeassistant.components.dsmr.*]
ignore_errors = true
[mypy-homeassistant.components.dynalite.*]
ignore_errors = true

1
script/hassfest/mypy_config.py
View File

@ -46,7 +46,6 @@ IGNORED_MODULES: Final[list[str]] = [
"homeassistant.components.dhcp.*",
"homeassistant.components.directv.*",
"homeassistant.components.doorbird.*",
"homeassistant.components.dsmr.*",
"homeassistant.components.dynalite.*",
"homeassistant.components.eafm.*",
"homeassistant.components.edl21.*",

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

@ -34,7 +34,7 @@ async def test_setup_network(hass, dsmr_connection_send_validate_fixture):
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@ -73,7 +73,7 @@ async def test_setup_serial(com_mock, hass, dsmr_connection_send_validate_fixtur
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@ -110,7 +110,7 @@ async def test_setup_serial_manual(
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@ -127,7 +127,7 @@ async def test_setup_serial_manual(
assert result["type"] == "form"
assert result["step_id"] == "setup_serial_manual_path"
assert result["errors"] == {}
assert result["errors"] is None
with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_configure(
@ -165,7 +165,7 @@ async def test_setup_serial_fail(com_mock, hass, dsmr_connection_send_validate_f
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@ -208,7 +208,7 @@ async def test_setup_serial_wrong_telegram(
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],