（优质长文警告⚠️，全文包含代码共计22000+字，完整阅读时间较长）
智能家居行业是指通过互联网、物联网、人工智能等先进技术将家庭设备和设施连接起来，实现智能化控制和管理的新兴产业。
本项目参考了github开源“博联智能家居”，采用博联旗下鸿雁系列智能插座和路由，算是二次开发。
目录
step1: 鸿雁插座有一个自带的app，可以实现一些基础操作，我们第一步要做的事，就是获知该app是怎么控制的。
step2: github上搜索broadlink，了解其基础功能。
step3: 成果转化，可以选择移植到微信小程序，或者PC端上，据我所知博联还是比较友好也是一个十分支持二次开发的厂商。
下面简单说说项目思路：
step1: 鸿雁插座有一个自带的app，可以实现一些基础操作，我们第一步要做的事，就是获知该app是怎么控制的。
主要工具：wireshark（截获数据）、socket套接字编程（监视和接收）
Wireshark是一个开源网络分析工具，用于捕获和分析网络数据包。它可以运行在Windows、Mac和Linux操作系统上，并且支持多种协议的分析，包括常见的TCP/IP协议、HTTP、FTP、DNS等。
通过Wireshark，用户可以实时捕获网络数据包，并对数据包进行深入分析。它提供了一系列强大的功能，包括：
数据包捕获：Wireshark可以通过网络接口捕获传输的数据包，用户可以选择感兴趣的数据包进行分析。
实时分析：Wireshark可以实时显示捕获的数据包，并提供详细的统计信息，如源IP地址、目的IP地址、协议类型、数据长度等。
数据包过滤：Wireshark支持灵活的过滤功能，用户可以根据协议类型、源/目的IP地址、端口号等条件过滤数据包，以便更好地分析感兴趣的数据。
统计功能：Wireshark提供了各种统计功能，如流量统计、协议分布图、包大小分布等，帮助用户更好地理解网络流量情况。
解码支持：Wireshark支持数百种协议的解码，用户可以查看和分析特定协议的数据包。
导出功能：Wireshark可以将捕获的数据包导出为多种格式，如文本文件、CSV文件、HTML文件等，方便用户进一步处理和保存。
截获相关数据之后，使用socket模拟发送过程，并进行测试。
step2: github上搜索broadlink，了解其基础功能。
pip install broadlink
这玩意没啥好说的，自己去看，我这一时半会也说不清楚，在文章的最后会附上相关代码。
主要工具：pip
broadlink的代码量不是很大（小几千行？），花个大半天就能理解的差不多，后续我们要做的算是一定程度的仿写，所以对其的了解程度非常重要。
下面附上实现其主要功能的程序段：
"""Support for Broadlink devices."""
import socket
import threading
import random
import time
import typing as t
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from . import exceptions as e
from .const import (
DEFAULT_BCAST_ADDR,
DEFAULT_PORT,
DEFAULT_RETRY_INTVL,
DEFAULT_TIMEOUT,
)
from .protocol import Datetime
HelloResponse = t.Tuple[int, t.Tuple[str, int], str, str, bool]
def scan(
timeout: int = DEFAULT_TIMEOUT,
local_ip_address: str = None,
discover_ip_address: str = DEFAULT_BCAST_ADDR,
discover_ip_port: int = DEFAULT_PORT,
) -> t.Generator[HelloResponse, None, None]:
"""Broadcast a hello message and yield responses."""
conn = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
conn.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
conn.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
if local_ip_address:
conn.bind((local_ip_address, 0))
port = conn.getsockname()[1]
else:
local_ip_address = "0.0.0.0"
port = 0
packet = bytearray(0x30)
packet[0x08:0x14] = Datetime.pack(Datetime.now())
packet[0x18:0x1C] = socket.inet_aton(local_ip_address)[::-1]
packet[0x1C:0x1E] = port.to_bytes(2, "little")
packet[0x26] = 6
checksum = sum(packet, 0xBEAF) & 0xFFFF
packet[0x20:0x22] = checksum.to_bytes(2, "little")
start_time = time.time()
discovered = []
try:
while (time.time() - start_time) < timeout:
time_left = timeout - (time.time() - start_time)
conn.settimeout(min(DEFAULT_RETRY_INTVL, time_left))
conn.sendto(packet, (discover_ip_address, discover_ip_port))
while True:
try:
resp, host = conn.recvfrom(1024)
except socket.timeout:
break
devtype = resp[0x34] | resp[0x35] << 8
mac = resp[0x3A:0x40][::-1]
if (host, mac, devtype) in discovered:
continue
discovered.append((host, mac, devtype))
name = resp[0x40:].split(b"\x00")[0].decode()
is_locked = bool(resp[0x7F])
yield devtype, host, mac, name, is_locked
finally:
conn.close()
def ping(address: str, port: int = DEFAULT_PORT) -> None:
"""Send a ping packet to an address.
This packet feeds the watchdog timer of firmwares >= v53.
Useful to prevent reboots when the cloud cannot be reached.
It must be sent every 2 minutes in such cases.
"""
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as conn:
conn.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
packet = bytearray(0x30)
packet[0x26] = 1
conn.sendto(packet, (address, port))
class Device:
"""Controls a Broadlink device."""
TYPE = "Unknown"
__INIT_KEY = "097628343fe99e23765c1513accf8b02"
__INIT_VECT = "562e17996d093d28ddb3ba695a2e6f58"
def __init__(
self,
host: t.Tuple[str, int],
mac: t.Union[bytes, str],
devtype: int,
timeout: int = DEFAULT_TIMEOUT,
name: str = "",
model: str = "",
manufacturer: str = "",
is_locked: bool = False,
) -> None:
"""Initialize the controller."""
self.host = host
self.mac = bytes.fromhex(mac) if isinstance(mac, str) else mac
self.devtype = devtype
self.timeout = timeout
self.name = name
self.model = model
self.manufacturer = manufacturer
self.is_locked = is_locked
self.count = random.randint(0x8000, 0xFFFF)
self.iv = bytes.fromhex(self.__INIT_VECT)
self.id = 0
self.type = self.TYPE # For backwards compatibility.
self.lock = threading.Lock()
self.aes = None
self.update_aes(bytes.fromhex(self.__INIT_KEY))
def __repr__(self) -> str:
"""Return a formal representation of the device."""
return (
"%s.%s(%s, mac=%r, devtype=%r, timeout=%r, name=%r, "
"model=%r, manufacturer=%r, is_locked=%r)"
) % (
self.__class__.__module__,
self.__class__.__qualname__,
self.host,
self.mac,
self.devtype,
self.timeout,
self.name,
self.model,
self.manufacturer,
self.is_locked,
)
def __str__(self) -> str:
"""Return a readable representation of the device."""
return "%s (%s / %s:%s / %s)" % (
self.name or "Unknown",
" ".join(filter(None, [self.manufacturer, self.model, hex(self.devtype)])),
*self.host,
":".join(format(x, "02X") for x in self.mac),
)
def update_aes(self, key: bytes) -> None:
"""Update AES."""
self.aes = Cipher(
algorithms.AES(bytes(key)), modes.CBC(self.iv), backend=default_backend()
)
def encrypt(self, payload: bytes) -> bytes:
"""Encrypt the payload."""
encryptor = self.aes.encryptor()
return encryptor.update(bytes(payload)) + encryptor.finalize()
def decrypt(self, payload: bytes) -> bytes:
"""Decrypt the payload."""
decryptor = self.aes.decryptor()
return decryptor.update(bytes(payload)) + decryptor.finalize()
def auth(self) -> bool:
"""Authenticate to the device."""
self.id = 0
self.update_aes(bytes.fromhex(self.__INIT_KEY))
packet = bytearray(0x50)
packet[0x04:0x14] = [0x31] * 16
packet[0x1E] = 0x01
packet[0x2D] = 0x01
packet[0x30:0x36] = "Test 1".encode()
response = self.send_packet(0x65, packet)
e.check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:])
self.id = int.from_bytes(payload[:0x4], "little")
self.update_aes(payload[0x04:0x14])
return True
def hello(self, local_ip_address=None) -> bool:
"""Send a hello message to the device.
Device information is checked before updating name and lock status.
"""
responses = scan(
timeout=self.timeout,
local_ip_address=local_ip_address,
discover_ip_address=self.host[0],
discover_ip_port=self.host[1],
)
try:
devtype, _, mac, name, is_locked = next(responses)
except StopIteration as err:
raise e.NetworkTimeoutError(
-4000,
"Network timeout",
f"No response received within {self.timeout}s",
) from err
if mac != self.mac:
raise e.DataValidationError(
-2040,
"Device information is not intact",
"The MAC address is different",
f"Expected {self.mac} and received {mac}",
)
if devtype != self.devtype:
raise e.DataValidationError(
-2040,
"Device information is not intact",
"The product ID is different",
f"Expected {self.devtype} and received {devtype}",
)
self.name = name
self.is_locked = is_locked
return True
def ping(self) -> None:
"""Ping the device.
This packet feeds the watchdog timer of firmwares >= v53.
Useful to prevent reboots when the cloud cannot be reached.
It must be sent every 2 minutes in such cases.
"""
ping(self.host[0], port=self.host[1])
def get_fwversion(self) -> int:
"""Get firmware version."""
packet = bytearray([0x68])
response = self.send_packet(0x6A, packet)
e.check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:])
return payload[0x4] | payload[0x5] << 8
def set_name(self, name: str) -> None:
"""Set device name."""
packet = bytearray(4)
packet += name.encode("utf-8")
packet += bytearray(0x50 - len(packet))
packet[0x43] = self.is_locked
response = self.send_packet(0x6A, packet)
e.check_error(response[0x22:0x24])
self.name = name
def set_lock(self, state: bool) -> None:
"""Lock/unlock the device."""
packet = bytearray(4)
packet += self.name.encode("utf-8")
packet += bytearray(0x50 - len(packet))
packet[0x43] = bool(state)
response = self.send_packet(0x6A, packet)
e.check_error(response[0x22:0x24])
self.is_locked = bool(state)
def get_type(self) -> str:
"""Return device type."""
return self.type
def send_packet(self, packet_type: int, payload: bytes) -> bytes:
"""Send a packet to the device."""
self.count = ((self.count + 1) | 0x8000) & 0xFFFF
packet = bytearray(0x38)
packet[0x00:0x08] = bytes.fromhex("5aa5aa555aa5aa55")
packet[0x24:0x26] = self.devtype.to_bytes(2, "little")
packet[0x26:0x28] = packet_type.to_bytes(2, "little")
packet[0x28:0x2A] = self.count.to_bytes(2, "little")
packet[0x2A:0x30] = self.mac[::-1]
packet[0x30:0x34] = self.id.to_bytes(4, "little")
p_checksum = sum(payload, 0xBEAF) & 0xFFFF
packet[0x34:0x36] = p_checksum.to_bytes(2, "little")
padding = (16 - len(payload)) % 16
payload = self.encrypt(payload + bytes(padding))
packet.extend(payload)
checksum = sum(packet, 0xBEAF) & 0xFFFF
packet[0x20:0x22] = checksum.to_bytes(2, "little")
with self.lock and socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as conn:
timeout = self.timeout
start_time = time.time()
while True:
time_left = timeout - (time.time() - start_time)
conn.settimeout(min(DEFAULT_RETRY_INTVL, time_left))
conn.sendto(packet, self.host)
try:
resp = conn.recvfrom(2048)[0]
break
except socket.timeout as err:
if (time.time() - start_time) > timeout:
raise e.NetworkTimeoutError(
-4000,
"Network timeout",
f"No response received within {timeout}s",
) from err
if len(resp) < 0x30:
raise e.DataValidationError(
-4007,
"Received data packet length error",
f"Expected at least 48 bytes and received {len(resp)}",
)
nom_checksum = int.from_bytes(resp[0x20:0x22], "little")
real_checksum = sum(resp, 0xBEAF) - sum(resp[0x20:0x22]) & 0xFFFF
if nom_checksum != real_checksum:
raise e.DataValidationError(
-4008,
"Received data packet check error",
f"Expected a checksum of {nom_checksum} and received {real_checksum}",
)
return resp
step3: 成果转化，可以选择移植到微信小程序，或者PC端上，据我所知博联还是比较友好也是一个十分支持二次开发的厂商。
对于微信小程序，我后续有时间会结合自己经历，写一写详细的引导性教学，感兴趣的小伙伴可以点下关注。
主要工具：微信开发者工具、QT、tkinter、pyinstaller……
大多数学生应该和我一样，大创不是为了做出什么真正有用的东西，主要追求的还是：听起来有那么一点创新点，又能混到学分，还能学到东西……
为什么说是成果转化呢，目前鸿雁的控制方式较为单一（仅支持手机app），且手机app上面的功能并为开发完全（就是看了代码知道的），完全可以在原有的功能基础上，加上一些自己的东西，反映出所谓“创新点”。
一下是我通过测试使用官方app了解到的功能：
远程控制：通过手机APP，用户可以随时随地远程控制插座的开关状态，无需亲自操作插座。
定时控制：鸿雁智能插座支持定时开关功能，用户可以设置定时开关时间，方便自动化控制电器设备。比如，可以设定插座在晚上10点关闭，早上7点开启，实现自动化的电器控制。
节能模式：鸿雁智能插座还支持节能模式，用户可以根据需要设定插座关闭时间，防止设备长时间待机造成能源浪费。
电量统计：插座内置电量统计功能，能够实时显示已使用的电量，方便用户了解电器设备的电量消耗情况。
安全保护：鸿雁智能插座具有过载保护和短路保护功能，可以确保使用过程中的安全性。
想一想人们还需要什么，结合现有包，扩充其功能。
所谓创新点，不仅仅体现在功能上，也可以是实现形式。
step4: 画大饼（bushi）
………………
部分代码展示
下面主要展示部分代码，涉及一些基础操作测试：
基础功能展示——使用python控制开闭
首先是使用我们的python程序实现插座的开闭功能（前提是保证你的电脑和插座处于同意局域网之下）：
device_ip = "设备IP"
device_port = 80
device_mac = "设备mac地址"
device_type = "broadlink.mp1"
try:
device = broadlink.mp1(host=(device_ip, device_port), mac=bytearray.fromhex(device_mac))
device.auth()
except ValueError:
pass
def test():
print("开始监听")
while True:
data, address = s.recvfrom(1024)
print(data)
print(address)
pass
t1 = t2 = t3 = t4 = "on"
def run1():
global t1
if t1 == "on":
print("您打开了s1")
t1 = "off"
else:
print("您关闭了s1")
t1 = "on"
if t1 == "on":
try:
device.set_power(1, True)
except:
pass
elif t1 == "off":
try:
device.set_power(1, False)
except:
pass
try:
if device.check_power()[socket]:
print("on")
else:
print("off")
except NameError:
pass
print(f"设备状态测试：{t1}\t")