uopi/internal/datasource/synthetic/array_graph_test.go

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)
		}
	}
}