From f7f297c3dfe2340fca4a3a9d10d7a138ed49020f Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Sat, 20 Jun 2026 17:06:55 +0200 Subject: [PATCH] Add synthetic array (waveform) DSP support + UX improvements Adds full array/waveform support through the synthetic DSP engine: a dsp.Sample value model (scalar or []float64), array ops (index, slice, sum, mean, min, max, length, fft) with an in-tree radix-2 FFT, and static type propagation (OpOutputType) that the editor mirrors to colour wires by data type and flag invalid wirings. Stateful filters and lua stay scalar-only. Adds a waveform plot mode (x-vs-index trace). Also: errored-node hover reasons, S/N add-signal/add-node HUD shortcuts in the synthetic editor, and view-mode widgets that blend with the canvas background (chrome kept in edit mode). Co-Authored-By: Claude Opus 4.6 --- TODO.md | 37 ++- docs/FUNCTIONAL_SPEC.md | 1 + docs/TECHNICAL_SPEC.md | 37 ++- .../datasource/synthetic/array_graph_test.go | 222 +++++++++++++++++ internal/datasource/synthetic/dsp_bridge.go | 24 ++ internal/datasource/synthetic/graph.go | 86 ++++++- internal/datasource/synthetic/synthetic.go | 62 ++++- internal/dsp/array_nodes.go | 230 ++++++++++++++++++ internal/dsp/array_nodes_test.go | 213 ++++++++++++++++ internal/dsp/fft.go | 58 +++++ internal/dsp/sample.go | 151 ++++++++++++ internal/dsp/types.go | 74 ++++++ web/src/Canvas.tsx | 2 +- web/src/PropertiesPane.tsx | 1 + web/src/SyntheticGraphEditor.tsx | 133 +++++++++- web/src/lib/synthTypes.ts | 86 +++++++ web/src/styles.css | 79 ++++++ web/src/widgets/PlotWidget.tsx | 31 ++- 18 files changed, 1470 insertions(+), 57 deletions(-) create mode 100644 internal/datasource/synthetic/array_graph_test.go create mode 100644 internal/dsp/array_nodes.go create mode 100644 internal/dsp/array_nodes_test.go create mode 100644 internal/dsp/fft.go create mode 100644 internal/dsp/sample.go create mode 100644 internal/dsp/types.go create mode 100644 web/src/lib/synthTypes.ts diff --git a/TODO.md b/TODO.md index fff23d4..e60d5b0 100644 --- a/TODO.md +++ b/TODO.md @@ -1,7 +1,6 @@ # TODO -- [ ] Implement git style versioning for: synthetic variable, panels, control logic -- [ ] Implement configuration manager: +- [ ] **MAJOR** Implement configuration manager: - configuration is a set of signals (that can be organized in group and subgroup) that are used to configure a system or a sub system - with configuration manager user should be able to: - Define and manage configuration sets: delete (keep server side copy of deleted config), create (specifying default to parameters, mandatory, optional etc), edit (with versioning git style) and compare set @@ -12,17 +11,31 @@ - user can fork an existing config instance to create a new one - user can compare two instances: show unified diff or side by side diff - etc... + - user can apply and save configuration instances + - in logic editor and control loop add nodes to read/write/create/apply config instances +- [ ] Improve UX: + - [x] Synthetic editor: + - [x] color code the node link by type + - [x] hover on a block in error should show the reason + - [x] add proper array functionality + - [x] add shortcut to add Signals (S) and nodes (N) with HUD + - [ ] Panels: + - [x] in view mode the widgets should have no border/bg but blend with background + - [ ] add widgets such +- [ ] Implement git style versioning for: synthetic variable, panels, control logic: + - possibility to fork any version + - click to view the version + - possibility to view graphical diff between versions (side by side or unified diff) + - simple slick versioning pane: + - vertical tree like + - each version represented by a circle + - active (the one currently view/edited) version has circle bigger then rest + - selected (the one that will be executed/showed by user) version has circle full, not active only border + - unsaved / new version appear with connection line dashed - [ ] Implement admin pane: create / manage groups, set users permits, manage auditors etc - [ ] Implement new datasources: + - [ ] Finalize alarm service - [ ] modbus tcp - [ ] scpi tcp -- [ ] Improve UX: - - [ ] Synthetic editor: - - color code the node link by type - - hover on a block in error should show the reason - - add proper array functionality - - add shortcut to add Signals (S) and nodes (N) with HUD - - [ ] Panels: - - in view mode the widgets should have no border/bg but blend with background - - add widgets such -- [ ] Implement proper distributed server side nodes to balance load and have redundancy (if a node is not available anymore all its clients migrate seamelessly to another) + - [ ] udp? other? +- [ ] **MAJOR** Implement proper distributed server side nodes to balance load and have redundancy (if a node is not available anymore all its clients migrate seamelessly to another) diff --git a/docs/FUNCTIONAL_SPEC.md b/docs/FUNCTIONAL_SPEC.md index cb15118..643bed6 100644 --- a/docs/FUNCTIONAL_SPEC.md +++ b/docs/FUNCTIONAL_SPEC.md @@ -192,6 +192,7 @@ When multiple widgets are selected: | Histogram | numeric scalar(s) | | Bar chart | numeric scalar(s) | | Logic analyser | boolean / integer (bitset) | +| Waveform | 1-D numeric array (latest sample, x-vs-index) | ### 4.5 Widget Properties (Properties Pane) diff --git a/docs/TECHNICAL_SPEC.md b/docs/TECHNICAL_SPEC.md index 1c6de44..4607651 100644 --- a/docs/TECHNICAL_SPEC.md +++ b/docs/TECHNICAL_SPEC.md @@ -41,7 +41,7 @@ | ---------------- | --------------------------------------------------------------- | | `preact` 10 | Virtual DOM UI framework | | `uPlot` | Extremely fast time-series/line plot (canvas-based, < 40 kB) | -| `Apache ECharts` | FFT, waterfall, histogram, bar, logic analyser plots | +| `Apache ECharts` | FFT, waterfall, histogram, bar, logic analyser, waveform plots | | `uplot.css` | uPlot default stylesheet | **Intentionally excluded:** React, Vue, Svelte, Konva, WebGPU, jQuery, npm at runtime. @@ -234,15 +234,30 @@ endpoints and for any change to a panel's `` block. **Built-in node types:** -| Node type | Parameters | Description | -| ------------ | ----------------------------------- | ------------------------------------------------ | -| `source` | `ds`, `name` | Reads a signal from any data source | -| `gain` | `factor` | Multiplies by a constant | -| `offset` | `value` | Adds a constant | -| `moving_avg` | `window` (samples) | Rolling mean | -| `lowpass` | `freq` (Hz), `order` (1–8) | Cascaded IIR Butterworth-style low-pass filter | -| `formula` | `expr` | Inline math expression (variables: `a`, `b`, …) | -| `lua` | `script` | Arbitrary Lua 5.1 code with persistent state | +| Node type | Parameters | In→Out | Description | +| ---------------- | --------------------------- | -------------- | ----------------------------------------------- | +| `source` | `ds`, `name` | — | Reads a signal from any data source | +| `gain` | `gain` | elementwise | Multiplies by a constant | +| `offset` | `offset` | elementwise | Adds a constant | +| `add`/`subtract` | — | elementwise | Sum of inputs / `a − b` | +| `multiply`/`divide` | — | elementwise | Product of inputs / `a ÷ b` | +| `clamp` | `min`, `max` | elementwise | Constrains to a range | +| `threshold` | `threshold`, `high`, `low` | elementwise | Comparator output | +| `moving_average` | `window` (samples) | scalar-only | Rolling mean | +| `rms` | `window` (samples) | scalar-only | Rolling RMS | +| `derivative` | — | scalar-only | Time derivative (per-sample `dt`) | +| `integrate` | — | scalar-only | Trapezoidal integral | +| `lowpass` | `freq` (Hz), `order` (1–8) | scalar-only | Cascaded IIR Butterworth-style low-pass filter | +| `expr` | `expr`, `vars` | elementwise | Inline math expression (named inputs) | +| `lua` | `script`, `vars` | scalar-only | Arbitrary Lua 5.1 code with persistent state | +| `index` | `i` | array→scalar | Element `i` of a waveform (bounds-checked) | +| `slice` | `start`, `end` | array→array | Sub-range of a waveform (clamped) | +| `sum`/`mean` | — | array→scalar | Σ / average of a waveform | +| `min`/`max` | — | array→scalar | Reduction of a waveform | +| `length` | — | array→scalar | Element count of a waveform | +| `fft` | — | array→array | Magnitude spectrum (zero-padded to next pow-2) | + +**Scalar vs waveform values:** A value flowing through the graph is a `dsp.Sample` — either a scalar `float64` or a `[]float64` waveform (the array-aware counterpart of EPICS `TypeFloat64Array`). *Elementwise* ops broadcast over arrays (scalar inputs act as constants; array inputs must share a length). *Reduction*/*producer* ops (`index`/`slice`/`sum`/…/`fft`) operate natively on waveforms. *Scalar-only* ops (stateful filters + `lua`) reject array inputs, since their per-evaluation state cannot be split across array lanes. `OpOutputType` (`internal/dsp/types.go`) propagates types statically at compile time to reject invalid wirings and to report the synthetic's metadata type; the editor mirrors these rules in `web/src/lib/synthTypes.ts` to colour wires by data type (scalar vs array) and flag type-incompatible links. Runtime `Sample` typing is authoritative. **Low-pass filter implementation:** Cascaded first-order IIR sections. Each stage computes `y = y_prev + α·(x − y_prev)` where `α = dt / (RC + dt)` and `RC = 1/(2π·fc)`. `dt` is computed per sample from source timestamps so the filter is correct for event-driven (non-uniform) data. @@ -353,7 +368,7 @@ The edit canvas is a free-form HTML div with absolutely positioned widget compon View mode renders widgets as absolutely positioned Preact components on a scrollable canvas div: - Each widget subscribes to its signal store(s) in a `useEffect` and re-renders only when values change. -- uPlot (time series) and ECharts (histogram, bar, FFT, waterfall, logic analyser) manage their own canvas elements inside their widget component. +- uPlot (time series) and ECharts (histogram, bar, FFT, waterfall, logic analyser, waveform) manage their own canvas elements inside their widget component. The `waveform` plot renders a waveform (array) signal's latest `[]float64` as an x-vs-index trace, replacing the trace on each update. - Plot widgets maintain a rolling ring buffer of 200,000 samples per signal for smooth long-window display. - Step-hold interpolation: when multiple signals at different update rates share a plot, the most recent value is carried forward to fill the shared time axis correctly. diff --git a/internal/datasource/synthetic/array_graph_test.go b/internal/datasource/synthetic/array_graph_test.go new file mode 100644 index 0000000..5dcca94 --- /dev/null +++ b/internal/datasource/synthetic/array_graph_test.go @@ -0,0 +1,222 @@ +package synthetic + +import ( + "context" + "log/slog" + "math" + "os" + "testing" + "time" + + "github.com/uopi/uopi/internal/broker" + "github.com/uopi/uopi/internal/datasource" + "github.com/uopi/uopi/internal/dsp" +) + +// evalSampleDef compiles a SignalDef and evaluates it against per-source +// Samples keyed by source node id, returning the output Sample. +func evalSampleDef(t *testing.T, def SignalDef, srcVals map[string]dsp.Sample) dsp.Sample { + t.Helper() + rg, err := compileGraph(def) + if err != nil { + t.Fatalf("compileGraph: %v", err) + } + out, err := rg.evalSample(srcVals) + if err != nil { + t.Fatalf("evalSample: %v", err) + } + return out +} + +// TestArrayElementwiseChain runs an array source through an elementwise op +// (gain) and asserts the output stays an array, broadcast per element. +func TestArrayElementwiseChain(t *testing.T) { + def := SignalDef{ + Name: "scaled", + Graph: &Graph{ + Output: "out", + Nodes: []GraphNode{ + {ID: "a", Kind: "source", DS: "x", Signal: "wave"}, + {ID: "g", Kind: "op", Op: "gain", Inputs: []string{"a"}, Params: map[string]any{"gain": 2.0}}, + {ID: "out", Kind: "output", Inputs: []string{"g"}}, + }, + }, + } + out := evalSampleDef(t, def, map[string]dsp.Sample{"a": dsp.Array([]float64{1, 2, 3})}) + if !out.IsArray { + t.Fatalf("want array output, got %v", out) + } + want := []float64{2, 4, 6} + for i, v := range want { + if out.Arr[i] != v { + t.Errorf("scaled[%d]: want %v, got %v", i, v, out.Arr[i]) + } + } +} + +// TestArrayReductionToScalar runs an array source into mean (array→scalar) and +// asserts a scalar output and an array-output compile type for the producer. +func TestArrayReductionToScalar(t *testing.T) { + def := SignalDef{ + Name: "avg", + Graph: &Graph{ + Output: "out", + Nodes: []GraphNode{ + {ID: "a", Kind: "source", DS: "x", Signal: "wave"}, + {ID: "m", Kind: "op", Op: "mean", Inputs: []string{"a"}}, + {ID: "out", Kind: "output", Inputs: []string{"m"}}, + }, + }, + } + out := evalSampleDef(t, def, map[string]dsp.Sample{"a": dsp.Array([]float64{2, 4, 6, 8})}) + if out.IsArray { + t.Fatalf("want scalar output, got array %v", out.Arr) + } + if math.Abs(out.F-5) > 1e-9 { + t.Errorf("mean: want 5, got %v", out.F) + } +} + +// TestArrayOutTypeMetadata verifies compileGraph reports an array output type +// for a pure-elementwise array graph and scalar for a reduction graph. +func TestArrayOutTypeMetadata(t *testing.T) { + arrayGraph := SignalDef{ + Name: "fftout", + Graph: &Graph{ + Output: "out", + Nodes: []GraphNode{ + {ID: "a", Kind: "source", DS: "x", Signal: "wave"}, + {ID: "f", Kind: "op", Op: "fft", Inputs: []string{"a"}}, + {ID: "out", Kind: "output", Inputs: []string{"f"}}, + }, + }, + } + rg, err := compileGraph(arrayGraph) + if err != nil { + t.Fatalf("compileGraph: %v", err) + } + if rg.outType != dsp.ValArray { + t.Errorf("fft graph outType: want ValArray, got %v", rg.outType) + } + + reduction := SignalDef{ + Name: "sumout", + Graph: &Graph{ + Output: "out", + Nodes: []GraphNode{ + {ID: "a", Kind: "source", DS: "x", Signal: "wave"}, + {ID: "s", Kind: "op", Op: "sum", Inputs: []string{"a"}}, + {ID: "out", Kind: "output", Inputs: []string{"s"}}, + }, + }, + } + rg2, err := compileGraph(reduction) + if err != nil { + t.Fatalf("compileGraph: %v", err) + } + if rg2.outType != dsp.ValScalar { + t.Errorf("sum graph outType: want ValScalar, got %v", rg2.outType) + } +} + +// TestStatefulRejectsArray verifies a stateful op (moving_average) errors when +// fed an array input at runtime. +func TestStatefulRejectsArray(t *testing.T) { + def := SignalDef{ + Name: "ma", + Graph: &Graph{ + Output: "out", + Nodes: []GraphNode{ + {ID: "a", Kind: "source", DS: "x", Signal: "wave"}, + {ID: "m", Kind: "op", Op: "moving_average", Inputs: []string{"a"}, Params: map[string]any{"window": 3.0}}, + {ID: "out", Kind: "output", Inputs: []string{"m"}}, + }, + }, + } + rg, err := compileGraph(def) + if err != nil { + t.Fatalf("compileGraph: %v", err) + } + if _, err := rg.evalSample(map[string]dsp.Sample{"a": dsp.Array([]float64{1, 2, 3})}); err == nil { + t.Error("expected moving_average to reject an array input") + } +} + +// TestSubscribeArrayPassthrough is an end-to-end check that a synthetic with an +// array-valued source and an elementwise op emits a []float64 over the broker, +// and that GetMetadata reports the waveform type. +func TestSubscribeArrayPassthrough(t *testing.T) { + log := slog.New(slog.NewTextHandler(os.Stderr, nil)) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + base := time.Date(2026, 6, 19, 10, 0, 0, 0, time.UTC) + src := &seqSource{name: "src", seq: []datasource.Value{ + {Timestamp: base.Add(1 * time.Second), Data: []float64{1, 2, 3}, Quality: datasource.QualityGood}, + {Timestamp: base.Add(2 * time.Second), Data: []float64{4, 5, 6}, Quality: datasource.QualityGood}, + }} + + brk := broker.New(ctx, log) + brk.Register(src) + syn := New(t.TempDir(), brk, log) + if err := syn.Connect(ctx); err != nil { + t.Fatal(err) + } + // gain x2 elementwise keeps the value an array. + if err := syn.AddSignal(SignalDef{ + Name: "scaled", + Graph: &Graph{Output: "out", Nodes: []GraphNode{ + {ID: "a", Kind: "source", DS: "src", Signal: "x"}, + {ID: "g", Kind: "op", Op: "gain", Inputs: []string{"a"}, Params: map[string]any{"gain": 2.0}}, + {ID: "out", Kind: "output", Inputs: []string{"g"}}, + }}, + }); err != nil { + t.Fatal(err) + } + // An fft-based signal has a statically-known array output type (the source's + // runtime type need not be known), so its metadata reports the waveform type. + if err := syn.AddSignal(SignalDef{ + Name: "spectrum", + Graph: &Graph{Output: "out", Nodes: []GraphNode{ + {ID: "a", Kind: "source", DS: "src", Signal: "x"}, + {ID: "f", Kind: "op", Op: "fft", Inputs: []string{"a"}}, + {ID: "out", Kind: "output", Inputs: []string{"f"}}, + }}, + }); err != nil { + t.Fatal(err) + } + + meta, err := syn.GetMetadata(ctx, "spectrum") + if err != nil { + t.Fatal(err) + } + if meta.Type != datasource.TypeFloat64Array { + t.Errorf("metadata type: want TypeFloat64Array, got %v", meta.Type) + } + + ch := make(chan datasource.Value, 8) + if _, err := syn.Subscribe(ctx, "scaled", ch); err != nil { + t.Fatal(err) + } + + want := [][]float64{{2, 4, 6}, {8, 10, 12}} + for i, w := range want { + select { + case v := <-ch: + arr, ok := v.Data.([]float64) + if !ok { + t.Fatalf("emit #%d: want []float64, got %T", i, v.Data) + } + if len(arr) != len(w) { + t.Fatalf("emit #%d: want len %d, got %d", i, len(w), len(arr)) + } + for k, val := range w { + if arr[k] != val { + t.Errorf("emit #%d [%d]: want %v, got %v", i, k, val, arr[k]) + } + } + case <-time.After(2 * time.Second): + t.Fatalf("timeout waiting for emit #%d", i) + } + } +} diff --git a/internal/datasource/synthetic/dsp_bridge.go b/internal/datasource/synthetic/dsp_bridge.go index 3d36da6..aa6ecac 100644 --- a/internal/datasource/synthetic/dsp_bridge.go +++ b/internal/datasource/synthetic/dsp_bridge.go @@ -126,6 +126,30 @@ func buildNode(d NodeDef) (dsp.Node, error) { case "lua": return &dsp.LuaNode{Script: stringParam(p, "script"), Vars: stringSliceParam(p, "vars")}, nil + case "index": + return &dsp.IndexNode{I: int(floatParam(p, "i"))}, nil + + case "slice": + return &dsp.SliceNode{Start: int(floatParam(p, "start")), End: int(floatParam(p, "end"))}, nil + + case "sum": + return &dsp.SumNode{}, nil + + case "mean": + return &dsp.MeanNode{}, nil + + case "min": + return &dsp.MinNode{}, nil + + case "max": + return &dsp.MaxNode{}, nil + + case "length": + return &dsp.LengthNode{}, nil + + case "fft": + return &dsp.FFTNode{}, nil + default: return nil, fmt.Errorf("unknown node type %q", d.Type) } diff --git a/internal/datasource/synthetic/graph.go b/internal/datasource/synthetic/graph.go index 982976f..16a68eb 100644 --- a/internal/datasource/synthetic/graph.go +++ b/internal/datasource/synthetic/graph.go @@ -13,9 +13,10 @@ import ( // each node's inputs already resolved. Op-node state maps persist across // evaluations (for stateful nodes like moving_average / lua). type runtimeGraph struct { - order []*rtNode // topological order (sources first, output last) - sources []rtSource // source nodes, in topological order - outputID string // id of the output node + order []*rtNode // topological order (sources first, output last) + sources []rtSource // source nodes, in topological order + outputID string // id of the output node + outType dsp.ValType // best-effort output type (scalar/array/unknown) } type rtNode struct { @@ -41,24 +42,30 @@ func (rg *runtimeGraph) sourceRefs() []broker.SignalRef { return refs } -// eval computes the output value given the latest value for each source node -// (keyed by source node id). Nodes are visited in topological order so every -// input is already present in vals by the time a node is processed. -func (rg *runtimeGraph) eval(sourceVals map[string]float64) (float64, error) { - vals := make(map[string]float64, len(rg.order)) +// evalSample computes the output Sample (scalar or array) given the latest +// value for each source node (keyed by source node id). Nodes are visited in +// topological order so every input is present by the time a node is processed. +// +// Op dispatch: +// - ArrayNode ops (reductions/producers) run natively on Samples. +// - stateless elementwise ops broadcast over array inputs. +// - stateful ops (filters) and lua are scalar-only; an array input errors, +// since their per-evaluation state cannot be split across array lanes. +func (rg *runtimeGraph) evalSample(sourceVals map[string]dsp.Sample) (dsp.Sample, error) { + vals := make(map[string]dsp.Sample, len(rg.order)) for id, v := range sourceVals { vals[id] = v } for _, n := range rg.order { switch n.kind { case "op": - in := make([]float64, len(n.inputs)) + in := make([]dsp.Sample, len(n.inputs)) for i, id := range n.inputs { in[i] = vals[id] } - r, err := n.op.Process(in, n.state) + r, err := evalOp(n, in) if err != nil { - return 0, fmt.Errorf("node %s (%s): %w", n.id, n.op.Type(), err) + return dsp.Sample{}, fmt.Errorf("node %s (%s): %w", n.id, n.op.Type(), err) } vals[n.id] = r case "output": @@ -70,6 +77,44 @@ func (rg *runtimeGraph) eval(sourceVals map[string]float64) (float64, error) { return vals[rg.outputID], nil } +// evalOp runs a single op node over its Sample inputs, choosing the right +// execution path for the node type. +func evalOp(n *rtNode, in []dsp.Sample) (dsp.Sample, error) { + if an, ok := n.op.(dsp.ArrayNode); ok { + return an.ProcessSample(in, n.state) + } + if dsp.StatelessElementwise(n.op.Type()) { + return dsp.ApplyElementwise(n.op, in, n.state) + } + // Stateful / lua: scalar-only. + row := make([]float64, len(in)) + for i, s := range in { + if s.IsArray { + return dsp.Sample{}, fmt.Errorf("does not accept an array input") + } + row[i] = s.F + } + r, err := n.op.Process(row, n.state) + if err != nil { + return dsp.Sample{}, err + } + return dsp.Scalar(r), nil +} + +// eval is the scalar wrapper around evalSample, kept so callers and tests that +// deal purely in float64 (legacy linear graphs, scalar sources) are unchanged. +func (rg *runtimeGraph) eval(sourceVals map[string]float64) (float64, error) { + sv := make(map[string]dsp.Sample, len(sourceVals)) + for id, v := range sourceVals { + sv[id] = dsp.Scalar(v) + } + out, err := rg.evalSample(sv) + if err != nil { + return 0, err + } + return out.F, nil +} + // compileGraph converts a SignalDef into an executable runtimeGraph. When the // def carries an explicit Graph it is used directly; otherwise the legacy // Inputs+Pipeline form is converted to an equivalent linear graph (see toGraph). @@ -83,23 +128,42 @@ func compileGraph(def SignalDef) (*runtimeGraph, error) { return nil, err } rg := &runtimeGraph{outputID: g.Output} + // nodeType tracks each node's best-effort output type for static + // propagation. Sources are unknown at compile time (their real type is + // only known once data flows), so type errors here are advisory; runtime + // Sample typing is authoritative. + nodeType := make(map[string]dsp.ValType, len(order)) for _, gn := range order { switch gn.Kind { case "source": rg.sources = append(rg.sources, rtSource{id: gn.ID, ref: broker.SignalRef{DS: gn.DS, Name: gn.Signal}}) + nodeType[gn.ID] = dsp.ValUnknown case "op": node, err := buildNode(NodeDef{Type: gn.Op, Params: gn.Params}) if err != nil { return nil, fmt.Errorf("node %q: %w", gn.ID, err) } + inTypes := make([]dsp.ValType, len(gn.Inputs)) + for i, id := range gn.Inputs { + inTypes[i] = nodeType[id] + } + ot, terr := dsp.OpOutputType(gn.Op, inTypes) + if terr != nil { + return nil, fmt.Errorf("node %q: %w", gn.ID, terr) + } + nodeType[gn.ID] = ot rg.order = append(rg.order, &rtNode{id: gn.ID, kind: "op", op: node, state: map[string]any{}, inputs: gn.Inputs}) case "output": rg.outputID = gn.ID + if len(gn.Inputs) > 0 { + nodeType[gn.ID] = nodeType[gn.Inputs[0]] + } rg.order = append(rg.order, &rtNode{id: gn.ID, kind: "output", inputs: gn.Inputs}) default: return nil, fmt.Errorf("node %q: unknown kind %q", gn.ID, gn.Kind) } } + rg.outType = nodeType[rg.outputID] return rg, nil } diff --git a/internal/datasource/synthetic/synthetic.go b/internal/datasource/synthetic/synthetic.go index 224b86e..e6658b9 100644 --- a/internal/datasource/synthetic/synthetic.go +++ b/internal/datasource/synthetic/synthetic.go @@ -13,6 +13,7 @@ import ( "github.com/uopi/uopi/internal/broker" "github.com/uopi/uopi/internal/datasource" + "github.com/uopi/uopi/internal/dsp" ) const definitionsFile = "synthetic.json" @@ -78,7 +79,7 @@ func (s *Synthetic) ListSignals(_ context.Context) ([]datasource.Metadata, error out := make([]datasource.Metadata, 0, len(s.signals)) for _, st := range s.signals { - out = append(out, defToMetadata(st.def)) + out = append(out, defToMetadata(st.def, outTypeOf(st))) } return out, nil } @@ -93,7 +94,7 @@ func (s *Synthetic) FilteredMetadata(keep func(SignalDef) bool) []datasource.Met out := make([]datasource.Metadata, 0, len(s.signals)) for _, st := range s.signals { if keep(st.def) { - out = append(out, defToMetadata(st.def)) + out = append(out, defToMetadata(st.def, outTypeOf(st))) } } return out @@ -108,7 +109,7 @@ func (s *Synthetic) GetMetadata(_ context.Context, signal string) (datasource.Me if !ok { return datasource.Metadata{}, datasource.ErrNotFound } - return defToMetadata(st.def), nil + return defToMetadata(st.def, outTypeOf(st)), nil } // Subscribe registers ch to receive computed values for the named signal. @@ -138,7 +139,7 @@ func (s *Synthetic) Subscribe(ctx context.Context, signal string, ch chan<- data defer cancel() // Latest value and timestamp per source node id. - latest := make(map[string]float64, len(refs)) + latest := make(map[string]dsp.Sample, len(refs)) latestTs := make([]time.Time, len(refs)) ready := make([]bool, len(refs)) @@ -163,7 +164,7 @@ func (s *Synthetic) Subscribe(ctx context.Context, signal string, ch chan<- data if !ok { return } - val := toFloat64(u.Value.Data) + val := toSample(u.Value.Data) select { case updateCh <- indexedUpdate{idx: idx, val: val, ts: u.Value.Timestamp}: default: @@ -224,7 +225,7 @@ func (s *Synthetic) Subscribe(ctx context.Context, signal string, ch chan<- data return } - result, err := cur.rg.eval(latest) + result, err := cur.rg.evalSample(latest) if err != nil { s.log.Warn("synthetic: pipeline error", "signal", signal, "err", err) continue @@ -232,7 +233,7 @@ func (s *Synthetic) Subscribe(ctx context.Context, signal string, ch chan<- data v := datasource.Value{ Timestamp: outTs, - Data: result, + Data: result.AsAny(), Quality: datasource.QualityGood, } select { @@ -421,11 +422,17 @@ func (s *Synthetic) startSignal(def SignalDef) error { return nil } -// defToMetadata converts a SignalDef into a datasource.Metadata. -func defToMetadata(def SignalDef) datasource.Metadata { +// defToMetadata converts a SignalDef into a datasource.Metadata. outType is the +// compiled graph's best-effort output type; an array output is reported as a +// waveform (TypeFloat64Array) so widgets can pick a compatible view. +func defToMetadata(def SignalDef, outType dsp.ValType) datasource.Metadata { + dt := datasource.TypeFloat64 + if outType == dsp.ValArray { + dt = datasource.TypeFloat64Array + } return datasource.Metadata{ Name: def.Name, - Type: datasource.TypeFloat64, + Type: dt, Unit: def.Meta.Unit, Description: def.Meta.Description, DisplayLow: def.Meta.DisplayLow, @@ -434,7 +441,38 @@ func defToMetadata(def SignalDef) datasource.Metadata { } } -// toFloat64 coerces any numeric value from a datasource.Value.Data to float64. +// outTypeOf returns the compiled output type for a signal state, or unknown. +func outTypeOf(st *signalState) dsp.ValType { + if st == nil || st.rg == nil { + return dsp.ValUnknown + } + return st.rg.outType +} + +// toSample coerces a datasource.Value.Data into a dsp.Sample: arrays become +// array Samples (waveforms), everything else a scalar Sample. +func toSample(v any) dsp.Sample { + switch val := v.(type) { + case []float64: + return dsp.Array(val) + case []float32: + out := make([]float64, len(val)) + for i, e := range val { + out[i] = float64(e) + } + return dsp.Array(out) + case []int: + out := make([]float64, len(val)) + for i, e := range val { + out[i] = float64(e) + } + return dsp.Array(out) + default: + return dsp.Scalar(toFloat64(v)) + } +} + +// toFloat64 coerces any numeric scalar value from a datasource.Value.Data to float64. func toFloat64(v any) float64 { switch val := v.(type) { case float64: @@ -460,6 +498,6 @@ func toFloat64(v any) float64 { // indexedUpdate carries a value from one upstream goroutine to the pipeline runner. type indexedUpdate struct { idx int - val float64 + val dsp.Sample ts time.Time } diff --git a/internal/dsp/array_nodes.go b/internal/dsp/array_nodes.go new file mode 100644 index 0000000..ff5019e --- /dev/null +++ b/internal/dsp/array_nodes.go @@ -0,0 +1,230 @@ +package dsp + +import ( + "errors" + "fmt" + "math" +) + +// This file holds ArrayNode ops: those that operate natively on waveform +// (float64 array) Samples — reductions (array→scalar), producers (array→array), +// and element access. Each also implements the legacy scalar Node interface +// (treating a scalar as a single-element array) so it remains usable from the +// scalar eval path. + +// reductionProcess adapts a scalar Process call to a reduction ArrayNode. +func reductionProcess(n ArrayNode, in []float64, st map[string]any) (float64, error) { + s, err := n.ProcessSample(scalarInputs(in), st) + if err != nil { + return 0, err + } + return s.F, nil +} + +// ── IndexNode ─────────────────────────────────────────────────────────────── + +// IndexNode extracts element I of an array input (array→scalar). +type IndexNode struct{ I int } + +func (n *IndexNode) Type() string { return "index" } +func (n *IndexNode) Process(in []float64, st map[string]any) (float64, error) { + return reductionProcess(n, in, st) +} +func (n *IndexNode) ProcessSample(in []Sample, _ map[string]any) (Sample, error) { + if len(in) == 0 { + return Sample{}, errors.New("index: no inputs") + } + arr := in[0].AsArray() + if n.I < 0 || n.I >= len(arr) { + return Sample{}, fmt.Errorf("index: %d out of range [0,%d)", n.I, len(arr)) + } + return Scalar(arr[n.I]), nil +} + +// ── SliceNode ─────────────────────────────────────────────────────────────── + +// SliceNode returns a sub-range [Start,End) of an array input (array→array), +// clamped to the array bounds. End <= 0 means "to the end". +type SliceNode struct{ Start, End int } + +func (n *SliceNode) Type() string { return "slice" } +func (n *SliceNode) Process(in []float64, st map[string]any) (float64, error) { + s, err := n.ProcessSample(scalarInputs(in), st) + if err != nil { + return 0, err + } + if len(s.Arr) == 0 { + return 0, nil + } + return s.Arr[0], nil +} +func (n *SliceNode) ProcessSample(in []Sample, _ map[string]any) (Sample, error) { + if len(in) == 0 { + return Sample{}, errors.New("slice: no inputs") + } + arr := in[0].AsArray() + start := n.Start + if start < 0 { + start = 0 + } + if start > len(arr) { + start = len(arr) + } + end := n.End + if end <= 0 || end > len(arr) { + end = len(arr) + } + if end < start { + end = start + } + out := make([]float64, end-start) + copy(out, arr[start:end]) + return Array(out), nil +} + +// ── reductions ──────────────────────────────────────────────────────────────── + +// SumNode sums an array input (array→scalar). +type SumNode struct{} + +func (n *SumNode) Type() string { return "sum" } +func (n *SumNode) Process(in []float64, st map[string]any) (float64, error) { + return reductionProcess(n, in, st) +} +func (n *SumNode) ProcessSample(in []Sample, _ map[string]any) (Sample, error) { + if len(in) == 0 { + return Sample{}, errors.New("sum: no inputs") + } + var s float64 + for _, v := range in[0].AsArray() { + s += v + } + return Scalar(s), nil +} + +// MeanNode averages an array input (array→scalar). +type MeanNode struct{} + +func (n *MeanNode) Type() string { return "mean" } +func (n *MeanNode) Process(in []float64, st map[string]any) (float64, error) { + return reductionProcess(n, in, st) +} +func (n *MeanNode) ProcessSample(in []Sample, _ map[string]any) (Sample, error) { + if len(in) == 0 { + return Sample{}, errors.New("mean: no inputs") + } + arr := in[0].AsArray() + if len(arr) == 0 { + return Scalar(0), nil + } + var s float64 + for _, v := range arr { + s += v + } + return Scalar(s / float64(len(arr))), nil +} + +// MinNode returns the minimum element of an array input (array→scalar). +type MinNode struct{} + +func (n *MinNode) Type() string { return "min" } +func (n *MinNode) Process(in []float64, st map[string]any) (float64, error) { + return reductionProcess(n, in, st) +} +func (n *MinNode) ProcessSample(in []Sample, _ map[string]any) (Sample, error) { + if len(in) == 0 { + return Sample{}, errors.New("min: no inputs") + } + arr := in[0].AsArray() + if len(arr) == 0 { + return Scalar(0), nil + } + m := arr[0] + for _, v := range arr[1:] { + if v < m { + m = v + } + } + return Scalar(m), nil +} + +// MaxNode returns the maximum element of an array input (array→scalar). +type MaxNode struct{} + +func (n *MaxNode) Type() string { return "max" } +func (n *MaxNode) Process(in []float64, st map[string]any) (float64, error) { + return reductionProcess(n, in, st) +} +func (n *MaxNode) ProcessSample(in []Sample, _ map[string]any) (Sample, error) { + if len(in) == 0 { + return Sample{}, errors.New("max: no inputs") + } + arr := in[0].AsArray() + if len(arr) == 0 { + return Scalar(0), nil + } + m := arr[0] + for _, v := range arr[1:] { + if v > m { + m = v + } + } + return Scalar(m), nil +} + +// LengthNode returns the element count of an array input (array→scalar). +type LengthNode struct{} + +func (n *LengthNode) Type() string { return "length" } +func (n *LengthNode) Process(in []float64, st map[string]any) (float64, error) { + return reductionProcess(n, in, st) +} +func (n *LengthNode) ProcessSample(in []Sample, _ map[string]any) (Sample, error) { + if len(in) == 0 { + return Sample{}, errors.New("length: no inputs") + } + return Scalar(float64(len(in[0].AsArray()))), nil +} + +// ── FFTNode ──────────────────────────────────────────────────────────────── + +// FFTNode computes the magnitude spectrum of an array input (array→array). The +// input is zero-padded to the next power of two; the output has that length and +// holds |X[k]| for each frequency bin. +type FFTNode struct{} + +func (n *FFTNode) Type() string { return "fft" } +func (n *FFTNode) Process(in []float64, st map[string]any) (float64, error) { + s, err := n.ProcessSample(scalarInputs(in), st) + if err != nil { + return 0, err + } + if len(s.Arr) == 0 { + return 0, nil + } + return s.Arr[0], nil +} +func (n *FFTNode) ProcessSample(in []Sample, _ map[string]any) (Sample, error) { + if len(in) == 0 { + return Sample{}, errors.New("fft: no inputs") + } + return Array(fftMagnitude(in[0].AsArray())), nil +} + +// fftMagnitude returns the magnitude spectrum of x, zero-padded to the next +// power of two. Returns an empty slice for empty input. +func fftMagnitude(x []float64) []float64 { + if len(x) == 0 { + return nil + } + n := nextPow2(len(x)) + re := make([]float64, n) + im := make([]float64, n) + copy(re, x) + fftRadix2(re, im) + mag := make([]float64, n) + for i := range mag { + mag[i] = math.Hypot(re[i], im[i]) + } + return mag +} diff --git a/internal/dsp/array_nodes_test.go b/internal/dsp/array_nodes_test.go new file mode 100644 index 0000000..688be99 --- /dev/null +++ b/internal/dsp/array_nodes_test.go @@ -0,0 +1,213 @@ +package dsp + +import ( + "math" + "testing" +) + +func TestSampleRoundTrip(t *testing.T) { + s := Scalar(3.5) + if s.IsArray { + t.Error("Scalar should not be an array") + } + if s.Type() != ValScalar { + t.Errorf("Scalar type: want ValScalar, got %v", s.Type()) + } + if s.AsAny() != 3.5 { + t.Errorf("Scalar AsAny: want 3.5, got %v", s.AsAny()) + } + if got := s.AsArray(); len(got) != 1 || got[0] != 3.5 { + t.Errorf("Scalar AsArray: want [3.5], got %v", got) + } + + a := Array([]float64{1, 2, 3}) + if !a.IsArray { + t.Error("Array should be an array") + } + if a.Type() != ValArray { + t.Errorf("Array type: want ValArray, got %v", a.Type()) + } + got, ok := a.AsAny().([]float64) + if !ok || len(got) != 3 { + t.Errorf("Array AsAny: want []float64 len 3, got %v", a.AsAny()) + } +} + +func TestApplyElementwiseAllScalar(t *testing.T) { + n := &AddNode{} + out, err := ApplyElementwise(n, []Sample{Scalar(2), Scalar(3)}, map[string]any{}) + if err != nil { + t.Fatal(err) + } + if out.IsArray || out.F != 5 { + t.Errorf("all-scalar add: want scalar 5, got %v", out) + } +} + +func TestApplyElementwiseBroadcast(t *testing.T) { + // array ⊕ scalar: scalar is a constant broadcast across the array. + n := &AddNode{} + out, err := ApplyElementwise(n, []Sample{Array([]float64{1, 2, 3}), Scalar(10)}, map[string]any{}) + if err != nil { + t.Fatal(err) + } + if !out.IsArray { + t.Fatalf("array+scalar: want array, got %v", out) + } + want := []float64{11, 12, 13} + for i, v := range want { + if out.Arr[i] != v { + t.Errorf("array+scalar[%d]: want %v, got %v", i, v, out.Arr[i]) + } + } +} + +func TestApplyElementwiseArrayArray(t *testing.T) { + n := &MultiplyNode{} + out, err := ApplyElementwise(n, []Sample{Array([]float64{1, 2, 3}), Array([]float64{4, 5, 6})}, map[string]any{}) + if err != nil { + t.Fatal(err) + } + want := []float64{4, 10, 18} + for i, v := range want { + if out.Arr[i] != v { + t.Errorf("array*array[%d]: want %v, got %v", i, v, out.Arr[i]) + } + } +} + +func TestApplyElementwiseLengthMismatch(t *testing.T) { + n := &AddNode{} + _, err := ApplyElementwise(n, []Sample{Array([]float64{1, 2}), Array([]float64{1, 2, 3})}, map[string]any{}) + if err == nil { + t.Error("expected length-mismatch error") + } +} + +func TestReductionNodes(t *testing.T) { + arr := []Sample{Array([]float64{2, 4, 6, 8})} + cases := []struct { + name string + node ArrayNode + want float64 + }{ + {"sum", &SumNode{}, 20}, + {"mean", &MeanNode{}, 5}, + {"min", &MinNode{}, 2}, + {"max", &MaxNode{}, 8}, + {"length", &LengthNode{}, 4}, + {"index", &IndexNode{I: 2}, 6}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + out, err := tc.node.ProcessSample(arr, map[string]any{}) + if err != nil { + t.Fatal(err) + } + if out.IsArray || out.F != tc.want { + t.Errorf("%s: want scalar %v, got %v", tc.name, tc.want, out) + } + }) + } +} + +func TestIndexNodeOutOfRange(t *testing.T) { + n := &IndexNode{I: 9} + _, err := n.ProcessSample([]Sample{Array([]float64{1, 2, 3})}, map[string]any{}) + if err == nil { + t.Error("expected out-of-range error") + } +} + +func TestSliceNode(t *testing.T) { + n := &SliceNode{Start: 1, End: 3} + out, err := n.ProcessSample([]Sample{Array([]float64{10, 20, 30, 40})}, map[string]any{}) + if err != nil { + t.Fatal(err) + } + want := []float64{20, 30} + if len(out.Arr) != len(want) { + t.Fatalf("slice: want len %d, got %d", len(want), len(out.Arr)) + } + for i, v := range want { + if out.Arr[i] != v { + t.Errorf("slice[%d]: want %v, got %v", i, v, out.Arr[i]) + } + } +} + +func TestFFTMagnitude(t *testing.T) { + // A constant signal has all energy in bin 0 (the DC term equals the sum). + x := []float64{1, 1, 1, 1} + mag := fftMagnitude(x) + if len(mag) != 4 { + t.Fatalf("fft len: want 4, got %d", len(mag)) + } + if math.Abs(mag[0]-4) > 1e-9 { + t.Errorf("fft DC bin: want 4, got %v", mag[0]) + } + for k := 1; k < len(mag); k++ { + if math.Abs(mag[k]) > 1e-9 { + t.Errorf("fft bin %d: want ~0, got %v", k, mag[k]) + } + } +} + +func TestFFTSingleTone(t *testing.T) { + // One full cycle of a cosine over 8 samples → energy in bins 1 and N-1. + n := 8 + x := make([]float64, n) + for i := range x { + x[i] = math.Cos(2 * math.Pi * float64(i) / float64(n)) + } + mag := fftMagnitude(x) + if math.Abs(mag[1]-float64(n)/2) > 1e-6 { + t.Errorf("fft tone bin 1: want %v, got %v", float64(n)/2, mag[1]) + } + if math.Abs(mag[n-1]-float64(n)/2) > 1e-6 { + t.Errorf("fft tone bin %d: want %v, got %v", n-1, float64(n)/2, mag[n-1]) + } +} + +func TestOpOutputType(t *testing.T) { + cases := []struct { + op string + in []ValType + want ValType + wantErr bool + }{ + // reductions → scalar regardless of input + {"sum", []ValType{ValArray}, ValScalar, false}, + {"mean", []ValType{ValScalar}, ValScalar, false}, + {"index", []ValType{ValUnknown}, ValScalar, false}, + // array producers require array, yield array + {"fft", []ValType{ValArray}, ValArray, false}, + {"slice", []ValType{ValUnknown}, ValArray, false}, + {"fft", []ValType{ValScalar}, ValUnknown, true}, + // scalar-only reject arrays + {"moving_average", []ValType{ValScalar}, ValScalar, false}, + {"lua", []ValType{ValArray}, ValUnknown, true}, + {"rms", []ValType{ValUnknown}, ValScalar, false}, + // elementwise: array if any array, scalar if all scalar, else unknown + {"add", []ValType{ValScalar, ValScalar}, ValScalar, false}, + {"add", []ValType{ValArray, ValScalar}, ValArray, false}, + {"gain", []ValType{ValUnknown}, ValUnknown, false}, + {"expr", []ValType{ValArray, ValScalar}, ValArray, false}, + } + for _, tc := range cases { + got, err := OpOutputType(tc.op, tc.in) + if tc.wantErr { + if err == nil { + t.Errorf("OpOutputType(%q,%v): expected error", tc.op, tc.in) + } + continue + } + if err != nil { + t.Errorf("OpOutputType(%q,%v): unexpected error %v", tc.op, tc.in, err) + continue + } + if got != tc.want { + t.Errorf("OpOutputType(%q,%v): want %v, got %v", tc.op, tc.in, tc.want, got) + } + } +} diff --git a/internal/dsp/fft.go b/internal/dsp/fft.go new file mode 100644 index 0000000..f0647f6 --- /dev/null +++ b/internal/dsp/fft.go @@ -0,0 +1,58 @@ +package dsp + +import "math" + +// nextPow2 returns the smallest power of two >= n (and at least 1). +func nextPow2(n int) int { + p := 1 + for p < n { + p <<= 1 + } + return p +} + +// fftRadix2 computes the in-place iterative radix-2 Cooley-Tukey FFT of the +// complex signal held in re/im. len(re) == len(im) must be a power of two. The +// transform overwrites re/im with the frequency-domain result. This is a small +// self-contained implementation (no external dependency) used by the synthetic +// fft op. +func fftRadix2(re, im []float64) { + n := len(re) + if n <= 1 { + return + } + + // Bit-reversal permutation. + for i, j := 1, 0; i < n; i++ { + bit := n >> 1 + for ; j&bit != 0; bit >>= 1 { + j ^= bit + } + j ^= bit + if i < j { + re[i], re[j] = re[j], re[i] + im[i], im[j] = im[j], im[i] + } + } + + // Danielson-Lanczos butterflies. + for length := 2; length <= n; length <<= 1 { + ang := -2 * math.Pi / float64(length) + wReal, wImag := math.Cos(ang), math.Sin(ang) + for i := 0; i < n; i += length { + curReal, curImag := 1.0, 0.0 + half := length >> 1 + for k := 0; k < half; k++ { + a := i + k + b := i + k + half + tReal := curReal*re[b] - curImag*im[b] + tImag := curReal*im[b] + curImag*re[b] + re[b] = re[a] - tReal + im[b] = im[a] - tImag + re[a] += tReal + im[a] += tImag + curReal, curImag = curReal*wReal-curImag*wImag, curReal*wImag+curImag*wReal + } + } + } +} diff --git a/internal/dsp/sample.go b/internal/dsp/sample.go new file mode 100644 index 0000000..3ef86fd --- /dev/null +++ b/internal/dsp/sample.go @@ -0,0 +1,151 @@ +package dsp + +import "fmt" + +// ValType is the data type of a Sample: a scalar float, a float array +// (waveform), or — at graph-compile time, before a source's real type is +// known — unknown. +type ValType uint8 + +const ( + ValUnknown ValType = iota + ValScalar + ValArray +) + +func (t ValType) String() string { + switch t { + case ValScalar: + return "scalar" + case ValArray: + return "array" + default: + return "unknown" + } +} + +// Sample is a value flowing through the synthetic DSP graph: either a scalar +// float64 or a float64 array (waveform). It is the array-aware counterpart of +// the bare float64 the legacy scalar Node interface uses. +type Sample struct { + F float64 + Arr []float64 + IsArray bool +} + +// Scalar wraps a float64 as a scalar Sample. +func Scalar(f float64) Sample { return Sample{F: f} } + +// Array wraps a []float64 as an array Sample. +func Array(a []float64) Sample { return Sample{Arr: a, IsArray: true} } + +// Type reports whether the sample is a scalar or an array. +func (s Sample) Type() ValType { + if s.IsArray { + return ValArray + } + return ValScalar +} + +// AsAny returns the value in the form datasource.Value.Data expects: a +// []float64 for arrays, a float64 for scalars. +func (s Sample) AsAny() any { + if s.IsArray { + return s.Arr + } + return s.F +} + +// AsArray returns the sample's data as a slice: the array itself, or a +// single-element slice for a scalar. Used by reductions that accept either. +func (s Sample) AsArray() []float64 { + if s.IsArray { + return s.Arr + } + return []float64{s.F} +} + +// ArrayNode is an optional extension of Node implemented by ops that operate +// natively on Samples (reductions array→scalar, producers array→array, etc.). +// eval prefers ProcessSample when a node implements it. +type ArrayNode interface { + Node + ProcessSample(inputs []Sample, state map[string]any) (Sample, error) +} + +// statelessElementwise lists scalar ops that are safe to broadcast element-wise +// over array inputs: they hold no per-evaluation state, so running the legacy +// Process once per array lane is well-defined. Stateful ops (moving_average, +// rms, lowpass, derivative, integrate) and lua are excluded — a single shared +// state map cannot be meaningfully split across lanes. +var statelessElementwise = map[string]bool{ + "gain": true, "offset": true, "add": true, "subtract": true, + "multiply": true, "divide": true, "clamp": true, "threshold": true, + "expr": true, +} + +// StatelessElementwise reports whether a scalar op type may be broadcast over +// array inputs via ApplyElementwise. +func StatelessElementwise(nodeType string) bool { return statelessElementwise[nodeType] } + +// scalarInputs wraps a legacy float64 input slice as scalar Samples. +func scalarInputs(in []float64) []Sample { + out := make([]Sample, len(in)) + for i, v := range in { + out[i] = Scalar(v) + } + return out +} + +// ApplyElementwise runs a stateless scalar Node over Sample inputs. If every +// input is scalar it calls Process once and wraps the result. If any input is +// an array it broadcasts: scalar inputs act as constants, all array inputs must +// share a common length (else an error), and Process is invoked once per index. +// +// The node MUST be stateless (see StatelessElementwise) — a shared state map +// cannot be split across array lanes. +func ApplyElementwise(n Node, inputs []Sample, state map[string]any) (Sample, error) { + // Determine the array length, if any input is an array. + length := -1 + for _, s := range inputs { + if !s.IsArray { + continue + } + if length == -1 { + length = len(s.Arr) + } else if len(s.Arr) != length { + return Sample{}, fmt.Errorf("%s: array length mismatch (%d vs %d)", n.Type(), length, len(s.Arr)) + } + } + + if length == -1 { + // All scalar — single legacy call. + row := make([]float64, len(inputs)) + for i, s := range inputs { + row[i] = s.F + } + r, err := n.Process(row, state) + if err != nil { + return Sample{}, err + } + return Scalar(r), nil + } + + out := make([]float64, length) + row := make([]float64, len(inputs)) + for i := 0; i < length; i++ { + for j, s := range inputs { + if s.IsArray { + row[j] = s.Arr[i] + } else { + row[j] = s.F + } + } + r, err := n.Process(row, state) + if err != nil { + return Sample{}, err + } + out[i] = r + } + return Array(out), nil +} diff --git a/internal/dsp/types.go b/internal/dsp/types.go new file mode 100644 index 0000000..6b42305 --- /dev/null +++ b/internal/dsp/types.go @@ -0,0 +1,74 @@ +package dsp + +import "fmt" + +// Op type categories for static (compile-time / editor) type propagation. +// These mirror the runtime dispatch in the synthetic graph evaluator and the +// frontend's inferNodeTypes (web/src/lib/synthTypes.ts) — keep the three in +// sync; a parity test guards the Go/TS pair. +var ( + // reductionOps collapse an array (or scalar) to a single scalar. + reductionOps = map[string]bool{ + "index": true, "length": true, "sum": true, + "mean": true, "min": true, "max": true, + } + // arrayProducerOps require an array input and yield an array. + arrayProducerOps = map[string]bool{ + "fft": true, "slice": true, + } + // scalarOnlyOps reject array inputs and yield a scalar. Stateful filters + // plus lua (whose state/closure cannot be broadcast per array lane). + scalarOnlyOps = map[string]bool{ + "moving_average": true, "rms": true, "lowpass": true, + "derivative": true, "integrate": true, "lua": true, + } +) + +// OpOutputType reports the output ValType of an op given its input types, and +// an error if the inputs are definitely incompatible with the op. Inputs may be +// ValUnknown (a source whose real type is not yet known at compile time); such +// inputs never trigger an error — runtime Sample typing is authoritative. +func OpOutputType(op string, in []ValType) (ValType, error) { + switch { + case reductionOps[op]: + return ValScalar, nil + + case arrayProducerOps[op]: + for _, t := range in { + if t == ValScalar { + return ValUnknown, fmt.Errorf("%s requires an array input", op) + } + } + return ValArray, nil + + case scalarOnlyOps[op]: + for _, t := range in { + if t == ValArray { + return ValUnknown, fmt.Errorf("%s does not accept an array input", op) + } + } + return ValScalar, nil + + default: + // Elementwise stateless ops (gain, offset, add, subtract, multiply, + // divide, clamp, threshold, expr): array if any input is an array, + // scalar if all inputs are definitely scalar, otherwise unknown. + anyArray, anyUnknown := false, false + for _, t := range in { + switch t { + case ValArray: + anyArray = true + case ValUnknown: + anyUnknown = true + } + } + switch { + case anyArray: + return ValArray, nil + case anyUnknown: + return ValUnknown, nil + default: + return ValScalar, nil + } + } +} diff --git a/web/src/Canvas.tsx b/web/src/Canvas.tsx index 47933cd..cab2497 100644 --- a/web/src/Canvas.tsx +++ b/web/src/Canvas.tsx @@ -155,7 +155,7 @@ export default function Canvas({ iface, onNavigate, timeRange }: Props) { return (
-
+
{iface.widgets.map(widget => { const Comp = COMPONENTS[widget.type]; const inner = Comp diff --git a/web/src/PropertiesPane.tsx b/web/src/PropertiesPane.tsx index 33e0683..e027b19 100644 --- a/web/src/PropertiesPane.tsx +++ b/web/src/PropertiesPane.tsx @@ -302,6 +302,7 @@ export default function PropertiesPane({ selected, multiCount, iface, onChange, + {(w.options['plotType'] ?? 'timeseries') === 'timeseries' && ( diff --git a/web/src/SyntheticGraphEditor.tsx b/web/src/SyntheticGraphEditor.tsx index 411527b..c76dc90 100644 --- a/web/src/SyntheticGraphEditor.tsx +++ b/web/src/SyntheticGraphEditor.tsx @@ -3,9 +3,10 @@ import { useState, useEffect, useRef } from 'preact/hooks'; import SignalPicker, { SignalOption } from './SignalPicker'; import LuaEditor from './LuaEditor'; import type { SignalDef, SynthGraph, SynthGraphNode } from './lib/types'; +import { inferNodeTypes, SynthType } from './lib/synthTypes'; interface DataSource { name: string; } -interface SignalInfo { name: string; } +interface SignalInfo { name: string; type?: string; } interface Props { // Existing signal name (edit mode). Omitted/empty in create mode. @@ -52,6 +53,18 @@ const OPS: OpDef[] = [ ]}, { type: 'expr', label: 'Formula', arity: { kind: 'named' }, params: [{ label: 'Expression', key: 'expr', type: 'text', default: 'a + b' }] }, { type: 'lua', label: 'Lua Script', arity: { kind: 'named' }, params: [{ label: 'Script', key: 'script', type: 'lua', default: 'return a + b' }] }, + // ── Array (waveform) ops ── + { type: 'index', label: 'Index (a[i])', arity: { kind: 'fixed', n: 1 }, params: [{ label: 'Index (i)', key: 'i', type: 'number', default: '0' }] }, + { type: 'slice', label: 'Slice', arity: { kind: 'fixed', n: 1 }, params: [ + { label: 'Start', key: 'start', type: 'number', default: '0' }, + { label: 'End (0 = to end)', key: 'end', type: 'number', default: '0' }, + ]}, + { type: 'sum', label: 'Sum (Σ array)', arity: { kind: 'fixed', n: 1 }, params: [] }, + { type: 'mean', label: 'Mean (array)', arity: { kind: 'fixed', n: 1 }, params: [] }, + { type: 'min', label: 'Min (array)', arity: { kind: 'fixed', n: 1 }, params: [] }, + { type: 'max', label: 'Max (array)', arity: { kind: 'fixed', n: 1 }, params: [] }, + { type: 'length', label: 'Length', arity: { kind: 'fixed', n: 1 }, params: [] }, + { type: 'fft', label: 'FFT (magnitude)', arity: { kind: 'fixed', n: 1 }, params: [] }, ]; const OP_BY_TYPE = new Map(OPS.map(o => [o.type, o])); function opLabel(type: string): string { return OP_BY_TYPE.get(type)?.label ?? type; } @@ -297,6 +310,9 @@ export default function SyntheticGraphEditor({ name, create, panelId, onClose, o const [loading, setLoading] = useState(true); const [saving, setSaving] = useState(false); const [error, setError] = useState(''); + // Quick-add HUD (S = signal, N = node); null when closed. + const [hud, setHud] = useState(null); + const [hudFilter, setHudFilter] = useState(''); // Identity fields — editable only when creating a new signal. const [newName, setNewName] = useState(''); @@ -308,7 +324,7 @@ export default function SyntheticGraphEditor({ name, create, panelId, onClose, o const sigName = create ? newName.trim() : (name ?? ''); const [dataSources, setDataSources] = useState([]); - const [dsSignals, setDsSignals] = useState>({}); + const [dsSignals, setDsSignals] = useState>({}); const canvasRef = useRef(null); const dragNode = useRef<{ id: string; dx: number; dy: number; pushed: boolean } | null>(null); @@ -338,7 +354,7 @@ export default function SyntheticGraphEditor({ name, create, panelId, onClose, o const res = await fetch(`/api/v1/signals?ds=${encodeURIComponent(ds)}`); if (!res.ok) return; const sigs: SignalInfo[] = await res.json(); - setDsSignals(prev => ({ ...prev, [ds]: sigs.map(s => s.name) })); + setDsSignals(prev => ({ ...prev, [ds]: sigs })); } catch {} } @@ -346,11 +362,36 @@ export default function SyntheticGraphEditor({ name, create, panelId, onClose, o // and a hook to lazily load all data sources' signals when a picker opens. function allSignalOptions(): SignalOption[] { const out: SignalOption[] = []; - for (const ds of dataSources) for (const name of (dsSignals[ds] ?? [])) out.push({ ds, name }); + for (const ds of dataSources) for (const s of (dsSignals[ds] ?? [])) out.push({ ds, name: s.name }); return out; } function openAllSignals() { dataSources.forEach(ds => loadSignals(ds)); } + // Resolve a source node's data type from the upstream signal's metadata. + // 'float64[]' is the only array type today; anything else is scalar, and an + // unresolved/unloaded source is 'unknown' (no error, runtime typing wins). + function sourceSynthType(n: SynthGraphNode): SynthType { + if (!n.ds || !n.signal) return 'unknown'; + const sigs = dsSignals[n.ds]; + if (!sigs) return 'unknown'; + const info = sigs.find(s => s.name === n.signal); + if (!info || !info.type) return 'unknown'; + return info.type === 'float64[]' ? 'array' : 'scalar'; + } + + // HUD filtering: case-insensitive substring over signal name/ds and op label/type. + function filteredSignalOptions(): SignalOption[] { + const q = hudFilter.trim().toLowerCase(); + const all = allSignalOptions(); + if (!q) return all; + return all.filter(o => o.name.toLowerCase().includes(q) || o.ds.toLowerCase().includes(q)); + } + function filteredOps(): OpDef[] { + const q = hudFilter.trim().toLowerCase(); + if (!q) return OPS; + return OPS.filter(o => o.label.toLowerCase().includes(q) || o.type.toLowerCase().includes(q)); + } + useEffect(() => { if (create) { const blank: SignalDef = { name: '', inputs: [], pipeline: [], meta: {} }; @@ -408,6 +449,20 @@ export default function SyntheticGraphEditor({ name, create, panelId, onClose, o commit({ nodes: [...graph.nodes, node], wires: graph.wires }); setSelected(node.id); setSelectedWire(null); } + // Add a source node pre-filled with a chosen ds:signal (used by the quick-add HUD). + function addSourceWith(ds: string, signal: string) { + const node: GNode = { id: genId(), kind: 'source', x: 2 * REM, y: (2 + graph.nodes.filter(n => n.kind === 'source').length * 5) * REM, ds, signal }; + if (ds) loadSignals(ds); + commit({ nodes: [...graph.nodes, node], wires: graph.wires }); + setSelected(node.id); setSelectedWire(null); + } + // Open the quick-add HUD; for signals, lazily load every source's signal list. + function openHud(kind: 'signal' | 'node') { + if (kind === 'signal') openAllSignals(); + setHudFilter(''); + setHud(kind); + } + function closeHud() { setHud(null); setHudFilter(''); } function addOp(op: OpDef, x?: number, y?: number) { const params: Record = {}; for (const p of op.params) params[p.key] = p.type === 'number' ? parseFloat(p.default) : p.default; @@ -582,22 +637,34 @@ export default function SyntheticGraphEditor({ name, create, panelId, onClose, o const mod = e.ctrlKey || e.metaKey; if (mod && e.key.toLowerCase() === 'z' && !e.shiftKey) { e.preventDefault(); undo(); return; } if (mod && (e.key.toLowerCase() === 'y' || (e.key.toLowerCase() === 'z' && e.shiftKey))) { e.preventDefault(); redo(); return; } + if (mod) return; + if (e.key === 'Escape') { if (hud) closeHud(); return; } + if (e.key.toLowerCase() === 's') { e.preventDefault(); openHud('signal'); return; } + if (e.key.toLowerCase() === 'n') { e.preventDefault(); openHud('node'); return; } if (e.key !== 'Delete' && e.key !== 'Backspace') return; if (selectedWire !== null) deleteWire(selectedWire); else if (selected) deleteNode(selected); } window.addEventListener('keydown', onKey); return () => window.removeEventListener('keydown', onKey); - }, [selected, selectedWire, graph]); + }, [selected, selectedWire, graph, hud]); const byId = new Map(graph.nodes.map(n => [n.id, n])); const sel = graph.nodes.find(n => n.id === selected) ?? null; const { errors: nodeErrors, first: validationError } = validate(graph); + // Static type propagation: infer each node's output type (scalar/array) to + // colour wires and surface type-incompatible wirings as node errors. Mirrors + // the backend dsp.OpOutputType rules (see lib/synthTypes.ts). + const { types: nodeTypes, errors: typeErrors } = inferNodeTypes(compile(graph), sourceSynthType); + for (const [id, msg] of typeErrors) if (!nodeErrors.has(id)) nodeErrors.set(id, msg); + const typeError = !validationError ? [...typeErrors.values()][0] : undefined; + const firstError = validationError ?? typeError; + async function handleSave() { if (!def) return; if (create && !sigName) { setError('Enter a name for the new signal.'); return; } - if (validationError) { setError(validationError); return; } + if (firstError) { setError(firstError); return; } setSaving(true); setError(''); try { @@ -692,9 +759,10 @@ export default function SyntheticGraphEditor({ name, create, panelId, onClose, o const a = byId.get(w.from); const b = byId.get(w.to); if (!a || !b) return null; const p1 = outAnchor(a); const p2 = inAnchor(b, w.toPort); + const tClass = nodeTypes.get(w.from) === 'array' ? ' synth-wire-array' : ' synth-wire-scalar'; return ( { e.stopPropagation(); setSelectedWire(idx); setSelected(null); }} /> ); @@ -712,6 +780,7 @@ export default function SyntheticGraphEditor({ name, create, panelId, onClose, o return (
startNodeDrag(e, node)} onMouseUp={() => { if (pendingRef.current) finishWire(node, firstFreePort(node)); }}> @@ -867,11 +936,57 @@ export default function SyntheticGraphEditor({ name, create, panelId, onClose, o
)} + {hud && ( +
+
e.stopPropagation()}> + setHudFilter((e.target as HTMLInputElement).value)} + onKeyDown={(e) => { + if (e.key === 'Escape') { e.preventDefault(); closeHud(); return; } + if (e.key === 'Enter') { + e.preventDefault(); + if (hud === 'signal') { + const opt = filteredSignalOptions()[0]; + if (opt) { addSourceWith(opt.ds, opt.name); closeHud(); } + } else { + const op = filteredOps()[0]; + if (op) { addOp(op); closeHud(); } + } + } + }} /> +
+ {hud === 'signal' + ? filteredSignalOptions().slice(0, 50).map(opt => ( + + )) + : filteredOps().map(op => ( + + ))} + {hud === 'signal' && filteredSignalOptions().length === 0 && ( +
No matching signals.
+ )} +
+
+
+ )} +
{error && {error}} - {!error && validationError && {validationError}} + {!error && firstError && {firstError}} -
diff --git a/web/src/lib/synthTypes.ts b/web/src/lib/synthTypes.ts new file mode 100644 index 0000000..3c815e6 --- /dev/null +++ b/web/src/lib/synthTypes.ts @@ -0,0 +1,86 @@ +// Static type propagation for the synthetic node-graph editor. This mirrors the +// Go runtime/compile rules in internal/dsp/types.go (OpOutputType) so the editor +// can colour wires by data type and flag type-incompatible wirings before save. +// Keep the two in sync — a parity test guards the pair. + +import type { SynthGraph, SynthGraphNode } from './types'; + +export type SynthType = 'scalar' | 'array' | 'unknown'; + +// reductionOps collapse an array (or scalar) to a single scalar. +const REDUCTION_OPS = new Set(['index', 'length', 'sum', 'mean', 'min', 'max']); +// arrayProducerOps require an array input and yield an array. +const ARRAY_PRODUCER_OPS = new Set(['fft', 'slice']); +// scalarOnlyOps reject array inputs and yield a scalar (stateful filters + lua). +const SCALAR_ONLY_OPS = new Set(['moving_average', 'rms', 'lowpass', 'derivative', 'integrate', 'lua']); + +// opOutputType reports an op's output type given its input types, plus an error +// message when the inputs are definitely incompatible with the op. 'unknown' +// inputs (a source whose real type isn't known yet) never trigger an error — +// runtime typing is authoritative. +export function opOutputType(op: string, inputs: SynthType[]): { type: SynthType; error?: string } { + if (REDUCTION_OPS.has(op)) return { type: 'scalar' }; + + if (ARRAY_PRODUCER_OPS.has(op)) { + if (inputs.some(t => t === 'scalar')) return { type: 'unknown', error: `${op} requires an array input` }; + return { type: 'array' }; + } + + if (SCALAR_ONLY_OPS.has(op)) { + if (inputs.some(t => t === 'array')) return { type: 'unknown', error: `${op} does not accept an array input` }; + return { type: 'scalar' }; + } + + // Elementwise stateless ops (gain, offset, add, subtract, multiply, divide, + // clamp, threshold, expr): array if any input is an array, scalar if all + // inputs are definitely scalar, otherwise unknown. + if (inputs.some(t => t === 'array')) return { type: 'array' }; + if (inputs.some(t => t === 'unknown')) return { type: 'unknown' }; + return { type: 'scalar' }; +} + +// inferNodeTypes walks the DAG in dependency order and assigns each node an +// output type. Source node types come from `sourceType` (resolved from upstream +// signal metadata); unresolved sources default to 'unknown'. `errors` collects +// per-node type-conflict messages for the editor's validation. +export function inferNodeTypes( + g: SynthGraph, + sourceType: (n: SynthGraphNode) => SynthType, +): { types: Map; errors: Map } { + const types = new Map(); + const errors = new Map(); + const byId = new Map(g.nodes.map(n => [n.id, n])); + + // Kahn topological order over node inputs. + const indeg = new Map(); + const succ = new Map(); + for (const n of g.nodes) indeg.set(n.id, 0); + for (const n of g.nodes) { + for (const from of n.inputs ?? []) { + if (!byId.has(from)) continue; + indeg.set(n.id, (indeg.get(n.id) ?? 0) + 1); + succ.set(from, [...(succ.get(from) ?? []), n.id]); + } + } + const queue = g.nodes.filter(n => (indeg.get(n.id) ?? 0) === 0).map(n => n.id); + while (queue.length) { + const id = queue.shift()!; + const n = byId.get(id)!; + if (n.kind === 'source') { + types.set(id, sourceType(n)); + } else if (n.kind === 'output') { + const from = (n.inputs ?? [])[0]; + types.set(id, (from && types.get(from)) || 'unknown'); + } else { + const inTypes = (n.inputs ?? []).map(f => types.get(f) ?? 'unknown'); + const { type, error } = opOutputType(n.op ?? '', inTypes); + types.set(id, type); + if (error) errors.set(id, error); + } + for (const s of succ.get(id) ?? []) { + indeg.set(s, (indeg.get(s) ?? 0) - 1); + if ((indeg.get(s) ?? 0) === 0) queue.push(s); + } + } + return { types, errors }; +} diff --git a/web/src/styles.css b/web/src/styles.css index 3c5a3fb..93b954f 100644 --- a/web/src/styles.css +++ b/web/src/styles.css @@ -375,6 +375,23 @@ body { /* width/height set inline from interface dimensions */ } +/* View mode: widgets shed their card chrome (border/background/shadow) so they + blend with the canvas background. Edit mode (EditCanvas) keeps the chrome so + widgets stay visible and selectable. Internal elements (gauge arc, bar track, + plot chart, inputs) keep their own styling. */ +.canvas-view-bare .textview, +.canvas-view-bare .gauge, +.canvas-view-bare .barh, +.canvas-view-bare .barv, +.canvas-view-bare .led-widget, +.canvas-view-bare .multiled-widget, +.canvas-view-bare .setvalue, +.canvas-view-bare .plot-widget { + background: transparent; + border-color: transparent; + box-shadow: none; +} + /* ── TextLabel widget ──────────────────────────────────────────────────────── */ .textlabel { @@ -1248,6 +1265,12 @@ body { .flow-wire-selected { stroke: #ef4444; stroke-width: 2.5; } .flow-wire-pending { stroke: #3b82f6; stroke-dasharray: 5 4; } +/* Synthetic-editor data-type wire colouring (scoped to .synth-graph so the + Logic / ControlLogic editors that share .flow-wire are unaffected). Scalar + keeps the default slate; array (waveform) signals are purple. */ +.synth-graph .flow-wire.synth-wire-array { stroke: #a855f7; } +.synth-graph .flow-wire.synth-wire-array:hover { stroke: #c084fc; } + /* Node block */ .flow-node { position: absolute; @@ -2002,8 +2025,64 @@ body { display: flex; flex-direction: column; box-shadow: 0 8px 32px rgba(0,0,0,0.5); + position: relative; } +/* Quick-add HUD (S = signal, N = node) */ +.synth-hud-backdrop { + position: absolute; + inset: 0; + background: rgba(10, 14, 22, 0.45); + display: flex; + align-items: flex-start; + justify-content: center; + padding-top: 12vh; + z-index: 20; +} +.synth-hud { + width: 26rem; + max-width: 80%; + background: #1a1f2e; + border: 1px solid #3a4660; + border-radius: 8px; + box-shadow: 0 12px 40px rgba(0,0,0,0.55); + display: flex; + flex-direction: column; + overflow: hidden; +} +.synth-hud-input { + border: none; + border-bottom: 1px solid #2d3748; + background: #0f1117; + color: #e2e8f0; + font-size: 0.9rem; + padding: 0.6rem 0.75rem; + outline: none; +} +.synth-hud-list { + max-height: 18rem; + overflow-y: auto; + display: flex; + flex-direction: column; +} +.synth-hud-item { + display: flex; + align-items: baseline; + justify-content: space-between; + gap: 0.5rem; + padding: 0.4rem 0.75rem; + background: none; + border: none; + color: #e2e8f0; + font-size: 0.85rem; + text-align: left; + cursor: pointer; +} +.synth-hud-item:hover { background: #243049; } +.synth-hud-item-name { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } +.synth-hud-item-meta { color: #64748b; font-size: 0.72rem; flex-shrink: 0; } +.synth-hud-empty { padding: 0.6rem 0.75rem; } + .wizard-header { display: flex; align-items: center; diff --git a/web/src/widgets/PlotWidget.tsx b/web/src/widgets/PlotWidget.tsx index 95a8814..5c908c1 100644 --- a/web/src/widgets/PlotWidget.tsx +++ b/web/src/widgets/PlotWidget.tsx @@ -84,6 +84,8 @@ export default function PlotWidget({ widget, onContextMenu, timeRange }: Props) const showLegend = legendPos !== 'none'; const buffers: SeriesBuffer[] = signals.map(() => ({ timestamps: [], values: [] })); + // Latest waveform (array) value per signal — used only by plotType 'waveform'. + const waveforms: number[][] = signals.map(() => []); const unsubs: (() => void)[] = []; const fmt = widget.options['format'] ?? ''; @@ -269,6 +271,24 @@ export default function PlotWidget({ widget, onContextMenu, timeRange }: Props) }), }; } + case 'waveform': { + // Latest waveform (array) sample per signal, drawn as an x-vs-index + // trace. Each update replaces the trace rather than appending. + return { + backgroundColor: ECHARTS_DARK.backgroundColor, + legend: legendOpt, + grid: { left: 60, right: 16, top: showLegend ? 28 : 12, bottom: 40 }, + xAxis: { type: 'value', name: 'Index', nameLocation: 'middle', nameGap: 24, min: 0, axisLine, axisLabel, splitLine }, + yAxis: { type: 'value', name: 'Value', axisLine, axisLabel, splitLine }, + series: signals.map((s, i) => ({ + name: s.name, + type: 'line', + data: (waveforms[i] ?? []).map((v, k) => [k, v]), + lineStyle: { color: s.color ?? COLORS[i % COLORS.length] }, + showSymbol: false, + })), + }; + } case 'logic': { // Logic-analyser style: each signal is a 0/1 step trace stacked on its // own lane, with a relative-time x-axis (seconds before now). @@ -423,7 +443,16 @@ export default function PlotWidget({ widget, onContextMenu, timeRange }: Props) if (sv.value === null || sv.value === undefined || sv.ts === null) return; const ts = new Date(sv.ts).getTime() / 1000; - if (plotType === 'timeseries') { + if (plotType === 'waveform') { + // Array-valued sample: keep only the latest waveform and redraw. + if (Array.isArray(sv.value)) { + waveforms[i] = sv.value.map((x: any) => typeof x === 'number' ? x : parseFloat(String(x))); + } else { + const v = typeof sv.value === 'number' ? sv.value : parseFloat(String(sv.value)); + waveforms[i] = isNaN(v) ? [] : [v]; + } + if (echart) echart.setOption(echartsOption(), { notMerge: true }); + } else if (plotType === 'timeseries') { const v = typeof sv.value === 'number' ? sv.value : parseFloat(String(sv.value)); if (isNaN(v)) return; pushSample(buffers[i], ts, v);