tcad-bodeplot/dea_api.py

from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile
from fastapi.responses import RedirectResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles

from dea.bode import calculate_bode_cascade_response, calculate_bode_compare_response, calculate_bode_response
from dea.config import (
    BODE_MAX_MULTIPLIER,
    BODE_POINTS,
    MAX_COEFFICIENTS,
    MAX_CSV_BYTES,
    MAX_PLOT_POINTS,
    NYQUIST_MARGIN,
)
from dea.csv_processing import (
    downsample_for_plot,
    filter_preview_response,
    filtered_csv_text,
    save_csv_upload,
)
from dea.filter_design import design_response
from dea.mcu import list_mcu_ports_response, write_mcu_command_response
from dea.schemas import BodeCascadeParams, BodeCompareParams, BodeParams, DesignParams, MCUWriteParams
from dea.security import (
    SECURITY_HEADERS,
    add_security_headers,
    is_lan_or_loopback,
    lan_denied_response,
)
import traceback
from dea.validation import (
    normalize_coefficients,
    parse_coefficients,
    parse_int_coefficients,
    require_finite,
    validate_coefficients,
    validate_frequency,
)

app = FastAPI(title="Difference Equation Analyzer API")


@app.middleware("http")
async def restrict_to_lan(request: Request, call_next):
    client_ip = request.client.host if request.client else None
    if not is_lan_or_loopback(client_ip):
        return lan_denied_response()
    response = await call_next(request)
    return add_security_headers(response)


# 掛載靜態網頁檔案，將預設首頁導向到 /ui/
app.mount("/ui", StaticFiles(directory="static", html=True), name="static")


@app.get("/")
def redirect_to_ui():
    return RedirectResponse(url="/ui/")


@app.post("/api/design")
def design_filter(params: DesignParams):
    try:
        return design_response(params)
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))


@app.post("/api/bode")
def calculate_bode(params: BodeParams):
    try:
        return calculate_bode_response(params)
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))


@app.post("/api/bode/compare")
def calculate_bode_compare(params: BodeCompareParams):
    try:
        return calculate_bode_compare_response(params)
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))


@app.post("/api/bode/compare_cascade")
def calculate_bode_compare_cascade(params: BodeCascadeParams):
    try:
        return calculate_bode_cascade_response(params)
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))



@app.post("/api/csv/upload")
async def upload_csv(file: UploadFile = File(...)):
    try:
        file_id = await save_csv_upload(file)
        return {"file_id": file_id}
    except HTTPException:
        raise
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=400, detail=f"CSV上傳失敗: {str(e)}")


def parse_stages_list(stages_str: str):
    if not stages_str:
        return None
    import json
    stages_raw = json.loads(stages_str)
    stages_list = []
    for i, s in enumerate(stages_raw):
        stages_list.append({
            "b": parse_coefficients(s.get("b_str", ""), f"stages[{i}].b"),
            "a": parse_coefficients(s.get("a_str", ""), f"stages[{i}].a"),
            "b_int": parse_int_coefficients(s.get("b_int_str", ""), f"stages[{i}].b_int") if s.get("b_int_str") else None,
            "a_int": parse_int_coefficients(s.get("a_int_str", ""), f"stages[{i}].a_int") if s.get("a_int_str") else None,
            "shift_in": int(s.get("shift_in", 14)),
            "shift_out": int(s.get("shift_out", 14)),
            "shift_b": int(s.get("shift_b", 14)),
            "shift_a": int(s.get("shift_a", 14)),
            "use_round": bool(s.get("use_round", False)),
            "isActive": bool(s.get("isActive", True))
        })
    return stages_list


@app.post("/api/filter")
async def filter_csv(
    file_id: str = Form(...),
    b: str = Form(...),
    a: str = Form(...),
    col_idx: int = Form(0),
    b_int: str = Form(None),
    a_int: str = Form(None),
    shift_in: int = Form(14),
    shift_out: int = Form(14),
    shift_b: int = Form(14),
    shift_a: int = Form(14),
    use_round: bool = Form(False),
    stages: str = Form(None),
    start_idx: int = Form(None),
    end_idx: int = Form(None),
):
    try:
        b_vals = parse_coefficients(b, "b")
        a_vals = parse_coefficients(a, "a")
        
        b_int_vals = parse_int_coefficients(b_int, "b_int") if b_int else None
        a_int_vals = parse_int_coefficients(a_int, "a_int") if a_int else None
        
        stages_list = parse_stages_list(stages)
        
        return filter_preview_response(
            file_id, b_vals, a_vals, col_idx, 
            b_int=b_int_vals, a_int=a_int_vals,
            shift_in=shift_in, shift_out=shift_out, 
            shift_b=shift_b, shift_a=shift_a,
            use_round=use_round,
            stages=stages_list,
            start_idx=start_idx,
            end_idx=end_idx
        )
    except HTTPException:
        raise
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=400, detail=f"CSV處理失敗: {str(e)}")


@app.get("/api/mcu/ports")
def list_mcu_ports():
    try:
        return list_mcu_ports_response()
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"MCU COM Port清單讀取失敗: {str(e)}")


@app.post("/api/mcu/write")
def write_mcu_command(params: MCUWriteParams):
    try:
        return write_mcu_command_response(params)
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"MCU 指令傳送失敗: {str(e)}")


@app.post("/api/filter/download")
async def filter_csv_download(
    file_id: str = Form(...),
    b: str = Form(...),
    a: str = Form(...),
    col_idx: int = Form(0),
    b_int: str = Form(None),
    a_int: str = Form(None),
    shift_in: int = Form(14),
    shift_out: int = Form(14),
    shift_b: int = Form(14),
    shift_a: int = Form(14),
    use_round: bool = Form(False),
    stages: str = Form(None),
    fs: float = Form(100000.0),
):
    try:
        b_vals = parse_coefficients(b, "b")
        a_vals = parse_coefficients(a, "a")
        
        b_int_vals = parse_int_coefficients(b_int, "b_int") if b_int else None
        a_int_vals = parse_int_coefficients(a_int, "a_int") if a_int else None
        
        stages_list = parse_stages_list(stages)
        
        csv_text = filtered_csv_text(
            file_id, b_vals, a_vals, col_idx,
            b_int=b_int_vals, a_int=a_int_vals,
            shift_in=shift_in, shift_out=shift_out,
            shift_b=shift_b, shift_a=shift_a,
            use_round=use_round,
            stages=stages_list,
            fs=fs
        )

        return StreamingResponse(
            iter([csv_text]),
            media_type="text/csv",
            headers={"Content-Disposition": "attachment; filename=filtered_output.csv"},
        )
    except HTTPException:
        raise
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=400, detail=f"CSV處理失敗: {str(e)}")