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, }