feat: integrate cascade workflow

tests/frontend/cascade-stage-smoke.mjs

@@ -0,0 +1,54 @@
+import assert from 'node:assert/strict';
+import { computeStageFixedCoeffs, createCascadeFilterPayload, normalizeCascadeStage } from '../../src/cascade-stage.js';
+
+const legacyStage = normalizeCascadeStage({
+  id: 'legacy-1',
+  filterType: 'Lowpass (低通)',
+  b_str: '1, 0',
+  a_str: '1',
+});
+
+assert.equal(legacyStage.id, 'legacy-1');
+assert.equal(legacyStage.isActive, true);
+assert.equal(legacyStage.isExpanded, false);
+assert.equal(legacyStage.shiftBitsB, 14);
+assert.deepEqual(Object.keys(legacyStage.fixedOverrides), ['a', 'b']);
+
+const fixedStage = normalizeCascadeStage({
+  b_str: '0.5, -0.25',
+  a_str: '1, -0.5',
+  shiftBitsB: 4,
+  shiftBitsA: 5,
+  fixedOverrides: { b: { 1: -9 }, a: {} },
+});
+const fixed = computeStageFixedCoeffs(fixedStage, (value) => value.split(',').map(Number));
+assert.deepEqual(fixed.b, [8, -9]);
+assert.deepEqual(fixed.a, [32, -16]);
+
+const stages = [
+  normalizeCascadeStage({ id: 'a', b_str: '1', a_str: '1', shiftBitsB: 12, shiftBitsA: 13, isActive: true }),
+  normalizeCascadeStage({ id: 'b', b_str: '2', a_str: '1', shiftBitsB: 10, shiftBitsA: 11, isActive: false }),
+];
+
+const payload = createCascadeFilterPayload({
+  stages,
+  fs: 100000,
+  shiftBitsIn: 9,
+  shiftBitsOut: 8,
+  useRound: true,
+  parseCoeffs: (value) => value.split(',').map(Number),
+  getStageFixedCoeffs: (stage) => ({
+    b: stage.b_str.split(',').map(Number),
+    a: stage.a_str.split(',').map(Number),
+  }),
+});
+
+assert.equal(payload.length, 2);
+assert.equal(payload[0].b_str, '1');
+assert.equal(payload[1].b_str, '2');
+assert.equal(payload[0].shift_in, 9);
+assert.equal(payload[0].shift_out, 12);
+assert.equal(payload[1].shift_in, 12);
+assert.equal(payload[1].shift_out, 8);
+assert.equal(payload[1].isActive, false);
+assert.equal(payload[0].use_round, true);

tests/test_dea_api.py

@@ -1,10 +1,15 @@
 import asyncio
 import json
+import tempfile
 import unittest
+from pathlib import Path
+from unittest.mock import patch
 
 from fastapi import HTTPException
 from fastapi.responses import JSONResponse
 
+import dea.csv_processing as csv_processing
+from dea.csv_processing import PRESET_FILE_ID
 from dea_api import (
     BodeCascadeParams,
     BodeCompareParams,
@@ -18,6 +23,7 @@ from dea_api import (
     design_filter,
     filter_csv,
     filter_csv_download,
+    preset_csv,
     upload_csv,
     write_mcu_command,
 )
@@ -159,6 +165,47 @@ class DeaApiTest(unittest.TestCase):
             raise AssertionError("Expected invalid MCU command validation to fail")
 
 
+
+    def test_preset_csv_endpoint_returns_local_ignored_file_metadata(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            preset_path = Path(tmp) / "preset_signals.csv"
+            preset_path.write_text("time,value\n0,1\n1,2\n", encoding="utf-8")
+
+            with patch.object(csv_processing, "PRESET_CSV_PATH", str(preset_path)):
+                body = preset_csv()
+
+        self.assertTrue(body["available"])
+        self.assertEqual(body["file_id"], PRESET_FILE_ID)
+        self.assertEqual(body["columns"], ["time", "value"])
+        self.assertEqual(body["default_col_idx"], 1)
+        self.assertEqual(body["rows"], 2)
+
+    def test_filter_can_use_preset_file_id_and_skip_blank_signal_cells(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            preset_path = Path(tmp) / "preset_signals.csv"
+            preset_path.write_text("time,value\n0,1\n1,\n2,3\n", encoding="utf-8")
+
+            with patch.object(csv_processing, "PRESET_CSV_PATH", str(preset_path)):
+                body = asyncio.run(
+                    filter_csv(
+                        file_id=PRESET_FILE_ID,
+                        b="1",
+                        a="1",
+                        col_idx=1,
+                        b_int=None,
+                        a_int=None,
+                        shift_in=14,
+                        shift_out=14,
+                        shift_b=14,
+                        shift_a=14,
+                        use_round=False,
+                    )
+                )
+
+        self.assertEqual(body["total_points"], 2)
+        self.assertEqual(body["index"], [0, 2])
+        self.assertEqual(body["original"], [1.0, 3.0])
+
     def test_filter_downsamples_plot_response_for_large_csv(self):
         rows = ["value"] + [str(i) for i in range(6001)]
         upload = InMemoryUpload(("\n".join(rows) + "\n").encode("utf-8"), "input.csv")