fix: fix type error, wrong use of any and Any

custom_components/xiaomi_home/climate.py

@@ -47,7 +47,7 @@
 """
 from __future__ import annotations
 import logging
-from typing import Optional
+from typing import Any, Optional
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
@@ -415,7 +415,7 @@ def swing_mode(self) -> Optional[str]:
                 return SWING_OFF
         return None
 
-    def __ac_state_changed(self, prop: MIoTSpecProperty, value: any) -> None:
+    def __ac_state_changed(self, prop: MIoTSpecProperty, value: Any) -> None:
         del prop
         if not isinstance(value, str):
             _LOGGER.error(

custom_components/xiaomi_home/event.py

@@ -46,6 +46,7 @@
 Event entities for Xiaomi Home.
 """
 from __future__ import annotations
+from typing import Any
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
@@ -84,6 +85,6 @@ def __init__(self, miot_device: MIoTDevice, spec: MIoTSpecEvent) -> None:
         # Set device_class
         self._attr_device_class = spec.device_class
 
-    def on_event_occurred(self, name: str, arguments: list[dict[int, any]]):
+    def on_event_occurred(self, name: str, arguments: list[dict[int, Any]]):
         """An event is occurred."""
         self._trigger_event(event_type=name, event_attributes=arguments)

custom_components/xiaomi_home/fan.py

@@ -93,7 +93,7 @@ class Fan(MIoTServiceEntity, FanEntity):
     _speed_min: Optional[int]
     _speed_max: Optional[int]
     _speed_step: Optional[int]
-    _mode_list: Optional[dict[any, any]]
+    _mode_list: Optional[dict[Any, Any]]
 
     def __init__(
         self, miot_device: MIoTDevice, entity_data: MIoTEntityData

custom_components/xiaomi_home/humidifier.py

@@ -47,7 +47,7 @@
 """
 from __future__ import annotations
 import logging
-from typing import Optional
+from typing import Any, Optional
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
@@ -97,7 +97,7 @@ class Humidifier(MIoTServiceEntity, HumidifierEntity):
     _prop_target_humidity: Optional[MIoTSpecProperty]
     _prop_humidity: Optional[MIoTSpecProperty]
 
-    _mode_list: dict[any, any]
+    _mode_list: dict[Any, Any]
 
     def __init__(
         self, miot_device: MIoTDevice,  entity_data: MIoTEntityData

custom_components/xiaomi_home/light.py

@@ -47,7 +47,7 @@
 """
 from __future__ import annotations
 import logging
-from typing import Optional
+from typing import Any, Optional
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
@@ -102,7 +102,7 @@ class Light(MIoTServiceEntity, LightEntity):
     _prop_mode: Optional[MIoTSpecProperty]
 
     _brightness_scale: Optional[tuple[int, int]]
-    _mode_list: Optional[dict[any, any]]
+    _mode_list: Optional[dict[Any, Any]]
 
     def __init__(
         self, miot_device: MIoTDevice,  entity_data: MIoTEntityData

custom_components/xiaomi_home/miot/common.py

@@ -48,9 +48,9 @@
 import json
 from os import path
 import random
-from typing import Optional
+from typing import Any, Optional
 import hashlib
-from paho.mqtt.client import MQTTMatcher
+from paho.mqtt.matcher import MQTTMatcher
 import yaml
 
 MIOT_ROOT_PATH: str = path.dirname(path.abspath(__file__))
@@ -87,7 +87,7 @@ def randomize_int(value: int, ratio: float) -> int:
 class MIoTMatcher(MQTTMatcher):
     """MIoT Pub/Sub topic matcher."""
 
-    def iter_all_nodes(self) -> any:
+    def iter_all_nodes(self) -> Any:
         """Return an iterator on all nodes with their paths and contents."""
         def rec(node, path_):
             # pylint: disable=protected-access
@@ -97,7 +97,7 @@ def rec(node, path_):
                 yield from rec(child, path_ + [part])
         return rec(self._root, [])
 
-    def get(self, topic: str) -> Optional[any]:
+    def get(self, topic: str) -> Optional[Any]:
         try:
             return self[topic]
         except KeyError:

custom_components/xiaomi_home/miot/miot_client.py

@@ -1,7 +1,52 @@
 # -*- coding: utf-8 -*-
-"""MIoT client instance."""
+"""
+Copyright (C) 2024 Xiaomi Corporation.
+
+The ownership and intellectual property rights of Xiaomi Home Assistant
+Integration and related Xiaomi cloud service API interface provided under this
+license, including source code and object code (collectively, "Licensed Work"),
+are owned by Xiaomi. Subject to the terms and conditions of this License, Xiaomi
+hereby grants you a personal, limited, non-exclusive, non-transferable,
+non-sublicensable, and royalty-free license to reproduce, use, modify, and
+distribute the Licensed Work only for your use of Home Assistant for
+non-commercial purposes. For the avoidance of doubt, Xiaomi does not authorize
+you to use the Licensed Work for any other purpose, including but not limited
+to use Licensed Work to develop applications (APP), Web services, and other
+forms of software.
+
+You may reproduce and distribute copies of the Licensed Work, with or without
+modifications, whether in source or object form, provided that you must give
+any other recipients of the Licensed Work a copy of this License and retain all
+copyright and disclaimers.
+
+Xiaomi provides the Licensed Work on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+CONDITIONS OF ANY KIND, either express or implied, including, without
+limitation, any warranties, undertakes, or conditions of TITLE, NO ERROR OR
+OMISSION, CONTINUITY, RELIABILITY, NON-INFRINGEMENT, MERCHANTABILITY, or
+FITNESS FOR A PARTICULAR PURPOSE. In any event, you are solely responsible
+for any direct, indirect, special, incidental, or consequential damages or
+losses arising from the use or inability to use the Licensed Work.
+
+Xiaomi reserves all rights not expressly granted to you in this License.
+Except for the rights expressly granted by Xiaomi under this License, Xiaomi
+does not authorize you in any form to use the trademarks, copyrights, or other
+forms of intellectual property rights of Xiaomi and its affiliates, including,
+without limitation, without obtaining other written permission from Xiaomi, you
+shall not use "Xiaomi", "Mijia" and other words related to Xiaomi or words that
+may make the public associate with Xiaomi in any form to publicize or promote
+the software or hardware devices that use the Licensed Work.
+
+Xiaomi has the right to immediately terminate all your authorization under this
+License in the event:
+1. You assert patent invalidation, litigation, or other claims against patents
+or other intellectual property rights of Xiaomi or its affiliates; or,
+2. You make, have made, manufacture, sell, or offer to sell products that knock
+off Xiaomi or its affiliates' products.
+
+MIoT client instance.
+"""
 from copy import deepcopy
-from typing import Callable, Optional, final
+from typing import Any, Callable, Optional, final
 import asyncio
 import json
 import logging
@@ -36,9 +81,9 @@
 @dataclass
 class MIoTClientSub:
     """MIoT client subscription."""
-    topic: str = None
-    handler: Callable[[dict, any], None] = None
-    handler_ctx: any = None
+    topic: Optional[str]
+    handler: Callable[[dict, Any], None]
+    handler_ctx: Any = None
 
     def __str__(self) -> str:
         return f'{self.topic}, {id(self.handler)}, {id(self.handler_ctx)}'
@@ -526,7 +571,7 @@ async def refresh_user_cert_async(self) -> bool:
         return False
 
     async def set_prop_async(
-        self, did: str, siid: int, piid: int, value: any
+        self, did: str, siid: int, piid: int, value: Any
     ) -> bool:
         if did not in self._device_list_cache:
             raise MIoTClientError(f'did not exist, {did}')
@@ -612,7 +657,7 @@ def request_refresh_prop(
             0.2, lambda: self._main_loop.create_task(
                 self.__refresh_props_handler()))
 
-    async def get_prop_async(self, did: str, siid: int, piid: int) -> any:
+    async def get_prop_async(self, did: str, siid: int, piid: int) -> Any:
         if did not in self._device_list_cache:
             raise MIoTClientError(f'did not exist, {did}')
 
@@ -717,8 +762,8 @@ async def action_async(
         return None
 
     def sub_prop(
-        self, did: str, handler: Callable[[dict, any], None],
-        siid: int = None, piid: int = None, handler_ctx: any = None
+        self, did: str, handler: Callable[[dict, Any], None],
+        siid: int = None, piid: int = None, handler_ctx: Any = None
     ) -> bool:
         if did not in self._device_list_cache:
             raise MIoTClientError(f'did not exist, {did}')
@@ -741,8 +786,8 @@ def unsub_prop(self, did: str, siid: int = None, piid: int = None) -> bool:
         return True
 
     def sub_event(
-        self, did: str, handler: Callable[[dict, any], None],
-        siid: int = None, eiid: int = None, handler_ctx: any = None
+        self, did: str, handler: Callable[[dict, Any], None],
+        siid: int = None, eiid: int = None, handler_ctx: Any = None
     ) -> bool:
         if did not in self._device_list_cache:
             raise MIoTClientError(f'did not exist, {did}')
@@ -764,8 +809,8 @@ def unsub_event(self, did: str, siid: int = None, eiid: int = None) -> bool:
         return True
 
     def sub_device_state(
-        self, did: str, handler: Callable[[str, MIoTDeviceState, any], None],
-        handler_ctx: any = None
+        self, did: str, handler: Callable[[str, MIoTDeviceState, Any], None],
+        handler_ctx: Any = None
     ) -> bool:
         """Call callback handler in main loop"""
         if did not in self._device_list_cache:
@@ -1060,7 +1105,7 @@ async def __on_miot_lan_state_change(self, state: bool) -> None:
 
     @final
     def __on_cloud_device_state_changed(
-        self, did: str, state: MIoTDeviceState, ctx: any
+        self, did: str, state: MIoTDeviceState, ctx: Any
     ) -> None:
         _LOGGER.info('cloud device state changed, %s, %s', did, state)
         cloud_device = self._device_list_cloud.get(did, None)
@@ -1110,7 +1155,7 @@ async def __on_gw_device_list_changed(
 
     @final
     async def __on_lan_device_state_changed(
-        self, did: str, state: dict, ctx: any
+        self, did: str, state: dict, ctx: Any
     ) -> None:
         _LOGGER.info('lan device state changed, %s, %s', did, state)
         lan_state_new: bool = state.get('online', False)
@@ -1146,7 +1191,7 @@ async def __on_lan_device_state_changed(
         self.__request_show_devices_changed_notify()
 
     @final
-    def __on_prop_msg(self, params: dict, ctx: any) -> None:
+    def __on_prop_msg(self, params: dict, ctx: Any) -> None:
         """params MUST contain did, siid, piid, value"""
         # BLE device has no online/offline msg
         try:
@@ -1158,7 +1203,7 @@ def __on_prop_msg(self, params: dict, ctx: any) -> None:
             _LOGGER.error('on prop msg error, %s, %s', params, err)
 
     @final
-    def __on_event_msg(self, params: dict, ctx: any) -> None:
+    def __on_event_msg(self, params: dict, ctx: Any) -> None:
         try:
             subs: list[MIoTClientSub] = list(self._sub_tree.iter_match(
                 f'{params["did"]}/e/{params["siid"]}/{params["eiid"]}'))

custom_components/xiaomi_home/miot/miot_cloud.py

@@ -51,7 +51,7 @@
 import logging
 import re
 import time
-from typing import Optional
+from typing import Any, Optional
 from urllib.parse import urlencode
 import aiohttp
 
@@ -97,7 +97,7 @@ def __init__(
         self._session = aiohttp.ClientSession()
 
     def __del__(self):
-        self._session.close()
+        self._main_loop.create_task(self._session.close())
 
     def set_redirect_url(self, redirect_url: str) -> None:
         if not isinstance(redirect_url, str) or redirect_url.strip() == '':
@@ -253,7 +253,7 @@ def __init__(
         self._session = aiohttp.ClientSession()
 
     def __del__(self):
-        self._session.close()
+        self._main_loop.create_task(self._session.close())
 
     def update_http_header(
         self, cloud_server: Optional[str] = None,
@@ -581,7 +581,7 @@ async def get_devices_async(
         self, home_ids: list[str] = None
     ) -> dict[str, dict]:
         homeinfos = await self.get_homeinfos_async()
-        homes: dict[str, dict[str, any]] = {}
+        homes: dict[str, dict[str, Any]] = {}
         devices: dict[str, dict] = {}
         for device_type in ['home_list', 'share_home_list']:
             homes.setdefault(device_type, {})
@@ -661,7 +661,7 @@ async def get_props_async(self, params: list) -> list:
             raise MIoTHttpError('invalid response result')
         return res_obj['result']
 
-    async def __get_prop_async(self, did: str, siid: int, piid: int) -> any:
+    async def __get_prop_async(self, did: str, siid: int, piid: int) -> Any:
         results = await self.get_props_async(
             params=[{'did': did, 'siid': siid, 'piid': piid}])
         if not results:
@@ -722,7 +722,7 @@ async def __get_prop_handler(self) -> bool:
 
     async def get_prop_async(
         self, did: str, siid: int, piid: int, immediately: bool = False
-    ) -> any:
+    ) -> Any:
         if immediately:
             return await self.__get_prop_async(did, siid, piid)
         key: str = f'{did}.{siid}.{piid}'