import numpy as np import pandas as pd from scipy import signal from fastapi import FastAPI, File, UploadFile, Form, HTTPException, Request from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from typing import List, Optional import io from fastapi.responses import StreamingResponse, RedirectResponse, JSONResponse import ipaddress 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 client_ip: try: ip_obj = ipaddress.ip_address(client_ip) if not (ip_obj.is_private or ip_obj.is_loopback): return JSONResponse(status_code=403, content={"detail": "Access Denied: Only LAN connections are allowed."}) except ValueError: pass return await call_next(request) # 掛載靜態網頁檔案，將預設首頁導向到 /ui/ app.mount("/ui", StaticFiles(directory="static", html=True), name="static") @app.get("/") def redirect_to_ui(): return RedirectResponse(url="/ui/") class DesignParams(BaseModel): filter_type: str fs: float lp_fc: float = 1000.0 lp_order: int = 1 hp_fc: float = 1000.0 hp_order: int = 1 bp_f_low: float = 500.0 bp_f_high: float = 2000.0 bp_order: int = 1 notch_f0: float = 120.0 notch_q: float = 1.0 opoz_fz: float = 15000.0 opoz_fp: float = 10.0 tptz_fz1: float = 200.0 tptz_fz2: float = 25000.0 tptz_fp1: float = 10.0 tptz_fp2: float = 5000.0 pid_kp: float = 0.003 pid_ki: float = 10.0 pid_kd: float = 0.000016 sogi_f0: float = 60.0 sogi_k: float = 1.414 class BodeParams(BaseModel): b: List[float] a: List[float] fs: float @app.post("/api/design") def design_filter(params: DesignParams): fs_val = params.fs f_type = params.filter_type b_new, a_new = [], [] try: if f_type == "Lowpass (低通)": b_new, a_new = signal.butter(params.lp_order, params.lp_fc, btype='low', fs=fs_val) elif f_type == "Highpass (高通)": b_new, a_new = signal.butter(params.hp_order, params.hp_fc, btype='high', fs=fs_val) elif f_type == "Bandpass (帶通)": f_low, f_high = params.bp_f_low, params.bp_f_high if f_low >= f_high: f_low = f_high * 0.99 b_new, a_new = signal.butter(params.bp_order, [f_low, f_high], btype='bandpass', fs=fs_val) elif f_type == "Notch (陷波器)": b_new, a_new = signal.iirnotch(params.notch_f0, params.notch_q, fs=fs_val) elif f_type == "1P1Z (一極一零)": b_s = [1.0, 2 * np.pi * params.opoz_fz] a_s = [1.0, 2 * np.pi * params.opoz_fp] b_new, a_new = signal.bilinear(b_s, a_s, fs=fs_val) dc_gain = np.sum(b_new) / np.sum(a_new) b_new = b_new / dc_gain elif f_type == "2P2Z (二極二零)": w_p1, w_p2 = 2 * np.pi * params.tptz_fp1, 2 * np.pi * params.tptz_fp2 w_z1, w_z2 = 2 * np.pi * params.tptz_fz1, 2 * np.pi * params.tptz_fz2 b_s = [1.0, w_z1 + w_z2, w_z1 * w_z2] a_s = [1.0, w_p1 + w_p2, w_p1 * w_p2] b_new, a_new = signal.bilinear(b_s, a_s, fs=fs_val) dc_gain = np.sum(b_new) / np.sum(a_new) b_new = b_new / dc_gain elif f_type == "PID 控制器": ki_term = params.pid_ki / (2.0 * fs_val) kd_term = params.pid_kd * fs_val b0 = params.pid_kp + ki_term + kd_term b1 = -params.pid_kp + ki_term - 2.0 * kd_term b2 = kd_term b_new = [b0, b1, b2] a_new = [1.0, -1.0, 0.0] elif f_type == "SOGI-Alpha (帶通)": w0 = 2 * np.pi * params.sogi_f0 b_s = [params.sogi_k * w0, 0.0] a_s = [1.0, params.sogi_k * w0, w0**2] b_new, a_new = signal.bilinear(b_s, a_s, fs=fs_val) elif f_type == "SOGI-Beta (低通)": w0 = 2 * np.pi * params.sogi_f0 b_s = [params.sogi_k * (w0**2)] a_s = [1.0, params.sogi_k * w0, w0**2] b_new, a_new = signal.bilinear(b_s, a_s, fs=fs_val) return {"b": b_new.tolist() if isinstance(b_new, np.ndarray) else b_new, "a": a_new.tolist() if isinstance(a_new, np.ndarray) else a_new} except Exception as e: raise HTTPException(status_code=400, detail=str(e)) @app.post("/api/bode") def calculate_bode(params: BodeParams): try: f_min = params.fs / 50000.0 f_max = params.fs * 3.162 f_eval = np.logspace(np.log10(f_min), np.log10(f_max), 500) # WorN determines frequencies to evaluate at w, h = signal.freqz(params.b, params.a, worN=f_eval, fs=params.fs) mag = 20 * np.log10(np.abs(h) + 1e-12) phase = np.angle(h, deg=True) return { "freq": f_eval.tolist(), "mag": mag.tolist(), "phase": phase.tolist() } except Exception as e: raise HTTPException(status_code=400, detail=str(e)) @app.post("/api/filter") async def filter_csv( file: UploadFile = File(...), b: str = Form(...), a: str = Form(...), col_idx: int = Form(0) ): try: b_vals = [float(x.strip()) for x in b.split(',') if x.strip()] a_vals = [float(x.strip()) for x in a.split(',') if x.strip()] contents = await file.read() df = pd.read_csv(io.BytesIO(contents)) if len(df.columns) <= col_idx: raise HTTPException(status_code=400, detail="欄位索引超出範圍") col_to_filter = df.columns[col_idx] x_signal = df[col_to_filter].values # 套用濾波 y_signal = signal.lfilter(b_vals, a_vals, x_signal) # 準備資料回傳前端進行繪圖 (為了防止過大，可以在此決定是否 downsample，但目前保留原貌) # 前端收到 JSON 後可自行繪製並產生 CSV return { "index": df.index.tolist(), "original": x_signal.tolist(), "filtered": y_signal.tolist(), "col_name": col_to_filter } except Exception as e: raise HTTPException(status_code=400, detail=f"CSV處理失敗: {str(e)}") @app.post("/api/filter/download") async def filter_csv_download( file: UploadFile = File(...), b: str = Form(...), a: str = Form(...), col_idx: int = Form(0) ): # 此端點專為產生包含輸出結果的 CSV 檔供下載 try: b_vals = [float(x.strip()) for x in b.split(',') if x.strip()] a_vals = [float(x.strip()) for x in a.split(',') if x.strip()] contents = await file.read() df = pd.read_csv(io.BytesIO(contents)) col_to_filter = df.columns[col_idx] x_signal = df[col_to_filter].values y_signal = signal.lfilter(b_vals, a_vals, x_signal) output_col_name = f"{col_to_filter}_filtered" df[output_col_name] = y_signal csv_buffer = io.StringIO() df.to_csv(csv_buffer, index=False) return StreamingResponse( iter([csv_buffer.getvalue()]), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=filtered_output.csv"} ) except Exception as e: raise HTTPException(status_code=400, detail=f"CSV處理失敗: {str(e)}")