ws50529
73 lines
2.7 KiB
Python
Executable File
73 lines
2.7 KiB
Python
Executable File
import io
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
from fastapi import HTTPException
|
|
from scipy import signal
|
|
|
|
from .config import MAX_CSV_BYTES, MAX_PLOT_POINTS
|
|
|
|
|
|
def downsample_for_plot(index, original, filtered):
|
|
total = len(index)
|
|
if total <= MAX_PLOT_POINTS:
|
|
return index, original, filtered, 1
|
|
step = int(np.ceil(total / MAX_PLOT_POINTS))
|
|
return index[::step], original[::step], filtered[::step], step
|
|
|
|
|
|
async def read_csv_upload(file):
|
|
filename = (file.filename or "").lower()
|
|
if filename and not filename.endswith(".csv"):
|
|
raise HTTPException(status_code=400, detail="請上傳 CSV 檔案")
|
|
contents = await file.read(MAX_CSV_BYTES + 1)
|
|
if len(contents) > MAX_CSV_BYTES:
|
|
raise HTTPException(status_code=413, detail=f"CSV 檔案不可超過 {MAX_CSV_BYTES // (1024 * 1024)}MB")
|
|
if not contents.strip():
|
|
raise HTTPException(status_code=400, detail="CSV 不可為空")
|
|
try:
|
|
df = pd.read_csv(io.BytesIO(contents))
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=f"CSV 讀取失敗: {str(e)}")
|
|
if df.empty:
|
|
raise HTTPException(status_code=400, detail="CSV 不可為空")
|
|
return df
|
|
|
|
|
|
def filtered_csv_data(df, b_vals, a_vals, col_idx):
|
|
if col_idx < 0 or len(df.columns) <= col_idx:
|
|
raise HTTPException(status_code=400, detail="欄位索引超出範圍")
|
|
col_to_filter = df.columns[col_idx]
|
|
x_signal = pd.to_numeric(df[col_to_filter], errors="coerce")
|
|
if x_signal.isna().any():
|
|
raise HTTPException(status_code=400, detail=f"欄位 {col_to_filter} 含有非數值資料")
|
|
x_values = x_signal.to_numpy(dtype=float)
|
|
if not np.isfinite(x_values).all():
|
|
raise HTTPException(status_code=400, detail=f"欄位 {col_to_filter} 含有非有限數值")
|
|
y_signal = signal.lfilter(b_vals, a_vals, x_values)
|
|
return col_to_filter, x_values, y_signal
|
|
|
|
|
|
def filter_preview_response(df, b_vals, a_vals, col_idx):
|
|
col_to_filter, x_signal, y_signal = filtered_csv_data(df, b_vals, a_vals, col_idx)
|
|
index, original, filtered, step = downsample_for_plot(df.index.to_numpy(), x_signal, y_signal)
|
|
|
|
return {
|
|
"index": index.tolist(),
|
|
"original": original.tolist(),
|
|
"filtered": filtered.tolist(),
|
|
"col_name": col_to_filter,
|
|
"total_points": int(len(df.index)),
|
|
"plot_points": int(len(index)),
|
|
"downsample_step": int(step),
|
|
}
|
|
|
|
|
|
def filtered_csv_text(df, b_vals, a_vals, col_idx):
|
|
col_to_filter, _, y_signal = filtered_csv_data(df, b_vals, a_vals, col_idx)
|
|
df[f"{col_to_filter}_filtered"] = y_signal
|
|
csv_buffer = io.StringIO()
|
|
df.to_csv(csv_buffer, index=False)
|
|
return csv_buffer.getvalue()
|
|
|