给上一篇的28BYJ-48三自由度机械臂加一个网页控制台，手机或电脑打开浏览器，拖动三个手柄就能指挥机械臂转动，不需要装任何软件。
1、机械臂简介
机械臂有三个关节：J1是底座，负责水平旋转；J2是肩、J3是肘，负责俯仰。网页上拖动蓝色手柄预览姿态，松手机械臂一步到位。角度逆时针为正，三轴行程都是±90°，0°是上电中间位。
2、控制原理
控制链路很短：浏览器发出HTTP请求，ESP32上的服务器接收后驱动步进电机。网页下发的是绝对角度，固件自己算出还差多少步再决定正转反转，连续下发也不会累加错乱。
3、软件介绍
软件由三个文件组成，全部上传到ESP32：
stepper3.py：三轴驱动，主循环补步输出脉冲，支持同步/独立两种运动模式，越界自动夹到±90°。（详见上一篇）
server.py：HTTP服务器，提供5个接口，控制页面在启动时读入内存。
index.html：网页控制台，纯HTML加SVG加原生JavaScript，不用任何前端框架。（详见附件）
服务器接口如下：
方法路径作用
| GET | / | 返回控制页面 |
| GET | /api/status | 返回三轴角度和运动状态 |
| GET | /api/move?j1=&j2=&j3=&hz=&sync= | 转到指定关节角 |
| POST | /api/stop | 急停 |
| GET | /api/home | 三轴回0° |
服务器参考代码如下：
import network
import socket
import time
import stepper3 as st
SSID = "SSID"
PASS = "PASS"
PORT = 80
LOOP_SLEEP_MS = 1 # 主循环空转间隔：1ms → tick 约 500~900Hz
def _wifi():
w = network.WLAN(network.STA_IF)
w.active(True)
time.sleep_ms(1500) # P4 的 WiFi 跑在板载 C6 协处理器上，start 是异步的，等它就绪
if not w.isconnected():
for i in range(3):
print("连接 WiFi: %s（第 %d 次）" % (SSID, i + 1))
try:
w.connect(SSID, PASS)
except OSError as e:
print(" connect 报错:", e)
for _ in range(150):
if w.isconnected():
break
time.sleep_ms(200)
if w.isconnected():
break
try:
w.disconnect() # 复位 STA 状态再重试（diag 实测第 2 次必成）
except OSError:
pass
time.sleep_ms(1000)
if w.isconnected():
print("WiFi 已连接，控制页地址: http://%s/" % w.ifconfig()[0])
else:
print("WiFi 连接失败：运行 diag_wifi.py 诊断，把输出发回来")
return w
_PAGE = None
_poll = 0 # /api/status 心跳计数（串口诊断用）
_last_moving = False # 上次轮询时的运动状态（检测"到位"时刻）
_hb_tick = 0 # 上次心跳时的 tick 次数
_hb_dt = 0 # 上次心跳时的累计流逝毫秒
def _page():
global _PAGE
if _PAGE is None:
with open("index.html", "rb") as f:
_PAGE = f.read()
return _PAGE
def _send_all(conn, buf):
"""非阻塞发送，期间持续 tick（大页面也不会卡住步进）"""
mv = memoryview(buf) if not isinstance(buf, memoryview) else buf
n = 0
t0 = time.ticks_ms()
while n < len(mv) and time.ticks_diff(time.ticks_ms(), t0) < 3000:
st.tick()
try:
k = conn.send(mv[n:])
except OSError:
k = 0
if k:
n += k
else:
time.sleep_ms(1)
def _send(conn, code, ctype, body):
if isinstance(body, str):
body = body.encode()
head = ("HTTP/1.1 %d OK\r\nContent-Type: %s\r\nContent-Length: %d"
"\r\nCache-Control: no-store\r\nConnection: close\r\n\r\n"
% (code, ctype, len(body))).encode()
try:
_send_all(conn, head)
_send_all(conn, body)
except OSError:
pass
try:
conn.close()
except OSError:
pass
def _query(qs):
d = {}
for kv in qs.split("&"):
if "=" in kv:
k, v = kv.split("=", 1)
d[k] = v
return d
def _handle(conn, req):
global _poll, _last_moving, _hb_tick, _hb_dt
line = req.split("\r\n", 1)[0]
parts = line.split(" ")
if len(parts) < 2:
return _send(conn, 400, "text/plain", "bad request")
route, _, qs = parts[1].partition("?")
q = _query(qs)
if route != "/api/status": # 状态轮询量大，单独用心跳节流
print("[请求] %s %s" % (parts[0], parts[1]))
if route == "/":
try:
return _send(conn, 200, "text/html; charset=utf-8", _page())
except OSError:
return _send(conn, 500, "text/plain", "index.html not found")
if route == "/api/status":
_poll += 1
if _poll % 20 == 0: # 页面每 0.5s 轮询一次 → 约 10s 打一条心跳
tk, dts, dtm = st.stats(True)
print("[心跳] tick %d 次(%d Hz) 累计 %d ms 最长间隔 %d ms（健康:>=200Hz, 最长<=5ms）"
% (tk - _hb_tick, (tk - _hb_tick) // 10, dts - _hb_dt, dtm))
_hb_tick, _hb_dt = tk, dts
a = st.angles()
m = st.moving()
if _last_moving and not m:
print("[到位] j1=%.1f j2=%.1f j3=%.1f" % (a[0], a[1], a[2]))
_last_moving = m
body = '{"j1":%.1f,"j2":%.1f,"j3":%.1f,"moving":%d}' % (
a[0], a[1], a[2], 1 if m else 0)
return _send(conn, 200, "application/json", body)
if route == "/api/move":
def f(k):
try:
return float(q.get(k, "0"))
except ValueError:
return 0.0
try:
hz = int(float(q.get("hz", "400")))
except ValueError:
hz = 400
sync = q.get("sync", "0") == "1"
t1, t2, t3 = f("j1"), f("j2"), f("j3")
st.set_targets(t1, t2, t3, hz, sync)
print("[move] -> 剩余步数 %s 运动中 %s" % (st.steps_left(), st.moving()))
a = st.angles()
body = '{"ok":1,"j1":%.1f,"j2":%.1f,"j3":%.1f}' % (a[0], a[1], a[2])
return _send(conn, 200, "application/json", body)
if route == "/api/stop":
st.stop_all()
print("[急停] 当前角度 %s" % (tuple("%.1f" % x for x in st.angles()),))
return _send(conn, 200, "application/json", '{"ok":1}')
if route == "/api/home":
print("[回中] 三轴回 0°")
st.home()
return _send(conn, 200, "application/json", '{"ok":1}')
return _send(conn, 404, "text/plain", "not found")
def _recv_req(conn, tmo_ms=1500):
"""非阻塞收请求，期间持续 tick；对端关闭返回已收到的内容"""
t0 = time.ticks_ms()
data = b""
while time.ticks_diff(time.ticks_ms(), t0) < tmo_ms:
st.tick()
try:
chunk = conn.recv(512)
except OSError: # EAGAIN：暂时没数据
time.sleep_ms(1)
continue
if not chunk: # 对端关闭
break
data += chunk
if b"\r\n\r\n" in data or len(data) > 2048:
break
return data
def _probe(s):
"""启动自测：量非阻塞 accept 单次耗时(ms)，用来决定多久碰一次 socket"""
try:
t0 = time.ticks_ms()
for _ in range(30):
try:
s.accept()
except OSError:
pass
return time.ticks_diff(time.ticks_ms(), t0) / 30.0
except Exception as e:
print("自测跳过:", e)
return 0.5
def run():
_wifi()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(("0.0.0.0", PORT))
s.listen(3)
s.setblocking(False) # 关键：非阻塞。阻塞超时路径实测要 ~65ms，主循环只有 15Hz
print("Web 服务已启动，端口 %d（非阻塞 + 主循环 tick 驱动步进）" % PORT)
cost = _probe(s)
# socket 调用会卡住 tick，所以不必每轮都碰 socket：
# 让 socket 开销占总时间 <20%，其余时间留给 tick
poll_n = 1
if cost > 2.0:
poll_n = min(100, int(cost / 2.0) + 1)
print("自测：accept 单次 %.2fms → 每 %d 轮轮询一次 socket，预计 tick %.0f Hz（步进需 >=200）"
% (cost, poll_n, 1000.0 / (LOOP_SLEEP_MS + cost / poll_n)))
st.tick(); st.stats(True) # 重置计时基准，避免把 WiFi 等待算进"最长间隔"
i = 0
while True:
st.tick()
i += 1
if i % poll_n == 0:
try:
conn, addr = s.accept()
except OSError:
conn = None
if conn:
conn.setblocking(False)
try:
req = _recv_req(conn)
if req:
_handle(conn, req.decode("utf-8", "ignore"))
else:
conn.close()
except Exception as e:
print("请求异常:", e)
try:
conn.close()
except Exception:
pass
time.sleep_ms(LOOP_SLEEP_MS)
run()
网页端每0.5秒轮询一次/api/status同步真实角度；拖动时只更新动画不发请求，松手才下发一次，无论拖多快机械臂只会收到一条指令。
4、使用方法
把三个文件上传到ESP32，填好WiFi账号密码，运行服务器后串口会打印出控制页地址，同一WiFi下的手机或电脑打开该地址即可。上电前先把机械臂摆在中间位置，那是它的0°。
我要赚赏金
STM32
MCU
通讯及无线技术
物联网技术
电子DIY
板卡试用
基础知识
软件与操作系统
我爱生活
小e食堂
