Phase 6

2026-04-25 22:56:09 +02:00
parent 8b548ba1c2
commit 986f6cd6d8
85 changed files with 11479 additions and 5050 deletions
@@ -0,0 +1,451 @@
+package synthetic
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/uopi/uopi/internal/broker"
+	"github.com/uopi/uopi/internal/datasource"
+	"github.com/uopi/uopi/internal/dsp"
+)
+
+const definitionsFile = "synthetic.json"
+
+// signalState holds everything needed to run one synthetic signal.
+type signalState struct {
+	def    SignalDef
+	nodes  []dsp.Node
+	states []map[string]any // one map per node, persistent across calls
+
+	// cancel stops the goroutine driving this signal.
+	cancel context.CancelFunc
+}
+
+// Synthetic is a DataSource that computes values from upstream signals via
+// configurable DSP pipelines.
+type Synthetic struct {
+	brk       *broker.Broker
+	storePath string
+	log       *slog.Logger
+
+	mu      sync.RWMutex
+	signals map[string]*signalState
+
+	// root context provided by Connect; used to start per-signal goroutines.
+	ctx context.Context
+}
+
+// New creates a Synthetic data source. storePath is the directory where
+// synthetic.json is stored. brk is used to subscribe to upstream signals.
+func New(storePath string, brk *broker.Broker, log *slog.Logger) *Synthetic {
+	return &Synthetic{
+		brk:       brk,
+		storePath: storePath,
+		log:       log,
+		signals:   make(map[string]*signalState),
+	}
+}
+
+// Name implements datasource.DataSource.
+func (s *Synthetic) Name() string { return "synthetic" }
+
+// Connect loads the definitions file and starts all signal goroutines.
+func (s *Synthetic) Connect(ctx context.Context) error {
+	s.ctx = ctx
+
+	defs, err := s.loadDefs()
+	if err != nil {
+		return fmt.Errorf("synthetic: load definitions: %w", err)
+	}
+
+	for _, def := range defs {
+		if err := s.startSignal(def); err != nil {
+			s.log.Warn("synthetic: failed to start signal", "name", def.Name, "err", err)
+		}
+	}
+	return nil
+}
+
+// ListSignals returns metadata for all defined synthetic signals.
+func (s *Synthetic) ListSignals(_ context.Context) ([]datasource.Metadata, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	out := make([]datasource.Metadata, 0, len(s.signals))
+	for _, st := range s.signals {
+		out = append(out, defToMetadata(st.def))
+	}
+	return out, nil
+}
+
+// GetMetadata returns metadata for a single named signal.
+func (s *Synthetic) GetMetadata(_ context.Context, signal string) (datasource.Metadata, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	st, ok := s.signals[signal]
+	if !ok {
+		return datasource.Metadata{}, datasource.ErrNotFound
+	}
+	return defToMetadata(st.def), nil
+}
+
+// Subscribe registers ch to receive computed values for the named signal.
+// Values are pushed whenever an upstream signal changes.
+func (s *Synthetic) Subscribe(ctx context.Context, signal string, ch chan<- datasource.Value) (datasource.CancelFunc, error) {
+	s.mu.RLock()
+	st, ok := s.signals[signal]
+	s.mu.RUnlock()
+	if !ok {
+		return nil, datasource.ErrNotFound
+	}
+
+	// Collect the upstream references for this signal.
+	refs := upstreamRefs(st.def)
+	if len(refs) == 0 {
+		return nil, fmt.Errorf("synthetic: signal %q has no upstream inputs", signal)
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+
+	go func() {
+		defer cancel()
+
+		// Latest value per upstream input index.
+		latest := make([]float64, len(refs))
+		ready := make([]bool, len(refs))
+
+		// Subscribe to every upstream ref via the broker.
+		updateCh := make(chan indexedUpdate, 32*len(refs))
+
+		unsubs := make([]func(), 0, len(refs))
+		for i, ref := range refs {
+			idx := i // capture
+			perCh := make(chan broker.Update, 16)
+			unsub, err := s.brk.Subscribe(ref, perCh)
+			if err != nil {
+				s.log.Warn("synthetic: upstream subscribe failed",
+					"signal", signal, "upstream", ref, "err", err)
+				// Continue; the slot will stay at zero.
+			} else {
+				unsubs = append(unsubs, unsub)
+				go func() {
+					for {
+						select {
+						case u, ok := <-perCh:
+							if !ok {
+								return
+							}
+							val := toFloat64(u.Value.Data)
+							select {
+							case updateCh <- indexedUpdate{idx: idx, val: val, ts: u.Value.Timestamp}:
+							default:
+							}
+						case <-ctx.Done():
+							return
+						}
+					}
+				}()
+			}
+		}
+
+		defer func() {
+			for _, unsub := range unsubs {
+				unsub()
+			}
+		}()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+
+			case upd := <-updateCh:
+				latest[upd.idx] = upd.val
+				ready[upd.idx] = true
+
+				// Only compute once we have at least one value for every input.
+				allReady := true
+				for _, r := range ready {
+					if !r {
+						allReady = false
+						break
+					}
+				}
+				if !allReady {
+					continue
+				}
+
+				// Run the pipeline.
+				s.mu.RLock()
+				cur, stillExists := s.signals[signal]
+				s.mu.RUnlock()
+				if !stillExists {
+					return
+				}
+
+				result, err := runPipeline(cur.nodes, cur.states, latest)
+				if err != nil {
+					s.log.Warn("synthetic: pipeline error", "signal", signal, "err", err)
+					continue
+				}
+
+				v := datasource.Value{
+					Timestamp: upd.ts,
+					Data:      result,
+					Quality:   datasource.QualityGood,
+				}
+				select {
+				case ch <- v:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}()
+
+	return datasource.CancelFunc(cancel), nil
+}
+
+// Write is not supported for synthetic signals.
+func (s *Synthetic) Write(_ context.Context, _ string, _ any) error {
+	return datasource.ErrNotWritable
+}
+
+// History is not supported for synthetic signals.
+func (s *Synthetic) History(_ context.Context, _ string, _, _ time.Time, _ int) ([]datasource.Value, error) {
+	return nil, datasource.ErrHistoryUnavailable
+}
+
+// AddSignal adds a new synthetic signal definition at runtime and persists it.
+func (s *Synthetic) AddSignal(def SignalDef) error {
+	if def.Name == "" {
+		return errors.New("signal name must not be empty")
+	}
+
+	nodes, err := BuildPipeline(def.Pipeline)
+	if err != nil {
+		return fmt.Errorf("build pipeline: %w", err)
+	}
+
+	s.mu.Lock()
+	if _, exists := s.signals[def.Name]; exists {
+		s.mu.Unlock()
+		return fmt.Errorf("signal %q already exists", def.Name)
+	}
+
+	states := make([]map[string]any, len(nodes))
+	for i := range states {
+		states[i] = make(map[string]any)
+	}
+
+	st := &signalState{
+		def:    def,
+		nodes:  nodes,
+		states: states,
+	}
+	s.signals[def.Name] = st
+	s.mu.Unlock()
+
+	if err := s.saveDefs(); err != nil {
+		// Roll back the in-memory addition.
+		s.mu.Lock()
+		delete(s.signals, def.Name)
+		s.mu.Unlock()
+		return fmt.Errorf("persist definitions: %w", err)
+	}
+
+	return nil
+}
+
+// RemoveSignal removes a synthetic signal definition at runtime and persists.
+func (s *Synthetic) RemoveSignal(name string) error {
+	s.mu.Lock()
+	st, ok := s.signals[name]
+	if !ok {
+		s.mu.Unlock()
+		return datasource.ErrNotFound
+	}
+	// Stop the goroutine if it is running.
+	if st.cancel != nil {
+		st.cancel()
+	}
+	delete(s.signals, name)
+	s.mu.Unlock()
+
+	if err := s.saveDefs(); err != nil {
+		return fmt.Errorf("persist definitions: %w", err)
+	}
+	return nil
+}
+
+// GetDefs returns a copy of all current signal definitions (for the REST API).
+func (s *Synthetic) GetDefs() []SignalDef {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	out := make([]SignalDef, 0, len(s.signals))
+	for _, st := range s.signals {
+		out = append(out, st.def)
+	}
+	return out
+}
+
+// ── internal helpers ──────────────────────────────────────────────────────────
+
+func (s *Synthetic) defsFilePath() string {
+	return filepath.Join(s.storePath, definitionsFile)
+}
+
+func (s *Synthetic) loadDefs() ([]SignalDef, error) {
+	path := s.defsFilePath()
+	data, err := os.ReadFile(path)
+	if errors.Is(err, os.ErrNotExist) {
+		// Create an empty definitions file.
+		if werr := os.WriteFile(path, []byte("[]\n"), 0o644); werr != nil {
+			s.log.Warn("synthetic: could not create empty definitions file", "path", path, "err", werr)
+		}
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	var defs []SignalDef
+	if err := json.Unmarshal(data, &defs); err != nil {
+		return nil, fmt.Errorf("parse %s: %w", path, err)
+	}
+	return defs, nil
+}
+
+func (s *Synthetic) saveDefs() error {
+	s.mu.RLock()
+	defs := make([]SignalDef, 0, len(s.signals))
+	for _, st := range s.signals {
+		defs = append(defs, st.def)
+	}
+	s.mu.RUnlock()
+
+	data, err := json.MarshalIndent(defs, "", "  ")
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(s.defsFilePath(), data, 0o644)
+}
+
+// startSignal builds the pipeline for def and registers the signalState.
+// The actual goroutines are started lazily by Subscribe.
+func (s *Synthetic) startSignal(def SignalDef) error {
+	nodes, err := BuildPipeline(def.Pipeline)
+	if err != nil {
+		return fmt.Errorf("build pipeline for %q: %w", def.Name, err)
+	}
+
+	states := make([]map[string]any, len(nodes))
+	for i := range states {
+		states[i] = make(map[string]any)
+	}
+
+	s.mu.Lock()
+	s.signals[def.Name] = &signalState{
+		def:    def,
+		nodes:  nodes,
+		states: states,
+	}
+	s.mu.Unlock()
+
+	s.log.Info("synthetic: signal registered", "name", def.Name)
+	return nil
+}
+
+// runPipeline executes all nodes in sequence. The output of node N becomes
+// input[0] of node N+1. For the first node, inputs is the full upstream slice.
+func runPipeline(nodes []dsp.Node, states []map[string]any, inputs []float64) (float64, error) {
+	if len(nodes) == 0 {
+		if len(inputs) == 0 {
+			return 0, nil
+		}
+		return inputs[0], nil
+	}
+
+	// First node receives all upstream inputs.
+	cur := inputs
+	var result float64
+	var err error
+
+	for i, node := range nodes {
+		result, err = node.Process(cur, states[i])
+		if err != nil {
+			return 0, fmt.Errorf("node %d (%s): %w", i, node.Type(), err)
+		}
+		// Subsequent nodes receive only the single output of the previous node.
+		cur = []float64{result}
+	}
+	return result, nil
+}
+
+// upstreamRefs returns the broker.SignalRef list for a SignalDef.
+// If Inputs is set, those take precedence; otherwise DS+Signal is used.
+func upstreamRefs(def SignalDef) []broker.SignalRef {
+	if len(def.Inputs) > 0 {
+		refs := make([]broker.SignalRef, len(def.Inputs))
+		for i, inp := range def.Inputs {
+			refs[i] = broker.SignalRef{DS: inp.DS, Name: inp.Signal}
+		}
+		return refs
+	}
+	if def.DS != "" && def.Signal != "" {
+		return []broker.SignalRef{{DS: def.DS, Name: def.Signal}}
+	}
+	return nil
+}
+
+// defToMetadata converts a SignalDef into a datasource.Metadata.
+func defToMetadata(def SignalDef) datasource.Metadata {
+	return datasource.Metadata{
+		Name:        def.Name,
+		Type:        datasource.TypeFloat64,
+		Unit:        def.Meta.Unit,
+		Description: def.Meta.Description,
+		DisplayLow:  def.Meta.DisplayLow,
+		DisplayHigh: def.Meta.DisplayHigh,
+		Writable:    def.Meta.Writable,
+	}
+}
+
+// toFloat64 coerces any numeric value from a datasource.Value.Data to float64.
+func toFloat64(v any) float64 {
+	switch val := v.(type) {
+	case float64:
+		return val
+	case float32:
+		return float64(val)
+	case int64:
+		return float64(val)
+	case int32:
+		return float64(val)
+	case int:
+		return float64(val)
+	case bool:
+		if val {
+			return 1
+		}
+		return 0
+	default:
+		return 0
+	}
+}
+
+// indexedUpdate carries a value from one upstream goroutine to the pipeline runner.
+type indexedUpdate struct {
+	idx int
+	val float64
+	ts  time.Time
+}