diff --git a/custom_components/midea_ac_lan/config_flow.py b/custom_components/midea_ac_lan/config_flow.py
index 675a441c..f2a416ec 100644
--- a/custom_components/midea_ac_lan/config_flow.py
+++ b/custom_components/midea_ac_lan/config_flow.py
@@ -1,17 +1,20 @@
+import logging
import os
-from typing import Any
-
-import voluptuous as vol
+from typing import Any, cast
try:
from homeassistant.helpers.json import save_json
except ImportError:
from homeassistant.util.json import save_json
-import logging
-
import homeassistant.helpers.config_validation as cv
-from homeassistant import config_entries
+import voluptuous as vol
+from homeassistant.config_entries import (
+ ConfigEntry,
+ ConfigFlow,
+ ConfigFlowResult,
+ OptionsFlow,
+)
from homeassistant.const import (
CONF_CUSTOMIZE,
CONF_DEVICE,
@@ -71,8 +74,8 @@
]
-class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
- available_device: list = []
+class MideaLanConfigFlow(ConfigFlow, domain=DOMAIN):
+ available_device: dict = {}
devices: dict = {}
found_device: dict = {}
supports: dict = {}
@@ -87,47 +90,20 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
for item in sorted_device_names:
supports[item[0]] = item[1]
- def _save_device_config(self, data: dict):
+ def _save_device_config(self, data: dict[str, Any]) -> None:
os.makedirs(self.hass.config.path(STORAGE_PATH), exist_ok=True)
record_file = self.hass.config.path(
f"{STORAGE_PATH}/{data[CONF_DEVICE_ID]}.json",
)
save_json(record_file, data)
- def _load_device_config(self, device_id):
+ def _load_device_config(self, device_id: str) -> Any:
record_file = self.hass.config.path(f"{STORAGE_PATH}/{device_id}.json")
json_data = load_json(record_file, default={})
return json_data
- def _save_account(self, account: dict):
- os.makedirs(self.hass.config.path(STORAGE_PATH), exist_ok=True)
- record_file = self.hass.config.path(f"{STORAGE_PATH}/account.json")
- account[CONF_PASSWORD] = format(
- (
- int(account[CONF_ACCOUNT].encode("utf-8").hex(), 16)
- ^ int(account[CONF_PASSWORD].encode("utf-8").hex(), 16)
- ),
- "x",
- )
- save_json(record_file, account)
-
- def _load_account(self):
- record_file = self.hass.config.path(f"{STORAGE_PATH}/account.json")
- json_data = load_json(record_file, default={})
- if CONF_ACCOUNT in json_data.keys():
- json_data[CONF_PASSWORD] = bytes.fromhex(
- format(
- (
- int(json_data[CONF_PASSWORD], 16)
- ^ int(json_data[CONF_ACCOUNT].encode("utf-8").hex(), 16)
- ),
- "X",
- ),
- ).decode("UTF-8")
- return json_data
-
@staticmethod
- def _check_storage_device(device: dict, storage_device: dict):
+ def _check_storage_device(device: dict, storage_device: dict) -> bool:
if storage_device.get(CONF_SUBTYPE) is None:
return False
if device.get(CONF_PROTOCOL) == 3 and (
@@ -137,7 +113,7 @@ def _check_storage_device(device: dict, storage_device: dict):
return False
return True
- def _already_configured(self, device_id, ip_address):
+ def _already_configured(self, device_id: str, ip_address: str) -> bool:
for entry in self._async_current_entries():
if device_id == entry.data.get(
CONF_DEVICE_ID,
@@ -145,7 +121,9 @@ def _already_configured(self, device_id, ip_address):
return True
return False
- async def async_step_user(self, user_input=None, error=None):
+ async def async_step_user(
+ self, user_input: dict[str, Any] | None = None, error: str | None = None
+ ) -> ConfigFlowResult:
if user_input is not None:
if user_input["action"] == "discovery":
return await self.async_step_discovery()
@@ -162,7 +140,9 @@ async def async_step_user(self, user_input=None, error=None):
errors={"base": error} if error else None,
)
- async def async_step_login(self, user_input=None, error=None):
+ async def async_step_login(
+ self, user_input: dict[str, Any] | None = None, error: str | None = None
+ ) -> ConfigFlowResult:
if user_input is not None:
if self.session is None:
self.session = async_create_clientsession(self.hass)
@@ -179,7 +159,6 @@ async def async_step_login(self, user_input=None, error=None):
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_SERVER: SERVERS[user_input[CONF_SERVER]],
}
- self._save_account(self.account)
return await self.async_step_auto()
else:
return await self.async_step_login(error="login_failed")
@@ -195,7 +174,9 @@ async def async_step_login(self, user_input=None, error=None):
errors={"base": error} if error else None,
)
- async def async_step_list(self, user_input=None, error=None):
+ async def async_step_list(
+ self, user_input: dict[str, Any] | None = None, error: str | None = None
+ ) -> ConfigFlowResult:
all_devices = discover()
if len(all_devices) > 0:
table = (
@@ -204,7 +185,8 @@ async def async_step_list(self, user_input=None, error=None):
for device_id, device in all_devices.items():
supported = device.get(CONF_TYPE) in self.supports.keys()
table += (
- f"\n{device_id}|{'%02X' % device.get(CONF_TYPE)}|{device.get(CONF_IP_ADDRESS)}|"
+ f"\n{device_id}|{'%02X' % device.get(CONF_TYPE)}|{
+ device.get(CONF_IP_ADDRESS)}|"
f"{device.get('sn')}|"
f"{'YES' if supported else 'NO'}"
)
@@ -216,7 +198,9 @@ async def async_step_list(self, user_input=None, error=None):
errors={"base": error} if error else None,
)
- async def async_step_discovery(self, user_input=None, error=None):
+ async def async_step_discovery(
+ self, user_input: dict[str, Any] | None = None, error: str | None = None
+ ) -> ConfigFlowResult:
if user_input is not None:
if user_input[CONF_IP_ADDRESS].lower() == "auto":
ip_address = None
@@ -226,9 +210,10 @@ async def async_step_discovery(self, user_input=None, error=None):
self.available_device = {}
for device_id, device in self.devices.items():
if not self._already_configured(device_id, device.get(CONF_IP_ADDRESS)):
- self.available_device[device_id] = (
- f"{device_id} ({self.supports.get(device.get(CONF_TYPE))})"
- )
+ self.available_device[
+ device_id
+ ] = f"{device_id} ({self.supports.get(
+ device.get(CONF_TYPE))})"
if len(self.available_device) > 0:
return await self.async_step_auto()
else:
@@ -241,10 +226,12 @@ async def async_step_discovery(self, user_input=None, error=None):
errors={"base": error} if error else None,
)
- async def async_step_auto(self, user_input=None, error=None):
+ async def async_step_auto(
+ self, user_input: dict[str, Any] | None = None, error: str | None = None
+ ) -> ConfigFlowResult:
if user_input is not None:
device_id = user_input[CONF_DEVICE]
- device = self.devices.get(device_id)
+ device = self.devices[device_id]
storage_device = self._load_device_config(device_id)
if self._check_storage_device(device, storage_device):
self.found_device = {
@@ -266,9 +253,7 @@ async def async_step_auto(self, user_input=None, error=None):
return await self.async_step_manually()
else:
if CONF_ACCOUNT not in self.account.keys():
- self.account = self._load_account()
- if CONF_ACCOUNT not in self.account.keys():
- return await self.async_step_login()
+ return await self.async_step_login()
if self.session is None:
self.session = async_create_clientsession(self.hass)
if self.cloud is None:
@@ -345,7 +330,9 @@ async def async_step_auto(self, user_input=None, error=None):
errors={"base": error} if error else None,
)
- async def async_step_manually(self, user_input=None, error=None):
+ async def async_step_manually(
+ self, user_input: dict[str, Any] | None = None, error: str | None = None
+ ) -> ConfigFlowResult:
if user_input is not None:
self.found_device = {
CONF_DEVICE_ID: user_input[CONF_DEVICE_ID],
@@ -483,12 +470,12 @@ async def async_step_manually(self, user_input=None, error=None):
@staticmethod
@callback
- def async_get_options_flow(config_entry):
- return OptionsFlowHandler(config_entry)
+ def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
+ return MideaLanOptionsFlowHandler(config_entry)
-class OptionsFlowHandler(config_entries.OptionsFlow):
- def __init__(self, config_entry: config_entries.ConfigEntry):
+class MideaLanOptionsFlowHandler(OptionsFlow):
+ def __init__(self, config_entry: ConfigEntry) -> None:
self._config_entry = config_entry
self._device_type = config_entry.data.get(CONF_TYPE)
if self._device_type is None:
@@ -502,16 +489,18 @@ def __init__(self, config_entry: config_entries.ConfigEntry):
if key not in MIDEA_DEVICES[self._device_type]["entities"]:
self._config_entry.options[CONF_SWITCHES].remove(key)
- async def async_step_init(self, user_input=None):
+ async def async_step_init(
+ self, user_input: dict[str, Any] | None = None
+ ) -> ConfigFlowResult:
if self._device_type == CONF_ACCOUNT:
return self.async_abort(reason="account_option")
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
sensors = {}
switches = {}
- for attribute, attribute_config in (
- MIDEA_DEVICES.get(self._device_type).get("entities").items()
- ):
+ for attribute, attribute_config in cast(
+ dict, MIDEA_DEVICES[cast(int, self._device_type)]["entities"]
+ ).items():
attribute_name = (
attribute if isinstance(attribute, str) else attribute.value
)