目录前言1.WIFI例程1.1station例程烧录1.2softAP例程烧录2.BLE例程2.1ble_ibeacon例程烧录2.2ble_master例程烧录2.3ble_slave例程烧录2.4blufi例程烧录3.协议例程3.1SNTP例程烧录3.2MQTT例程烧录3.3HTTP协议实现3.3.1HTTP的超时时间前言本文在学习期间所著如有侵权小编会直接删除首先我们从官网拉取一个SDK文件Ai-Thinker-Open/Ai-Thinker-WB2以下我用的Ubuntu来进行的编译和烧录1.WIFI例程1.1 station例程烧录station模式是让开发板去连接家里的wifi连接成功后开发板会从路由器那里获得一个ip地址。整个工作流程就是CPU从Flash读取已保存的WiFi配置SSID和密码至RAM缓冲区随后停止重试计数器并清理断开状态的附属定时器与标志位将状态机从“断开”切换为“连接中”并填充迁移参数接着打印RAM中的SSID和密码作为确认选择第0号虚拟接口启动本地扫描缓存查询未命中目标SSID完成查询后正式将状态标志写入“连接中”并执行该状态初始化动作同时向上层应用发送重连和连接中的事件回调然后启动空口扫描及后续认证关联流程。把路径改到wifi下的station目录下。然后我们打开main.c文件将wifi的名称和密码改成我们自己的wifi名称和密码然后保存文件然后把开发板插入连接到虚拟机上可以通过上面虚拟机选项查询有没有连接出现如下图的显示证明开发板与虚拟机建立连接了此处的断开连接连接主机就证明开发板连接到了虚拟机上然后输入命令ls /dev | grep ttyUSB来查询我们开发板连接到虚拟机的哪个端口上了然后我们输入命令:make flash p/dev/ttyUSB0 b115200开始进行编译烧录然后出现Please Press Reset Key! ,我们点击一下开发板上的EN键出现[All Success]证明编译烧录完成然后我们可以输入make clean来清除一下编译产生的文件然后断开开发板与虚拟机的连接连接到主机出现如下图的显示证明开发板与主机建立连接了然后打开串口调试助手查看开发板发送的信息我们再点一次开发板的EN键wifi连接成功1.2softAP例程烧录softAP模式就是开发板自己充当热点可以让如手机等设备来进行连接整个工作流程就是初始化硬件读取flash中的AP配置将状态机由”接口关闭“切换为”空闲“启动AP模式将预设信息给硬件设备并且配置好ip地址然后设置好连接成功后的回调函数将wifi热点在空中广播cpu收到硬件层有外部连接触发中断然后解析请求帧同意后建立会话.然后我们打开ubuntu打开softAP例程进行编译和烧录烧录之后我们可以打开main.c来查看开发板的名称和密码然后打开我们的手机进行连接开发板后打开串口调试助手可以看到连接成功的日志2.BLE例程2.1ble_ibeacon例程烧录ble_ibeacon模式就是低功耗蓝牙的一种广播者模式主要是让设备化做一个只发送信号、不建立连接的”信标“整个过程就是:CPU启动BLE信标功能执行蓝牙控制器射频自校准与初始化随后初始化HC驱动层最后启用蓝牙协议栈使设备具备蓝牙广播能力。我们打开ubuntu来进入到ble_ibeacon的目录下进行烧录程序然后打开串口查看打印信息我们打开手机的蓝牙我这里用的nRF Connect发现可以扫描到我们的开发板2.2ble_master例程烧录ble_master模式是低功耗蓝牙中的中央设备模式主要作用是主动扫描、发起连接并作为连接建立后的”控制方“进行双向数据交换整个过程就是CPU进入BLE主机模式初始化蓝牙扫描事件驱动框架启动硬件扫描并持续接收广播数据包解析出每个周边设备的MAC地址、信号强度和名称后逐条打印输出。我们发现开发板可以进行扫描axk_mqtt2.3ble_slave例程烧录ble_slave模式是低功耗蓝牙中的被动响应方主要是广播自己的存在等待其他设备来连接连接建立后它主要作为数据和服务提供方等待从机来读取或者写入数据整个过程CPU执行蓝牙射频自校准然后读取并打印本机蓝牙MAC地址完成蓝牙硬件初始化然后CPU完成与远端蓝牙设备的物理连接建立协商MTU大小为247字节并更新连接参数间隔30ms、延迟0、超时6000ms此后双方可正常进行数据收发。然后烧录程序之后然后点击EN键重启模块。然后我们手机端可以下载一个蓝牙连接助手然后通过蓝牙连接助手去扫描模块找到对应的蓝牙名称然后进行连接如果没有找到就重新刷新列表重启模块然后打开串口调试助手就会看到蓝牙连接成功通过点击断开连接我们也能中断连接打开串口调试助手就会看到蓝牙断开连接2.4blufi例程烧录blufi就是通过蓝牙将wifi名称和密码安全的传输给开发板然后连接wifi的整体工作流程CPU从Flash读取MTD分区信息并校验ROMFS(校验失败)执行射频自校准初始化Wi-Fi动并将状态机切至空闲设置BLE为从机角色并启动广播初始化BLUFI配网服务最后周期性打印剩余内存供开发者监控。烧录程序后打开串口调试助手查看发送的信息。然后我们可以通过手机端的来进行蓝牙连接和配网我们打开手机端的EspBlufi下拉空白处进行刷新扫描然后开发板可以按EN键重启然后扫描后点击打开我们的设备名然后点击进去点击连接选项然后可以在手机端和串口调试助手都看到提示信息然后我们可以点击配网将开发板的模式调为STA模式然后把wifi名称和密码发给它进行配网然后可以查看串口调试助手发现配置WiFi的指令发送出去了然后一会串口调试助手会发送信息配网成功了然后可以点解断开选项可以通过串口调试助手发现蓝牙已经断开的提示最后针对ios手机可能最开始会扫描不到开发板以我这款为例Ai-Thinker-WB2我们可以去查找一下蓝牙名称然后修改一下开发板上的蓝牙名称打开vscode我们可以通过查找蓝牙名称来进行查找这个字符串在哪里然后打开字符串所在文件的.c文件然后我的目录是在这个目录下有一个axk_blufi.c文件我们打开这个文件找到这个字符串然后将本来的AXK_BLUFI改为BLUFI_AXK主要原因就是ios端的软件进行扫描的时候会直接扫描BLUFI开头的设备过滤到非BLUFI开头的设备所以就直接扫描不到然后保存代码重新编译烧录即可。3.协议例程3.1SNTP例程烧录SNTP就是让设备联网后自动从互联网时间服务器获取并校准系统时钟的协议用来解决断电后时间重置的问题。工作过程DNS解析:将域名ntp.aliyun.com转为IP地址。创建UDP套接字:向目标IP的123端口发起连接。构造请求包:按照NTP协议(RFC5905)打包一个48字节的数据包(设置模式为3客户端)。发送并等待响应:将包发出去阻塞或异步等待服务器回包。解析时间戳:从回包的第40~43字节提取秒数换算后写入芯片的系统时钟。首先我们打开SNTP_demo的main.c将SSID和PWD改为自己wifi的名称和密码然后我们在烧录SNTP_demo例程之后打开串口调试助手可以看到我们的开发板的CPU调用SNTP客户端向ntp.aliyun.com发送UDP请求接收响应报文并解析出UTC时间更新系统本地时钟随后循环每秒打印一次当前时间。3.2MQTT例程烧录基于TCP的MQTT就是MQTT协议在传输层跑在TCP协议之上主要作用是作用:利用TCP的确认重传和按序交付机制为MQTT的发布/订阅通信提供一条绝对可靠、不丢包、不乱序的双向数据通道。首先我们打开基于TCP的MQTT协议的例程文件然后打开main.c文件修改wifi的名称和密码然后保存main.c文件然后打开demo.c文件我们修改一下MQTT的服务器找到函数并且把字符串里的服务器换成自己的服务器然后我们保存文件进行编译烧录这里的MQTT我们选择是建立在TCP连接之上烧录完成后可以打开串口调试助手查看信息。出现此处消息提醒说明我们的MQTT连接建立成功了然后以ubuntu为例子我将开发板和ubuntu建立MQTT连接之后在终端打开MQTT服务器就能看到开发板发送的信息了。如果没有就保证ubuntu服务器打开然后重启模块即可3.3HTTP协议实现3.3.1HTTP协议实现以下面一个简单的python脚本为例建立http服务器然后通过我们wb2去连接import socket import time import json import random HOST 0.0.0.0 PORT 8080 led_status off sensor_data {temperature: 25.0, humidity: 60.0} server socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((HOST, PORT)) server.listen(5) print(f 服务器已启动监听 {HOST}:{PORT}) def build_response(status_code, content_type, body_text): status_messages { 200: OK, 400: Bad Request, 404: Not Found, 405: Method Not Allowed, } status_line fHTTP/1.1 {status_code} {status_messages.get(status_code, Unknown)} headers [ fContent-Type: {content_type}, fContent-Length: {len(body_text)}, Connection: close, ] header_section \r\n.join(headers) return f{status_line}\r\n{header_section}\r\n\r\n{body_text}.encode() def repair_json_body(raw_request_str): 修复因 AT 指令逗号分隔导致的 JSON 分裂 lines raw_request_str.split(\r\n) if not lines: return None # 找空行位置 blank_line_index -1 for i, line in enumerate(lines): if line : blank_line_index i break if blank_line_index -1: return None header_lines lines[1:blank_line_index] body_raw \r\n.join(lines[blank_line_index1:]) standard_headers {} json_fragments [] for line in header_lines: if : in line: key, value line.split(: , 1) if key.lower() in [host, content-type, content-length, connection, user-agent, accept]: standard_headers[key.lower()] value else: json_fragments.append(line.strip()) else: json_fragments.append(line.strip()) combined_body body_raw.strip() if json_fragments: if combined_body.endswith(,): combined_body combined_body[:-1] if combined_body.startswith({) and not combined_body.endswith(}): combined_body , combined_body ,.join(json_fragments) if combined_body.startswith({) and not combined_body.endswith(}): combined_body } if content-length in standard_headers: standard_headers[content-length] str(len(combined_body)) return combined_body, standard_headers def handle_request(method, path, full_path, headers, body): global led_status, sensor_data print(f Body (repr): {repr(body)}) print(f Body length: {len(body)}) if method GET: # LED 控制 if path /led/on: led_status on return build_response(200, text/plain, LED ON) elif path /led/off: led_status off return build_response(200, text/plain, LED OFF) elif path /led/status: return build_response(200, application/json, f{{led:{led_status}}}) # 传感器数据 elif path /sensor: sensor_data[temperature] round(25 (random.random() - 0.5) * 2, 1) sensor_data[humidity] round(60 (random.random() - 0.5) * 5, 1) return build_response(200, application/json, json.dumps(sensor_data)) # 时间 elif path /time: now time.strftime(%Y-%m-%d %H:%M:%S) return build_response(200, application/json, f{{time:{now}}}) # 根路径 elif path /: html fhtmlbodyh1WB2 Test/h1pLED: {led_status}/p/body/html return build_response(200, text/html, html) # 带查询参数的 /data?typetempunitc elif path /data and ? in full_path: query_str full_path.split(?, 1)[1] params {} for item in query_str.split(): if in item: k, v item.split(, 1) params[k] v print(f查询参数: {params}) if params.get(type) temp: val sensor_data[temperature] unit params.get(unit, c) return build_response(200, application/json, json.dumps({temperature: val, unit: unit})) else: return build_response(200, application/json, json.dumps({error: unknown type})) else: return build_response(404, text/plain, Not Found) elif method POST: if path /report: try: data json.loads(body) print(f✅ 解析成功: {data}) return build_response(200, application/json, json.dumps({status: received, data: data})) except json.JSONDecodeError as e: print(f❌ JSON 解析失败: {e}) return build_response(400, text/plain, Invalid JSON) else: return build_response(404, text/plain, Not Found) else: return build_response(405, text/plain, Method Not Allowed) while True: try: conn, addr server.accept() print(f 新连接来自: {addr}) request_data conn.recv(8192) if not request_data: conn.close() continue raw_str request_data.decode(errorsignore) print( 原始请求字节 (repr):, repr(request_data)) lines raw_str.split(\r\n) if not lines: conn.close() continue first_line lines[0].split( ) if len(first_line) 2: conn.send(bHTTP/1.1 400 Bad Request\r\n\r\n) conn.close() continue method first_line[0].upper() full_path first_line[1] path full_path.split(?)[0] if ? in full_path else full_path print(f {method} {full_path}) # 解析头部 headers {} blank_line_index raw_str.find(\r\n\r\n) if blank_line_index -1: conn.send(bHTTP/1.1 400 Bad Request\r\n\r\n) conn.close() continue header_text raw_str[:blank_line_index] for line in header_text.split(\r\n)[1:]: if : in line: key, value line.split(: , 1) headers[key.lower()] value body raw_str[blank_line_index 4:] # 如果是 POST /report尝试修复 JSON if method POST and path /report: repaired repair_json_body(raw_str) if repaired: repaired_body, repaired_headers repaired print(f 修复后的 body: {repr(repaired_body)}) body repaired_body if content-length in headers: headers[content-length] str(len(body)) # 根据 Content-Length 截取 body如果有 if content-length in headers: try: cl int(headers[content-length]) body body[:cl] except: pass response handle_request(method, path, full_path, headers, body) conn.send(response) status_line response.split(b\r\n)[0].decode() print(f✅ 响应: {status_line}) except (BrokenPipeError, ConnectionResetError): print(⚠️ 客户端断开) except Exception as e: print(f❌ 异常: {e}) finally: conn.close() print(--- 连接关闭 ---\n)然后将我们的wb2连接到我们的脚本所处的电脑的同一个局域网下然后进行AT指令连接指令官方指令combo模组通用指令_v4.18p_3.8.0.pdfATHTTPCLIENTLINE1,2,text/xml,《自己的ipv4地址》,8080,/,testhttp连接成功这个http也有其他好玩的可以请求一下命令ATHTTPCLIENTLINE1,2,text/xml,《自己的ipv4地址》,8080,/timeATHTTPCLIENTLINE1,2,text/xml,《自己的ipv4地址》,8080,/led/onATHTTPCLIENTLINE1,2,text/xml,《自己的ipv4地址》,8080,/sensor路径方法功能/led/onGET开启模拟 LED/led/offGET关闭模拟 LED/led/statusGET返回 LED 状态 JSON/sensorGET返回随机温湿度 JSON/timeGET返回服务器当前时间 JSON/data?type...unit...GET带参数查询可自由扩展/GET返回简单 HTML 欢迎页/reportPOST接收 JSON 数据并返回确认3.3.2HTTP的超时时间以我的WB2-12F为例在我们以AT指令发送httpget的命令后服务器接收到http请求一直延迟不响应的环境下我们wb2的超时时间是20秒左右。环境测试以python的脚本为例子(延迟时间可以自己来订制)import socket import time # 监听所有网卡的 8080 端口 server socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.bind((0.0.0.0, 8080)) server.listen(1) print(服务器已启动等待 WB2 连接...) conn, addr server.accept() print(fWB2 已连接IP: {addr}) # 接收模组发来的 HTTP 请求确保模组发送阶段完成 request_data conn.recv(2048) print(f收到模组 HTTP 请求:\n{request_data.decode()}) # 故意延迟 6 秒大于模组默认 5 秒超时 delay_seconds 25 print(f模拟服务器处理中故意延迟 {delay_seconds} 秒再回复...) time.sleep(delay_seconds) # 发送正确的 HTTP 响应 response ( HTTP/1.1 200 OK\r\n Content-Type: application/json\r\n Content-Length: 20\r\n \r\n {\result\:\success\} ) conn.send(response.encode()) print(响应已发送测试结束) conn.close() server.close()我们启动这个脚本然后打开我们的串口调试助手将我们的wb2连接到我们的脚本所处的电脑的同一个局域网下然后发送AT指令去请求连接指令ATHTTPCLIENTLINE1,2,text/xml,《自己的ipv4地址》,8080,/,test然后打开我们的串口调试助手我们可以看到已经因为超时而退出然后报ERROR
