ws50529
128 lines
3.5 KiB
Python
Executable File
128 lines
3.5 KiB
Python
Executable File
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_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_data,
|
|
filtered_csv_text,
|
|
read_csv_upload,
|
|
)
|
|
from dea.filter_design import design_response
|
|
from dea.schemas import BodeCompareParams, BodeParams, DesignParams
|
|
from dea.security import (
|
|
SECURITY_HEADERS,
|
|
add_security_headers,
|
|
is_lan_or_loopback,
|
|
lan_denied_response,
|
|
)
|
|
from dea.validation import (
|
|
normalize_coefficients,
|
|
parse_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/filter")
|
|
async def filter_csv(
|
|
file: UploadFile = File(...),
|
|
b: str = Form(...),
|
|
a: str = Form(...),
|
|
col_idx: int = Form(0),
|
|
):
|
|
try:
|
|
b_vals = parse_coefficients(b, "b")
|
|
a_vals = parse_coefficients(a, "a")
|
|
df = await read_csv_upload(file)
|
|
return filter_preview_response(df, b_vals, a_vals, col_idx)
|
|
except HTTPException:
|
|
raise
|
|
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),
|
|
):
|
|
try:
|
|
b_vals = parse_coefficients(b, "b")
|
|
a_vals = parse_coefficients(a, "a")
|
|
df = await read_csv_upload(file)
|
|
csv_text = filtered_csv_text(df, b_vals, a_vals, col_idx)
|
|
|
|
return StreamingResponse(
|
|
iter([csv_text]),
|
|
media_type="text/csv",
|
|
headers={"Content-Disposition": "attachment; filename=filtered_output.csv"},
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=f"CSV處理失敗: {str(e)}")
|