Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

chore: config_flow typing #52

Merged
merged 2 commits into from
Jun 1, 2024
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 51 additions & 62 deletions custom_components/midea_ac_lan/config_flow.py
Original file line number Diff line number Diff line change
@@ -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,
Expand Down Expand Up @@ -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 = {}
Expand All @@ -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 (
Expand All @@ -137,15 +113,17 @@ 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,
) or ip_address == entry.data.get(CONF_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()
Expand All @@ -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)
Expand All @@ -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")
Expand All @@ -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 = (
Expand All @@ -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"{'<font color=gree>YES</font>' if supported else '<font color=red>NO</font>'}"
)
Expand All @@ -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
Expand All @@ -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:
Expand All @@ -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 = {
Expand All @@ -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:
Expand Down Expand Up @@ -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],
Expand Down Expand Up @@ -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:
Expand All @@ -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
)
Expand Down
Loading