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

change internal class scope

Browse Source
This commit is contained in:
Feng Wang 2024-12-22 20:16:45 +08:00
parent d3c39f38e1
commit fb46bc6b19
1 changed files with 50 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -76,51 +76,51 @@ _LOGGER = logging.getLogger(__name__)
@dataclass @dataclass
class MIoTLanGetDevListData: class _MIoTLanGetDevListData:
handler: Callable[[dict, Any], None] handler: Callable[[dict, Any], None]
handler_ctx: Any handler_ctx: Any
timeout_ms: int timeout_ms: int
@dataclass @dataclass
class MIoTLanUnregisterBroadcastData: class _MIoTLanUnregisterBroadcastData:
key: str key: str
@dataclass @dataclass
class MIoTLanRegisterBroadcastData: class _MIoTLanRegisterBroadcastData:
key: str key: str
handler: Callable[[dict, Any], None] handler: Callable[[dict, Any], None]
handler_ctx: Any handler_ctx: Any
@dataclass @dataclass
class MIoTLanUnsubDeviceData: class _MIoTLanUnsubDeviceData:
key: str key: str
@dataclass @dataclass
class MIoTLanSubDeviceData: class _MIoTLanSubDeviceData:
key: str key: str
handler: Callable[[str, dict, Any], Coroutine] handler: Callable[[str, dict, Any], Coroutine]
handler_ctx: Any handler_ctx: Any
@dataclass @dataclass
class MIoTLanNetworkUpdateData: class _MIoTLanNetworkUpdateData:
status: InterfaceStatus status: InterfaceStatus
if_name: str if_name: str
@dataclass @dataclass
class MIoTLanRequestData: class _MIoTLanRequestData:
msg_id: int msg_id: int
handler: Optional[Callable[[dict, Any], None]] handler: Optional[Callable[[dict, Any], None]]
handler_ctx: Any handler_ctx: Any
timeout: Optional[asyncio.TimerHandle] timeout: Optional[asyncio.TimerHandle]
class MIoTLanDeviceState(Enum): class _MIoTLanDeviceState(Enum):
FRESH = 0 FRESH = 0
PING1 = auto() PING1 = auto()
PING2 = auto() PING2 = auto()
@ -128,7 +128,7 @@ class MIoTLanDeviceState(Enum):
DEAD = auto() DEAD = auto()
class MIoTLanDevice: class _MIoTLanDevice:
"""MIoT lan device.""" """MIoT lan device."""
# pylint: disable=unused-argument # pylint: disable=unused-argument
OT_HEADER: int = 0x2131 OT_HEADER: int = 0x2131
@ -154,7 +154,7 @@ class MIoTLanDevice:
_manager: 'MIoTLan' _manager: 'MIoTLan'
_if_name: Optional[str] _if_name: Optional[str]
_sub_locked: bool _sub_locked: bool
_state: MIoTLanDeviceState _state: _MIoTLanDeviceState
_online: bool _online: bool
_online_offline_history: list[dict[str, Any]] _online_offline_history: list[dict[str, Any]]
_online_offline_timer: Optional[asyncio.TimerHandle] _online_offline_timer: Optional[asyncio.TimerHandle]
@ -185,14 +185,14 @@ class MIoTLanDevice:
self.supported_wildcard_sub = False self.supported_wildcard_sub = False
self._if_name = None self._if_name = None
self._sub_locked = False self._sub_locked = False
self._state = MIoTLanDeviceState.DEAD self._state = _MIoTLanDeviceState.DEAD
self._online = False self._online = False
self._online_offline_history = [] self._online_offline_history = []
self._online_offline_timer = None self._online_offline_timer = None
def ka_init_handler() -> None: def ka_init_handler() -> None:
self._ka_internal = self.KA_INTERVAL_MIN self._ka_internal = self.KA_INTERVAL_MIN
self.__update_keep_alive(state=MIoTLanDeviceState.DEAD) self.__update_keep_alive(state=_MIoTLanDeviceState.DEAD)
self._ka_timer = self._manager.internal_loop.call_later( self._ka_timer = self._manager.internal_loop.call_later(
randomize_float(self.CONSTRUCT_STATE_PENDING, 0.5), randomize_float(self.CONSTRUCT_STATE_PENDING, 0.5),
ka_init_handler,) ka_init_handler,)
@ -204,7 +204,7 @@ class MIoTLanDevice:
self._if_name = if_name self._if_name = if_name
_LOGGER.info( _LOGGER.info(
'device if_name change, %s, %s', self._if_name, self.did) 'device if_name change, %s, %s', self._if_name, self.did)
self.__update_keep_alive(state=MIoTLanDeviceState.FRESH) self.__update_keep_alive(state=_MIoTLanDeviceState.FRESH)
@property @property
def online(self) -> bool: def online(self) -> bool:
@ -352,31 +352,31 @@ class MIoTLanDevice:
return return
_LOGGER.info('unsubscribe success, %s, %s', self._if_name, self.did) _LOGGER.info('unsubscribe success, %s, %s', self._if_name, self.did)
def __update_keep_alive(self, state: MIoTLanDeviceState) -> None: def __update_keep_alive(self, state: _MIoTLanDeviceState) -> None:
last_state: MIoTLanDeviceState = self._state last_state: _MIoTLanDeviceState = self._state
self._state = state self._state = state
if self._state != MIoTLanDeviceState.FRESH: if self._state != _MIoTLanDeviceState.FRESH:
_LOGGER.debug('device status, %s, %s', self.did, self._state) _LOGGER.debug('device status, %s, %s', self.did, self._state)
if self._ka_timer: if self._ka_timer:
self._ka_timer.cancel() self._ka_timer.cancel()
self._ka_timer = None self._ka_timer = None
match state: match state:
case MIoTLanDeviceState.FRESH: case _MIoTLanDeviceState.FRESH:
if last_state == MIoTLanDeviceState.DEAD: if last_state == _MIoTLanDeviceState.DEAD:
self._ka_internal = self.KA_INTERVAL_MIN self._ka_internal = self.KA_INTERVAL_MIN
self.__change_online(True) self.__change_online(True)
self._ka_timer = self._manager.internal_loop.call_later( self._ka_timer = self._manager.internal_loop.call_later(
self.__get_next_ka_timeout(), self.__update_keep_alive, self.__get_next_ka_timeout(), self.__update_keep_alive,
MIoTLanDeviceState.PING1) _MIoTLanDeviceState.PING1)
case ( case (
MIoTLanDeviceState.PING1 _MIoTLanDeviceState.PING1
| MIoTLanDeviceState.PING2 | _MIoTLanDeviceState.PING2
| MIoTLanDeviceState.PING3 | _MIoTLanDeviceState.PING3
): ):
# Set the timer first to avoid Any early returns # Set the timer first to avoid Any early returns
self._ka_timer = self._manager.internal_loop.call_later( self._ka_timer = self._manager.internal_loop.call_later(
self.FAST_PING_INTERVAL, self.__update_keep_alive, self.FAST_PING_INTERVAL, self.__update_keep_alive,
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)
@ -385,8 +385,8 @@ class MIoTLanDevice:
_LOGGER.error('ip is Not set for device, %s', self.did) _LOGGER.error('ip is Not set for device, %s', self.did)
return return
self._manager.ping(if_name=self._if_name, target_ip=self.ip) self._manager.ping(if_name=self._if_name, target_ip=self.ip)
case MIoTLanDeviceState.DEAD: case _MIoTLanDeviceState.DEAD:
if last_state == MIoTLanDeviceState.PING3: if last_state == _MIoTLanDeviceState.PING3:
self._ka_internal = self.KA_INTERVAL_MIN self._ka_internal = self.KA_INTERVAL_MIN
self.__change_online(False) self.__change_online(False)
case _: case _:
@ -451,7 +451,7 @@ class MIoTLan:
_network: MIoTNetwork _network: MIoTNetwork
_mips_service: MipsService _mips_service: MipsService
_enable_subscribe: bool _enable_subscribe: bool
_lan_devices: dict[str, MIoTLanDevice] _lan_devices: dict[str, _MIoTLanDevice]
_virtual_did: str _virtual_did: str
_probe_msg: bytes _probe_msg: bytes
_write_buffer: bytearray _write_buffer: bytearray
@ -466,9 +466,9 @@ class MIoTLan:
_scan_timer: Optional[asyncio.TimerHandle] _scan_timer: Optional[asyncio.TimerHandle]
_last_scan_interval: Optional[float] _last_scan_interval: Optional[float]
_msg_id_counter: int _msg_id_counter: int
_pending_requests: dict[int, MIoTLanRequestData] _pending_requests: dict[int, _MIoTLanRequestData]
_device_msg_matcher: MIoTMatcher _device_msg_matcher: MIoTMatcher
_device_state_sub_map: dict[str, MIoTLanSubDeviceData] _device_state_sub_map: dict[str, _MIoTLanSubDeviceData]
_reply_msg_buffer: dict[str, asyncio.TimerHandle] _reply_msg_buffer: dict[str, asyncio.TimerHandle]
_lan_state_sub_map: dict[str, Callable[[bool], Coroutine]] _lan_state_sub_map: dict[str, Callable[[bool], Coroutine]]
@ -702,14 +702,14 @@ class MIoTLan:
) -> bool: ) -> bool:
self._internal_loop.call_soon_threadsafe( self._internal_loop.call_soon_threadsafe(
self.__sub_device_state, self.__sub_device_state,
MIoTLanSubDeviceData( _MIoTLanSubDeviceData(
key=key, handler=handler, handler_ctx=handler_ctx)) key=key, handler=handler, handler_ctx=handler_ctx))
return True return True
@final @final
def unsub_device_state(self, key: str) -> bool: def unsub_device_state(self, key: str) -> bool:
self._internal_loop.call_soon_threadsafe( self._internal_loop.call_soon_threadsafe(
self.__unsub_device_state, MIoTLanUnsubDeviceData(key=key)) self.__unsub_device_state, _MIoTLanUnsubDeviceData(key=key))
return True return True
@final @final
@ -728,7 +728,7 @@ class MIoTLan:
f'{"#" if siid is None or piid is None else f"{siid}/{piid}"}') f'{"#" if siid is None or piid is None else f"{siid}/{piid}"}')
self._internal_loop.call_soon_threadsafe( self._internal_loop.call_soon_threadsafe(
self.__sub_broadcast, self.__sub_broadcast,
MIoTLanRegisterBroadcastData( _MIoTLanRegisterBroadcastData(
key=key, handler=handler, handler_ctx=handler_ctx)) key=key, handler=handler, handler_ctx=handler_ctx))
return True return True
@ -746,7 +746,7 @@ class MIoTLan:
f'{"#" if siid is None or piid is None else f"{siid}/{piid}"}') f'{"#" if siid is None or piid is None else f"{siid}/{piid}"}')
self._internal_loop.call_soon_threadsafe( self._internal_loop.call_soon_threadsafe(
self.__unsub_broadcast, self.__unsub_broadcast,
MIoTLanUnregisterBroadcastData(key=key)) _MIoTLanUnregisterBroadcastData(key=key))
return True return True
@final @final
@ -765,7 +765,7 @@ class MIoTLan:
f'{"#" if siid is None or eiid is None else f"{siid}/{eiid}"}') f'{"#" if siid is None or eiid is None else f"{siid}/{eiid}"}')
self._internal_loop.call_soon_threadsafe( self._internal_loop.call_soon_threadsafe(
self.__sub_broadcast, self.__sub_broadcast,
MIoTLanRegisterBroadcastData( _MIoTLanRegisterBroadcastData(
key=key, handler=handler, handler_ctx=handler_ctx)) key=key, handler=handler, handler_ctx=handler_ctx))
return True return True
@ -783,7 +783,7 @@ class MIoTLan:
f'{"#" if siid is None or eiid is None else f"{siid}/{eiid}"}') f'{"#" if siid is None or eiid is None else f"{siid}/{eiid}"}')
self._internal_loop.call_soon_threadsafe( self._internal_loop.call_soon_threadsafe(
self.__unsub_broadcast, self.__unsub_broadcast,
MIoTLanUnregisterBroadcastData(key=key)) _MIoTLanUnregisterBroadcastData(key=key))
return True return True
@final @final
@ -865,7 +865,7 @@ class MIoTLan:
fut: asyncio.Future = self._main_loop.create_future() fut: asyncio.Future = self._main_loop.create_future()
self._internal_loop.call_soon_threadsafe( self._internal_loop.call_soon_threadsafe(
self.__get_dev_list, self.__get_dev_list,
MIoTLanGetDevListData( _MIoTLanGetDevListData(
handler=get_device_list_handler, handler=get_device_list_handler,
handler_ctx=fut, handler_ctx=fut,
timeout_ms=timeout_ms)) timeout_ms=timeout_ms))
@ -908,7 +908,7 @@ class MIoTLan:
return return
self._internal_loop.call_soon_threadsafe( self._internal_loop.call_soon_threadsafe(
self.__on_network_info_chnage, self.__on_network_info_chnage,
MIoTLanNetworkUpdateData(status=status, if_name=info.name)) _MIoTLanNetworkUpdateData(status=status, if_name=info.name))
async def __on_mips_service_change( async def __on_mips_service_change(
self, group_id: str, state: MipsServiceState, data: dict self, group_id: str, state: MipsServiceState, data: dict
@ -940,7 +940,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: _MIoTLanDevice | None = 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:
@ -975,7 +975,7 @@ class MIoTLan:
handler_ctx: Any = None, handler_ctx: Any = None,
timeout_ms: Optional[int] = None timeout_ms: Optional[int] = None
) -> None: ) -> None:
def request_timeout_handler(req_data: MIoTLanRequestData): def request_timeout_handler(req_data: _MIoTLanRequestData):
self._pending_requests.pop(req_data.msg_id, None) self._pending_requests.pop(req_data.msg_id, None)
if req_data and req_data.handler: if req_data and req_data.handler:
req_data.handler({ req_data.handler({
@ -984,7 +984,7 @@ class MIoTLan:
req_data.handler_ctx) req_data.handler_ctx)
timer: Optional[asyncio.TimerHandle] = None timer: Optional[asyncio.TimerHandle] = None
request_data = MIoTLanRequestData( request_data = _MIoTLanRequestData(
msg_id=msg_id, msg_id=msg_id,
handler=handler, handler=handler,
handler_ctx=handler_ctx, handler_ctx=handler_ctx,
@ -1032,22 +1032,22 @@ class MIoTLan:
'error': str(err)}, 'error': str(err)},
handler_ctx) handler_ctx)
def __sub_device_state(self, data: MIoTLanSubDeviceData) -> None: def __sub_device_state(self, data: _MIoTLanSubDeviceData) -> None:
self._device_state_sub_map[data.key] = data self._device_state_sub_map[data.key] = data
def __unsub_device_state(self, data: MIoTLanUnsubDeviceData) -> None: def __unsub_device_state(self, data: _MIoTLanUnsubDeviceData) -> None:
self._device_state_sub_map.pop(data.key, None) self._device_state_sub_map.pop(data.key, None)
def __sub_broadcast(self, data: MIoTLanRegisterBroadcastData) -> None: def __sub_broadcast(self, data: _MIoTLanRegisterBroadcastData) -> None:
self._device_msg_matcher[data.key] = data self._device_msg_matcher[data.key] = data
_LOGGER.debug('lan register broadcast, %s', data.key) _LOGGER.debug('lan register broadcast, %s', data.key)
def __unsub_broadcast(self, data: MIoTLanUnregisterBroadcastData) -> None: def __unsub_broadcast(self, data: _MIoTLanUnregisterBroadcastData) -> None:
if self._device_msg_matcher.get(topic=data.key): if self._device_msg_matcher.get(topic=data.key):
del self._device_msg_matcher[data.key] del self._device_msg_matcher[data.key]
_LOGGER.debug('lan unregister broadcast, %s', data.key) _LOGGER.debug('lan unregister broadcast, %s', data.key)
def __get_dev_list(self, data: MIoTLanGetDevListData) -> None: def __get_dev_list(self, data: _MIoTLanGetDevListData) -> None:
dev_list = { dev_list = {
device.did: { device.did: {
'online': device.online, 'online': device.online,
@ -1082,7 +1082,7 @@ class MIoTLan:
_LOGGER.error( _LOGGER.error(
'invalid device token, %s, %s', did, info) 'invalid device token, %s, %s', did, info)
continue continue
self._lan_devices[did] = MIoTLanDevice( self._lan_devices[did] = _MIoTLanDevice(
manager=self, did=did, token=info['token'], manager=self, did=did, token=info['token'],
ip=info.get('ip', None)) ip=info.get('ip', None))
else: else:
@ -1095,7 +1095,7 @@ class MIoTLan:
continue continue
lan_device.on_delete() lan_device.on_delete()
def __on_network_info_chnage(self, data: MIoTLanNetworkUpdateData) -> None: def __on_network_info_chnage(self, data: _MIoTLanNetworkUpdateData) -> None:
if data.status == InterfaceStatus.ADD: if data.status == InterfaceStatus.ADD:
self._available_net_ifs.add(data.if_name) self._available_net_ifs.add(data.if_name)
if data.if_name in self._net_ifs: if data.if_name in self._net_ifs:
@ -1207,7 +1207,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: _MIoTLanDevice | None = 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]
@ -1247,7 +1247,7 @@ 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: _MIoTLanRequestData | None = \
self._pending_requests.pop(msg['id'], None) self._pending_requests.pop(msg['id'], None)
if req: if req:
if req.timeout: if req.timeout:
@ -1274,7 +1274,7 @@ class MIoTLan:
'invalid message, no siid or piid, %s, %s', did, msg) 'invalid message, no siid or piid, %s, %s', did, msg)
continue continue
key = f'{did}/p/{param["siid"]}/{param["piid"]}' key = f'{did}/p/{param["siid"]}/{param["piid"]}'
subs: list[MIoTLanRegisterBroadcastData] = list( subs: list[_MIoTLanRegisterBroadcastData] = list(
self._device_msg_matcher.iter_match(key)) self._device_msg_matcher.iter_match(key))
for sub in subs: for sub in subs:
self._main_loop.call_soon_threadsafe( self._main_loop.call_soon_threadsafe(
@ -1285,7 +1285,7 @@ class MIoTLan:
and 'eiid' in msg['params'] and 'eiid' in msg['params']
): ):
key = f'{did}/e/{msg["params"]["siid"]}/{msg["params"]["eiid"]}' key = f'{did}/e/{msg["params"]["siid"]}/{msg["params"]["eiid"]}'
subs: list[MIoTLanRegisterBroadcastData] = list( subs: list[_MIoTLanRegisterBroadcastData] = list(
self._device_msg_matcher.iter_match(key)) self._device_msg_matcher.iter_match(key))
for sub in subs: for sub in subs:
self._main_loop.call_soon_threadsafe( self._main_loop.call_soon_threadsafe(