chickliu/ha_xiaomi_home
1
0
Fork 0
mirror of https://github.com/XiaoMi/ha_xiaomi_home.git synced 2026-01-17 07:10:44 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix: fix some type error

Browse Source
This commit is contained in:
topsworld 2024-12-23 22:01:06 +08:00
parent 13d76b66de
commit ab5fac9f45
1 changed files with 246 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

367
custom_components/xiaomi_home/config_flow.py
View File

@ -101,37 +101,39 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
# pylint: disable=unused-argument, inconsistent-quotes # pylint: disable=unused-argument, inconsistent-quotes
VERSION = 1 VERSION = 1
MINOR_VERSION = 1 MINOR_VERSION = 1
DEFAULT_AREA_NAME_RULE = 'room'
_main_loop: asyncio.AbstractEventLoop _main_loop: asyncio.AbstractEventLoop
_mips_service: Optional[MipsService] _miot_network: MIoTNetwork
_miot_storage: Optional[MIoTStorage] _mips_service: MipsService
_miot_network: Optional[MIoTNetwork] _miot_storage: MIoTStorage
_miot_i18n: Optional[MIoTI18n] _miot_i18n: MIoTI18n
_integration_language: Optional[str] _integration_language: str
_storage_path: Optional[str] _storage_path: str
_virtual_did: Optional[str] _virtual_did: str
_uid: Optional[str] _uid: str
_uuid: Optional[str] _uuid: str
_ctrl_mode: Optional[str] _ctrl_mode: str
_area_name_rule: Optional[str] _area_name_rule: str
_action_debug: bool _action_debug: bool
_hide_non_standard_entities: bool _hide_non_standard_entities: bool
_display_devices_changed_notify: bool _display_devices_changed_notify: bool
_auth_info: Optional[dict]
_nick_name: Optional[str] _auth_info: dict
_nick_name: str
_home_selected: dict _home_selected: dict
_home_info_buffer: Optional[dict[str, str | dict[str, dict]]] _home_info_buffer: dict
_home_list: Optional[dict] _home_list_show: dict
_device_list_sorted: dict _device_list_sorted: dict
_device_list_filter: dict _device_list_filter: dict
_cloud_server: Optional[str] _cloud_server: str
_oauth_redirect_url: Optional[str] _oauth_redirect_url_full: str
_miot_oauth: Optional[MIoTOauthClient] _miot_oauth: Optional[MIoTOauthClient]
_miot_http: Optional[MIoTHttpClient] _miot_http: Optional[MIoTHttpClient]
_user_cert_state: bool _user_cert_state: bool
_oauth_auth_url: Optional[str] _oauth_auth_url: str
_task_oauth: Optional[asyncio.Task[None]] _task_oauth: Optional[asyncio.Task[None]]
_config_error_reason: Optional[str] _config_error_reason: Optional[str]
@ -139,70 +141,65 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self) -> None: def __init__(self) -> None:
self._main_loop = asyncio.get_running_loop() self._main_loop = asyncio.get_running_loop()
self._mips_service = None self._integration_language = DEFAULT_INTEGRATION_LANGUAGE
self._miot_storage = None self._storage_path = ''
self._miot_network = None self._virtual_did = ''
self._miot_i18n = None self._uid = ''
self._uuid = '' # MQTT client id
self._integration_language = None self._ctrl_mode = DEFAULT_CTRL_MODE
self._storage_path = None self._area_name_rule = self.DEFAULT_AREA_NAME_RULE
self._virtual_did = None
self._uid = None
self._uuid = None # MQTT client id
self._ctrl_mode = None
self._area_name_rule = None
self._action_debug = False self._action_debug = False
self._hide_non_standard_entities = False self._hide_non_standard_entities = False
self._display_devices_changed_notify = True self._display_devices_changed_notify = True
self._auth_info = None self._auth_info = {}
self._nick_name = None self._nick_name = DEFAULT_NICK_NAME
self._home_selected = {} self._home_selected = {}
self._home_info_buffer = None self._home_info_buffer = {}
self._home_list = None self._home_list_show = {}
self._device_list_sorted = None self._device_list_sorted = {}
self._cloud_server = None self._cloud_server = DEFAULT_CLOUD_SERVER
self._oauth_redirect_url = None self._oauth_redirect_url_full = ''
self._miot_oauth = None self._miot_oauth = None
self._miot_http = None self._miot_http = None
self._user_cert_state = False
self._oauth_auth_url = None self._user_cert_state = False
self._oauth_auth_url = ''
self._task_oauth = None self._task_oauth = None
self._config_error_reason = None self._config_error_reason = None
self._fut_oauth_code = None self._fut_oauth_code = None
async def async_step_user(self, user_input=None): async def async_step_user(
self, user_input: Optional[dict] = None
):
self.hass.data.setdefault(DOMAIN, {}) self.hass.data.setdefault(DOMAIN, {})
loop: asyncio.AbstractEventLoop = asyncio.get_running_loop() if not self._virtual_did:
if self._virtual_did is None:
self._virtual_did = str(secrets.randbits(64)) self._virtual_did = str(secrets.randbits(64))
self.hass.data[DOMAIN].setdefault(self._virtual_did, {}) self.hass.data[DOMAIN].setdefault(self._virtual_did, {})
if self._storage_path is None: if not self._storage_path:
self._storage_path = self.hass.config.path('.storage', DOMAIN) self._storage_path = self.hass.config.path('.storage', DOMAIN)
# MIoT network # MIoT network
self._miot_network = self.hass.data[DOMAIN].get('miot_network', None) self._miot_network = self.hass.data[DOMAIN].get('miot_network', None)
if self._miot_network is None: if not self._miot_network:
self._miot_network = MIoTNetwork(loop=loop) self._miot_network = MIoTNetwork(loop=self._main_loop)
self.hass.data[DOMAIN]['miot_network'] = self._miot_network self.hass.data[DOMAIN]['miot_network'] = self._miot_network
await self._miot_network.init_async( await self._miot_network.init_async(
refresh_interval=NETWORK_REFRESH_INTERVAL) refresh_interval=NETWORK_REFRESH_INTERVAL)
_LOGGER.info('async_step_user, create miot network') _LOGGER.info('async_step_user, create miot network')
# Mips server # Mips server
self._mips_service = self.hass.data[DOMAIN].get('mips_service', None) self._mips_service = self.hass.data[DOMAIN].get('mips_service', None)
if self._mips_service is None: if not self._mips_service:
aiozc: HaAsyncZeroconf = await zeroconf.async_get_async_instance( aiozc: HaAsyncZeroconf = await zeroconf.async_get_async_instance(
self.hass) self.hass)
self._mips_service = MipsService(aiozc=aiozc, loop=loop) self._mips_service = MipsService(aiozc=aiozc, loop=self._main_loop)
self.hass.data[DOMAIN]['mips_service'] = self._mips_service self.hass.data[DOMAIN]['mips_service'] = self._mips_service
await self._mips_service.init_async() await self._mips_service.init_async()
_LOGGER.info('async_step_user, create mips service') _LOGGER.info('async_step_user, create mips service')
# MIoT storage # MIoT storage
self._miot_storage = self.hass.data[DOMAIN].get('miot_storage', None) self._miot_storage = self.hass.data[DOMAIN].get('miot_storage', None)
if self._miot_storage is None: if not self._miot_storage:
self._miot_storage = MIoTStorage( self._miot_storage = MIoTStorage(
root_path=self._storage_path, loop=loop) root_path=self._storage_path, loop=self._main_loop)
self.hass.data[DOMAIN]['miot_storage'] = self._miot_storage self.hass.data[DOMAIN]['miot_storage'] = self._miot_storage
_LOGGER.info( _LOGGER.info(
'async_step_user, create miot storage, %s', self._storage_path) 'async_step_user, create miot storage, %s', self._storage_path)
@ -214,7 +211,9 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_eula(user_input) return await self.async_step_eula(user_input)
async def async_step_eula(self, user_input=None): async def async_step_eula(
self, user_input: Optional[dict] = None
):
if user_input: if user_input:
if user_input.get('eula', None) is True: if user_input.get('eula', None) is True:
return await self.async_step_auth_config() return await self.async_step_auth_config()
@ -225,16 +224,18 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form( return self.async_show_form(
step_id='eula', step_id='eula',
data_schema=vol.Schema({ data_schema=vol.Schema({
vol.Required('eula', default=False): bool, vol.Required('eula', default=False): bool, # type: ignore
}), }),
last_step=False, last_step=False,
errors={'base': reason}, errors={'base': reason},
) )
async def async_step_auth_config(self, user_input=None): async def async_step_auth_config(
self, user_input: Optional[dict] = None
):
if user_input: if user_input:
self._cloud_server = user_input.get( self._cloud_server = user_input.get(
'cloud_server', DEFAULT_CLOUD_SERVER) 'cloud_server', self._cloud_server)
self._integration_language = user_input.get( self._integration_language = user_input.get(
'integration_language', DEFAULT_INTEGRATION_LANGUAGE) 'integration_language', DEFAULT_INTEGRATION_LANGUAGE)
self._miot_i18n = MIoTI18n( self._miot_i18n = MIoTI18n(
@ -242,7 +243,7 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
await self._miot_i18n.init_async() await self._miot_i18n.init_async()
webhook_path = webhook_async_generate_path( webhook_path = webhook_async_generate_path(
webhook_id=self._virtual_did) webhook_id=self._virtual_did)
self._oauth_redirect_url = ( self._oauth_redirect_url_full = (
f'{user_input.get("oauth_redirect_url")}{webhook_path}') f'{user_input.get("oauth_redirect_url")}{webhook_path}')
return await self.async_step_oauth(user_input) return await self.async_step_oauth(user_input)
# Generate default language from HomeAssistant config (not user config) # Generate default language from HomeAssistant config (not user config)
@ -257,33 +258,38 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
data_schema=vol.Schema({ data_schema=vol.Schema({
vol.Required( vol.Required(
'cloud_server', 'cloud_server',
default=DEFAULT_CLOUD_SERVER): vol.In(CLOUD_SERVERS), default=self._cloud_server # type: ignore
): vol.In(CLOUD_SERVERS),
vol.Required( vol.Required(
'integration_language', 'integration_language',
default=default_language): vol.In(INTEGRATION_LANGUAGES), default=default_language # type: ignore
): vol.In(INTEGRATION_LANGUAGES),
vol.Required( vol.Required(
'oauth_redirect_url', 'oauth_redirect_url',
default=OAUTH_REDIRECT_URL): vol.In([OAUTH_REDIRECT_URL]), default=OAUTH_REDIRECT_URL # type: ignore
): vol.In([OAUTH_REDIRECT_URL]),
}), }),
last_step=False, last_step=False,
) )
async def async_step_oauth(self, user_input=None): async def async_step_oauth(
self, user_input: Optional[dict] = None
):
# 1: Init miot_oauth, generate auth url # 1: Init miot_oauth, generate auth url
try: try:
if self._miot_oauth is None: if not self._miot_oauth:
_LOGGER.info( _LOGGER.info(
'async_step_oauth, redirect_url: %s', 'async_step_oauth, redirect_url: %s',
self._oauth_redirect_url) self._oauth_redirect_url_full)
miot_oauth = MIoTOauthClient( miot_oauth = MIoTOauthClient(
client_id=OAUTH2_CLIENT_ID, client_id=OAUTH2_CLIENT_ID,
redirect_url=self._oauth_redirect_url, redirect_url=self._oauth_redirect_url_full,
cloud_server=self._cloud_server cloud_server=self._cloud_server
) )
state = str(secrets.randbits(64)) state = str(secrets.randbits(64))
self.hass.data[DOMAIN][self._virtual_did]['oauth_state'] = state self.hass.data[DOMAIN][self._virtual_did]['oauth_state'] = state
self._oauth_auth_url = miot_oauth.gen_auth_url( self._oauth_auth_url = miot_oauth.gen_auth_url(
redirect_url=self._oauth_redirect_url, state=state) redirect_url=self._oauth_redirect_url_full, state=state)
_LOGGER.info( _LOGGER.info(
'async_step_oauth, oauth_url: %s', self._oauth_auth_url) 'async_step_oauth, oauth_url: %s', self._oauth_auth_url)
webhook_async_unregister( webhook_async_unregister(
@ -298,7 +304,7 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
) )
self._fut_oauth_code = self.hass.data[DOMAIN][ self._fut_oauth_code = self.hass.data[DOMAIN][
self._virtual_did].get('fut_oauth_code', None) self._virtual_did].get('fut_oauth_code', None)
if self._fut_oauth_code is None: if not self._fut_oauth_code:
self._fut_oauth_code = self._main_loop.create_future() self._fut_oauth_code = self._main_loop.create_future()
self.hass.data[DOMAIN][self._virtual_did][ self.hass.data[DOMAIN][self._virtual_did][
'fut_oauth_code'] = self._fut_oauth_code 'fut_oauth_code'] = self._fut_oauth_code
@ -337,11 +343,16 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def __check_oauth_async(self) -> None: async def __check_oauth_async(self) -> None:
# TASK 1: Get oauth code # TASK 1: Get oauth code
if not self._fut_oauth_code:
raise MIoTConfigError('oauth_code_fut_error')
oauth_code: Optional[str] = await self._fut_oauth_code oauth_code: Optional[str] = await self._fut_oauth_code
if not oauth_code:
raise MIoTConfigError('oauth_code_error')
# TASK 2: Get access_token and user_info from miot_oauth # TASK 2: Get access_token and user_info from miot_oauth
if not self._auth_info: if not self._auth_info:
try: try:
if not self._miot_oauth:
raise MIoTConfigError('oauth_client_error')
auth_info = await self._miot_oauth.get_access_token_async( auth_info = await self._miot_oauth.get_access_token_async(
code=oauth_code) code=oauth_code)
if not self._miot_http: if not self._miot_http:
@ -363,7 +374,7 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
try: try:
self._nick_name = ( self._nick_name = (
await self._miot_http.get_user_info_async() or {} await self._miot_http.get_user_info_async() or {}
).get('miliaoNick', DEFAULT_NICK_NAME) ).get('miliaoNick', self._nick_name)
except (MIoTOauthError, json.JSONDecodeError): except (MIoTOauthError, json.JSONDecodeError):
self._nick_name = DEFAULT_NICK_NAME self._nick_name = DEFAULT_NICK_NAME
_LOGGER.error('get nick name failed') _LOGGER.error('get nick name failed')
@ -374,6 +385,8 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
# TASK 3: Get home info # TASK 3: Get home info
try: try:
if not self._miot_http:
raise MIoTConfigError('http_client_error')
self._home_info_buffer = ( self._home_info_buffer = (
await self._miot_http.get_devices_async()) await self._miot_http.get_devices_async())
_LOGGER.info('get_homeinfos response: %s', self._home_info_buffer) _LOGGER.info('get_homeinfos response: %s', self._home_info_buffer)
@ -433,7 +446,7 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
f'{home_info["home_name"]} ' f'{home_info["home_name"]} '
f'[ {len(dev_list)} {tip_devices} {tip_central} ]') f'[ {len(dev_list)} {tip_devices} {tip_central} ]')
self._home_list = dict(sorted(home_list.items())) self._home_list_show = dict(sorted(home_list.items()))
# TASK 7: Get user's MiHome certificate # TASK 7: Get user's MiHome certificate
if self._cloud_server in SUPPORT_CENTRAL_GATEWAY_CTRL: if self._cloud_server in SUPPORT_CENTRAL_GATEWAY_CTRL:
@ -454,6 +467,8 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
user_key=user_key, did=self._virtual_did) user_key=user_key, did=self._virtual_did)
crt_str = await self._miot_http.get_central_cert_async( crt_str = await self._miot_http.get_central_cert_async(
csr_str) csr_str)
if not crt_str:
raise MIoTError('get_central_cert_async failed')
if not await miot_cert.update_user_cert_async( if not await miot_cert.update_user_cert_async(
cert=crt_str): cert=crt_str):
raise MIoTError('update_user_cert_async failed') raise MIoTError('update_user_cert_async failed')
@ -492,10 +507,12 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors={'base': error_reason}, errors={'base': error_reason},
) )
async def async_step_homes_select(self, user_input=None): async def async_step_homes_select(
self, user_input: Optional[dict] = None
):
_LOGGER.debug('async_step_homes_select') _LOGGER.debug('async_step_homes_select')
try: try:
if user_input is None: if not user_input:
return await self.__display_homes_select_form('') return await self.__display_homes_select_form('')
home_selected: list = user_input.get('home_infos', []) home_selected: list = user_input.get('home_infos', [])
@ -506,7 +523,8 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
'homes']['home_list'].items(): 'homes']['home_list'].items():
if home_id in home_selected: if home_id in home_selected:
self._home_selected[home_id] = home_info self._home_selected[home_id] = home_info
self._area_name_rule = user_input.get('area_name_rule') self._area_name_rule = user_input.get(
'area_name_rule', self._area_name_rule)
# Storage device list # Storage device list
devices_list: dict[str, dict] = { devices_list: dict[str, dict] = {
did: dev_info did: dev_info
@ -544,10 +562,16 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form( return self.async_show_form(
step_id='homes_select', step_id='homes_select',
data_schema=vol.Schema({ data_schema=vol.Schema({
vol.Required('home_infos'): cv.multi_select(self._home_list), vol.Required('home_infos'): cv.multi_select(
vol.Required('area_name_rule', default='room'): vol.In( self._home_list_show),
self._miot_i18n.translate(key='config.room_name_rule')), vol.Required(
vol.Required('advanced_options', default=False): bool, 'area_name_rule',
default=self._area_name_rule # type: ignore
): vol.In(self._miot_i18n.translate(
key='config.room_name_rule')),
vol.Required(
'advanced_options', default=False # type: ignore
): bool,
}), }),
errors={'base': reason}, errors={'base': reason},
description_placeholders={ description_placeholders={
@ -556,7 +580,9 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
last_step=False, last_step=False,
) )
async def async_step_advanced_options(self, user_input: dict = None): async def async_step_advanced_options(
self, user_input: Optional[dict] = None
):
if user_input: if user_input:
self._ctrl_mode = user_input.get('ctrl_mode', self._ctrl_mode) self._ctrl_mode = user_input.get('ctrl_mode', self._ctrl_mode)
self._action_debug = user_input.get( self._action_debug = user_input.get(
@ -570,16 +596,22 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form( return self.async_show_form(
step_id='advanced_options', step_id='advanced_options',
data_schema=vol.Schema({ data_schema=vol.Schema({
vol.Required('devices_filter', default=False): bool, vol.Required(
vol.Required('ctrl_mode', default=DEFAULT_CTRL_MODE): vol.In( 'devices_filter', default=False): bool, # type: ignore
self._miot_i18n.translate(key='config.control_mode')), vol.Required(
vol.Required('action_debug', default=self._action_debug): bool, 'ctrl_mode', default=self._ctrl_mode # type: ignore
): vol.In(self._miot_i18n.translate(key='config.control_mode')),
vol.Required(
'action_debug', default=self._action_debug # type: ignore
): bool,
vol.Required( vol.Required(
'hide_non_standard_entities', 'hide_non_standard_entities',
default=self._hide_non_standard_entities): bool, default=self._hide_non_standard_entities # type: ignore
): bool,
vol.Required( vol.Required(
'display_devices_changed_notify', 'display_devices_changed_notify',
default=self._display_devices_changed_notify): bool, default=self._display_devices_changed_notify # type: ignore
): bool,
}), }),
last_step=False, last_step=False,
) )
@ -645,16 +677,17 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return await self.__display_devices_filter_form(reason='') return await self.__display_devices_filter_form(reason='')
async def __display_devices_filter_form(self, reason: str): async def __display_devices_filter_form(self, reason: str):
tip_devices: str = self._miot_i18n.translate( tip_devices: str = self._miot_i18n.translate(
key='config.other.devices') key='config.other.devices') # type: ignore
tip_without_room: str = self._miot_i18n.translate( tip_without_room: str = self._miot_i18n.translate(
key='config.other.without_room') key='config.other.without_room') # type: ignore
trans_statistics_logic: dict = self._miot_i18n.translate( trans_statistics_logic: dict = self._miot_i18n.translate(
key='config.statistics_logic') key='config.statistics_logic') # type: ignore
trans_filter_mode: dict = self._miot_i18n.translate( trans_filter_mode: dict = self._miot_i18n.translate(
key='config.filter_mode') key='config.filter_mode') # type: ignore
trans_connect_type: dict = self._miot_i18n.translate( trans_connect_type: dict = self._miot_i18n.translate(
key='config.connect_type') key='config.connect_type') # type: ignore
room_device_count: dict = {} room_device_count: dict = {}
model_device_count: dict = {} model_device_count: dict = {}
@ -692,22 +725,27 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form( return self.async_show_form(
step_id='devices_filter', step_id='devices_filter',
data_schema=vol.Schema({ data_schema=vol.Schema({
vol.Required('room_filter_mode', default='exclude'): vol.Required(
vol.In(trans_filter_mode), 'room_filter_mode', default='exclude' # type: ignore
): vol.In(trans_filter_mode),
vol.Optional('room_list'): cv.multi_select(room_list), vol.Optional('room_list'): cv.multi_select(room_list),
vol.Required('type_filter_mode', default='exclude'): vol.Required(
vol.In(trans_filter_mode), 'type_filter_mode', default='exclude' # type: ignore
): vol.In(trans_filter_mode),
vol.Optional('type_list'): cv.multi_select(type_list), vol.Optional('type_list'): cv.multi_select(type_list),
vol.Required('model_filter_mode', default='exclude'): vol.Required(
vol.In(trans_filter_mode), 'model_filter_mode', default='exclude' # type: ignore
): vol.In(trans_filter_mode),
vol.Optional('model_list'): cv.multi_select(dict(sorted( vol.Optional('model_list'): cv.multi_select(dict(sorted(
model_list.items(), key=lambda item: item[0]))), model_list.items(), key=lambda item: item[0]))),
vol.Required('devices_filter_mode', default='exclude'): vol.Required(
vol.In(trans_filter_mode), 'devices_filter_mode', default='exclude' # type: ignore
): vol.In(trans_filter_mode),
vol.Optional('device_list'): cv.multi_select(dict(sorted( vol.Optional('device_list'): cv.multi_select(dict(sorted(
device_list.items(), key=lambda device: device[1]))), device_list.items(), key=lambda device: device[1]))),
vol.Required('statistics_logic', default='or'): vol.Required(
vol.In(trans_statistics_logic), 'statistics_logic', default='or' # type: ignore
): vol.In(trans_statistics_logic),
}), }),
errors={'base': reason}, errors={'base': reason},
last_step=False last_step=False
@ -759,7 +797,7 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
'uid': self._uid, 'uid': self._uid,
'nick_name': self._nick_name, 'nick_name': self._nick_name,
'cloud_server': self._cloud_server, 'cloud_server': self._cloud_server,
'oauth_redirect_url': self._oauth_redirect_url, 'oauth_redirect_url': self._oauth_redirect_url_full,
'ctrl_mode': self._ctrl_mode, 'ctrl_mode': self._ctrl_mode,
'home_selected': self._home_selected, 'home_selected': self._home_selected,
'area_name_rule': self._area_name_rule, 'area_name_rule': self._area_name_rule,
@ -784,7 +822,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
# pylint: disable=inconsistent-quotes # pylint: disable=inconsistent-quotes
_config_entry: config_entries.ConfigEntry _config_entry: config_entries.ConfigEntry
_main_loop: asyncio.AbstractEventLoop _main_loop: asyncio.AbstractEventLoop
_miot_client: Optional[MIoTClient] _miot_client: MIoTClient
_miot_network: Optional[MIoTNetwork] _miot_network: Optional[MIoTNetwork]
_miot_storage: Optional[MIoTStorage] _miot_storage: Optional[MIoTStorage]
@ -799,7 +837,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
_uid: Optional[str] _uid: Optional[str]
_storage_path: Optional[str] _storage_path: Optional[str]
_cloud_server: Optional[str] _cloud_server: Optional[str]
_oauth_redirect_url: Optional[str] _oauth_redirect_url_full: str
_integration_language: Optional[str] _integration_language: Optional[str]
_ctrl_mode: Optional[str] _ctrl_mode: Optional[str]
_nick_name: Optional[str] _nick_name: Optional[str]
@ -811,7 +849,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
_home_selected_dict: Optional[dict] _home_selected_dict: Optional[dict]
_home_info_buffer: Optional[dict[str, str | dict[str, dict]]] _home_info_buffer: Optional[dict[str, str | dict[str, dict]]]
_home_list: Optional[dict] _home_list: Optional[dict]
_device_list: Optional[dict[str, dict]] _device_list: dict[str, dict]
_devices_add: list[str] _devices_add: list[str]
_devices_remove: list[str] _devices_remove: list[str]
@ -851,7 +889,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
self._uid = self._entry_data['uid'] self._uid = self._entry_data['uid']
self._storage_path = self._entry_data['storage_path'] self._storage_path = self._entry_data['storage_path']
self._cloud_server = self._entry_data['cloud_server'] self._cloud_server = self._entry_data['cloud_server']
self._oauth_redirect_url = self._entry_data['oauth_redirect_url'] self._oauth_redirect_url_full = ''
self._ctrl_mode = self._entry_data['ctrl_mode'] self._ctrl_mode = self._entry_data['ctrl_mode']
self._integration_language = self._entry_data['integration_language'] self._integration_language = self._entry_data['integration_language']
self._nick_name = self._entry_data['nick_name'] self._nick_name = self._entry_data['nick_name']
@ -959,7 +997,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
if user_input: if user_input:
webhook_path = webhook_async_generate_path( webhook_path = webhook_async_generate_path(
webhook_id=self._virtual_did) webhook_id=self._virtual_did)
self._oauth_redirect_url = ( self._oauth_redirect_url_full = (
f'{user_input.get("oauth_redirect_url")}{webhook_path}') f'{user_input.get("oauth_redirect_url")}{webhook_path}')
return await self.async_step_oauth(user_input) return await self.async_step_oauth(user_input)
return self.async_show_form( return self.async_show_form(
@ -981,9 +1019,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
state = str(secrets.randbits(64)) state = str(secrets.randbits(64))
self.hass.data[DOMAIN][self._virtual_did]['oauth_state'] = state self.hass.data[DOMAIN][self._virtual_did]['oauth_state'] = state
self._miot_oauth.set_redirect_url( self._miot_oauth.set_redirect_url(
redirect_url=self._oauth_redirect_url) redirect_url=self._oauth_redirect_url_full)
self._oauth_auth_url = self._miot_oauth.gen_auth_url( self._oauth_auth_url = self._miot_oauth.gen_auth_url(
redirect_url=self._oauth_redirect_url, state=state) redirect_url=self._oauth_redirect_url_full, state=state)
_LOGGER.info( _LOGGER.info(
'async_step_oauth, oauth_url: %s', 'async_step_oauth, oauth_url: %s',
self._oauth_auth_url) self._oauth_auth_url)
@ -1166,7 +1204,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
self._nick_name_new = user_input.get('nick_name') self._nick_name_new = user_input.get('nick_name')
return await self.async_step_homes_select() return await self.async_step_homes_select()
async def async_step_homes_select(self, user_input=None): async def async_step_homes_select(
self, user_input: Optional[dict] = None
):
if not self._update_devices: if not self._update_devices:
return await self.async_step_update_trans_rules() return await self.async_step_update_trans_rules()
if not user_input: if not user_input:
@ -1236,27 +1276,19 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
if dev_info['home_id'] in self._home_selected_list} if dev_info['home_id'] in self._home_selected_list}
if not self._device_list: if not self._device_list:
return await self.__display_homes_select_form('no_devices') return await self.__display_homes_select_form('no_devices')
# Statistics devices changed self._device_list_sorted = dict(sorted(
self._devices_add = [] self._device_list.items(), key=lambda item:
self._devices_remove = [] item[1].get('home_id', '')+item[1].get('room_id', '')))
local_devices = await self._miot_storage.load_async(
domain='miot_devices',
name=f'{self._uid}_{self._cloud_server}',
type_=dict) or {}
self._devices_add = [ if user_input.get('devices_filter', False):
did for did in self._device_list.keys() if did not in local_devices] return await self.async_step_devices_filter()
self._devices_remove = [ return await self.update_devices_done_async()
did for did in local_devices.keys() if did not in self._device_list]
_LOGGER.debug(
'devices update, add->%s, remove->%s',
self._devices_add, self._devices_remove)
return await self.async_step_update_trans_rules()
async def __display_homes_select_form(self, reason: str): async def __display_homes_select_form(self, reason: str):
return self.async_show_form( return self.async_show_form(
step_id='homes_select', step_id='homes_select',
data_schema=vol.Schema({ data_schema=vol.Schema({
vol.Required('devices_filter', default=False): bool,
vol.Required('home_infos', default=self._home_selected_list): vol.Required('home_infos', default=self._home_selected_list):
cv.multi_select(self._home_list), cv.multi_select(self._home_list),
vol.Required('ctrl_mode', default=self._ctrl_mode): vol.In( vol.Required('ctrl_mode', default=self._ctrl_mode): vol.In(
@ -1269,6 +1301,100 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
last_step=False last_step=False
) )
async def async_step_devices_filter(
self, user_input: Optional[dict] = None
):
if user_input:
return await self.update_devices_done_async()
return await self.__display_devices_filter_form(reason='')
async def __display_devices_filter_form(self, reason: str):
tip_devices: str = self._miot_i18n.translate(
key='config.other.devices')
tip_without_room: str = self._miot_i18n.translate(
key='config.other.without_room')
trans_statistics_logic: dict = self._miot_i18n.translate(
key='config.statistics_logic')
trans_filter_mode: dict = self._miot_i18n.translate(
key='config.filter_mode')
trans_connect_type: dict = self._miot_i18n.translate(
key='config.connect_type')
room_device_count: dict = {}
model_device_count: dict = {}
connect_type_count: dict = {}
device_list: dict = {}
for did, info in self._device_list_sorted.items():
device_list[did] = (
f'[ {info["home_name"]} {info["room_name"]} ] ' +
f'{info["name"]}, {did}')
room_device_count.setdefault(info['room_id'], 0)
room_device_count[info['room_id']] += 1
model_device_count.setdefault(info['model'], 0)
model_device_count[info['model']] += 1
connect_type_count.setdefault(str(info['connect_type']), 0)
connect_type_count[str(info['connect_type'])] += 1
model_list: dict = {}
for model, count in model_device_count.items():
model_list[model] = f'{model} [ {count} {tip_devices} ]'
type_list: dict = {
k: f'{trans_connect_type.get(k, f"Connect Type ({k})")} '
f'[ {v} {tip_devices} ]'
for k, v in connect_type_count.items()}
room_list: dict = {}
for home_id, home_info in self._home_selected_dict.items():
for room_id, room_name in home_info['room_info'].items():
if room_id not in room_device_count:
continue
room_list[room_id] = (
f'{home_info["home_name"]} {room_name}'
f' [ {room_device_count[room_id]}{tip_devices} ]')
if home_id in room_device_count:
room_list[home_id] = (
f'{home_info["home_name"]} {tip_without_room}'
f' [ {room_device_count[home_id]}{tip_devices} ]')
return self.async_show_form(
step_id='devices_filter',
data_schema=vol.Schema({
vol.Required('room_filter_mode', default='exclude'):
vol.In(trans_filter_mode),
vol.Optional('room_list'): cv.multi_select(room_list),
vol.Required('type_filter_mode', default='exclude'):
vol.In(trans_filter_mode),
vol.Optional('type_list'): cv.multi_select(type_list),
vol.Required('model_filter_mode', default='exclude'):
vol.In(trans_filter_mode),
vol.Optional('model_list'): cv.multi_select(dict(sorted(
model_list.items(), key=lambda item: item[0]))),
vol.Required('devices_filter_mode', default='exclude'):
vol.In(trans_filter_mode),
vol.Optional('device_list'): cv.multi_select(dict(sorted(
device_list.items(), key=lambda device: device[1]))),
vol.Required('statistics_logic', default='or'):
vol.In(trans_statistics_logic),
}),
errors={'base': reason},
last_step=False
)
async def update_devices_done_async(self):
# Statistics devices changed
self._devices_add = []
self._devices_remove = []
local_devices: dict = await self._miot_storage.load_async(
domain='miot_devices',
name=f'{self._uid}_{self._cloud_server}',
type_=dict) or {} # type: ignore
self._devices_add = [
did for did in self._device_list.keys() if did not in local_devices]
self._devices_remove = [
did for did in local_devices.keys() if did not in self._device_list]
_LOGGER.debug(
'devices update, add->%s, remove->%s',
self._devices_add, self._devices_remove)
return await self.async_step_update_trans_rules()
async def async_step_update_trans_rules(self, user_input=None): async def async_step_update_trans_rules(self, user_input=None):
if not self._update_trans_rules: if not self._update_trans_rules:
return await self.async_step_update_lan_ctrl_config() return await self.async_step_update_lan_ctrl_config()
@ -1397,7 +1523,6 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
last_step=True last_step=True
) )
self._entry_data['oauth_redirect_url'] = self._oauth_redirect_url
if self._lang_new != self._integration_language: if self._lang_new != self._integration_language:
self._entry_data['integration_language'] = self._lang_new self._entry_data['integration_language'] = self._lang_new
self._need_reload = True self._need_reload = True