Files
shuizhunyi/shouji/main.py
2026-06-08 15:17:52 +08:00

762 lines
26 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- 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:
"""
手机通信主服务。
- 后台线程处理串口读写 + 协议解析
- 手机发 ?0100 轮询时自动发送排队测量数据
"""
def __init__(self):
self.serial: Optional[BtSerial] = None
self.proto = ProtocolHandler()
self._pending: list[tuple[bytes, bytes]] = [] # 待发送测量数据
self._sent_count = 0
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
@property
def sent_count(self) -> int:
return self._sent_count
@property
def pending_count(self) -> int:
return len(self._pending)
# ── 连接 ──
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, has_query = self.proto.feed(data)
# 发送协议响应 (握手/控制码)
for rsp in responses:
self.serial.write(rsp)
# 手机轮询 (±0100) + 有待发送数据 → 发送
if has_query and self._pending:
self._send_data()
time.sleep(0.05)
except Exception:
time.sleep(0.1)
# ── 数据发送 ──
def queue_measurement(self, staff_reading: float, distance: float):
"""将测量数据加入发送队列"""
line1, line2 = MeasurementBuilder.build(staff_reading, distance)
self._pending.append((line1, line2))
self._log(f"+ 队列: R={staff_reading:.5f}m HD={distance:.3f}m "
f"(共{len(self._pending)}条)")
def _send_data(self):
"""发送队列中第一条 (由轮询触发)"""
if not self._pending or not self.serial:
return
line1, line2 = self._pending.pop(0)
self.serial.write(line1)
time.sleep(0.03) # 两帧间隔 (模拟水准仪)
self.serial.write(line2)
self._sent_count += 1
self._log(f"↑ 发送 #{self._sent_count}")
def force_send(self, ignore_phone: bool = False) -> bool:
"""强制发送 (不等轮询)"""
if not self.serial or not self.serial.running:
self._log("⚠ 串口未打开")
return False
if not self.serial.phone_connected and not ignore_phone:
self._log("⚠ 手机未连接")
return False
if not self._pending:
self._log("⚠ 无待发送数据")
return False
self._send_data()
return True
def clear_queue(self):
self._pending.clear()
# ════════════════════════════════════════════════════════════════
# 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,
"pending_count": svc.pending_count,
"sent_count": svc.sent_count,
}
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,
"pending_count": svc.pending_count,
"sent_count": svc.sent_count,
"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"
force → 强制发送队列中数据 (不等轮询)
clear → 清空待发送队列
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 _queue():
with svc_lock:
if not svc.connected:
raise HTTPException(status_code=400,
detail="未连接, 请先 POST /connect")
svc.queue_measurement(r, hd)
return {
"command": "send",
"staff_reading": r,
"distance": hd,
"pending_count": svc.pending_count,
}
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, _queue)
if action == "force":
def _force():
with svc_lock:
ok = svc.force_send(ignore_phone=False)
return {
"command": "force",
"status": "ok" if ok else "skipped",
"pending_count": svc.pending_count,
}
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, _force)
if action == "clear":
with svc_lock:
svc.clear_queue()
return {"command": "clear", "pending_count": svc.pending_count}
# ── 原始数据发送 ──
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()