XiaoMi/custom_components/xiaomi_home/miot/miot_mips.py

# -*- coding: utf-8 -*-
"""
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 Pub/Sub client.
"""
import asyncio
import json
import logging
import os
import queue
import random
import re
import ssl
import struct
import threading
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum, auto
from typing import Callable, Optional, final

from paho.mqtt.client import (
    MQTT_ERR_SUCCESS,
    MQTT_ERR_UNKNOWN,
    Client,
    MQTTv5)

# pylint: disable=relative-beyond-top-level
from .common import MIoTMatcher
from .const import MIHOME_MQTT_KEEPALIVE
from .miot_error import MIoTErrorCode, MIoTMipsError
from .miot_ev import MIoTEventLoop, TimeoutHandle

_LOGGER = logging.getLogger(__name__)


class MipsMsgTypeOptions(Enum):
    """MIoT Pub/Sub message type."""
    ID = 0
    RET_TOPIC = auto()
    PAYLOAD = auto()
    FROM = auto()
    MAX = auto()


class MipsMessage:
    """MIoT Pub/Sub message."""
    mid: int = 0
    msg_from: str = None
    ret_topic: str = None
    payload: str = None

    @staticmethod
    def unpack(data: bytes):
        mips_msg = MipsMessage()
        data_len = len(data)
        data_start = 0
        data_end = 0
        while data_start < data_len:
            data_end = data_start+5
            unpack_len, unpack_type = struct.unpack(
                '<IB', data[data_start:data_end])
            unpack_data = data[data_end:data_end+unpack_len]
            #  string end with \x00
            match unpack_type:
                case MipsMsgTypeOptions.ID.value:
                    mips_msg.mid = int.from_bytes(
                        unpack_data, byteorder='little')
                case MipsMsgTypeOptions.RET_TOPIC.value:
                    mips_msg.ret_topic = str(
                        unpack_data.strip(b'\x00'), 'utf-8')
                case MipsMsgTypeOptions.PAYLOAD.value:
                    mips_msg.payload = str(unpack_data.strip(b'\x00'), 'utf-8')
                case MipsMsgTypeOptions.FROM.value:
                    mips_msg.msg_from = str(
                        unpack_data.strip(b'\x00'), 'utf-8')
                case _:
                    pass
            data_start = data_end+unpack_len
        return mips_msg

    @staticmethod
    def pack(
        mid: int, payload: str, msg_from: str = None, ret_topic: str = None
    ) -> bytes:
        if mid is None or payload is None:
            raise MIoTMipsError('invalid mid or payload')
        pack_msg: bytes = b''
        # mid
        pack_msg += struct.pack('<IBI', 4, MipsMsgTypeOptions.ID.value, mid)
        # msg_from
        if msg_from:
            pack_len = len(msg_from)
            pack_msg += struct.pack(
                f'<IB{pack_len}sx', pack_len+1,
                MipsMsgTypeOptions.FROM.value, msg_from.encode('utf-8'))
        # ret_topic
        if ret_topic:
            pack_len = len(ret_topic)
            pack_msg += struct.pack(
                f'<IB{pack_len}sx', pack_len+1,
                MipsMsgTypeOptions.RET_TOPIC.value, ret_topic.encode('utf-8'))
        # payload
        pack_len = len(payload)
        pack_msg += struct.pack(
            f'<IB{pack_len}sx', pack_len+1,
            MipsMsgTypeOptions.PAYLOAD.value, payload.encode('utf-8'))
        return pack_msg

    def __str__(self) -> str:
        return f'{self.mid}, {self.msg_from}, {self.ret_topic}, {self.payload}'


class MipsCmdType(Enum):
    """MIoT Pub/Sub command type."""
    CONNECT = 0
    DISCONNECT = auto()
    DEINIT = auto()
    SUB = auto()
    UNSUB = auto()
    CALL_API = auto()
    REG_BROADCAST = auto()
    UNREG_BROADCAST = auto()

    REG_MIPS_STATE = auto()
    UNREG_MIPS_STATE = auto()
    REG_DEVICE_STATE = auto()
    UNREG_DEVICE_STATE = auto()


@dataclass
class MipsCmd:
    """MIoT Pub/Sub command."""
    type_: MipsCmdType
    data: any

    def __init__(self, type_: MipsCmdType, data: any) -> None:
        self.type_ = type_
        self.data = data


@dataclass
class MipsRequest:
    """MIoT Pub/Sub request."""
    mid: int = None
    on_reply: Callable[[str, any], None] = None
    on_reply_ctx: any = None
    timer: TimeoutHandle = None


@dataclass
class MipsRequestData:
    """MIoT Pub/Sub request data."""
    topic: str = None
    payload: str = None
    on_reply: Callable[[str, any], None] = None
    on_reply_ctx: any = None
    timeout_ms: int = None


@dataclass
class MipsSendBroadcastData:
    """MIoT Pub/Sub send broadcast data."""
    topic: str = None
    payload: str = None


@dataclass
class MipsIncomingApiCall:
    """MIoT Pub/Sub incoming API call."""
    mid: int = None
    ret_topic: str = None
    timer: TimeoutHandle = None


@dataclass
class MipsApi:
    """MIoT Pub/Sub API."""
    topic: str = None
    """
    param1: session
    param2: payload
    param3: handler_ctx
    """
    handler: Callable[[MipsIncomingApiCall, str, any], None] = None
    handler_ctx: any = None


class MipsRegApi(MipsApi):
    """.MIoT Pub/Sub register API."""


@dataclass
class MipsReplyData:
    """MIoT Pub/Sub reply data."""
    session: MipsIncomingApiCall = None
    payload: str = None


@dataclass
class MipsBroadcast:
    """MIoT Pub/Sub broadcast."""
    topic: str = None
    """
    param 1: msg topic
    param 2: msg payload
    param 3: handle_ctx
    """
    handler: Callable[[str, str, any], None] = None
    handler_ctx: any = None

    def __str__(self) -> str:
        return f'{self.topic}, {id(self.handler)}, {id(self.handler_ctx)}'


class MipsRegBroadcast(MipsBroadcast):
    """MIoT Pub/Sub register broadcast."""


@dataclass
class MipsState:
    """MIoT Pub/Sub state."""
    key: str = None
    """
    str: key
    bool: mips connect state
    any: ctx
    """
    handler: Callable[[str, bool], asyncio.Future] = None


class MipsRegState(MipsState):
    """MIoT Pub/Sub register state."""


class MIoTDeviceState(Enum):
    """MIoT device state define."""
    DISABLE = 0
    OFFLINE = auto()
    ONLINE = auto()


@dataclass
class MipsDeviceState:
    """MIoT Pub/Sub device state."""
    did: str = None
    """handler
    str: did
    MIoTDeviceState: online/offline/disable
    any: ctx
    """
    handler: Callable[[str, MIoTDeviceState, any], None] = None
    handler_ctx: any = None


class MipsRegDeviceState(MipsDeviceState):
    """MIoT Pub/Sub register device state."""


class MipsClient(ABC):
    """MIoT Pub/Sub client."""
    # pylint: disable=unused-argument
    MQTT_INTERVAL_MS = 1000
    MIPS_QOS: int = 2
    UINT32_MAX: int = 0xFFFFFFFF
    MIPS_RECONNECT_INTERVAL_MIN: int = 30000
    MIPS_RECONNECT_INTERVAL_MAX: int = 600000
    MIPS_SUB_PATCH: int = 300
    MIPS_SUB_INTERVAL: int = 1000
    main_loop: asyncio.AbstractEventLoop
    _logger: logging.Logger
    _client_id: str
    _host: str
    _port: int
    _username: str
    _password: str
    _ca_file: str
    _cert_file: str
    _key_file: str

    _mqtt_logger: logging.Logger
    _mqtt: Client
    _mqtt_fd: int
    _mqtt_timer: TimeoutHandle
    _mqtt_state: bool

    _event_connect: asyncio.Event
    _event_disconnect: asyncio.Event
    _mev: MIoTEventLoop
    _mips_thread: threading.Thread
    _mips_queue: queue.Queue
    _cmd_event_fd: os.eventfd
    _mips_reconnect_tag: bool
    _mips_reconnect_interval: int
    _mips_reconnect_timer: Optional[TimeoutHandle]
    _mips_state_sub_map: dict[str, MipsState]
    _mips_sub_pending_map: dict[str, int]
    _mips_sub_pending_timer: Optional[TimeoutHandle]

    _on_mips_cmd: Callable[[MipsCmd], None]
    _on_mips_message: Callable[[str, bytes], None]
    _on_mips_connect: Callable[[int, dict], None]
    _on_mips_disconnect: Callable[[int, dict], None]

    def __init__(
            self, client_id: str, host: str, port: int,
            username: str = None, password: str = None,
            ca_file: str = None, cert_file: str = None, key_file: str = None,
            loop: Optional[asyncio.AbstractEventLoop] = None
    ) -> None:
        # MUST run with running loop
        self.main_loop = loop or asyncio.get_running_loop()
        self._logger = None
        self._client_id = client_id
        self._host = host
        self._port = port
        self._username = username
        self._password = password
        self._ca_file = ca_file
        self._cert_file = cert_file
        self._key_file = key_file

        self._mqtt_logger = None
        self._mqtt_fd = -1
        self._mqtt_timer = None
        self._mqtt_state = False
        # mqtt init for API_VERSION2,
        # callback_api_version=CallbackAPIVersion.VERSION2,
        self._mqtt = Client(client_id=self._client_id, protocol=MQTTv5)
        self._mqtt.enable_logger(logger=self._mqtt_logger)

        # Mips init
        self._event_connect = asyncio.Event()
        self._event_disconnect = asyncio.Event()
        self._mips_reconnect_tag = False
        self._mips_reconnect_interval = 0
        self._mips_reconnect_timer = None
        self._mips_state_sub_map = {}
        self._mips_sub_pending_map = {}
        self._mips_sub_pending_timer = None
        self._mev = MIoTEventLoop()
        self._mips_queue = queue.Queue()
        self._cmd_event_fd = os.eventfd(0, os.O_NONBLOCK)
        self.mev_set_read_handler(
            self._cmd_event_fd, self.__mips_cmd_read_handler, None)
        self._mips_thread = threading.Thread(target=self.__mips_loop_thread)
        self._mips_thread.daemon = True
        self._mips_thread.name = self._client_id
        self._mips_thread.start()

        self._on_mips_cmd = None
        self._on_mips_message = None
        self._on_mips_connect = None
        self._on_mips_disconnect = None

    @property
    def client_id(self) -> str:
        return self._client_id

    @property
    def host(self) -> str:
        return self._host

    @property
    def port(self) -> int:
        return self._port

    @final
    @property
    def mips_state(self) -> bool:
        """mips connect state.

        Returns:
            bool: True: connected, False: disconnected
        """
        return self._mqtt and self._mqtt.is_connected()

    @final
    def mips_deinit(self) -> None:
        self._mips_send_cmd(type_=MipsCmdType.DEINIT, data=None)
        self._mips_thread.join()
        self._mips_thread = None

        self._logger = None
        self._client_id = None
        self._host = None
        self._port = None
        self._username = None
        self._password = None
        self._ca_file = None
        self._cert_file = None
        self._key_file = None
        self._mqtt_logger = None
        self._mips_state_sub_map = None
        self._mips_sub_pending_map = None
        self._mips_sub_pending_timer = None

        self._event_connect = None
        self._event_disconnect = None

    def update_mqtt_password(self, password: str) -> None:
        self._password = password
        self._mqtt.username_pw_set(
            username=self._username, password=self._password)

    def log_debug(self, msg, *args, **kwargs) -> None:
        if self._logger:
            self._logger.debug(f'{self._client_id}, '+msg, *args, **kwargs)

    def log_info(self, msg, *args, **kwargs) -> None:
        if self._logger:
            self._logger.info(f'{self._client_id}, '+msg, *args, **kwargs)

    def log_error(self, msg, *args, **kwargs) -> None:
        if self._logger:
            self._logger.error(f'{self._client_id}, '+msg, *args, **kwargs)

    def enable_logger(self, logger: Optional[logging.Logger] = None) -> None:
        self._logger = logger

    def enable_mqtt_logger(
        self, logger: Optional[logging.Logger] = None
    ) -> None:
        if logger:
            self._mqtt.enable_logger(logger=logger)
        else:
            self._mqtt.disable_logger()

    @final
    def mips_connect(self) -> None:
        """mips connect."""
        return self._mips_send_cmd(type_=MipsCmdType.CONNECT, data=None)

    @final
    async def mips_connect_async(self) -> None:
        """mips connect async."""
        self._mips_send_cmd(type_=MipsCmdType.CONNECT, data=None)
        return await self._event_connect.wait()

    @final
    def mips_disconnect(self) -> None:
        """mips disconnect."""
        return self._mips_send_cmd(type_=MipsCmdType.DISCONNECT, data=None)

    @final
    async def mips_disconnect_async(self) -> None:
        """mips disconnect async."""
        self._mips_send_cmd(type_=MipsCmdType.DISCONNECT, data=None)
        return await self._event_disconnect.wait()

    @final
    def sub_mips_state(
        self, key: str, handler: Callable[[str, bool], asyncio.Future]
    ) -> bool:
        """Subscribe mips state.
        NOTICE: callback to main loop thread
        """
        if isinstance(key, str) is False or handler is None:
            raise MIoTMipsError('invalid params')
        return self._mips_send_cmd(
            type_=MipsCmdType.REG_MIPS_STATE,
            data=MipsRegState(key=key, handler=handler))

    @final
    def unsub_mips_state(self, key: str) -> bool:
        """Unsubscribe mips state."""
        if isinstance(key, str) is False:
            raise MIoTMipsError('invalid params')
        return self._mips_send_cmd(
            type_=MipsCmdType.UNREG_MIPS_STATE, data=MipsRegState(key=key))

    @final
    def mev_set_timeout(
        self, timeout_ms: int, handler: Callable[[any], None],
        handler_ctx: any = None
    ) -> Optional[TimeoutHandle]:
        """set timeout.
        NOTICE: Internal function, only mips threads are allowed to call
        """
        if self._mev is None:
            return None
        return self._mev.set_timeout(
            timeout_ms=timeout_ms,  handler=handler, handler_ctx=handler_ctx)

    @final
    def mev_clear_timeout(self, handle: TimeoutHandle) -> None:
        """clear timeout.
        NOTICE: Internal function, only mips threads are allowed to call
        """
        if self._mev is None:
            return
        self._mev.clear_timeout(handle)

    @final
    def mev_set_read_handler(
        self, fd: int, handler: Callable[[any], None], handler_ctx: any
    ) -> bool:
        """set read handler.
        NOTICE: Internal function, only mips threads are allowed to call
        """
        if self._mev is None:
            return False
        return self._mev.set_read_handler(
            fd=fd, handler=handler, handler_ctx=handler_ctx)

    @final
    def mev_set_write_handler(
        self, fd: int, handler: Callable[[any], None], handler_ctx: any
    ) -> bool:
        """set write handler.
        NOTICE: Internal function, only mips threads are allowed to call
        """
        if self._mev is None:
            return False
        return self._mev.set_write_handler(
            fd=fd, handler=handler, handler_ctx=handler_ctx)

    @property
    def on_mips_cmd(self) -> Callable[[MipsCmd], None]:
        return self._on_mips_cmd

    @on_mips_cmd.setter
    def on_mips_cmd(self, handler: Callable[[MipsCmd], None]) -> None:
        """MUST set after __init__ done.
        NOTICE thread safe, this function will be called at the **mips** thread
        """
        self._on_mips_cmd = handler

    @property
    def on_mips_message(self) -> Callable[[str, bytes], None]:
        return self._on_mips_message

    @on_mips_message.setter
    def on_mips_message(self, handler: Callable[[str, bytes], None]) -> None:
        """MUST set after __init__ done.
        NOTICE thread safe, this function will be called at the **mips** thread
        """
        self._on_mips_message = handler

    @property
    def on_mips_connect(self) -> Callable[[int, dict], None]:
        return self._on_mips_connect

    @on_mips_connect.setter
    def on_mips_connect(self, handler: Callable[[int, dict], None]) -> None:
        """MUST set after __init__ done.
        NOTICE thread safe, this function will be called at the
        **main loop** thread
        """
        self._on_mips_connect = handler

    @property
    def on_mips_disconnect(self) -> Callable[[int, dict], None]:
        return self._on_mips_disconnect

    @on_mips_disconnect.setter
    def on_mips_disconnect(self, handler: Callable[[int, dict], None]) -> None:
        """MUST set after __init__ done.
        NOTICE thread safe, this function will be called at the
        **main loop** thread
        """
        self._on_mips_disconnect = handler

    @abstractmethod
    def sub_prop(
        self, did: str, handler: Callable[[dict, any], None],
        siid: int = None, piid: int = None, handler_ctx: any = None
    ) -> bool: ...

    @abstractmethod
    def unsub_prop(
        self, did: str, siid: int = None, piid: int = None
    ) -> bool: ...

    @abstractmethod
    def sub_event(
        self, did: str, handler: Callable[[dict, any], None],
        siid: int = None, eiid: int = None, handler_ctx: any = None
    ) -> bool: ...

    @abstractmethod
    def unsub_event(
        self, did: str, siid: int = None, eiid: int = None
    ) -> bool: ...

    @abstractmethod
    async def get_dev_list_async(
        self, payload: str = None, timeout_ms: int = 10000
    ) -> dict[str, dict]: ...

    @abstractmethod
    async def get_prop_async(
        self, did: str, siid: int, piid: int, timeout_ms: int = 10000
    ) -> any: ...

    @abstractmethod
    async def set_prop_async(
        self, did: str, siid: int, piid: int, value: any,
        timeout_ms: int = 10000
    ) -> bool: ...

    @abstractmethod
    async def action_async(
        self, did: str, siid: int, aiid: int, in_list: list,
        timeout_ms: int = 10000
    ) -> tuple[bool, list]: ...

    @final
    def _mips_sub_internal(self, topic: str) -> None:
        """mips subscribe.
        NOTICE: Internal function, only mips threads are allowed to call
        """
        self.__thread_check()
        if not self._mqtt or not self._mqtt.is_connected():
            return
        try:
            if topic not in self._mips_sub_pending_map:
                self._mips_sub_pending_map[topic] = 0
            if not self._mips_sub_pending_timer:
                self._mips_sub_pending_timer = self.mev_set_timeout(
                    10, self.__mips_sub_internal_pending_handler, topic)
        except Exception as err:  # pylint: disable=broad-exception-caught
            # Catch all exception
            self.log_error(f'mips sub internal error, {topic}. {err}')

    @final
    def _mips_unsub_internal(self, topic: str) -> None:
        """mips unsubscribe.
        NOTICE: Internal function, only mips threads are allowed to call
        """
        self.__thread_check()
        if not self._mqtt or not self._mqtt.is_connected():
            return
        try:
            result, mid = self._mqtt.unsubscribe(topic=topic)
            if result == MQTT_ERR_SUCCESS:
                self.log_debug(
                    f'mips unsub internal success, {result}, {mid}, {topic}')
                return
            self.log_error(
                f'mips unsub internal error, {result}, {mid}, {topic}')
        except Exception as err:  # pylint: disable=broad-exception-caught
            # Catch all exception
            self.log_error(f'mips unsub internal error, {topic}, {err}')

    @final
    def _mips_publish_internal(
        self, topic: str, payload: str | bytes,
        wait_for_publish: bool = False, timeout_ms: int = 10000
    ) -> bool:
        """mips publish message.
        NOTICE: Internal function, only mips threads are allowed to call

        """
        self.__thread_check()
        if not self._mqtt or not self._mqtt.is_connected():
            return False
        try:
            handle = self._mqtt.publish(
                topic=topic, payload=payload, qos=self.MIPS_QOS)
            # self.log_debug(f'_mips_publish_internal, {topic}, {payload}')
            if wait_for_publish is True:
                handle.wait_for_publish(timeout_ms/1000.0)
            return True
        except Exception as err:  # pylint: disable=broad-exception-caught
            # Catch other exception
            self.log_error(f'mips publish internal error, {err}')
        return False

    @final
    def _mips_send_cmd(self, type_: MipsCmdType, data: any) -> bool:
        if self._mips_queue is None or self._cmd_event_fd is None:
            raise MIoTMipsError('send mips cmd disable')
        # Put data to queue
        self._mips_queue.put(MipsCmd(type_=type_, data=data))
        # Write event fd
        os.eventfd_write(self._cmd_event_fd, 1)
        # self.log_debug(f'send mips cmd, {type}, {data}')
        return True

    def __thread_check(self) -> None:
        if threading.current_thread() is not self._mips_thread:
            raise MIoTMipsError('illegal call')

    def __mips_cmd_read_handler(self, ctx: any) -> None:
        fd_value = os.eventfd_read(self._cmd_event_fd)
        if fd_value == 0:
            return
        while self._mips_queue.empty() is False:
            mips_cmd: MipsCmd = self._mips_queue.get(block=False)
            if mips_cmd.type_ == MipsCmdType.CONNECT:
                self._mips_reconnect_tag = True
                self.__mips_try_reconnect(immediately=True)
            elif mips_cmd.type_ == MipsCmdType.DISCONNECT:
                self._mips_reconnect_tag = False
                self.__mips_disconnect()
            elif mips_cmd.type_ == MipsCmdType.DEINIT:
                self.log_info('mips client recv deinit cmd')
                self.__mips_disconnect()
                # Close cmd event fd
                if self._cmd_event_fd:
                    self.mev_set_read_handler(
                        self._cmd_event_fd, None, None)
                    os.close(self._cmd_event_fd)
                    self._cmd_event_fd = None
                if self._mips_queue:
                    self._mips_queue = None
                # ev loop stop
                if self._mev:
                    self._mev.loop_stop()
                    self._mev = None
                break
            elif mips_cmd.type_ == MipsCmdType.REG_MIPS_STATE:
                state: MipsState = mips_cmd.data
                self._mips_state_sub_map[state.key] = state
                self.log_debug(f'mips register mips state, {state.key}')
            elif mips_cmd.type_ == MipsCmdType.UNREG_MIPS_STATE:
                state: MipsState = mips_cmd.data
                del self._mips_state_sub_map[state.key]
                self.log_debug(f'mips unregister mips state, {state.key}')
            else:
                if self._on_mips_cmd:
                    self._on_mips_cmd(mips_cmd=mips_cmd)

    def __mqtt_read_handler(self, ctx: any) -> None:
        self.__mqtt_loop_handler(ctx=ctx)

    def __mqtt_write_handler(self, ctx: any) -> None:
        self.mev_set_write_handler(self._mqtt_fd, None, None)
        self.__mqtt_loop_handler(ctx=ctx)

    def __mqtt_timer_handler(self, ctx: any) -> None:
        self.__mqtt_loop_handler(ctx=ctx)
        if self._mqtt:
            self._mqtt_timer = self.mev_set_timeout(
                self.MQTT_INTERVAL_MS, self.__mqtt_timer_handler, None)

    def __mqtt_loop_handler(self, ctx: any) -> None:
        try:
            if self._mqtt:
                self._mqtt.loop_read()
            if self._mqtt:
                self._mqtt.loop_write()
            if self._mqtt:
                self._mqtt.loop_misc()
            if self._mqtt and self._mqtt.want_write():
                self.mev_set_write_handler(
                    self._mqtt_fd, self.__mqtt_write_handler, None)
        except Exception as err:  # pylint: disable=broad-exception-caught
            # Catch all exception
            self.log_error(f'__mqtt_loop_handler, {err}')
            raise err

    def __mips_loop_thread(self) -> None:
        self.log_info('mips_loop_thread start')
        # Set mqtt config
        if self._username:
            self._mqtt.username_pw_set(
                username=self._username, password=self._password)
        if (
            self._ca_file
            and self._cert_file
            and self._key_file
        ):
            self._mqtt.tls_set(
                tls_version=ssl.PROTOCOL_TLS_CLIENT,
                ca_certs=self._ca_file,
                certfile=self._cert_file,
                keyfile=self._key_file)
        else:
            self._mqtt.tls_set(tls_version=ssl.PROTOCOL_TLS_CLIENT)
        self._mqtt.tls_insecure_set(True)
        self._mqtt.on_connect = self.__on_connect
        self._mqtt.on_connect_fail = self.__on_connect_failed
        self._mqtt.on_disconnect = self.__on_disconnect
        self._mqtt.on_message = self.__on_message
        # Run event loop
        self._mev.loop_forever()
        self.log_info('mips_loop_thread exit!')

    def __on_connect(self, client, user_data, flags, rc, props) -> None:
        if not self._mqtt.is_connected():
            return
        self.log_info(f'mips connect, {flags}, {rc}, {props}')
        self._mqtt_state = True
        if self._on_mips_connect:
            self.mev_set_timeout(
                timeout_ms=0,
                handler=lambda ctx:
                    self._on_mips_connect(rc, props))
        for item in self._mips_state_sub_map.values():
            if item.handler is None:
                continue
            self.main_loop.call_soon_threadsafe(
                self.main_loop.create_task,
                item.handler(item.key, True))
        # Resolve future
        self._event_connect.set()
        self._event_disconnect.clear()

    def __on_connect_failed(self, client, user_data, flags, rc) -> None:
        self.log_error(f'mips connect failed, {flags}, {rc}')
        # Try to reconnect
        self.__mips_try_reconnect()

    def __on_disconnect(self,  client, user_data, rc, props) -> None:
        if self._mqtt_state:
            self.log_error(f'mips disconnect, {rc}, {props}')
            self._mqtt_state = False
            if self._mqtt_timer:
                self.mev_clear_timeout(self._mqtt_timer)
                self._mqtt_timer = None
            if self._mqtt_fd != -1:
                self.mev_set_read_handler(self._mqtt_fd, None, None)
                self.mev_set_write_handler(self._mqtt_fd, None, None)
                self._mqtt_fd = -1
            # Clear retry sub
            if self._mips_sub_pending_timer:
                self.mev_clear_timeout(self._mips_sub_pending_timer)
                self._mips_sub_pending_timer = None
            self._mips_sub_pending_map = {}
            if self._on_mips_disconnect:
                self.mev_set_timeout(
                    timeout_ms=0,
                    handler=lambda ctx:
                        self._on_mips_disconnect(rc, props))
            # Call state sub handler
            for item in self._mips_state_sub_map.values():
                if item.handler is None:
                    continue
                self.main_loop.call_soon_threadsafe(
                    self.main_loop.create_task,
                    item.handler(item.key, False))

        # Try to reconnect
        self.__mips_try_reconnect()
        # Set event
        self._event_disconnect.set()
        self._event_connect.clear()

    def __on_message(self, client, user_data, msg) -> None:
        self._on_mips_message(topic=msg.topic, payload=msg.payload)

    def __mips_try_reconnect(self, immediately: bool = False) -> None:
        if self._mips_reconnect_timer:
            self.mev_clear_timeout(self._mips_reconnect_timer)
            self._mips_reconnect_timer = None
        if not self._mips_reconnect_tag:
            return
        interval: int = 0
        if not immediately:
            interval = self.__get_next_reconnect_time()
            self.log_error(
                'mips try reconnect after %sms', interval)
        self._mips_reconnect_timer = self.mev_set_timeout(
            interval, self.__mips_connect, None)

    def __mips_sub_internal_pending_handler(self, ctx: any) -> None:
        subbed_count = 1
        for topic in list(self._mips_sub_pending_map.keys()):
            if subbed_count > self.MIPS_SUB_PATCH:
                break
            count = self._mips_sub_pending_map[topic]
            if count > 3:
                self._mips_sub_pending_map.pop(topic)
                self.log_error(f'retry mips sub internal error, {topic}')
                continue
            subbed_count += 1
            result, mid = self._mqtt.subscribe(topic, qos=self.MIPS_QOS)
            if result == MQTT_ERR_SUCCESS:
                self._mips_sub_pending_map.pop(topic)
                self.log_debug(f'mips sub internal success, {topic}')
                continue
            self._mips_sub_pending_map[topic] = count+1
            self.log_error(
                f'retry mips sub internal, {count}, {topic}, {result}, {mid}')

        if len(self._mips_sub_pending_map):
            self._mips_sub_pending_timer = self.mev_set_timeout(
                self.MIPS_SUB_INTERVAL,
                self.__mips_sub_internal_pending_handler, None)
        else:
            self._mips_sub_pending_timer = None

    def __mips_connect(self, ctx: any = None) -> None:
        result = MQTT_ERR_UNKNOWN
        if self._mips_reconnect_timer:
            self.mev_clear_timeout(self._mips_reconnect_timer)
            self._mips_reconnect_timer = None
        try:
            # Try clean mqtt fd before mqtt connect
            if self._mqtt_timer:
                self.mev_clear_timeout(self._mqtt_timer)
                self._mqtt_timer = None
            if self._mqtt_fd != -1:
                self.mev_set_read_handler(self._mqtt_fd, None, None)
                self.mev_set_write_handler(self._mqtt_fd, None, None)
                self._mqtt_fd = -1
            result = self._mqtt.connect(
                host=self._host, port=self._port,
                clean_start=True, keepalive=MIHOME_MQTT_KEEPALIVE)
            self.log_info(f'__mips_connect success, {result}')
        except (TimeoutError, OSError) as error:
            self.log_error('__mips_connect, connect error, %s', error)

        if result == MQTT_ERR_SUCCESS:
            self._mqtt_fd = self._mqtt.socket()
            self.log_debug(f'__mips_connect, _mqtt_fd, {self._mqtt_fd}')
            self.mev_set_read_handler(
                self._mqtt_fd, self.__mqtt_read_handler, None)
            if self._mqtt.want_write():
                self.mev_set_write_handler(
                    self._mqtt_fd, self.__mqtt_write_handler, None)
            self._mqtt_timer = self.mev_set_timeout(
                self.MQTT_INTERVAL_MS, self.__mqtt_timer_handler, None)
        else:
            self.log_error(f'__mips_connect error result, {result}')
            self.__mips_try_reconnect()

    def __mips_disconnect(self) -> None:
        if self._mips_reconnect_timer:
            self.mev_clear_timeout(self._mips_reconnect_timer)
            self._mips_reconnect_timer = None
        if self._mqtt_timer:
            self.mev_clear_timeout(self._mqtt_timer)
            self._mqtt_timer = None
        if self._mqtt_fd != -1:
            self.mev_set_read_handler(self._mqtt_fd, None, None)
            self.mev_set_write_handler(self._mqtt_fd, None, None)
            self._mqtt_fd = -1
        self._mqtt.disconnect()

    def __get_next_reconnect_time(self) -> int:
        if self._mips_reconnect_interval == 0:
            self._mips_reconnect_interval = self.MIPS_RECONNECT_INTERVAL_MIN
        else:
            self._mips_reconnect_interval = min(
                self._mips_reconnect_interval*2,
                self.MIPS_RECONNECT_INTERVAL_MAX)
        return self._mips_reconnect_interval


class MipsCloudClient(MipsClient):
    """MIoT Pub/Sub Cloud Client."""
    # pylint: disable=unused-argument
    # pylint: disable=inconsistent-quotes
    _msg_matcher: MIoTMatcher

    def __init__(
            self, uuid: str, cloud_server: str, app_id: str,
            token: str, port: int = 8883,
            loop: Optional[asyncio.AbstractEventLoop] = None
    ) -> None:
        self._msg_matcher = MIoTMatcher()
        super().__init__(
            client_id=f'ha.{uuid}', host=f'{cloud_server}-ha.mqtt.io.mi.com',
            port=port, username=app_id, password=token, loop=loop)

        self.on_mips_cmd = self.__on_mips_cmd_handler
        self.on_mips_message = self.__on_mips_message_handler
        self.on_mips_connect = self.__on_mips_connect_handler
        self.on_mips_disconnect = self.__on_mips_disconnect_handler

    def deinit(self) -> None:
        self.mips_deinit()
        self._msg_matcher = None
        self.on_mips_cmd = None
        self.on_mips_message = None
        self.on_mips_connect = None

    @final
    def connect(self) -> None:
        self.mips_connect()

    @final
    async def connect_async(self) -> None:
        await self.mips_connect_async()

    @final
    def disconnect(self) -> None:
        self.mips_disconnect()
        self._msg_matcher = MIoTMatcher()

    @final
    async def disconnect_async(self) -> None:
        await self.mips_disconnect_async()
        self._msg_matcher = MIoTMatcher()

    def update_access_token(self, access_token: str) -> bool:
        if not isinstance(access_token, str):
            raise MIoTMipsError('invalid token')
        return self.update_mqtt_password(password=access_token)

    @final
    def sub_prop(
        self, did: str, handler: Callable[[dict, any], None],
        siid: int = None, piid: int = None, handler_ctx: any = None
    ) -> bool:
        if not isinstance(did, str) or handler is None:
            raise MIoTMipsError('invalid params')

        topic: str = (
            f'device/{did}/up/properties_changed/'
            f'{"#" if siid is None or piid is None else f"{siid}/{piid}"}')

        def on_prop_msg(topic: str, payload: str, ctx: any) -> bool:
            try:
                msg: dict = json.loads(payload)
            except json.JSONDecodeError:
                self.log_error(
                    f'on_prop_msg, invalid msg, {topic}, {payload}')
                return
            if (
                not isinstance(msg.get('params', None), dict)
                or 'siid' not in msg['params']
                or 'piid' not in msg['params']
                or 'value' not in msg['params']
            ):
                self.log_error(
                    f'on_prop_msg, invalid msg, {topic}, {payload}')
                return
            if handler:
                self.log_debug('on properties_changed, %s', payload)
                handler(msg['params'], ctx)
        return self.__reg_broadcast(
            topic=topic, handler=on_prop_msg, handler_ctx=handler_ctx)

    @final
    def unsub_prop(self, did: str, siid: int = None, piid: int = None) -> bool:
        if not isinstance(did, str):
            raise MIoTMipsError('invalid params')
        topic: str = (
            f'device/{did}/up/properties_changed/'
            f'{"#" if siid is None or piid is None else f"{siid}/{piid}"}')
        return self.__unreg_broadcast(topic=topic)

    @final
    def sub_event(
        self, did: str, handler: Callable[[dict, any], None],
        siid: int = None, eiid: int = None, handler_ctx: any = None
    ) -> bool:
        if not isinstance(did, str) or handler is None:
            raise MIoTMipsError('invalid params')
        # Spelling error: event_occured
        topic: str = (
            f'device/{did}/up/event_occured/'
            f'{"#" if siid is None or eiid is None else f"{siid}/{eiid}"}')

        def on_event_msg(topic: str, payload: str, ctx: any) -> bool:
            try:
                msg: dict = json.loads(payload)
            except json.JSONDecodeError:
                self.log_error(
                    f'on_event_msg, invalid msg, {topic}, {payload}')
                return
            if (
                not isinstance(msg.get('params', None), dict)
                or 'siid' not in msg['params']
                or 'eiid' not in msg['params']
                or 'arguments' not in msg['params']
            ):
                self.log_error(
                    f'on_event_msg, invalid msg, {topic}, {payload}')
                return
            if handler:
                self.log_debug('on on_event_msg, %s', payload)
                msg['params']['from'] = 'cloud'
                handler(msg['params'], ctx)
        return self.__reg_broadcast(
            topic=topic, handler=on_event_msg, handler_ctx=handler_ctx)

    @final
    def unsub_event(self, did: str, siid: int = None, eiid: int = None) -> bool:
        if not isinstance(did, str):
            raise MIoTMipsError('invalid params')
        # Spelling error: event_occured
        topic: str = (
            f'device/{did}/up/event_occured/'
            f'{"#" if siid is None or eiid is None else f"{siid}/{eiid}"}')
        return self.__unreg_broadcast(topic=topic)

    @final
    def sub_device_state(
        self, did: str, handler: Callable[[str, MIoTDeviceState, any], None],
        handler_ctx: any = None
    ) -> bool:
        """subscribe online state."""
        if not isinstance(did, str) or handler is None:
            raise MIoTMipsError('invalid params')
        topic: str = f'device/{did}/state/#'

        def on_state_msg(topic: str, payload: str, ctx: any) -> None:
            msg: dict = json.loads(payload)
            # {"device_id":"xxxx","device_name":"米家智能插座3   ","event":"online",
            # "model": "cuco.plug.v3","timestamp":1709001070828,"uid":xxxx}
            if msg is None or 'device_id' not in msg or 'event' not in msg:
                self.log_error(f'on_state_msg, recv unknown msg, {payload}')
                return
            if msg['device_id'] != did:
                self.log_error(
                    f'on_state_msg, err msg, {did}!={msg["device_id"]}')
                return
            if handler:
                self.log_debug('cloud, device state changed, %s', payload)
                handler(
                    did, MIoTDeviceState.ONLINE if msg['event'] == 'online'
                    else MIoTDeviceState.OFFLINE, ctx)
        return self.__reg_broadcast(
            topic=topic, handler=on_state_msg, handler_ctx=handler_ctx)

    @final
    def unsub_device_state(self, did: str) -> bool:
        if not isinstance(did, str):
            raise MIoTMipsError('invalid params')
        topic: str = f'device/{did}/state/#'
        return self.__unreg_broadcast(topic=topic)

    async def get_dev_list_async(
        self, payload: str = None, timeout_ms: int = 10000
    ) -> dict[str, dict]:
        raise NotImplementedError('please call in http client')

    async def get_prop_async(
        self, did: str, siid: int, piid: int,  timeout_ms: int = 10000
    ) -> any:
        raise NotImplementedError('please call in http client')

    async def set_prop_async(
        self, did: str, siid: int, piid: int, value: any,
        timeout_ms: int = 10000
    ) -> bool:
        raise NotImplementedError('please call in http client')

    async def action_async(
        self, did: str, siid: int, aiid: int, in_list: list,
        timeout_ms: int = 10000
    ) -> tuple[bool, list]:
        raise NotImplementedError('please call in http client')

    def __on_mips_cmd_handler(self, mips_cmd: MipsCmd) -> None:
        """
        NOTICE thread safe, this function will be called at the **mips** thread
        """
        if mips_cmd.type_ == MipsCmdType.REG_BROADCAST:
            reg_bc: MipsRegBroadcast = mips_cmd.data
            if not self._msg_matcher.get(topic=reg_bc.topic):
                sub_bc: MipsBroadcast = MipsBroadcast(
                    topic=reg_bc.topic, handler=reg_bc.handler,
                    handler_ctx=reg_bc.handler_ctx)
                self._msg_matcher[reg_bc.topic] = sub_bc
                self._mips_sub_internal(topic=reg_bc.topic)
            else:
                self.log_debug(f'mips cloud re-reg broadcast, {reg_bc.topic}')
        elif mips_cmd.type_ == MipsCmdType.UNREG_BROADCAST:
            unreg_bc: MipsRegBroadcast = mips_cmd.data
            if self._msg_matcher.get(topic=unreg_bc.topic):
                del self._msg_matcher[unreg_bc.topic]
                self._mips_unsub_internal(topic=unreg_bc.topic)

    def __reg_broadcast(
        self, topic: str, handler: Callable[[str, str, any], None],
        handler_ctx: any = None
    ) -> bool:
        return self._mips_send_cmd(
            type_=MipsCmdType.REG_BROADCAST,
            data=MipsRegBroadcast(
                topic=topic, handler=handler, handler_ctx=handler_ctx))

    def __unreg_broadcast(self, topic: str) -> bool:
        return self._mips_send_cmd(
            type_=MipsCmdType.UNREG_BROADCAST,
            data=MipsRegBroadcast(topic=topic))

    def __on_mips_connect_handler(self, rc, props) -> None:
        """sub topic."""
        for topic, _ in list(
                self._msg_matcher.iter_all_nodes()):
            self._mips_sub_internal(topic=topic)

    def __on_mips_disconnect_handler(self, rc, props) -> None:
        """unsub topic."""
        pass

    def __on_mips_message_handler(self, topic: str, payload) -> None:
        """
        NOTICE thread safe, this function will be called at the **mips** thread
        """
        # broadcast
        bc_list: list[MipsBroadcast] = list(
            self._msg_matcher.iter_match(topic))
        if not bc_list:
            return
        # self.log_debug(f"on broadcast, {topic}, {payload}")
        for item in bc_list or []:
            if item.handler is None:
                continue
            # NOTICE: call threadsafe
            self.main_loop.call_soon_threadsafe(
                item.handler, topic, payload, item.handler_ctx)


class MipsLocalClient(MipsClient):
    """MIoT Pub/Sub Local Client."""
    # pylint: disable=unused-argument
    # pylint: disable=inconsistent-quotes
    MIPS_RECONNECT_INTERVAL_MIN: int = 6000
    MIPS_RECONNECT_INTERVAL_MAX: int = 60000
    MIPS_SUB_PATCH: int = 1000
    MIPS_SUB_INTERVAL: int = 100
    _did: str
    _group_id: str
    _home_name: str
    _mips_seed_id: int
    _reply_topic: str
    _dev_list_change_topic: str
    _request_map: dict[str, MipsRequest]
    _msg_matcher: MIoTMatcher
    _device_state_sub_map: dict[str, MipsDeviceState]
    _get_prop_queue: dict[str, list]
    _get_prop_timer: asyncio.TimerHandle
    _on_dev_list_changed: Callable[[any, list[str]], asyncio.Future]

    def __init__(
        self, did: str, host: str, group_id: str,
        ca_file: str, cert_file: str, key_file: str,
        port: int = 8883, home_name: str = '',
        loop: Optional[asyncio.AbstractEventLoop] = None
    ) -> None:
        self._did = did
        self._group_id = group_id
        self._home_name = home_name
        self._mips_seed_id = random.randint(0, self.UINT32_MAX)
        self._reply_topic = f'{did}/reply'
        self._dev_list_change_topic = f'{did}/appMsg/devListChange'
        self._request_map = {}
        self._msg_matcher = MIoTMatcher()
        self._device_state_sub_map = {}
        self._get_prop_queue = {}
        self._get_prop_timer = None
        self._on_dev_list_changed = None

        super().__init__(
            client_id=did, host=host, port=port,
            ca_file=ca_file, cert_file=cert_file, key_file=key_file, loop=loop)
        # MIPS local thread name use group_id
        self._mips_thread.name = self._group_id

        self.on_mips_cmd = self.__on_mips_cmd_handler
        self.on_mips_message = self.__on_mips_message_handler
        self.on_mips_connect = self.__on_mips_connect_handler

    @property
    def group_id(self) -> str:
        return self._group_id

    def deinit(self) -> None:
        self.mips_deinit()
        self._did = None
        self._mips_seed_id = None
        self._reply_topic = None
        self._dev_list_change_topic = None
        self._request_map = None
        self._msg_matcher = None
        self._device_state_sub_map = None
        self._get_prop_queue = None
        self._get_prop_timer = None
        self._on_dev_list_changed = None

        self.on_mips_cmd = None
        self.on_mips_message = None
        self.on_mips_connect = None

    def log_debug(self, msg, *args, **kwargs) -> None:
        if self._logger:
            self._logger.debug(f'{self._home_name}, '+msg, *args, **kwargs)

    def log_info(self, msg, *args, **kwargs) -> None:
        if self._logger:
            self._logger.info(f'{self._home_name}, '+msg, *args, **kwargs)

    def log_error(self, msg, *args, **kwargs) -> None:
        if self._logger:
            self._logger.error(f'{self._home_name}, '+msg, *args, **kwargs)

    @final
    def connect(self) -> None:
        self.mips_connect()

    @final
    async def connect_async(self) -> None:
        await self.mips_connect_async()

    @final
    def disconnect(self) -> None:
        self.mips_disconnect()
        self._request_map = {}
        self._msg_matcher = MIoTMatcher()
        self._device_state_sub_map = {}

    @final
    async def disconnect_async(self) -> None:
        await self.mips_disconnect_async()
        self._request_map = {}
        self._msg_matcher = MIoTMatcher()
        self._device_state_sub_map = {}

    @final
    def sub_prop(
        self, did: str, handler: Callable[[dict, any], None],
        siid: int = None, piid: int = None, handler_ctx: any = None
    ) -> bool:
        topic: str = (
            f'appMsg/notify/iot/{did}/property/'
            f'{"#" if siid is None or piid is None else f"{siid}.{piid}"}')

        def on_prop_msg(topic: str, payload: str, ctx: any):
            msg: dict = json.loads(payload)
            if (
                msg is None
                or 'did' not in msg
                or 'siid' not in msg
                or 'piid' not in msg
                or 'value' not in msg
            ):
                # self.log_error(f'on_prop_msg, recv unknown msg, {payload}')
                return
            if handler:
                self.log_debug('local, on properties_changed, %s', payload)
                handler(msg, ctx)
        return self.__reg_broadcast(
            topic=topic, handler=on_prop_msg, handler_ctx=handler_ctx)

    @final
    def unsub_prop(self, did: str, siid: int = None, piid: int = None) -> bool:
        topic: str = (
            f'appMsg/notify/iot/{did}/property/'
            f'{"#" if siid is None or piid is None else f"{siid}.{piid}"}')
        return self.__unreg_broadcast(topic=topic)

    @final
    def sub_event(
        self, did: str, handler: Callable[[dict, any], None],
        siid: int = None, eiid: int = None, handler_ctx: any = None
    ) -> bool:
        topic: str = (
            f'appMsg/notify/iot/{did}/event/'
            f'{"#" if siid is None or eiid is None else f"{siid}.{eiid}"}')

        def on_event_msg(topic: str, payload: str, ctx: any):
            msg: dict = json.loads(payload)
            if (
                msg is None
                or 'did' not in msg
                or 'siid' not in msg
                or 'eiid' not in msg
                or 'arguments' not in msg
            ):
                # self.log_error(f'on_event_msg, recv unknown msg, {payload}')
                return
            if handler:
                self.log_debug('local, on event_occurred, %s', payload)
                handler(msg, ctx)
        return self.__reg_broadcast(
            topic=topic, handler=on_event_msg, handler_ctx=handler_ctx)

    @final
    def unsub_event(self, did: str, siid: int = None, eiid: int = None) -> bool:
        topic: str = (
            f'appMsg/notify/iot/{did}/event/'
            f'{"#" if siid is None or eiid is None else f"{siid}.{eiid}"}')
        return self.__unreg_broadcast(topic=topic)

    @final
    async def get_prop_safe_async(
        self, did: str, siid: int, piid: int, timeout_ms: int = 10000
    ) -> any:
        self._get_prop_queue.setdefault(did, [])
        fut: asyncio.Future = self.main_loop.create_future()
        self._get_prop_queue[did].append({
            'param': json.dumps({
                'did': did,
                'siid': siid,
                'piid': piid
            }),
            'fut': fut,
            'timeout_ms': timeout_ms
        })
        if self._get_prop_timer is None:
            self._get_prop_timer = self.main_loop.create_task(
                self.__get_prop_timer_handle())
        return await fut

    @final
    async def get_prop_async(
        self, did: str, siid: int, piid: int, timeout_ms: int = 10000
    ) -> any:
        result_obj = await self.__request_async(
            topic='proxy/get',
            payload=json.dumps({
                'did': did,
                'siid': siid,
                'piid': piid
            }),
            timeout_ms=timeout_ms)
        if not isinstance(result_obj, dict) or 'value' not in result_obj:
            return None
        return result_obj['value']

    @final
    async def set_prop_async(
        self, did: str, siid: int, piid: int, value: any,
        timeout_ms: int = 10000
    ) -> dict:
        payload_obj: dict = {
            'did': did,
            'rpc': {
                'id': self.__gen_mips_id,
                'method': 'set_properties',
                'params': [{
                    'did': did,
                    'siid': siid,
                    'piid': piid,
                    'value': value
                }]
            }
        }
        result_obj = await self.__request_async(
            topic='proxy/rpcReq',
            payload=json.dumps(payload_obj),
            timeout_ms=timeout_ms)
        if result_obj:
            if (
                'result' in result_obj
                and len(result_obj['result']) == 1
                and 'did' in result_obj['result'][0]
                and result_obj['result'][0]['did'] == did
                and 'code' in result_obj['result'][0]
            ):
                return result_obj['result'][0]
            if 'error' in result_obj:
                return result_obj['error']
        return {
            'code': MIoTErrorCode.CODE_INTERNAL_ERROR.value,
            'message': 'Invalid result'}

    @final
    async def action_async(
        self, did: str, siid: int, aiid: int, in_list: list,
        timeout_ms: int = 10000
    ) -> dict:
        payload_obj: dict = {
            'did': did,
            'rpc': {
                'id': self.__gen_mips_id,
                'method': 'action',
                'params': {
                    'did': did,
                    'siid': siid,
                    'aiid': aiid,
                    'in': in_list
                }
            }
        }
        result_obj = await self.__request_async(
            topic='proxy/rpcReq', payload=json.dumps(payload_obj),
            timeout_ms=timeout_ms)
        if result_obj:
            if 'result' in result_obj and 'code' in result_obj['result']:
                return result_obj['result']
            if 'error' in result_obj:
                return result_obj['error']
        return {
            'code': MIoTErrorCode.CODE_INTERNAL_ERROR.value,
            'message': 'Invalid result'}

    @final
    async def get_dev_list_async(
        self, payload: str = None, timeout_ms: int = 10000
    ) -> dict[str, dict]:
        result_obj = await self.__request_async(
            topic='proxy/getDevList', payload=payload or '{}',
            timeout_ms=timeout_ms)
        if not result_obj or 'devList' not in result_obj:
            return None
        device_list = {}
        for did, info in result_obj['devList'].items():
            name: str = info.get('name', None)
            urn: str = info.get('urn', None)
            model: str = info.get('model', None)
            if name is None or urn is None or model is None:
                self.log_error(f'invalid device info, {did}, {info}')
                continue
            device_list[did] = {
                'did': did,
                'name': name,
                'urn': urn,
                'model': model,
                'online': info.get('online', False),
                'icon': info.get('icon', None),
                'fw_version': None,
                'home_id': '',
                'home_name': '',
                'room_id': info.get('roomId', ''),
                'room_name': info.get('roomName', ''),
                'specv2_access': info.get('specV2Access', False),
                'push_available': info.get('pushAvailable', False),
                'manufacturer': model.split('.')[0],
            }
        return device_list

    @final
    async def get_action_group_list_async(
        self, timeout_ms: int = 10000
    ) -> list[str]:
        result_obj = await self.__request_async(
            topic='proxy/getMijiaActionGroupList',
            payload='{}',
            timeout_ms=timeout_ms)
        if not result_obj or 'result' not in result_obj:
            return None
        return result_obj['result']

    @final
    async def exec_action_group_list_async(
        self, ag_id: str, timeout_ms: int = 10000
    ) -> dict:
        result_obj = await self.__request_async(
            topic='proxy/execMijiaActionGroup',
            payload=f'{{"id":"{ag_id}"}}',
            timeout_ms=timeout_ms)
        if result_obj:
            if 'result' in result_obj:
                return result_obj['result']
            if 'error' in result_obj:
                return result_obj['error']
        return {
            'code': MIoTErrorCode.CODE_MIPS_INVALID_RESULT.value,
            'message': 'invalid result'}

    @final
    @property
    def on_dev_list_changed(self) -> Callable[[any, list[str]], asyncio.Future]:
        return self._on_dev_list_changed

    @final
    @on_dev_list_changed.setter
    def on_dev_list_changed(
        self, func: Callable[[any, list[str]], asyncio.Future]
    ) -> None:
        """run in main loop."""
        self._on_dev_list_changed = func

    @final
    def __on_mips_cmd_handler(self, mips_cmd: MipsCmd) -> None:
        if mips_cmd.type_ == MipsCmdType.CALL_API:
            req_data: MipsRequestData = mips_cmd.data
            req = MipsRequest()
            req.mid = self.__gen_mips_id
            req.on_reply = req_data.on_reply
            req.on_reply_ctx = req_data.on_reply_ctx
            pub_topic: str = f'master/{req_data.topic}'
            result = self.__mips_publish(
                topic=pub_topic, payload=req_data.payload, mid=req.mid,
                ret_topic=self._reply_topic)
            self.log_debug(
                f'mips local call api, {result}, {req.mid}, {pub_topic}, '
                f'{req_data.payload}')

            def on_request_timeout(req: MipsRequest):
                self.log_error(
                    f'on mips request timeout, {req.mid}, {pub_topic}'
                    f', {req_data.payload}')
                self._request_map.pop(str(req.mid), None)
                req.on_reply(
                    '{"error":{"code":-10006, "message":"timeout"}}',
                    req.on_reply_ctx)
            req.timer = self.mev_set_timeout(
                req_data.timeout_ms, on_request_timeout, req)
            self._request_map[str(req.mid)] = req
        elif mips_cmd.type_ == MipsCmdType.REG_BROADCAST:
            reg_bc: MipsRegBroadcast = mips_cmd.data
            sub_topic: str = f'{self._did}/{reg_bc.topic}'
            if not self._msg_matcher.get(sub_topic):
                sub_bc: MipsBroadcast = MipsBroadcast(
                    topic=sub_topic, handler=reg_bc.handler,
                    handler_ctx=reg_bc.handler_ctx)
                self._msg_matcher[sub_topic] = sub_bc
                self._mips_sub_internal(topic=f'master/{reg_bc.topic}')
            else:
                self.log_debug(f'mips re-reg broadcast, {sub_topic}')
        elif mips_cmd.type_ == MipsCmdType.UNREG_BROADCAST:
            unreg_bc: MipsRegBroadcast = mips_cmd.data
            # Central hub gateway needs to add prefix
            unsub_topic: str = f'{self._did}/{unreg_bc.topic}'
            if self._msg_matcher.get(unsub_topic):
                del self._msg_matcher[unsub_topic]
                self._mips_unsub_internal(
                    topic=re.sub(f'^{self._did}', 'master', unsub_topic))
        elif mips_cmd.type_ == MipsCmdType.REG_DEVICE_STATE:
            reg_dev_state: MipsRegDeviceState = mips_cmd.data
            self._device_state_sub_map[reg_dev_state.did] = reg_dev_state
            self.log_debug(
                f'mips local reg device state, {reg_dev_state.did}')
        elif mips_cmd.type_ == MipsCmdType.UNREG_DEVICE_STATE:
            unreg_dev_state: MipsRegDeviceState = mips_cmd.data
            del self._device_state_sub_map[unreg_dev_state.did]
            self.log_debug(
                f'mips local unreg device state, {unreg_dev_state.did}')
        else:
            self.log_error(
                f'mips local recv unknown cmd, {mips_cmd.type_}, '
                f'{mips_cmd.data}')

    def __on_mips_connect_handler(self, rc, props) -> None:
        self.log_debug('__on_mips_connect_handler')
        # Sub did/#, include reply topic
        self._mips_sub_internal(f'{self._did}/#')
        # Sub device list change
        self._mips_sub_internal('master/appMsg/devListChange')
        # Do not need to subscribe api topics, for they are covered by did/#
        # Sub api topic.
        # Sub broadcast topic
        for topic, _ in list(self._msg_matcher.iter_all_nodes()):
            self._mips_sub_internal(
                topic=re.sub(f'^{self._did}', 'master', topic))

    @final
    def __on_mips_message_handler(self, topic: str, payload: bytes) -> None:
        mips_msg: MipsMessage = MipsMessage.unpack(payload)
        # self.log_debug(
        #     f"mips local client, on_message, {topic} -> {mips_msg}")
        # Reply
        if topic == self._reply_topic:
            self.log_debug(f'on request reply, {mips_msg}')
            req: MipsRequest = self._request_map.pop(str(mips_msg.mid), None)
            if req:
                # Cancel timer
                self.mev_clear_timeout(req.timer)
                if req.on_reply:
                    self.main_loop.call_soon_threadsafe(
                        req.on_reply, mips_msg.payload or '{}',
                        req.on_reply_ctx)
            return
        # Broadcast
        bc_list: list[MipsBroadcast] = list(self._msg_matcher.iter_match(
            topic=topic))
        if bc_list:
            self.log_debug(f'on broadcast, {topic}, {mips_msg}')
            for item in bc_list or []:
                if item.handler is None:
                    continue
                self.main_loop.call_soon_threadsafe(
                    item.handler, topic[topic.find('/')+1:],
                    mips_msg.payload or '{}', item.handler_ctx)
            return
        # Device list change
        if topic == self._dev_list_change_topic:
            payload_obj: dict = json.loads(mips_msg.payload)
            dev_list = payload_obj.get('devList', None)
            if not isinstance(dev_list, list) or not dev_list:
                _LOGGER.error(
                    'unknown devListChange msg, %s', mips_msg.payload)
                return
            if self._on_dev_list_changed:
                self.main_loop.call_soon_threadsafe(
                    self.main_loop.create_task,
                    self._on_dev_list_changed(self, payload_obj['devList']))
            return

        self.log_debug(
            f'mips local client, recv unknown msg, {topic} -> {mips_msg}')

    @property
    def __gen_mips_id(self) -> int:
        mips_id: int = self._mips_seed_id
        self._mips_seed_id = int((self._mips_seed_id+1) % self.UINT32_MAX)
        return mips_id

    def __mips_publish(
            self, topic: str, payload: str | bytes, mid: int = None,
            ret_topic: str = None, wait_for_publish: bool = False,
            timeout_ms: int = 10000
    ) -> bool:
        mips_msg: bytes = MipsMessage.pack(
            mid=mid or self.__gen_mips_id, payload=payload,
            msg_from='local', ret_topic=ret_topic)
        return self._mips_publish_internal(
            topic=topic.strip(), payload=mips_msg,
            wait_for_publish=wait_for_publish, timeout_ms=timeout_ms)

    def __request(
            self, topic: str, payload: str,
            on_reply: Callable[[str, any], None],
            on_reply_ctx: any = None, timeout_ms: int = 10000
    ) -> bool:
        if topic is None or payload is None or on_reply is None:
            raise MIoTMipsError('invalid params')
        req_data: MipsRequestData = MipsRequestData()
        req_data.topic = topic
        req_data.payload = payload
        req_data.on_reply = on_reply
        req_data.on_reply_ctx = on_reply_ctx
        req_data.timeout_ms = timeout_ms
        return self._mips_send_cmd(type_=MipsCmdType.CALL_API, data=req_data)

    def __reg_broadcast(
        self, topic: str, handler: Callable[[str, str, any], None],
        handler_ctx: any
    ) -> bool:
        return self._mips_send_cmd(
            type_=MipsCmdType.REG_BROADCAST,
            data=MipsRegBroadcast(
                topic=topic, handler=handler, handler_ctx=handler_ctx))

    def __unreg_broadcast(self, topic) -> bool:
        return self._mips_send_cmd(
            type_=MipsCmdType.UNREG_BROADCAST,
            data=MipsRegBroadcast(topic=topic))

    @final
    async def __request_async(
        self, topic: str, payload: str, timeout_ms: int = 10000
    ) -> dict:
        fut_handler: asyncio.Future = self.main_loop.create_future()

        def on_msg_reply(payload: str, ctx: any):
            fut: asyncio.Future = ctx
            if fut:
                self.main_loop.call_soon_threadsafe(fut.set_result, payload)
        if not self.__request(
                topic=topic,
                payload=payload,
                on_reply=on_msg_reply,
                on_reply_ctx=fut_handler,
                timeout_ms=timeout_ms):
            # Request error
            fut_handler.set_result('internal request error')

        result = await fut_handler
        try:
            return json.loads(result)
        except json.JSONDecodeError:
            return {
                'code': MIoTErrorCode.CODE_MIPS_INVALID_RESULT.value,
                'message': f'Error: {result}'}

    async def __get_prop_timer_handle(self) -> None:
        for did in list(self._get_prop_queue.keys()):
            item = self._get_prop_queue[did].pop()
            _LOGGER.debug('get prop, %s, %s', did, item)
            result_obj = await self.__request_async(
                topic='proxy/get',
                payload=item['param'],
                timeout_ms=item['timeout_ms'])
            if result_obj is None or 'value' not in result_obj:
                item['fut'].set_result(None)
            else:
                item['fut'].set_result(result_obj['value'])

            if not self._get_prop_queue[did]:
                self._get_prop_queue.pop(did, None)

        if self._get_prop_queue:
            self._get_prop_timer = self.main_loop.call_later(
                0.1, lambda: self.main_loop.create_task(
                    self.__get_prop_timer_handle()))
        else:
            self._get_prop_timer = None