当前位置：首页 > news >正文

树莓派 AT 指令串口助手

news 2025/9/4 8:03:06

简约中文界面

#!/usr/bin/env python3
# -*- coding: utf-8 -*-HTTP_PORT = 12345
SER_BAUD = 115200
IFACE_NAME = "wlan0"from flask import Flask, request, jsonify, render_template_string
import serial, serial.tools.list_ports as list_ports
import threading, time, socket, fcntl, structapp = Flask(__name__)
ser = None
recv_buf = []
lock = threading.Lock()def get_ip(iface):try:s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)return socket.inet_ntoa(fcntl.ioctl(s.fileno(), 0x8915, struct.pack('256s', iface[:15].encode()))[20:24])except Exception:return "0.0.0.0"def reader():global serwhile ser and ser.is_open:try:data = ser.read(4096)if data:text = data.decode(errors='replace')with lock:recv_buf.append(text)time.sleep(0.05)except:breakINDEX_HTML = """
<!doctype html>
<html lang="zh">
<head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1"><title>串口调试助手</title><style>body { font-family: Arial, sans-serif; max-width: 900px; margin: 0 auto; padding: 20px; background-color: #f5f5f5; }h1 { color: #333; text-align: center; }.card { background: white; border-radius: 8px; padding: 20px; margin-bottom: 20px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }input, button { padding: 8px 12px; margin: 5px; border-radius: 4px; border: 1px solid #ddd; }button { background: #4CAF50; color: white; border: none; cursor: pointer; }button:hover { background: #45a049; }#recv { background: #f8f8f8; border: 1px solid #ddd; height: 300px; overflow: auto; padding: 10px; white-space: pre-wrap; }.status { padding: 10px; margin: 10px 0; border-radius: 4px; }.connected { background: #d4edda; color: #155724; }.disconnected { background: #f8d7da; color: #721c24; }</style>
</head>
<body><h1>串口调试助手</h1><div class="card"><p>访问地址: {{ url }}</p><div id="status" class="status disconnected">未连接串口</div></div><div class="card"><h2>串口设置</h2><div><select id="portSelect"><option value="">选择串口</option></select><select id="baudSelect"><option value="9600">9600</option><option value="19200">19200</option><option value="38400">38400</option><option value="57600">57600</option><option value="115200" selected>115200</option></select><button onclick="refreshPorts()">刷新串口</button><button id="connectBtn" onclick="openPort()">连接</button></div></div><div class="card"><h2>发送数据</h2><input id="sendText" placeholder="输入要发送的内容" style="width: 300px;" /><label><input type="checkbox" id="crlf" checked /> 添加CRLF</label><button onclick="sendData()">发送</button></div><div class="card"><h2>接收数据 <button onclick="clearRecv()">清空</button></h2><div id="recv"></div></div><script>let isConnected = false;async function refreshPorts() {const response = await fetch('/ports');const ports = await response.json();const select = document.getElementById('portSelect');select.innerHTML = '<option value="">选择串口</option>';ports.forEach(port => {select.innerHTML += `<option value="${port}">${port}</option>`;});}async function openPort() {const port = document.getElementById('portSelect').value;const baud = document.getElementById('baudSelect').value;if (!port) {alert('请选择串口');return;}try {await fetch('/open', {method: 'POST',headers: {'Content-Type': 'application/json'},body: JSON.stringify({port, baud})});isConnected = true;document.getElementById('status').className = 'status connected';document.getElementById('status').textContent = `已连接: ${port} @ ${baud}bps`;document.getElementById('connectBtn').textContent = '断开';document.getElementById('connectBtn').onclick = closePort;} catch (error) {alert('连接失败: ' + error);}}async function closePort() {isConnected = false;document.getElementById('status').className = 'status disconnected';document.getElementById('status').textContent = '未连接串口';document.getElementById('connectBtn').textContent = '连接';document.getElementById('connectBtn').onclick = openPort;}async function sendData() {if (!isConnected) {alert('请先连接串口');return;}const text = document.getElementById('sendText').value;const crlf = document.getElementById('crlf').checked;await fetch('/send', {method: 'POST',headers: {'Content-Type': 'application/json'},body: JSON.stringify({text, crlf})});}function clearRecv() {document.getElementById('recv').textContent = '';}async function poll() {try {const response = await fetch('/recv');const data = await response.json();if (data.data) {const recvElement = document.getElementById('recv');recvElement.textContent += data.data;recvElement.scrollTop = recvElement.scrollHeight;}} catch (error) {console.error('接收错误:', error);}setTimeout(poll, 200);}// 初始化refreshPorts();poll();</script>
</body>
</html>
"""@app.route("/")
def index():return render_template_string(INDEX_HTML, url=f"http://{get_ip(IFACE_NAME)}:{HTTP_PORT}")@app.route("/ports")
def ports():return jsonify([p.device for p in list_ports.comports()])@app.route("/open", methods=["POST"])
def open_port():global sertry:if ser and ser.is_open:ser.close()data = request.jsonport = data.get("port")baud = int(data.get("baud", SER_BAUD))ser = serial.Serial(port, baud, timeout=0)threading.Thread(target=reader, daemon=True).start()return jsonify({"ok": True})except Exception as e:return jsonify({"ok": False, "error": str(e)})@app.route("/send", methods=["POST"])
def send():global serif not ser or not ser.is_open:return jsonify({"ok": False, "error": "串口未打开"})try:data = request.jsontxt = data.get("text", "")if data.get("crlf", True):txt += "\r\n"ser.write(txt.encode())return jsonify({"ok": True})except Exception as e:return jsonify({"ok": False, "error": str(e)})@app.route("/recv")
def recv():with lock:data = "".join(recv_buf)recv_buf.clear()return jsonify({"data": data})if __name__ == "__main__":print(f"服务已启动: http://{get_ip(IFACE_NAME)}:{HTTP_PORT}")app.run(host="0.0.0.0", port=HTTP_PORT, debug=False)

在这里插入图片描述

炫酷中文界面

#!/usr/bin/env python3
# -*- coding: utf-8 -*-HTTP_PORT = 12345
SER_BAUD = 115200
IFACE_NAME = "wlan0"from flask import Flask, request, jsonify, render_template_string
import serial, serial.tools.list_ports as list_ports
import threading, time, socket, fcntl, struct
import json
import osapp = Flask(__name__)
ser = None
recv_buf = []
send_history = []
lock = threading.Lock()
HISTORY_FILE = 'send_history.json'# Load send history from file
def load_history():global send_historytry:if os.path.exists(HISTORY_FILE):with open(HISTORY_FILE, 'r') as f:send_history = json.load(f)except:send_history = []# Save send history to file
def save_history():try:with open(HISTORY_FILE, 'w') as f:json.dump(send_history, f)except:pass# Initialize history
load_history()def get_ip(iface):try:s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)return socket.inet_ntoa(fcntl.ioctl(s.fileno(), 0x8915, struct.pack('256s', iface[:15].encode()))[20:24])except Exception:return "0.0.0.0"def reader():global serwhile ser and ser.is_open:try:data = ser.read(4096)if data:text = data.decode(errors='replace')with lock:recv_buf.append(text)time.sleep(0.05)except:breakINDEX_HTML = """
<!doctype html>
<html lang="en">
<head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1"><title>Serial Terminal</title><style>:root {--primary: #00f3ff;--secondary: #007bff;--accent: #7e42ff;--dark: #0a1929;--darker: #061020;--light: #1a2b45;}body { font-family: 'Segoe UI', 'Roboto Mono', monospace; max-width: 1200px; margin: 0 auto; padding: 20px; background-color: var(--darker);color: #e0e0e0;background-image: radial-gradient(circle at 15% 50%, rgba(0, 243, 255, 0.05) 0%, transparent 20%),radial-gradient(circle at 85% 30%, rgba(126, 66, 255, 0.05) 0%, transparent 20%),linear-gradient(to bottom, var(--darker), var(--dark));}h1 { color: var(--primary);text-align: center; font-weight: 300;letter-spacing: 2px;text-transform: uppercase;margin-bottom: 30px;text-shadow: 0 0 10px var(--primary), 0 0 20px var(--primary);position: relative;}h1:after {content: '';position: absolute;bottom: -10px;left: 50%;transform: translateX(-50%);width: 100px;height: 2px;background: linear-gradient(90deg, transparent, var(--primary), transparent);}h2 {color: var(--secondary);font-weight: 400;margin-top: 0;border-left: 3px solid var(--accent);padding-left: 15px;}.card { background: rgba(26, 43, 69, 0.6); border-radius: 8px; padding: 25px; margin-bottom: 25px; border: 1px solid rgba(0, 243, 255, 0.1);box-shadow: 0 4px 20px rgba(0, 0, 0, 0.3);backdrop-filter: blur(10px);position: relative;overflow: hidden;}.card:before {content: '';position: absolute;top: 0;left: -100%;width: 100%;height: 100%;background: linear-gradient(90deg, transparent, rgba(0, 243, 255, 0.1), transparent);transition: 0.5s;}.card:hover:before {left: 100%;}input, select, button, textarea { padding: 12px 18px; margin: 8px 5px; border-radius: 6px; border: 1px solid rgba(0, 243, 255, 0.3);background: rgba(10, 25, 41, 0.8);color: #fff;font-family: inherit;font-size: 14px;transition: all 0.3s ease;}input:focus, select:focus, textarea:focus {outline: none;border-color: var(--primary);box-shadow: 0 0 15px rgba(0, 243, 255, 0.3);}button { background: linear-gradient(135deg, var(--secondary), var(--accent));color: white; border: none; cursor: pointer; font-weight: 500;letter-spacing: 0.5px;text-transform: uppercase;font-size: 12px;box-shadow: 0 4px 15px rgba(0, 123, 255, 0.3);}button:hover { transform: translateY(-2px);box-shadow: 0 6px 20px rgba(126, 66, 255, 0.4);}button.secondary {background: rgba(10, 25, 41, 0.8);border: 1px solid rgba(0, 243, 255, 0.3);}#recv { background: rgba(10, 25, 41, 0.8); border: 1px solid rgba(0, 243, 255, 0.2); height: 300px; overflow: auto; padding: 15px; white-space: pre-wrap;border-radius: 6px;font-family: 'Fira Code', 'Cascadia Code', monospace;font-size: 13px;color: var(--primary);text-shadow: 0 0 5px rgba(0, 243, 255, 0.5);}.status { padding: 15px; margin: 15px 0; border-radius: 6px; text-align: center;font-weight: 500;letter-spacing: 1px;border: 1px solid;}.connected { background: rgba(0, 200, 83, 0.1);color: #00c853;border-color: rgba(0, 200, 83, 0.3);box-shadow: 0 0 15px rgba(0, 200, 83, 0.2);}.disconnected { background: rgba(255, 68, 68, 0.1);color: #ff4444;border-color: rgba(255, 68, 68, 0.3);box-shadow: 0 0 15px rgba(255, 68, 68, 0.2);}.grid {display: grid;grid-template-columns: 1fr 1fr;gap: 20px;}.history-panel {max-height: 200px;overflow-y: auto;margin-top: 15px;padding: 10px;background: rgba(10, 25, 41, 0.6);border-radius: 6px;border: 1px solid rgba(0, 243, 255, 0.2);}.history-item {padding: 8px;margin: 5px 0;background: rgba(0, 243, 255, 0.1);border-radius: 4px;cursor: pointer;transition: all 0.2s ease;font-family: 'Fira Code', monospace;font-size: 12px;word-break: break-all;}.history-item:hover {background: rgba(0, 243, 255, 0.2);transform: translateX(5px);}.history-actions {display: flex;gap: 10px;margin-top: 10px;}@media (max-width: 968px) {.grid {grid-template-columns: 1fr;}}.pulse {animation: pulse 2s infinite;}@keyframes pulse {0% { opacity: 0.7; }50% { opacity: 1; }100% { opacity: 0.7; }}.glow-text {text-shadow: 0 0 10px currentColor;}</style>
</head>
<body><h1>Serial Terminal</h1><div class="card"><p>Access URL: <span class="glow-text">{{ url }}</span></p><div id="status" class="status disconnected">🔴 Disconnected</div></div><div class="grid"><div class="card"><h2>Serial Configuration</h2><select id="portSelect"><option value="">Select serial port...</option></select><select id="baudSelect"><option value="9600">9600 baud</option><option value="19200">19200 baud</option><option value="38400">38400 baud</option><option value="57600">57600 baud</option><option value="115200" selected>115200 baud</option></select><button onclick="refreshPorts()">🔄 Scan Ports</button><button id="connectBtn" onclick="openPort()">🚀 Connect</button></div><div class="card"><h2>Send Data</h2><input id="sendText" placeholder="Enter data to send..." style="width: 100%;" /><label><input type="checkbox" id="crlf" checked /> Add CRLF</label><button onclick="sendData()">📨 Send</button><h3 style="margin-top: 20px;">Send History</h3><div class="history-panel" id="historyPanel"><div id="historyList"></div></div><div class="history-actions"><button onclick="clearHistory()" class="secondary">Clear History</button><button onclick="saveHistoryToFile()" class="secondary">Export</button></div></div></div><div class="card"><h2>Receive Data <button onclick="clearRecv()">Clear Buffer</button></h2><div id="recv">>> Waiting for data...</div></div><script>let isConnected = false;let sendHistory = [];// Load history from serverasync function loadHistory() {try {const response = await fetch('/history');const data = await response.json();sendHistory = data.history || [];updateHistoryDisplay();} catch (error) {console.error('Failed to load history:', error);}}// Update history list displayfunction updateHistoryDisplay() {const historyList = document.getElementById('historyList');historyList.innerHTML = '';if (sendHistory.length === 0) {historyList.innerHTML = '<div style="text-align: center; color: #888; padding: 20px;">No history yet</div>';return;}// Show latest firstsendHistory.slice().reverse().forEach((item, index) => {const div = document.createElement('div');div.className = 'history-item';div.textContent = item.text + (item.crlf ? '\\r\\n' : '');div.onclick = () => {document.getElementById('sendText').value = item.text;document.getElementById('crlf').checked = item.crlf;};historyList.appendChild(div);});}// Clear historyasync function clearHistory() {if (confirm('Are you sure you want to clear all send history?')) {try {await fetch('/clear_history', { method: 'POST' });sendHistory = [];updateHistoryDisplay();} catch (error) {console.error('Failed to clear history:', error);}}}// Export history to filefunction saveHistoryToFile() {const data = sendHistory.map(item => item.text + (item.crlf ? '\\r\\n' : '')).join('\\n');const blob = new Blob([data], { type: 'text/plain' });const url = URL.createObjectURL(blob);const a = document.createElement('a');a.href = url;a.download = 'serial_history.txt';a.click();URL.revokeObjectURL(url);}async function refreshPorts() {try {const response = await fetch('/ports');const ports = await response.json();const select = document.getElementById('portSelect');select.innerHTML = '<option value="">Select serial port...</option>';ports.forEach(port => {select.innerHTML += `<option value="${port}">${port}</option>`;});} catch (error) {console.error('Port scan failed:', error);}}async function openPort() {const port = document.getElementById('portSelect').value;const baud = document.getElementById('baudSelect').value;if (!port) {alert('Please select a valid serial port');return;}try {const response = await fetch('/open', {method: 'POST',headers: {'Content-Type': 'application/json'},body: JSON.stringify({port, baud})});const result = await response.json();if (result.ok) {isConnected = true;document.getElementById('status').className = 'status connected';document.getElementById('status').textContent = `🟢 Connected: ${port} @ ${baud} baud`;document.getElementById('connectBtn').textContent = '🔌 Disconnect';document.getElementById('connectBtn').onclick = closePort;} else {alert(`Connection failed: ${result.error}`);}} catch (error) {alert('Connection error: ' + error);}}async function closePort() {isConnected = false;document.getElementById('status').className = 'status disconnected';document.getElementById('status').textContent = '🔴 Disconnected';document.getElementById('connectBtn').textContent = '🚀 Connect';document.getElementById('connectBtn').onclick = openPort;}async function sendData() {if (!isConnected) {alert('Please connect to a serial port first');return;}const text = document.getElementById('sendText').value;if (!text) {alert('Please enter data to send');return;}const crlf = document.getElementById('crlf').checked;try {const response = await fetch('/send', {method: 'POST',headers: {'Content-Type': 'application/json'},body: JSON.stringify({text, crlf})});const result = await response.json();if (result.ok) {// Add to history and reloadawait fetch('/add_history', {method: 'POST',headers: {'Content-Type': 'application/json'},body: JSON.stringify({text, crlf})});await loadHistory();document.getElementById('sendText').value = '';}} catch (error) {alert('Send failed: ' + error);}}function clearRecv() {document.getElementById('recv').textContent = '>> Buffer cleared';}async function poll() {try {const response = await fetch('/recv');const data = await response.json();if (data.data) {const recvElement = document.getElementById('recv');recvElement.textContent += data.data;recvElement.scrollTop = recvElement.scrollHeight;}} catch (error) {console.error('Receive error:', error);}setTimeout(poll, 100);}// Initializedocument.addEventListener('DOMContentLoaded', function() {refreshPorts();loadHistory();poll();});</script>
</body>
</html>
"""@app.route("/")
def index():return render_template_string(INDEX_HTML, url=f"http://{get_ip(IFACE_NAME)}:{HTTP_PORT}")@app.route("/ports")
def ports():return jsonify([p.device for p in list_ports.comports()])@app.route("/open", methods=["POST"])
def open_port():global sertry:if ser and ser.is_open:ser.close()data = request.jsonport = data.get("port")baud = int(data.get("baud", SER_BAUD))ser = serial.Serial(port, baud, timeout=0)threading.Thread(target=reader, daemon=True).start()return jsonify({"ok": True})except Exception as e:return jsonify({"ok": False, "error": str(e)})@app.route("/send", methods=["POST"])
def send():global serif not ser or not ser.is_open:return jsonify({"ok": False, "error": "Serial port not open"})try:data = request.jsontxt = data.get("text", "")if data.get("crlf", True):txt += "\r\n"ser.write(txt.encode())return jsonify({"ok": True})except Exception as e:return jsonify({"ok": False, "error": str(e)})@app.route("/recv")
def recv():with lock:data = "".join(recv_buf)recv_buf.clear()return jsonify({"data": data})@app.route("/history")
def get_history():return jsonify({"history": send_history})@app.route("/add_history", methods=["POST"])
def add_history():global send_historydata = request.jsontext = data.get("text", "")crlf = data.get("crlf", True)# Add to history (limit to 50 items)send_history.append({"text": text, "crlf": crlf})if len(send_history) > 50:send_history = send_history[-50:]save_history()return jsonify({"ok": True})@app.route("/clear_history", methods=["POST"])
def clear_history():global send_historysend_history = []save_history()return jsonify({"ok": True})if __name__ == "__main__":print(f"Serial Terminal started: http://{get_ip(IFACE_NAME)}:{HTTP_PORT}")app.run(host="0.0.0.0", port=HTTP_PORT, debug=False)