ws50529
74 lines
2.0 KiB
Python
Executable File
74 lines
2.0 KiB
Python
Executable File
import re
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from .config import MCU_BAUD_RATE, MCU_SERIAL_PORT, MCU_WRITE_TIMEOUT
|
|
from .schemas import MCUWriteParams
|
|
|
|
|
|
MCU_CMD_PATTERN = re.compile(r"^bodeplot=-?\d+,-?\d+,-?\d+,-?\d+,-?\d+$")
|
|
|
|
|
|
def _serial_module():
|
|
try:
|
|
import serial
|
|
except ImportError as exc:
|
|
raise HTTPException(status_code=500, detail="MCU COM Port功能需要安裝 pyserial") from exc
|
|
return serial
|
|
|
|
|
|
def write_mcu_command_response(params: MCUWriteParams):
|
|
cmd = params.command.strip()
|
|
if not MCU_CMD_PATTERN.fullmatch(cmd):
|
|
raise HTTPException(status_code=400, detail="MCU 指令格式錯誤")
|
|
|
|
target_port = (params.port or MCU_SERIAL_PORT).strip()
|
|
if not target_port:
|
|
raise HTTPException(status_code=400, detail="MCU COM Port不可為空")
|
|
|
|
serial = _serial_module()
|
|
payload = (cmd + "\r\n").encode("ascii", errors="strict")
|
|
|
|
try:
|
|
with serial.Serial(
|
|
port=target_port,
|
|
baudrate=MCU_BAUD_RATE,
|
|
timeout=MCU_WRITE_TIMEOUT,
|
|
write_timeout=MCU_WRITE_TIMEOUT,
|
|
) as ser:
|
|
written = ser.write(payload)
|
|
ser.flush()
|
|
except serial.SerialException as exc:
|
|
raise HTTPException(status_code=500, detail=f"MCU COM Port傳送失敗: {str(exc)}") from exc
|
|
|
|
return {
|
|
"status": "ok",
|
|
"command": cmd,
|
|
"port": target_port,
|
|
"baudrate": MCU_BAUD_RATE,
|
|
"bytes_written": int(written),
|
|
}
|
|
|
|
|
|
def list_mcu_ports_response():
|
|
_serial_module()
|
|
try:
|
|
from serial.tools import list_ports
|
|
except ImportError as exc:
|
|
raise HTTPException(status_code=500, detail="MCU COM Port清單功能需要安裝 pyserial") from exc
|
|
|
|
ports = [
|
|
{
|
|
"device": port.device,
|
|
"description": port.description or "",
|
|
"hwid": port.hwid or "",
|
|
}
|
|
for port in list_ports.comports()
|
|
]
|
|
|
|
return {
|
|
"default_port": MCU_SERIAL_PORT,
|
|
"baudrate": MCU_BAUD_RATE,
|
|
"ports": ports,
|
|
}
|