Midea/midea_ac_lan/midea/core/discover.py

import logging
import socket
import ifaddr
from ipaddress import IPv4Network
from .security import LocalSecurity
try:
    import xml.etree.cElementTree as ET
except ImportError:
    import xml.etree.ElementTree as ET

_LOGGER = logging.getLogger(__name__)

BROADCAST_MSG = bytearray([
    0x5a, 0x5a, 0x01, 0x11, 0x48, 0x00, 0x92, 0x00,
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
    0x7f, 0x75, 0xbd, 0x6b, 0x3e, 0x4f, 0x8b, 0x76,
    0x2e, 0x84, 0x9c, 0x6e, 0x57, 0x8d, 0x65, 0x90,
    0x03, 0x6e, 0x9d, 0x43, 0x42, 0xa5, 0x0f, 0x1f,
    0x56, 0x9e, 0xb8, 0xec, 0x91, 0x8e, 0x92, 0xe5
])

DEVICE_INFO_MSG = bytearray([
    0x5a, 0x5a, 0x15, 0x00, 0x00, 0x38, 0x00, 0x04,
    0x00, 0x00, 0x00, 0x00, 0x00, 0x27, 0x33, 0x05,
    0x13, 0x06, 0x14, 0x14, 0x00, 0x00, 0x00, 0x00,
    0x00, 0x00, 0x03, 0xe8, 0x00, 0x00, 0x00, 0x00,
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
    0xca, 0x8d, 0x9b, 0xf9, 0xa0, 0x30, 0x1a, 0xe3,
    0xb7, 0xe4, 0x2d, 0x53, 0x49, 0x47, 0x62, 0xbe
])


def discover(discover_type=None, ip_address=None):
    if discover_type is None:
        discover_type = []
    security = LocalSecurity()
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    sock.settimeout(5)
    found_devices = {}
    if ip_address is None:
        addrs = enum_all_broadcast()
    else:
        addrs = [ip_address]
    _LOGGER.debug(f"All addresses for broadcast: {addrs}")
    for v in range(0, 3):
        for addr in addrs:
            sock.sendto(BROADCAST_MSG, (addr, 6445))
            sock.sendto(BROADCAST_MSG, (addr, 20086))
    while True:
        try:
            data, addr = sock.recvfrom(512)
            ip = addr[0]
            _LOGGER.debug(f"Received response from {addr}: {data.hex()}")
            if len(data) >= 104 and (data[:2].hex() == "5a5a" or data[8:10].hex() == "5a5a"):
                if data[:2].hex() == "5a5a":
                    protocol = 2
                elif data[:2].hex() == "8370":
                    protocol = 3
                    if data[8:10].hex() == "5a5a":
                        data = data[8:-16]
                else:
                    continue
                device_id = int.from_bytes(bytearray.fromhex(data[20:26].hex()), "little")
                if device_id in found_devices:
                    continue
                encrypt_data = data[40:-16]
                reply = security.aes_decrypt(encrypt_data)
                _LOGGER.debug(f"Declassified reply: {reply.hex()}")
                ssid = reply[41:41 + reply[40]].decode("utf-8")
                device_type = ssid.split("_")[1]
                port = bytes2port(reply[4:8])
                model = reply[17:25].decode("utf-8")
                sn = reply[8:40].decode("utf-8")
            elif data[:6].hex() == "3c3f786d6c20":
                protocol = 1
                root = ET.fromstring(data.decode(
                    encoding="utf-8", errors="replace"))
                child = root.find("body/device")
                m = child.attrib
                port, sn, device_type = int(m["port"]), m["apc_sn"], str(
                    hex(int(m["apc_type"])))[2:]
                response = get_device_info(ip, int(port))
                device_id = get_id_from_response(response)
                if len(sn) == 32:
                    model = sn[9:17]
                elif len(sn) == 22:
                    model = sn[3:11]
                else:
                    model = ""
            else:
                continue
            device = {
                "device_id": device_id,
                "type": int(device_type, 16),
                "ip_address": ip,
                "port": port,
                "model": model,
                "sn": sn,
                "protocol": protocol
            }
            if len(discover_type) == 0 or device.get("type") in discover_type:
                found_devices[device_id] = device
                _LOGGER.debug(f"Found a supported device: {device}")
            else:
                _LOGGER.debug(f"Found a unsupported device: {device}")
        except socket.timeout:
            break
        except socket.error as e:
            _LOGGER.error(f"Socket error: {repr(e)}")
    return found_devices


def get_id_from_response(response):
    if response[64:-16][:6].hex() == "3c3f786d6c20":
        xml = response[64:-16]
        root = ET.fromstring(xml.decode(encoding="utf-8", errors="replace"))
        child = root.find("smartDevice")
        m = child.attrib
        return int.from_bytes(bytearray.fromhex(m["devId"]), "little")
    else:
        return 0


def bytes2port(paramArrayOfbyte):
    if paramArrayOfbyte is None:
        return 0
    b, i = 0, 0
    while b < 4:
        if b < len(paramArrayOfbyte):
            b1 = paramArrayOfbyte[b] & 0xFF
        else:
            b1 = 0
        i |= b1 << b * 8
        b += 1
    return i


def get_device_info(device_ip, device_port: int):
    response = bytearray(0)
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.settimeout(8)
            device_address = (device_ip, device_port)
            sock.connect(device_address)
            _LOGGER.debug(f"Sending to {device_ip}:{device_port} {DEVICE_INFO_MSG.hex()}")
            sock.sendall(DEVICE_INFO_MSG)
            response = sock.recv(512)
    except socket.timeout:
        _LOGGER.warning(f"Connect the device {device_ip}:{device_port} timed out for 8s. "
                        f"Don't care about a small amount of this. if many maybe not support."
                        )
    except socket.error:
        _LOGGER.warning(f"Can't connect to Device {device_ip}:{device_port}")
    return response


def enum_all_broadcast():
    nets = []
    adapters = ifaddr.get_adapters()
    for adapter in adapters:
        for ip in adapter.ips:
            if ip.is_IPv4 and ip.network_prefix < 32:
                localNet = IPv4Network(f"{ip.ip}/{ip.network_prefix}", strict=False)
                if localNet.is_private and not localNet.is_loopback and not localNet.is_link_local:
                    addr = str(localNet.broadcast_address)
                    if addr not in nets:
                        nets.append(addr)
    return nets