网页后端服务运行在树莓派上，首先设计项目结构
web-control/
├── app.py # Flask Web服务器
├── requirements.txt # Python依赖包
├── static/
│ ├── css/
│ │ └── style.css # 样式表
│ └── js/
│ └── script.js # 前端JavaScript
└── templates/
└── index.html # 网页模板
后端用python设计，流程如下：
#!/usr/bin/env python3
"""
树莓派5 Flask Web服务器
"""
import socket
import threading
import time
from flask import Flask, render_template, request, jsonify, Response
app = Flask(__name__)
# 配置
ESP32_IP = "192.168.1.100" # 修改为你的ESP32 IP地址
ESP32_PORT = 5000
ESP32_TIMEOUT = 3
# 全局变量
tcp_socket = None
connection_status = False
status_lock = threading.Lock()
# 默认配置
default_config = {
"frequency": 1000,
"waveform": "SINE",
"output": True,
"ip": ESP32_IP,
"port": ESP32_PORT
}
# 波形名称映射
WAVEFORM_MAP = {
"SINE": "正弦波",
"SQUARE1": "方波1",
"SQUARE2": "方波2",
"TRIANGLE": "三角波"
}
class AD9833Controller:
def __init__(self, ip, port):
self.ip = ip
self.port = port
self.socket = None
self.connected = False
def connect(self):
"""连接到ESP32 TCP服务器"""
try:
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.settimeout(ESP32_TIMEOUT)
self.socket.connect((self.ip, self.port))
# 读取欢迎消息
welcome = self.receive_response()
print(f"Connected to ESP32: {welcome}")
self.connected = True
with status_lock:
global connection_status
connection_status = True
return True, "连接成功"
except Exception as e:
error_msg = f"连接失败: {str(e)}"
print(error_msg)
self.connected = False
with status_lock:
connection_status = False
return False, error_msg
def disconnect(self):
"""断开连接"""
if self.socket:
try:
self.send_command("EXIT")
self.socket.close()
except:
pass
finally:
self.socket = None
self.connected = False
with status_lock:
global connection_status
connection_status = False
def send_command(self, command):
"""发送命令到ESP32"""
if not self.socket or not self.connected:
success, message = self.connect()
if not success:
return False, "未连接到设备"
try:
# 发送命令（添加换行符）
self.socket.sendall((command + "\n").encode('utf-8'))
return True, "命令已发送"
except Exception as e:
self.connected = False
with status_lock:
connection_status = False
return False, f"发送失败: {str(e)}"
def receive_response(self, timeout=2):
"""接收ESP32的响应"""
if not self.socket:
return ""
try:
self.socket.settimeout(timeout)
response = b""
while True:
try:
chunk = self.socket.recv(1024)
if not chunk:
break
response += chunk
# 如果收到换行符或响应结束
if b'\n' in chunk or len(chunk) < 1024:
break
except socket.timeout:
break
return response.decode('utf-8', errors='ignore').strip()
except Exception as e:
print(f"接收响应错误: {e}")
return ""
def send_and_receive(self, command):
"""发送命令并接收响应"""
success, msg = self.send_command(command)
if not success:
return success, msg
time.sleep(0.1) # 等待ESP32处理
response = self.receive_response()
return True, response
def set_frequency(self, frequency):
"""设置频率"""
try:
freq = int(frequency)
if freq < 0 or freq > 12500000:
return False, "频率超出范围 (0-12,500,000 Hz)"
success, response = self.send_and_receive(f"FREQ={freq}")
return success, response
except ValueError:
return False, "无效的频率值"
def set_waveform(self, waveform):
"""设置波形"""
waveform = waveform.upper()
if waveform not in ["SINE", "SQUARE1", "SQUARE2", "TRIANGLE"]:
return False, "无效的波形类型"
success, response = self.send_and_receive(f"WAVE={waveform}")
return success, response
def set_output(self, enable):
"""启用/禁用输出"""
command = "ON" if enable else "OFF"
success, response = self.send_and_receive(command)
return success, response
def get_status(self):
"""获取设备状态"""
success, response = self.send_and_receive("STATUS")
if success and response:
# 解析状态响应
status_info = {"connected": True}
lines = response.split('\n')
for line in lines:
line = line.strip()
if "Frequency:" in line:
try:
freq_str = line.split(":")[1].strip().split(" ")[0]
status_info["frequency"] = int(freq_str)
except:
status_info["frequency"] = 0
elif "Waveform:" in line:
wave_str = line.split(":")[1].strip()
status_info["waveform"] = wave_str
elif "Output:" in line:
output_str = line.split(":")[1].strip()
status_info["output"] = (output_str == "ON")
return True, status_info
return success, response
def test_device(self):
"""运行测试序列"""
success, response = self.send_and_receive("TEST")
return success, response
def reset_device(self):
"""重置设备"""
success, response = self.send_and_receive("RESET")
return success, response
# 创建控制器实例
controller = AD9833Controller(ESP32_IP, ESP32_PORT)
# Flask路由
@app.route('/')
def index():
"""主页"""
return render_template('index.html',
ip=ESP32_IP,
port=ESP32_PORT,
default_freq=default_config["frequency"])
@app.route('/api/connect', methods=['POST'])
def api_connect():
"""连接设备API"""
data = request.json
ip = data.get('ip', ESP32_IP)
port = data.get('port', ESP32_PORT)
# 更新IP和端口
global ESP32_IP, ESP32_PORT
ESP32_IP = ip
ESP32_PORT = int(port)
# 创建新的控制器实例
global controller
controller = AD9833Controller(ESP32_IP, ESP32_PORT)
success, message = controller.connect()
if success:
# 获取初始状态
controller.get_status()
return jsonify({
"success": success,
"message": message,
"connected": success
})
@app.route('/api/disconnect', methods=['POST'])
def api_disconnect():
"""断开连接API"""
controller.disconnect()
return jsonify({
"success": True,
"message": "已断开连接",
"connected": False
})
@app.route('/api/status', methods=['GET'])
def api_status():
"""获取状态API"""
with status_lock:
connected = connection_status
if connected:
success, response = controller.get_status()
if success and isinstance(response, dict):
return jsonify({
"success": True,
"connected": True,
"data": response
})
elif success:
return jsonify({
"success": True,
"connected": True,
"data": {"message": response}
})
else:
return jsonify({
"success": False,
"connected": False,
"message": response
})
else:
return jsonify({
"success": False,
"connected": False,
"message": "未连接到设备"
})
@app.route('/api/set_frequency', methods=['POST'])
def api_set_frequency():
"""设置频率API"""
data = request.json
frequency = data.get('frequency', 1000)
success, message = controller.set_frequency(frequency)
return jsonify({
"success": success,
"message": message,
"frequency": frequency if success else None
})
@app.route('/api/set_waveform', methods=['POST'])
def api_set_waveform():
"""设置波形API"""
data = request.json
waveform = data.get('waveform', 'SINE')
success, message = controller.set_waveform(waveform)
return jsonify({
"success": success,
"message": message,
"waveform": waveform if success else None,
"waveform_cn": WAVEFORM_MAP.get(waveform, "未知")
})
@app.route('/api/set_output', methods=['POST'])
def api_set_output():
"""设置输出API"""
data = request.json
enable = data.get('enable', True)
success, message = controller.set_output(enable)
return jsonify({
"success": success,
"message": message,
"output": enable if success else None
})
@app.route('/api/test', methods=['POST'])
def api_test():
"""测试设备API"""
success, message = controller.test_device()
return jsonify({
"success": success,
"message": message
})
@app.route('/api/reset', methods=['POST'])
def api_reset():
"""重置设备API"""
success, message = controller.reset_device()
return jsonify({
"success": success,
"message": message
})
@app.route('/api/ping', methods=['GET'])
def api_ping():
"""心跳检测API"""
with status_lock:
return jsonify({
"success": True,
"connected": connection_status,
"timestamp": time.time()
})
# 错误处理
@app.errorhandler(404)
def not_found(error):
return jsonify({"success": False, "message": "资源未找到"}), 404
@app.errorhandler(500)
def server_error(error):
return jsonify({"success": False, "message": "服务器内部错误"}), 500
def start_status_monitor():
"""状态监控线程"""
def monitor():
while True:
time.sleep(5) # 每5秒检查一次
with status_lock:
if connection_status and controller.connected:
try:
# 发送PING命令保持连接
controller.send_command("PING")
except:
connection_status = False
monitor_thread = threading.Thread(target=monitor, daemon=True)
monitor_thread.start()
if __name__ == '__main__':
print("=" * 50)
print("Web控制程序")
print("=" * 50)
print(f"目标设备: {ESP32_IP}:{ESP32_PORT}")
print("Web界面: http://0.0.0.0:8080")
print("=" * 50)
# 启动状态监控
start_status_monitor()
# 启动Flask服务器
app.run(host='0.0.0.0', port=8080, debug=False, threaded=True)样式表、html文件
树莓派终端运行方法：
source venv/bin/activate
# 运行Flask应用
python app.py
# 或者以后台方式运行
nohup python app.py > webapp.log 2>&1 &
效果如下：
我要赚赏金
STM32
MCU
通讯及无线技术
物联网技术
电子DIY
板卡试用
基础知识
软件与操作系统
我爱生活
小e食堂
