703 lines
24 KiB
Python
703 lines
24 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
手机蓝牙通信 — FastAPI 服务 (单文件版)
|
||
=======================================
|
||
PC端通过蓝牙 SPP 模拟 Trimble DiNi 水准仪，等待手机APP连接。
|
||
当手机发送 ?0100 轮询时，自动将排队的测量数据以 M5 格式发送给手机。
|
||
|
||
用法:
|
||
python main.py # 默认 0.0.0.0:58100
|
||
python main.py --port 8080 # 指定 HTTP 端口
|
||
python main.py --bt-port COM3 # 启动时自动连接蓝牙 COM 口
|
||
|
||
API 接口:
|
||
GET / 服务信息 + 连接状态 + COM 口列表
|
||
GET /test 测试接口 (未连接→COM列表, 已连接→手机状态)
|
||
POST /connect {port} 打开蓝牙 COM 口, 等待手机连接
|
||
POST /disconnect 断开
|
||
POST /command {cmd} 命令: send <r> <hd> 发送测量 | disconnect 断开
|
||
GET /status 手机连接状态 + 待发送队列
|
||
|
||
协议说明:
|
||
手机→PC: 0x02F0... 握手 | ?0100 轮询 | 0x050B8D 控制码 | KENC 编码
|
||
PC→手机: 0x02E0... 握手 | M5 测量数据行
|
||
|
||
依赖: pip install fastapi uvicorn pyserial
|
||
"""
|
||
|
||
import sys
|
||
import time
|
||
import datetime
|
||
import hashlib
|
||
import threading
|
||
import asyncio
|
||
import argparse
|
||
from contextlib import asynccontextmanager
|
||
from typing import Optional
|
||
|
||
import serial
|
||
import serial.tools.list_ports
|
||
from fastapi import FastAPI, HTTPException, Query
|
||
from pydantic import BaseModel
|
||
|
||
# ════════════════════════════════════════════════════════════════
|
||
# 常量
|
||
# ════════════════════════════════════════════════════════════════
|
||
|
||
DEFAULT_HTTP_PORT = 58100
|
||
DEFAULT_BAUDRATE = 9600
|
||
|
||
# ── 协议常量 ──
|
||
HANDSHAKE_PHONE = bytes([0x02, 0xF0, 0x00, 0x00, 0xDE, 0x03, 0x00, 0x07])
|
||
HANDSHAKE_PC = bytes([0x02, 0xE0, 0x00, 0x00, 0x7F, 0x00, 0x00, 0x07])
|
||
CONTROL_CODE = bytes([0x05, 0x0B, 0x8D])
|
||
QUERY_MARKER = b"?0100"
|
||
KENC_ON_MARKER = b"KENC | 1 bit"
|
||
KENC_OFF_MARKER = b"KENC | 0 bit"
|
||
END_ACK = b"\r\n@"
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════
|
||
# COM 口扫描
|
||
# ════════════════════════════════════════════════════════════════
|
||
|
||
def list_com_ports() -> list[dict]:
|
||
"""列出所有 COM 口并标注蓝牙类型"""
|
||
result = []
|
||
for p in serial.tools.list_ports.comports():
|
||
hwid = (p.hwid or "").upper()
|
||
desc_raw = p.description or ""
|
||
|
||
is_bt_spp = "00001101" in hwid
|
||
is_incoming = is_bt_spp and "000000000000" in hwid
|
||
is_outgoing = is_bt_spp and not is_incoming
|
||
is_bluetooth = "BTHENUM" in hwid or is_bt_spp or \
|
||
any(kw in desc_raw.lower() for kw in
|
||
["bluetooth", "spp", "rfcomm", "蓝牙"])
|
||
|
||
result.append({
|
||
"device": p.device,
|
||
"description": desc_raw,
|
||
"hwid": p.hwid or "",
|
||
"is_bluetooth": bool(is_bluetooth),
|
||
"is_spp_incoming": is_incoming,
|
||
"is_spp_outgoing": is_outgoing,
|
||
})
|
||
return result
|
||
|
||
|
||
def find_bt_spp_ports() -> tuple[list[str], list[str]]:
|
||
"""返回 (incoming, outgoing) — incoming是手机接入端口, 推荐"""
|
||
incoming, outgoing = [], []
|
||
for p in serial.tools.list_ports.comports():
|
||
hwid = (p.hwid or "").upper()
|
||
if "BTHENUM" in hwid and "00001101" in hwid:
|
||
if "000000000000" in hwid:
|
||
incoming.append(p.device)
|
||
else:
|
||
outgoing.append(p.device)
|
||
return incoming, outgoing
|
||
|
||
|
||
def print_com_list():
|
||
"""打印可连接 COM 口到控制台"""
|
||
ports = list_com_ports()
|
||
incoming, outgoing = find_bt_spp_ports()
|
||
|
||
print("=" * 60)
|
||
print(" 可连接 COM 口列表 (手机通信)")
|
||
print("-" * 60)
|
||
if not ports:
|
||
print(" (未检测到任何 COM 口)")
|
||
else:
|
||
rec_ports = set(incoming)
|
||
bt = [p for p in ports if p.get("is_bluetooth")]
|
||
other = [p for p in ports if not p.get("is_bluetooth")]
|
||
if bt:
|
||
print(f"\n 蓝牙端口 ({len(bt)} 个):")
|
||
for p in bt:
|
||
rec = " ★ 推荐(手机接入)" if p["device"] in rec_ports else ""
|
||
print(f" {p['device']:<8} — {p['description']}{rec}")
|
||
if other:
|
||
print(f"\n 其他 COM 口 ({len(other)} 个):")
|
||
for p in other:
|
||
print(f" {p['device']:<8} — {p['description']}")
|
||
print("=" * 60)
|
||
return ports
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════
|
||
# 协议处理器 — 解析手机发来的二进制/文本数据
|
||
# ════════════════════════════════════════════════════════════════
|
||
|
||
class ProtocolHandler:
|
||
"""处理手机 APP 通过蓝牙模块发来的数据，提取握手/轮询/控制命令"""
|
||
|
||
def __init__(self):
|
||
self.buf = b""
|
||
|
||
def feed(self, data: bytes) -> tuple[list[bytes], bool]:
|
||
"""
|
||
喂入数据 → (待发送响应列表, 是否有 ?0100 轮询)
|
||
"""
|
||
self.buf += data
|
||
responses = []
|
||
has_query = False
|
||
|
||
while self.buf:
|
||
first = self.buf[:1]
|
||
|
||
# ── 二进制握手 (8字节) ──
|
||
if first == b"\x02":
|
||
if len(self.buf) >= 8 and self.buf[:8] == HANDSHAKE_PHONE:
|
||
responses.append(HANDSHAKE_PC)
|
||
self.buf = self.buf[8:]
|
||
continue
|
||
break
|
||
|
||
# ── 短控制码 (3字节) ──
|
||
elif first == b"\x05":
|
||
if len(self.buf) >= 3 and self.buf[:3] == CONTROL_CODE:
|
||
responses.append(CONTROL_CODE)
|
||
self.buf = self.buf[3:]
|
||
continue
|
||
break
|
||
|
||
# ── ?0100 轮询 ──
|
||
elif first in (b"\x3f", b"?"):
|
||
idx = self.buf.find(QUERY_MARKER)
|
||
if idx >= 0:
|
||
if idx > 0:
|
||
self.buf = self.buf[idx:]
|
||
has_query = True
|
||
self.buf = self.buf[len(QUERY_MARKER):]
|
||
if self.buf[:2] == b"\r\n":
|
||
self.buf = self.buf[2:]
|
||
continue
|
||
break
|
||
|
||
# ── KENC 编码设置 ──
|
||
elif KENC_ON_MARKER in self.buf:
|
||
idx = self.buf.find(KENC_ON_MARKER)
|
||
self.buf = self.buf[idx + len(KENC_ON_MARKER):]
|
||
if self.buf[:2] == b"\r\n":
|
||
self.buf = self.buf[2:]
|
||
continue
|
||
|
||
elif KENC_OFF_MARKER in self.buf:
|
||
idx = self.buf.find(KENC_OFF_MARKER)
|
||
responses.append(END_ACK)
|
||
self.buf = self.buf[idx + len(KENC_OFF_MARKER):]
|
||
if self.buf[:2] == b"\r\n":
|
||
self.buf = self.buf[2:]
|
||
continue
|
||
|
||
# ── 非打印字节 → 跳过 (RFCOMM Credit 残留) ──
|
||
elif self.buf[0] < 0x20:
|
||
self.buf = self.buf[1:]
|
||
continue
|
||
|
||
else:
|
||
break
|
||
|
||
# 缓冲区保护
|
||
if len(self.buf) > 4096:
|
||
self.buf = b""
|
||
|
||
return responses, has_query
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════
|
||
# 测量数据生成 — 模拟 DiNi M5 格式
|
||
# ════════════════════════════════════════════════════════════════
|
||
|
||
class MeasurementBuilder:
|
||
"""生成 M5 格式测量数据 (模拟 Trimble DiNi 03 输出)"""
|
||
|
||
@staticmethod
|
||
def build(staff_reading: float, distance: float) -> tuple[bytes, bytes]:
|
||
"""返回 (line1_bytes, line2_bytes)"""
|
||
now = datetime.datetime.now()
|
||
ts = now.strftime("%H:%M:%S") + str(now.microsecond // 100000)
|
||
r, hd = staff_reading, distance
|
||
|
||
h = hashlib.md5(
|
||
f"Trimble DiNi 03|{ts}|{r:.5f}|{hd:.3f}|salt".encode()
|
||
).hexdigest()
|
||
|
||
line1 = (
|
||
f"For M5|Adr |KD1 {ts} "
|
||
f"|R {r:.5f} m "
|
||
f"|HD {hd:.3f} m "
|
||
f"| @"
|
||
)
|
||
line2 = f" | <Trimble DiNi 03>{h}\r\n@"
|
||
|
||
return line1.encode("ascii"), line2.encode("ascii")
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════
|
||
# 蓝牙串口 — 后台线程读写
|
||
# ════════════════════════════════════════════════════════════════
|
||
|
||
class BtSerial:
|
||
"""蓝牙虚拟串口 — 非阻塞读, 后台线程持续收数据"""
|
||
|
||
def __init__(self, port: str):
|
||
self.port = port
|
||
self.ser: Optional[serial.Serial] = None
|
||
self.running = False
|
||
self._phone_connected = False
|
||
self._rx_queue = b""
|
||
self._read_thread: Optional[threading.Thread] = None
|
||
self._lock = threading.Lock()
|
||
self._log_cb = None
|
||
self._last_data_time = 0.0
|
||
self._total_bytes_rx = 0
|
||
self._total_bytes_tx = 0
|
||
|
||
def _log(self, msg: str):
|
||
if self._log_cb:
|
||
self._log_cb(msg)
|
||
|
||
def open(self) -> bool:
|
||
try:
|
||
self.ser = serial.Serial(
|
||
self.port, baudrate=DEFAULT_BAUDRATE,
|
||
timeout=0, # 非阻塞
|
||
write_timeout=1.0,
|
||
)
|
||
except Exception as e:
|
||
self._log(f"✗ 打开 {self.port} 失败: {e}")
|
||
return False
|
||
|
||
self.running = True
|
||
self._log(f"✓ {self.port} 已打开, 等待手机连接...")
|
||
|
||
self._read_thread = threading.Thread(target=self._read_loop, daemon=True)
|
||
self._read_thread.start()
|
||
return True
|
||
|
||
def close(self):
|
||
self.running = False
|
||
if self._read_thread and self._read_thread.is_alive():
|
||
self._read_thread.join(timeout=2)
|
||
if self.ser:
|
||
try:
|
||
self.ser.close()
|
||
except Exception:
|
||
pass
|
||
self.ser = None
|
||
self._phone_connected = False
|
||
|
||
@property
|
||
def phone_connected(self) -> bool:
|
||
return self._phone_connected
|
||
|
||
@property
|
||
def total_rx(self) -> int:
|
||
return self._total_bytes_rx
|
||
|
||
@property
|
||
def total_tx(self) -> int:
|
||
return self._total_bytes_tx
|
||
|
||
def _read_loop(self):
|
||
"""后台读线程 — 非阻塞轮询"""
|
||
buf = b""
|
||
while self.running:
|
||
try:
|
||
if not self.ser or not self.ser.is_open:
|
||
time.sleep(0.1)
|
||
continue
|
||
|
||
chunk = self.ser.read(4096)
|
||
if chunk:
|
||
buf += chunk
|
||
self._total_bytes_rx += len(chunk)
|
||
self._last_data_time = time.time()
|
||
|
||
if not self._phone_connected:
|
||
self._phone_connected = True
|
||
self._log("*** 手机已连接 ***")
|
||
|
||
# 按已知协议标记对齐数据
|
||
while True:
|
||
idx = -1
|
||
for marker in [b"\x02", b"\x05", b"?", b"\x3f",
|
||
b"KENC", b"!KENC"]:
|
||
pos = buf.find(marker)
|
||
if pos >= 0 and (idx < 0 or pos < idx):
|
||
idx = pos
|
||
if idx > 0:
|
||
buf = buf[idx:]
|
||
elif idx < 0:
|
||
if len(buf) > 512:
|
||
buf = b""
|
||
break
|
||
# 将对齐后的数据放入队列
|
||
with self._lock:
|
||
self._rx_queue += buf
|
||
buf = b""
|
||
break
|
||
|
||
# 超时断开检测 (30秒无数据)
|
||
if self._phone_connected and \
|
||
(time.time() - self._last_data_time) > 30:
|
||
self._phone_connected = False
|
||
self._log("*** 手机已断开 ***")
|
||
|
||
time.sleep(0.05)
|
||
except OSError:
|
||
self._phone_connected = False
|
||
time.sleep(0.5)
|
||
except Exception:
|
||
time.sleep(0.1)
|
||
|
||
def read_all(self) -> bytes:
|
||
"""主线程读取累积数据"""
|
||
with self._lock:
|
||
if self._rx_queue:
|
||
data = self._rx_queue
|
||
self._rx_queue = b""
|
||
return data
|
||
return b""
|
||
|
||
def write(self, data: bytes) -> bool:
|
||
if not self.ser or not self.ser.is_open:
|
||
return False
|
||
try:
|
||
with self._lock:
|
||
self.ser.write(data)
|
||
self.ser.flush()
|
||
self._total_bytes_tx += len(data)
|
||
return True
|
||
except Exception as e:
|
||
self._log(f"发送失败: {e}")
|
||
return False
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════
|
||
# 手机通信服务 — 串口 + 协议 + 测量队列
|
||
# ════════════════════════════════════════════════════════════════
|
||
|
||
class MobileService:
|
||
"""
|
||
手机通信主服务。
|
||
- 后台线程处理串口读写 + 协议解析 (握手/控制码自动响应)
|
||
- 收到 send 命令时直接发送数据，不等手机轮询
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.serial: Optional[BtSerial] = None
|
||
self.proto = ProtocolHandler()
|
||
self._poll_thread: Optional[threading.Thread] = None
|
||
self._log_cb = None
|
||
|
||
def _log(self, msg: str):
|
||
if self._log_cb:
|
||
self._log_cb(msg)
|
||
|
||
@property
|
||
def connected(self) -> bool:
|
||
return self.serial is not None and self.serial.running
|
||
|
||
@property
|
||
def phone_connected(self) -> bool:
|
||
return self.serial is not None and self.serial.phone_connected
|
||
|
||
# ── 连接 ──
|
||
|
||
def connect(self, port: str) -> bool:
|
||
if self.serial:
|
||
self.disconnect()
|
||
self.serial = BtSerial(port)
|
||
self.serial._log_cb = self._log
|
||
if not self.serial.open():
|
||
self.serial = None
|
||
return False
|
||
# 启动轮询线程
|
||
self._start_poll()
|
||
return True
|
||
|
||
def disconnect(self):
|
||
self._stop_poll()
|
||
if self.serial:
|
||
self.serial.close()
|
||
self.serial = None
|
||
|
||
# ── 后台轮询 ──
|
||
|
||
def _start_poll(self):
|
||
self._poll_thread = threading.Thread(
|
||
target=self._poll_loop, daemon=True)
|
||
self._poll_thread.start()
|
||
|
||
def _stop_poll(self):
|
||
if self._poll_thread and self._poll_thread.is_alive():
|
||
self._poll_thread.join(timeout=2)
|
||
self._poll_thread = None
|
||
|
||
def _poll_loop(self):
|
||
"""后台轮询 — 处理收数据 + 自动响应握手/控制码"""
|
||
while self.serial and self.serial.running:
|
||
try:
|
||
data = self.serial.read_all()
|
||
if data:
|
||
responses, _ = self.proto.feed(data)
|
||
|
||
# 发送协议响应 (握手/控制码)
|
||
for rsp in responses:
|
||
self.serial.write(rsp)
|
||
|
||
time.sleep(0.05)
|
||
except Exception:
|
||
time.sleep(0.1)
|
||
|
||
# ── 数据发送 (直接发，不等轮询) ──
|
||
|
||
def send_measurement(self, staff_reading: float, distance: float) -> bool:
|
||
"""直接构建 M5 两帧并通过串口发送"""
|
||
if not self.serial or not self.serial.running:
|
||
self._log("⚠ 串口未打开")
|
||
return False
|
||
|
||
line1, line2 = MeasurementBuilder.build(staff_reading, distance)
|
||
|
||
self.serial.write(line1)
|
||
time.sleep(0.03) # 两帧间隔 (模拟水准仪)
|
||
self.serial.write(line2)
|
||
|
||
self._log(f"↑ 直接发送: R={staff_reading:.5f}m HD={distance:.3f}m")
|
||
return True
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════
|
||
# FastAPI — 全局状态
|
||
# ════════════════════════════════════════════════════════════════
|
||
|
||
svc = MobileService()
|
||
svc_lock = threading.Lock()
|
||
|
||
|
||
class ConnectRequest(BaseModel):
|
||
port: str
|
||
|
||
|
||
class CommandRequest(BaseModel):
|
||
cmd: str
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
print("[mobile] 手机蓝牙通信服务启动")
|
||
yield
|
||
with svc_lock:
|
||
svc.disconnect()
|
||
print("[mobile] 手机蓝牙通信服务已停止")
|
||
|
||
|
||
app = FastAPI(
|
||
title="DiNi Mobile Bluetooth Service",
|
||
description="PC蓝牙SPP服务端 — 模拟水准仪, 向手机APP发送M5测量数据",
|
||
version="4.0",
|
||
lifespan=lifespan,
|
||
)
|
||
|
||
|
||
# ── GET / ──
|
||
|
||
@app.get("/")
|
||
async def root():
|
||
return {
|
||
"service": "DiNi Mobile Bluetooth Service v4.0",
|
||
"connected": svc.connected,
|
||
"port": svc.serial.port if svc.serial else None,
|
||
"phone_connected": svc.phone_connected,
|
||
"com_ports": list_com_ports() if not svc.connected else None,
|
||
}
|
||
|
||
|
||
# ── GET /status ──
|
||
|
||
@app.get("/status")
|
||
async def status():
|
||
info = {
|
||
"connected": svc.connected,
|
||
"phone_connected": svc.phone_connected,
|
||
}
|
||
if svc.serial:
|
||
info["port"] = svc.serial.port
|
||
info["total_rx"] = svc.serial.total_rx
|
||
info["total_tx"] = svc.serial.total_tx
|
||
return info
|
||
|
||
|
||
# ── GET/POST /test ──
|
||
|
||
@app.api_route("/test", methods=["GET", "POST"])
|
||
async def test():
|
||
if not svc.connected:
|
||
return {
|
||
"connected": False,
|
||
"message": "未连接 — 返回可连接的 COM 口列表",
|
||
"com_ports": list_com_ports(),
|
||
}
|
||
return {
|
||
"connected": True,
|
||
"port": svc.serial.port,
|
||
"phone_connected": svc.phone_connected,
|
||
"total_rx": svc.serial.total_rx,
|
||
"total_tx": svc.serial.total_tx,
|
||
}
|
||
|
||
|
||
# ── POST /connect ──
|
||
|
||
@app.post("/connect")
|
||
async def connect(req: ConnectRequest):
|
||
def _do():
|
||
with svc_lock:
|
||
ok = svc.connect(req.port)
|
||
if not ok:
|
||
raise HTTPException(
|
||
status_code=500,
|
||
detail=f"无法打开 {req.port}，请检查 COM 口是否存在或已被占用"
|
||
)
|
||
return {"status": "connected", "port": req.port}
|
||
|
||
loop = asyncio.get_running_loop()
|
||
return await loop.run_in_executor(None, _do)
|
||
|
||
|
||
# ── POST /disconnect ──
|
||
|
||
@app.post("/disconnect")
|
||
async def disconnect():
|
||
def _do():
|
||
with svc_lock:
|
||
svc.disconnect()
|
||
return {"status": "disconnected"}
|
||
|
||
loop = asyncio.get_running_loop()
|
||
return await loop.run_in_executor(None, _do)
|
||
|
||
|
||
# ── POST /command ──
|
||
|
||
@app.post("/command")
|
||
async def send_command(req: CommandRequest):
|
||
"""
|
||
发送命令。
|
||
|
||
支持的命令:
|
||
send <R> <HD> → 直接发送测量数据 如: "send 0.89182 3.323"
|
||
disconnect → 断开蓝牙连接
|
||
"""
|
||
cmd = req.cmd.strip()
|
||
parts = cmd.split()
|
||
action = parts[0].lower() if parts else ""
|
||
|
||
if action == "disconnect":
|
||
def _d():
|
||
with svc_lock:
|
||
svc.disconnect()
|
||
return {"status": "disconnected", "command": "disconnect"}
|
||
loop = asyncio.get_running_loop()
|
||
return await loop.run_in_executor(None, _d)
|
||
|
||
if action == "send":
|
||
# 格式: send <staff_reading> <distance>
|
||
if len(parts) < 3:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="格式: send <标尺读数> <距离> 如: send 0.89182 3.323"
|
||
)
|
||
try:
|
||
r = float(parts[1])
|
||
hd = float(parts[2])
|
||
except ValueError:
|
||
raise HTTPException(status_code=400, detail="参数必须是数字")
|
||
|
||
def _send():
|
||
with svc_lock:
|
||
if not svc.connected:
|
||
raise HTTPException(status_code=400,
|
||
detail="未连接, 请先 POST /connect")
|
||
ok = svc.send_measurement(r, hd)
|
||
return {
|
||
"command": "send",
|
||
"staff_reading": r,
|
||
"distance": hd,
|
||
"status": "ok" if ok else "error",
|
||
}
|
||
loop = asyncio.get_running_loop()
|
||
return await loop.run_in_executor(None, _send)
|
||
|
||
# ── 原始数据发送 ──
|
||
if not svc.connected:
|
||
raise HTTPException(status_code=400,
|
||
detail="未连接, 请先 POST /connect")
|
||
|
||
def _raw():
|
||
# 支持 hex: "hex:AABBCC" 或纯文本直接发送
|
||
if cmd.startswith("hex:"):
|
||
raw = bytes.fromhex(cmd[4:].replace(" ", ""))
|
||
else:
|
||
raw = cmd.encode("ascii")
|
||
ok = svc.serial.write(raw)
|
||
return {
|
||
"command": "raw",
|
||
"status": "ok" if ok else "error",
|
||
"bytes_sent": len(raw) if ok else 0,
|
||
}
|
||
|
||
loop = asyncio.get_running_loop()
|
||
return await loop.run_in_executor(None, _raw)
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════
|
||
# 启动入口
|
||
# ════════════════════════════════════════════════════════════════
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description="手机蓝牙通信 — FastAPI 服务 (模拟 DiNi 水准仪)")
|
||
parser.add_argument("--host", default="0.0.0.0", help="监听地址 (默认 0.0.0.0)")
|
||
parser.add_argument("--port", type=int, default=DEFAULT_HTTP_PORT,
|
||
help=f"HTTP 服务端口 (默认 {DEFAULT_HTTP_PORT})")
|
||
parser.add_argument("--bt-port", default=None,
|
||
help="启动时自动连接的蓝牙 COM 口 (如 COM3)")
|
||
args = parser.parse_args()
|
||
|
||
# ── 打印可连接 COM 口 ──
|
||
print_com_list()
|
||
|
||
# ── 可选: 启动时自动连接 ──
|
||
if args.bt_port:
|
||
print(f"\n启动时自动连接 → {args.bt_port} ...")
|
||
try:
|
||
ok = svc.connect(args.bt_port)
|
||
if ok:
|
||
print(f"✓ 启动时已自动连接 {args.bt_port}")
|
||
else:
|
||
print(f"✗ 自动连接 {args.bt_port} 失败, 服务仍正常启动")
|
||
except Exception as e:
|
||
print(f"✗ 自动连接失败: {e}")
|
||
|
||
# ── 启动 uvicorn ──
|
||
import uvicorn
|
||
print(f"\n▶ 手机蓝牙通信服务启动中...")
|
||
print(f" HTTP 地址: http://{args.host}:{args.port}")
|
||
print(f" 测试接口: http://{args.host}:{args.port}/test")
|
||
print(f" 命令接口: http://{args.host}:{args.port}/command")
|
||
print(f" API 文档: http://{args.host}:{args.port}/docs")
|
||
print(f" 按 Ctrl+C 停止服务\n")
|
||
uvicorn.run(app, host=args.host, port=args.port, workers=1, log_level="info")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|