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 <noreply@anthropic.com>
This commit is contained in:
Martino Ferrari
2026-06-20 17:06:55 +02:00
parent 446de7f1ee
commit f7f297c3df
18 changed files with 1470 additions and 57 deletions
@@ -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)
}
}
}
@@ -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)
}
+75 -11
View File
@@ -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
}
+50 -12
View File
@@ -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
}
+230
View File
@@ -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
}
+213
View File
@@ -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)
}
}
}
+58
View File
@@ -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
}
}
}
}
+151
View File
@@ -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
}
+74
View File
@@ -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
}
}
}