feat: add MCU communication support to write filter coefficients via serial port

dea/config.py

@@ -1,7 +1,11 @@
+import os
+
 MAX_COEFFICIENTS = 64
 MAX_CSV_BYTES = 32 * 1024 * 1024
 MAX_PLOT_POINTS = 5000
 BODE_POINTS = 500
 BODE_MAX_MULTIPLIER = 3.162
 NYQUIST_MARGIN = 0.999999
-
+MCU_SERIAL_PORT = os.getenv("MCU_SERIAL_PORT", "/dev/ttyUSB2")
+MCU_BAUD_RATE = int(os.getenv("MCU_BAUD_RATE", "115200"))
+MCU_WRITE_TIMEOUT = float(os.getenv("MCU_WRITE_TIMEOUT", "1.0"))

dea/mcu.py

@@ -0,0 +1,73 @@
+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 串口功能需要安裝 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 串口不可為空")
+
+    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 串口傳送失敗: {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 串口清單功能需要安裝 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,
+    }

dea/schemas.py

@@ -1,4 +1,4 @@
-from typing import List
+from typing import List, Optional
 
 from pydantic import BaseModel, Field
 
@@ -42,3 +42,8 @@ class BodeParams(BaseModel):
 class BodeCompareParams(BaseModel):
     ideal: BodeParams
     fixed: BodeParams
+
+
+class MCUWriteParams(BaseModel):
+    command: str = Field(min_length=1, max_length=128)
+    port: Optional[str] = Field(default=None, max_length=256)