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

style: param init

Browse Source
This commit is contained in:
topsworld 2025-01-09 17:33:33 +08:00
parent 43fc8c4d56
commit 29102b6536
3 changed files with 38 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
custom_components/xiaomi_home/miot/miot_i18n.py
View File

@ -48,7 +48,7 @@ MIoT internationalization translation.
import asyncio import asyncio
import logging import logging
import os import os
from typing import Optional from typing import Optional, Union
# pylint: disable=relative-beyond-top-level # pylint: disable=relative-beyond-top-level
from .common import load_json_file from .common import load_json_file
@ -98,7 +98,7 @@ class MIoTI18n:
def translate( def translate(
self, key: str, replace: Optional[dict[str, str]] = None self, key: str, replace: Optional[dict[str, str]] = None
) -> str | dict | None: ) -> Union[str, dict, None]:
result = self._data result = self._data
for item in key.split('.'): for item in key.split('.'):
if item not in result: if item not in result:

27
custom_components/xiaomi_home/miot/miot_lan.py
View File

@ -381,7 +381,8 @@ class _MIoTLanDevice:
_MIoTLanDeviceState(state.value+1)) _MIoTLanDeviceState(state.value+1))
# Fast ping # Fast ping
if self._if_name is None: if self._if_name is None:
_LOGGER.error('if_name is Not set for device, %s', self.did) _LOGGER.error(
'if_name is Not set for device, %s', self.did)
return return
if self.ip is None: if self.ip is None:
_LOGGER.error('ip is Not set for device, %s', self.did) _LOGGER.error('ip is Not set for device, %s', self.did)
@ -419,10 +420,10 @@ class _MIoTLanDevice:
self.online = True self.online = True
else: else:
_LOGGER.info('unstable device detected, %s', self.did) _LOGGER.info('unstable device detected, %s', self.did)
self._online_offline_timer = \ self._online_offline_timer = (
self._manager.internal_loop.call_later( self._manager.internal_loop.call_later(
self.NETWORK_UNSTABLE_RESUME_TH, self.NETWORK_UNSTABLE_RESUME_TH,
self.__online_resume_handler) self.__online_resume_handler))
def __online_resume_handler(self) -> None: def __online_resume_handler(self) -> None:
_LOGGER.info('unstable resume threshold past, %s', self.did) _LOGGER.info('unstable resume threshold past, %s', self.did)
@ -508,9 +509,9 @@ class MIoTLan:
key='miot_lan', group_id='*', key='miot_lan', group_id='*',
handler=self.__on_mips_service_change) handler=self.__on_mips_service_change)
self._enable_subscribe = enable_subscribe self._enable_subscribe = enable_subscribe
self._virtual_did = str(virtual_did) \ self._virtual_did = (
if (virtual_did is not None) \ str(virtual_did) if (virtual_did is not None)
else str(secrets.randbits(64)) else str(secrets.randbits(64)))
# Init socket probe message # Init socket probe message
probe_bytes = bytearray(self.OT_PROBE_LEN) probe_bytes = bytearray(self.OT_PROBE_LEN)
probe_bytes[:20] = ( probe_bytes[:20] = (
@ -948,7 +949,7 @@ class MIoTLan:
# The following methods SHOULD ONLY be called in the internal loop # The following methods SHOULD ONLY be called in the internal loop
def ping(self, if_name: str | None, target_ip: str) -> None: def ping(self, if_name: Optional[str], target_ip: str) -> None:
if not target_ip: if not target_ip:
return return
self.__sendto( self.__sendto(
@ -964,7 +965,7 @@ class MIoTLan:
) -> None: ) -> None:
if timeout_ms and not handler: if timeout_ms and not handler:
raise ValueError('handler is required when timeout_ms is set') raise ValueError('handler is required when timeout_ms is set')
device: _MIoTLanDevice | None = self._lan_devices.get(did) device: Optional[_MIoTLanDevice] = self._lan_devices.get(did)
if not device: if not device:
raise ValueError('invalid device') raise ValueError('invalid device')
if not device.cipher: if not device.cipher:
@ -1232,7 +1233,7 @@ class MIoTLan:
return return
# Keep alive message # Keep alive message
did: str = str(struct.unpack('>Q', data[4:12])[0]) did: str = str(struct.unpack('>Q', data[4:12])[0])
device: _MIoTLanDevice | None = self._lan_devices.get(did) device: Optional[_MIoTLanDevice] = self._lan_devices.get(did)
if not device: if not device:
return return
timestamp: int = struct.unpack('>I', data[12:16])[0] timestamp: int = struct.unpack('>I', data[12:16])[0]
@ -1272,8 +1273,8 @@ class MIoTLan:
_LOGGER.warning('invalid message, no id, %s, %s', did, msg) _LOGGER.warning('invalid message, no id, %s, %s', did, msg)
return return
# Reply # Reply
req: _MIoTLanRequestData | None = \ req: Optional[_MIoTLanRequestData] = (
self._pending_requests.pop(msg['id'], None) self._pending_requests.pop(msg['id'], None))
if req: if req:
if req.timeout: if req.timeout:
req.timeout.cancel() req.timeout.cancel()
@ -1334,7 +1335,7 @@ class MIoTLan:
return False return False
def __sendto( def __sendto(
self, if_name: str | None, data: bytes, address: str, port: int self, if_name: Optional[str], data: bytes, address: str, port: int
) -> None: ) -> None:
if if_name is None: if if_name is None:
# Broadcast # Broadcast
@ -1356,7 +1357,7 @@ class MIoTLan:
try: try:
# Scan devices # Scan devices
self.ping(if_name=None, target_ip='255.255.255.255') self.ping(if_name=None, target_ip='255.255.255.255')
except Exception as err: # pylint: disable=broad-exception-caught except Exception as err: # pylint: disable=broad-exception-caught
# Ignore any exceptions to avoid blocking the loop # Ignore any exceptions to avoid blocking the loop
_LOGGER.error('ping device error, %s', err) _LOGGER.error('ping device error, %s', err)
pass pass

43
custom_components/xiaomi_home/miot/miot_mips.py
View File

@ -85,9 +85,9 @@ class _MipsMsgTypeOptions(Enum):
class _MipsMessage: class _MipsMessage:
"""MIoT Pub/Sub message.""" """MIoT Pub/Sub message."""
mid: int = 0 mid: int = 0
msg_from: str | None = None msg_from: Optional[str] = None
ret_topic: str | None = None ret_topic: Optional[str] = None
payload: str | None = None payload: Optional[str] = None
@staticmethod @staticmethod
def unpack(data: bytes) -> '_MipsMessage': def unpack(data: bytes) -> '_MipsMessage':
@ -122,8 +122,8 @@ class _MipsMessage:
def pack( def pack(
mid: int, mid: int,
payload: str, payload: str,
msg_from: str | None = None, msg_from: Optional[str] = None,
ret_topic: str | None = None ret_topic: Optional[str] = None
) -> bytes: ) -> bytes:
if mid is None or payload is None: if mid is None or payload is None:
raise MIoTMipsError('invalid mid or payload') raise MIoTMipsError('invalid mid or payload')
@ -159,7 +159,7 @@ class _MipsRequest:
mid: int mid: int
on_reply: Callable[[str, Any], None] on_reply: Callable[[str, Any], None]
on_reply_ctx: Any on_reply_ctx: Any
timer: asyncio.TimerHandle | None timer: Optional[asyncio.TimerHandle]
@dataclass @dataclass
@ -199,13 +199,13 @@ class MIoTDeviceState(Enum):
@dataclass @dataclass
class MipsDeviceState: class MipsDeviceState:
"""MIoT Pub/Sub device state.""" """MIoT Pub/Sub device state."""
did: str | None = None did: Optional[str] = None
"""handler """handler
str: did str: did
MIoTDeviceState: online/offline/disable MIoTDeviceState: online/offline/disable
Any: ctx Any: ctx
""" """
handler: Callable[[str, MIoTDeviceState, Any], None] | None = None handler: Optional[Callable[[str, MIoTDeviceState, Any], None]] = None
handler_ctx: Any = None handler_ctx: Any = None
@ -220,26 +220,26 @@ class _MipsClient(ABC):
MIPS_SUB_PATCH: int = 300 MIPS_SUB_PATCH: int = 300
MIPS_SUB_INTERVAL: float = 1 MIPS_SUB_INTERVAL: float = 1
main_loop: asyncio.AbstractEventLoop main_loop: asyncio.AbstractEventLoop
_logger: logging.Logger | None _logger: Optional[logging.Logger]
_client_id: str _client_id: str
_host: str _host: str
_port: int _port: int
_username: str | None _username: Optional[str]
_password: str | None _password: Optional[str]
_ca_file: str | None _ca_file: Optional[str]
_cert_file: str | None _cert_file: Optional[str]
_key_file: str | None _key_file: Optional[str]
_mqtt_logger: logging.Logger | None _mqtt_logger: Optional[logging.Logger]
_mqtt: Client _mqtt: Client
_mqtt_fd: int _mqtt_fd: int
_mqtt_timer: asyncio.TimerHandle | None _mqtt_timer: Optional[asyncio.TimerHandle]
_mqtt_state: bool _mqtt_state: bool
_event_connect: asyncio.Event _event_connect: asyncio.Event
_event_disconnect: asyncio.Event _event_disconnect: asyncio.Event
_internal_loop: asyncio.AbstractEventLoop _internal_loop: asyncio.AbstractEventLoop
_mips_thread: threading.Thread | None = None _mips_thread: Optional[threading.Thread]
_mips_reconnect_tag: bool _mips_reconnect_tag: bool
_mips_reconnect_interval: float _mips_reconnect_interval: float
_mips_reconnect_timer: Optional[asyncio.TimerHandle] _mips_reconnect_timer: Optional[asyncio.TimerHandle]
@ -284,6 +284,7 @@ class _MipsClient(ABC):
# Mips init # Mips init
self._event_connect = asyncio.Event() self._event_connect = asyncio.Event()
self._event_disconnect = asyncio.Event() self._event_disconnect = asyncio.Event()
self._mips_thread = None
self._mips_reconnect_tag = False self._mips_reconnect_tag = False
self._mips_reconnect_interval = 0 self._mips_reconnect_interval = 0
self._mips_reconnect_timer = None self._mips_reconnect_timer = None
@ -1056,8 +1057,8 @@ class MipsLocalClient(_MipsClient):
_request_map: dict[str, _MipsRequest] _request_map: dict[str, _MipsRequest]
_msg_matcher: MIoTMatcher _msg_matcher: MIoTMatcher
_get_prop_queue: dict[str, list] _get_prop_queue: dict[str, list]
_get_prop_timer: asyncio.TimerHandle | None _get_prop_timer: Optional[asyncio.TimerHandle]
_on_dev_list_changed: Callable[[Any, list[str]], Coroutine] | None _on_dev_list_changed: Optional[Callable[[Any, list[str]], Coroutine]]
def __init__( def __init__(
self, did: str, host: str, group_id: str, self, did: str, host: str, group_id: str,
@ -1365,7 +1366,7 @@ class MipsLocalClient(_MipsClient):
@property @property
def on_dev_list_changed( def on_dev_list_changed(
self self
) -> Callable[[Any, list[str]], Coroutine] | None: ) -> Optional[Callable[[Any, list[str]], Coroutine]]:
return self._on_dev_list_changed return self._on_dev_list_changed
@final @final
@ -1454,7 +1455,7 @@ class MipsLocalClient(_MipsClient):
# Reply # Reply
if topic == self._reply_topic: if topic == self._reply_topic:
self.log_debug(f'on request reply, {mips_msg}') self.log_debug(f'on request reply, {mips_msg}')
req: _MipsRequest | None = self._request_map.pop( req: Optional[_MipsRequest] = self._request_map.pop(
str(mips_msg.mid), None) str(mips_msg.mid), None)
if req: if req:
# Cancel timer # Cancel timer