diff --git a/homeassistant/components/dsmr/__init__.py b/homeassistant/components/dsmr/__init__.py index 61c7660fc69..3ebac7b42fc 100644 --- a/homeassistant/components/dsmr/__init__.py +++ b/homeassistant/components/dsmr/__init__.py @@ -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) diff --git a/homeassistant/components/dsmr/config_flow.py b/homeassistant/components/dsmr/config_flow.py index 2312cad215b..1ddbdb480c1 100644 --- a/homeassistant/components/dsmr/config_flow.py +++ b/homeassistant/components/dsmr/config_flow.py @@ -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)), diff --git a/mypy.ini b/mypy.ini index 50257141179..505d02b1111 100644 --- a/mypy.ini +++ b/mypy.ini @@ -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 diff --git a/script/hassfest/mypy_config.py b/script/hassfest/mypy_config.py index b934bed3bc1..21600257ffb 100644 --- a/script/hassfest/mypy_config.py +++ b/script/hassfest/mypy_config.py @@ -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.*", diff --git a/tests/components/dsmr/test_config_flow.py b/tests/components/dsmr/test_config_flow.py index 70f2b16e0b0..006893a81e8 100644 --- a/tests/components/dsmr/test_config_flow.py +++ b/tests/components/dsmr/test_config_flow.py @@ -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"],