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)}")