feat: implement dea module

dea/csv_processing.py

@@ -0,0 +1,72 @@
+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()
+