tcad-bodeplot/dea_api.py

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