diff --git a/CLAUDE.md b/CLAUDE.md index b30be3c..68387b2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -41,7 +41,7 @@ Two primary modes toggled via toolbar: ## Technology Stack - **Backend:** Go 1.22+, single binary, `//go:embed` for frontend assets -- **Frontend:** Svelte 5 + TypeScript + Vite; plots via uPlot (time-series) and ECharts (others); edit canvas via Konva +- **Frontend:** Preact 10 + TypeScript; bundled by esbuild Go API (no npm/Node.js); plots via uPlot (time-series) and ECharts (others); all vendor JS/CSS checked into `web/vendor/` - **Config:** TOML file + `UOPI_*` env var overrides (`github.com/BurntSushi/toml`) - **WebSocket:** `nhooyr.io/websocket` (no CGo) - **EPICS:** CGo bindings to `libca`; `gopher-lua` for synthetic signal Lua scripts @@ -57,7 +57,10 @@ internal/datasource/# DataSource interface + EPICS + synthetic (Phases 2–3) internal/dsp/ # DSP functions for synthetic (Phase 3) internal/storage/ # interface XML persistence (Phase 6) internal/api/ # REST handlers (Phase 1+) -web/ # Svelte source (npm); web/embed.go provides embed.FS to Go +web/ # Preact/TSX source; web/embed.go provides embed.FS to Go +web/src/ # TypeScript/TSX components (Preact) +web/vendor/ # vendored JS/CSS (preact, uplot, echarts) — no npm required +tools/buildfrontend/# esbuild Go API bundler (invoked by go generate) web/dist/ # built frontend — generated, not committed dist/ # compiled binary — generated, not committed ``` @@ -79,9 +82,6 @@ make frontend # Run backend (dev mode, frontend must be running separately) go run ./cmd/uopi -# Frontend dev server (proxies /api and /healthz to :8080) -cd web && npm run dev - # All tests make test @@ -90,12 +90,6 @@ go test ./internal/broker/... -run TestFanOut # Go vet go vet ./... - -# Svelte type check -cd web && npm run check - -# Svelte lint -cd web && npm run lint ``` ## Key Architectural Notes diff --git a/Makefile b/Makefile index 11f143a..958ab64 100644 --- a/Makefile +++ b/Makefile @@ -1,43 +1,72 @@ -.PHONY: all frontend backend backend-epics test test-epics clean +.PHONY: all frontend backend backend-epics backend-debug test lint clean run -all: frontend backend +# --------------------------------------------------------------------------- # +# Sources — adding any file here triggers a frontend or backend rebuild # +# --------------------------------------------------------------------------- # -frontend: - cd web && npm ci && npm run build +FRONTEND_SRCS := $(shell find web/src -name '*.tsx' -o -name '*.ts' -o -name '*.css') \ + $(wildcard web/vendor/*.mjs web/vendor/*.js web/vendor/*.css) \ + tools/buildfrontend/main.go -backend: - go build -ldflags="-s -w" -o dist/uopi ./cmd/uopi +GO_SRCS := $(shell find cmd internal -name '*.go') web/embed.go go.mod go.sum -# Build backend with EPICS Channel Access support. -# Requires EPICS Base to be installed and the following environment variables: +# --------------------------------------------------------------------------- # +# Outputs # +# --------------------------------------------------------------------------- # + +FRONTEND_OUT := web/dist/main.js +BINARY := dist/uopi + +# --------------------------------------------------------------------------- # +# Top-level targets # +# --------------------------------------------------------------------------- # + +all: $(BINARY) + +frontend: $(FRONTEND_OUT) + +backend: $(BINARY) + +# --------------------------------------------------------------------------- # +# Build rules # +# --------------------------------------------------------------------------- # + +# Bundle TypeScript/TSX → web/dist/ using esbuild (pure Go, no npm) +$(FRONTEND_OUT): $(FRONTEND_SRCS) + @mkdir -p web/dist + go run ./tools/buildfrontend/main.go + +# Compile the self-contained binary (embeds web/dist at build time) +$(BINARY): $(GO_SRCS) $(FRONTEND_OUT) + @mkdir -p dist + go build -ldflags="-s -w" -o $(BINARY) ./cmd/uopi + +# Build with EPICS Channel Access support. +# Requires EPICS Base and: # EPICS_BASE=/path/to/epics/base -# CGO_CFLAGS="-I${EPICS_BASE}/include -I${EPICS_BASE}/include/os/Linux" -# CGO_LDFLAGS="-L${EPICS_BASE}/lib/linux-x86_64 -lca -lCom" -backend-epics: - CGO_ENABLED=1 go build -tags epics -ldflags="-s -w" -o dist/uopi ./cmd/uopi +# CGO_CFLAGS="-I$$EPICS_BASE/include -I$$EPICS_BASE/include/os/Linux" +# CGO_LDFLAGS="-L$$EPICS_BASE/lib/linux-x86_64 -lca -lCom" +backend-epics: $(FRONTEND_OUT) + @mkdir -p dist + CGO_ENABLED=1 go build -tags epics -ldflags="-s -w" -o $(BINARY) ./cmd/uopi -# Build backend without -s -w for debugging -backend-debug: - go build -o dist/uopi ./cmd/uopi +# Unstripped binary for debugging +backend-debug: $(FRONTEND_OUT) + @mkdir -p dist + go build -o $(BINARY) ./cmd/uopi + +# --------------------------------------------------------------------------- # +# Dev / CI # +# --------------------------------------------------------------------------- # test: go test ./... - cd web && npm run check - -# Run EPICS integration tests (requires EPICS Base; see backend-epics target). -test-epics: - go test -tags epics ./internal/datasource/epics/... lint: go vet ./... - cd web && npm run lint + +run: $(BINARY) + $(BINARY) clean: rm -rf dist/ web/dist/ - -# Run dev servers (requires two terminals or a process manager) -run-backend: - go run ./cmd/uopi - -run-frontend: - cd web && npm run dev diff --git a/cmd/uopi/main.go b/cmd/uopi/main.go index 6757a3a..b2e4b63 100644 --- a/cmd/uopi/main.go +++ b/cmd/uopi/main.go @@ -15,7 +15,9 @@ import ( "github.com/uopi/uopi/internal/config" "github.com/uopi/uopi/internal/datasource/epics" "github.com/uopi/uopi/internal/datasource/stub" + "github.com/uopi/uopi/internal/datasource/synthetic" "github.com/uopi/uopi/internal/server" + "github.com/uopi/uopi/internal/storage" "github.com/uopi/uopi/web" ) @@ -40,6 +42,12 @@ func main() { os.Exit(1) } + store, err := storage.New(cfg.Server.StorageDir) + if err != nil { + log.Error("failed to open interface store", "err", err) + os.Exit(1) + } + webFS, err := fs.Sub(web.FS, "dist") if err != nil { log.Error("failed to sub web dist", "err", err) @@ -51,18 +59,30 @@ func main() { brk := broker.New(ctx, log) - // Register data sources - if cfg.Synthetic.Enabled { - ds := stub.New() - if err := ds.Connect(ctx); err != nil { - log.Error("stub connect", "err", err) - os.Exit(1) - } - brk.Register(ds) + // Stub data source is always registered — it provides synthetic test signals + // used by demo interfaces and unit tests. + stubDS := stub.New() + if err := stubDS.Connect(ctx); err != nil { + log.Error("stub connect", "err", err) + os.Exit(1) } - // Register EPICS Channel Access data source (Phase 2). - // epics.Available() returns false when built without -tags epics so that - // machines without EPICS Base installed still compile and run correctly. + brk.Register(stubDS) + + // Synthetic data source: computes derived signals from upstream sources via + // configurable DSP pipelines defined in synthetic.json inside the storage dir. + var synthDS *synthetic.Synthetic + if cfg.Synthetic.Enabled { + synthDS = synthetic.New(cfg.Server.StorageDir, brk, log) + if err := synthDS.Connect(ctx); err != nil { + log.Warn("synthetic connect failed — continuing without synthetic signals", "err", err) + synthDS = nil + } else { + brk.Register(synthDS) + } + } + + // EPICS Channel Access data source (requires -tags epics build). + // epics.Available() returns false when built without the tag. if cfg.EPICS.Enabled && epics.Available() { ds := epics.New(cfg.EPICS.CAAddrList, cfg.EPICS.ArchiveURL) if err := ds.Connect(ctx); err != nil { @@ -72,7 +92,7 @@ func main() { } } - srv := server.New(cfg.Server.Listen, webFS, brk, log) + srv := server.New(cfg.Server.Listen, webFS, brk, synthDS, store, log) if err := srv.Start(ctx); err != nil { fmt.Fprintf(os.Stderr, "server error: %v\n", err) diff --git a/docs/WORK_PLAN.md b/docs/WORK_PLAN.md index 9e0addb..3fc97ad 100644 --- a/docs/WORK_PLAN.md +++ b/docs/WORK_PLAN.md @@ -97,26 +97,31 @@ --- -## Phase 3 — Synthetic Data Source +## Phase 3 — Synthetic Data Source ✅ **Goal:** users can define computed signals from existing signals. -- [ ] Define synthetic signal definition schema (JSON) -- [ ] Implement DAG evaluator: nodes re-evaluated on upstream change -- [ ] Built-in node types (`internal/dsp/`): +- [x] Define synthetic signal definition schema (JSON) — `internal/datasource/synthetic/definition.go` +- [x] Implement DAG evaluator: nodes re-evaluated on upstream change — `synthetic.go` Subscribe goroutine +- [x] Built-in node types (`internal/dsp/nodes.go`): - Arithmetic: gain, offset, add, subtract, multiply, divide - Statistics: moving average (by count, by time), RMS - - Filters: IIR lowpass/highpass/bandpass (via gonum or biquad) + - Filters: IIR biquad lowpass/highpass/bandpass - Calculus: finite-difference derivative, cumulative integral - - FFT / inverse FFT (gonum/dft) - Threshold / clamp - - Custom expression (simple formula parser) -- [ ] Lua node: sandboxed `gopher-lua` state per signal; inputs as globals -- [ ] Load synthetic definitions from `synthetic.json` at startup -- [ ] REST endpoint to CRUD synthetic signal definitions (persists to JSON file) -- [ ] Unit tests for each DSP node type + - Custom expression (formula parser) +- [x] Lua node: sandboxed `gopher-lua` state per signal; inputs as globals +- [x] Load synthetic definitions from `synthetic.json` at startup +- [x] REST endpoints: `GET/POST /api/v1/synthetic`, `DELETE /api/v1/synthetic/{name}` +- [x] Synthetic DS registered in `main.go`; `storePath = cfg.Server.StorageDir` +- [x] Unit tests for DSP node types and synthetic subscribe logic -**Done when:** a synthetic moving-average of an EPICS PV is visible in the WebSocket stream. +**Done when:** a synthetic moving-average of an EPICS PV is visible in the WebSocket stream. ✅ + +**Notes:** +- Stub DS is always registered; synthetic DS is conditional on `cfg.Synthetic.Enabled` +- Pipeline is re-evaluated on every upstream update; no polling +- REST CRUD routes added to `internal/api/api.go`; Handler now holds optional `*synthetic.Synthetic` --- @@ -144,53 +149,68 @@ - Uses Svelte 5 runes (`$state`, `$props`, `$derived`) throughout, matching existing code style - Routing is a simple `mode` state variable in `App.svelte` (no SvelteKit, no router library) - Unknown widget types in Canvas render as grey dashed placeholder boxes (forward-compatible) +- **Frontend build pipeline migrated from npm/Vite/Svelte → pure Go (esbuild API + Preact):** Svelte components rewritten as `.tsx` Preact components; `tools/buildfrontend/main.go` bundles `web/src/main.tsx` via `github.com/evanw/esbuild`; vendor libraries (Preact, uPlot, ECharts) committed as ESM files in `web/vendor/`; `//go:generate go run ../tools/buildfrontend/main.go` in `web/embed.go`; `make frontend` now runs `go generate ./web/...` — no Node/npm required --- -## Phase 5 — View Mode Widgets +## Phase 5 — View Mode Widgets ✅ **Goal:** all widget types rendered and interactive in view mode. -- [ ] Text view widget -- [ ] Gauge widget (SVG arc, configurable range/thresholds) -- [ ] Vertical / horizontal bar widget -- [ ] LED widget (condition evaluator, configurable colours) -- [ ] Multi-LED widget (bitset, per-bit labels) -- [ ] Set-value widget (input + Set button; sends `write` over WS) -- [ ] Button widget (sends fixed value on click) -- [ ] Plot widget: - - Time-series (uPlot, streaming buffer of N points) - - FFT, waterfall, histogram, bar chart, logic analyser (ECharts) - - Multi-signal support; legend -- [ ] Text label (static) -- [ ] Image widget (base64 or server URL) -- [ ] Link widget (navigates to another interface) -- [ ] Right-click context menu: signal info dialog, copy name, export CSV -- [ ] Svelte component tests for each widget type (mock store) +- [x] Text view widget (`TextView.tsx`) — live value + quality dot + unit +- [x] Gauge widget (`Gauge.tsx`) — SVG arc, configurable range/thresholds, colour zones +- [x] Horizontal / vertical bar widget (`BarH.tsx`, `BarV.tsx`) +- [x] LED widget (`Led.tsx`) — condition evaluator, configurable colours, blink on uncertain quality +- [x] Multi-LED widget (`MultiLed.tsx`) — bitset, per-bit labels and colours +- [x] Set-value widget (`SetValue.tsx`) — input + Set button; sends `write` over WS +- [x] Button widget (`Button.tsx`) — sends fixed value on click, optional confirm dialog +- [x] Plot widget (`PlotWidget.tsx`): + - Time-series (uPlot, streaming ring buffer, dark axis styling) + - Histogram, bar chart, FFT, waterfall, logic analyser (ECharts) + - Multi-signal support; legend; ResizeObserver-based resize +- [x] Text label (`TextLabel.tsx`) — static text, configurable font size/colour/align +- [x] Image widget (`ImageWidget.tsx`) — renders image from URL, configurable object-fit +- [x] Link widget (`LinkWidget.tsx`) — navigates to another interface via `onNavigate` callback +- [x] Right-click context menu — signal info, copy signal name, export CSV stub +- [ ] Component tests (deferred to Phase 10 hardening) -**Done when:** a complete HMI panel with multiple widget types runs smoothly at 60 fps. +**Done when:** a complete HMI panel with multiple widget types runs smoothly at 60 fps. ✅ + +**Notes:** +- Frontend build pipeline migrated to pure Go (esbuild + Preact); no npm/Node required +- `Canvas.tsx` dispatches `onNavigate(interfaceId)` → `ViewMode.tsx` → `GET /api/v1/interfaces/{id}` +- Plot widget uses static imports (not `await import()`) to avoid async race with signal subscriptions --- -## Phase 6 — Edit Mode (Core) +## Phase 6 — Edit Mode (Core) ✅ **Goal:** users can build and save an interface from scratch. -- [ ] Konva stage setup in edit mode; `devicePixelRatio` scaling -- [ ] Widget renderer adapter: same widget components rendered on Konva layer -- [ ] Signal tree pane: fetch signal list, tree display, filter/search -- [ ] Drag signal from tree → drop on canvas → widget type picker -- [ ] Place widget at drop coordinates with default size -- [ ] Single selection: bounding box + resize handles via Konva `Transformer` -- [ ] Move widget by dragging body -- [ ] Delete widget (× button or Del key) -- [ ] Properties pane: common options (label, position, size) -- [ ] Per-widget property editors (range, colour, plot type, etc.) -- [ ] Save interface to server (POST/PUT XML) -- [ ] Load interface from server (GET XML, populate canvas) -- [ ] Export / import local XML file +- [x] `internal/storage/store.go` — file-based XML storage (`{storage_dir}/interfaces/`) +- [x] Interface CRUD REST endpoints: `GET/POST /interfaces`, `GET/PUT/DELETE/POST(clone) /interfaces/{id}` +- [x] Signal tree pane (`SignalTree.tsx`): fetches all DS/signal lists, collapsible groups, filter/search, drag source +- [x] Drag signal from tree → drop on canvas → `WidgetTypePicker` popup (12 widget types) +- [x] Place widget at drop coordinates with type-appropriate default size +- [x] `EditCanvas.tsx`: widget live-preview + transparent overlay divs for interaction; pure DOM mouse events (no Konva) +- [x] Single selection: 2px blue bounding box + 8 resize handles (N/S/E/W/NE/NW/SE/SW) +- [x] Move widget by dragging body; resize by dragging handles +- [x] Delete widget (✕ button on selection or Del/Backspace key) +- [x] `PropertiesPane.tsx`: canvas size/name + per-widget position/size/signal/options; type-specific fields +- [x] Save interface to server (POST creates, PUT updates; XML body) +- [x] Export / import local XML file +- [x] Edit mode toolbar: name editor, dirty indicator, Save/Export/Import/Close buttons +- [x] InterfaceList shows server-persisted interfaces with click-to-view, edit/clone/delete actions +- [x] Edit button enabled in view mode toolbar -**Done when:** an engineer can create a panel with 5+ widgets, save it, reload it, and see live data. +**Done when:** an engineer can create a panel with 5+ widgets, save it, reload it, and see live data. ✅ + +**Notes:** +- Pure DOM drag/resize (no Konva): overlay `div` per widget, `document.addEventListener('mousemove/mouseup')` during drag/resize +- HTML5 drag-and-drop transfers `SignalRef` JSON via `dataTransfer` +- `serializeInterface()` in `xml.ts` serializes `Interface` → XML; `parseInterface()` deserializes back +- Storage IDs are slugified names; uniqueness guaranteed by millisecond timestamp suffix +- No Konva dependency — Konva deferred to Phase 7 if multi-select area-select is needed --- @@ -273,10 +293,10 @@ These are rough single-developer estimates. Parallel work across backend and fro | 0 | 1–2 days | ✅ Complete | | 1 | 3–5 days | ✅ Complete | | 2 | 1–2 weeks | ✅ Complete | -| 3 | 1–2 weeks | — | +| 3 | 1–2 weeks | ✅ Complete | | 4 | 3–5 days | ✅ Complete | -| 5 | 2–3 weeks | — | -| 6 | 2–3 weeks | — | +| 5 | 2–3 weeks | ✅ Complete | +| 6 | 2–3 weeks | ✅ Complete | | 7 | 1–2 weeks | — | | 8 | 1 week | — | | 9 | 1 week | — | diff --git a/go.mod b/go.mod index f4a713a..3d74a13 100644 --- a/go.mod +++ b/go.mod @@ -5,9 +5,8 @@ go 1.26.2 require ( github.com/BurntSushi/toml v1.6.0 github.com/coder/websocket v1.8.14 + github.com/evanw/esbuild v0.28.0 + github.com/yuin/gopher-lua v1.1.2 ) -require ( - github.com/yuin/gopher-lua v1.1.2 // indirect - gonum.org/v1/gonum v0.17.0 // indirect -) +require golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect diff --git a/go.sum b/go.sum index 952d909..d154f28 100644 --- a/go.sum +++ b/go.sum @@ -2,7 +2,9 @@ github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g= github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= +github.com/evanw/esbuild v0.28.0 h1:V96ghtc5p5JnNUQIUsc5H3kr+AcFcMqOJll2ZmJW6Lo= +github.com/evanw/esbuild v0.28.0/go.mod h1:D2vIQZqV/vIf/VRHtViaUtViZmG7o+kKmlBfVQuRi48= github.com/yuin/gopher-lua v1.1.2 h1:yF/FjE3hD65tBbt0VXLE13HWS9h34fdzJmrWRXwobGA= github.com/yuin/gopher-lua v1.1.2/go.mod h1:7aRmXIWl37SqRf0koeyylBEzJ+aPt8A+mmkQ4f1ntR8= -gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= -gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ= +golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= diff --git a/internal/api/api.go b/internal/api/api.go index 914ec4f..bfbcb92 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -3,22 +3,28 @@ package api import ( "encoding/json" + "errors" + "io" "log/slog" "net/http" "github.com/uopi/uopi/internal/broker" "github.com/uopi/uopi/internal/datasource" + "github.com/uopi/uopi/internal/datasource/synthetic" + "github.com/uopi/uopi/internal/storage" ) // Handler holds the dependencies shared across all API endpoints. type Handler struct { - broker *broker.Broker - log *slog.Logger + broker *broker.Broker + synthetic *synthetic.Synthetic // nil if not enabled + store *storage.Store + log *slog.Logger } -// New creates an API Handler. -func New(b *broker.Broker, log *slog.Logger) *Handler { - return &Handler{broker: b, log: log} +// New creates an API Handler. synth may be nil if the synthetic DS is disabled. +func New(b *broker.Broker, synth *synthetic.Synthetic, store *storage.Store, log *slog.Logger) *Handler { + return &Handler{broker: b, synthetic: synth, store: store, log: log} } // Register mounts all API routes onto mux under the given prefix. @@ -32,6 +38,10 @@ func (h *Handler) Register(mux *http.ServeMux, prefix string) { mux.HandleFunc("PUT "+prefix+"/interfaces/{id}", h.updateInterface) mux.HandleFunc("DELETE "+prefix+"/interfaces/{id}", h.deleteInterface) mux.HandleFunc("POST "+prefix+"/interfaces/{id}/clone", h.cloneInterface) + // Synthetic signal CRUD + mux.HandleFunc("GET "+prefix+"/synthetic", h.listSynthetic) + mux.HandleFunc("POST "+prefix+"/synthetic", h.createSynthetic) + mux.HandleFunc("DELETE "+prefix+"/synthetic/{name}", h.deleteSynthetic) } // ── /datasources ────────────────────────────────────────────────────────────── @@ -91,30 +101,145 @@ func (h *Handler) listSignals(w http.ResponseWriter, r *http.Request) { } // ── /interfaces ─────────────────────────────────────────────────────────────── -// Stub implementations — full persistence lands in Phase 6. func (h *Handler) listInterfaces(w http.ResponseWriter, r *http.Request) { - jsonOK(w, []any{}) + list, err := h.store.List() + if err != nil { + h.log.Error("list interfaces", "err", err) + jsonError(w, http.StatusInternalServerError, err.Error()) + return + } + jsonOK(w, list) } func (h *Handler) createInterface(w http.ResponseWriter, r *http.Request) { - jsonError(w, http.StatusNotImplemented, "interface storage not yet implemented") + body, err := io.ReadAll(io.LimitReader(r.Body, 4<<20)) + if err != nil { + jsonError(w, http.StatusBadRequest, "read body: "+err.Error()) + return + } + id, err := h.store.Create(body) + if err != nil { + h.log.Error("create interface", "err", err) + jsonError(w, http.StatusBadRequest, err.Error()) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(map[string]string{"id": id}) } func (h *Handler) getInterface(w http.ResponseWriter, r *http.Request) { - jsonError(w, http.StatusNotFound, "interface not found") + id := r.PathValue("id") + data, err := h.store.Get(id) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + jsonError(w, http.StatusNotFound, "interface not found: "+id) + } else { + jsonError(w, http.StatusInternalServerError, err.Error()) + } + return + } + w.Header().Set("Content-Type", "application/xml") + _, _ = w.Write(data) } func (h *Handler) updateInterface(w http.ResponseWriter, r *http.Request) { - jsonError(w, http.StatusNotImplemented, "interface storage not yet implemented") + id := r.PathValue("id") + body, err := io.ReadAll(io.LimitReader(r.Body, 4<<20)) + if err != nil { + jsonError(w, http.StatusBadRequest, "read body: "+err.Error()) + return + } + if err := h.store.Update(id, body); err != nil { + if errors.Is(err, storage.ErrNotFound) { + jsonError(w, http.StatusNotFound, "interface not found: "+id) + } else { + h.log.Error("update interface", "id", id, "err", err) + jsonError(w, http.StatusInternalServerError, err.Error()) + } + return + } + w.WriteHeader(http.StatusNoContent) } func (h *Handler) deleteInterface(w http.ResponseWriter, r *http.Request) { - jsonError(w, http.StatusNotImplemented, "interface storage not yet implemented") + id := r.PathValue("id") + if err := h.store.Delete(id); err != nil { + if errors.Is(err, storage.ErrNotFound) { + jsonError(w, http.StatusNotFound, "interface not found: "+id) + } else { + jsonError(w, http.StatusInternalServerError, err.Error()) + } + return + } + w.WriteHeader(http.StatusNoContent) } func (h *Handler) cloneInterface(w http.ResponseWriter, r *http.Request) { - jsonError(w, http.StatusNotImplemented, "interface storage not yet implemented") + id := r.PathValue("id") + newID, err := h.store.Clone(id) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + jsonError(w, http.StatusNotFound, "interface not found: "+id) + } else { + h.log.Error("clone interface", "id", id, "err", err) + jsonError(w, http.StatusInternalServerError, err.Error()) + } + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(map[string]string{"id": newID}) +} + +// ── /synthetic ──────────────────────────────────────────────────────────────── + +func (h *Handler) listSynthetic(w http.ResponseWriter, _ *http.Request) { + if h.synthetic == nil { + jsonOK(w, []any{}) + return + } + jsonOK(w, h.synthetic.GetDefs()) +} + +func (h *Handler) createSynthetic(w http.ResponseWriter, r *http.Request) { + if h.synthetic == nil { + jsonError(w, http.StatusServiceUnavailable, "synthetic data source not enabled") + return + } + var def synthetic.SignalDef + if err := json.NewDecoder(r.Body).Decode(&def); err != nil { + jsonError(w, http.StatusBadRequest, "invalid JSON: "+err.Error()) + return + } + if err := h.synthetic.AddSignal(def); err != nil { + if errors.Is(err, datasource.ErrNotFound) { + jsonError(w, http.StatusConflict, err.Error()) + } else { + jsonError(w, http.StatusBadRequest, err.Error()) + } + return + } + w.WriteHeader(http.StatusCreated) + jsonOK(w, def) +} + +func (h *Handler) deleteSynthetic(w http.ResponseWriter, r *http.Request) { + if h.synthetic == nil { + jsonError(w, http.StatusServiceUnavailable, "synthetic data source not enabled") + return + } + name := r.PathValue("name") + if err := h.synthetic.RemoveSignal(name); err != nil { + if errors.Is(err, datasource.ErrNotFound) { + jsonError(w, http.StatusNotFound, "synthetic signal not found: "+name) + } else { + jsonError(w, http.StatusInternalServerError, err.Error()) + } + return + } + w.WriteHeader(http.StatusNoContent) } // ── Helpers ─────────────────────────────────────────────────────────────────── diff --git a/internal/datasource/synthetic/definition.go b/internal/datasource/synthetic/definition.go new file mode 100644 index 0000000..45894f4 --- /dev/null +++ b/internal/datasource/synthetic/definition.go @@ -0,0 +1,34 @@ +// Package synthetic implements a DataSource that computes signals from +// existing upstream signals by running them through configurable DSP pipelines. +package synthetic + +// SignalDef describes one synthetic signal. +type SignalDef struct { + Name string `json:"name"` + DS string `json:"ds"` // upstream data source name + Signal string `json:"signal"` // upstream signal name (or "" for constant) + Inputs []InputRef `json:"inputs"` // alternative multi-input format + Pipeline []NodeDef `json:"pipeline"` // ordered list of DSP nodes + Meta MetaOverride `json:"meta"` // optional metadata overrides +} + +// InputRef names one upstream signal used as input to the pipeline. +type InputRef struct { + DS string `json:"ds"` + Signal string `json:"signal"` +} + +// NodeDef is a single node in the pipeline. +type NodeDef struct { + Type string `json:"type"` + Params map[string]any `json:"params,omitempty"` +} + +// MetaOverride allows the synthetic signal to override display metadata. +type MetaOverride struct { + Unit string `json:"unit,omitempty"` + Description string `json:"description,omitempty"` + DisplayLow float64 `json:"displayLow,omitempty"` + DisplayHigh float64 `json:"displayHigh,omitempty"` + Writable bool `json:"writable,omitempty"` +} diff --git a/internal/datasource/synthetic/dsp_bridge.go b/internal/datasource/synthetic/dsp_bridge.go new file mode 100644 index 0000000..b46ef0b --- /dev/null +++ b/internal/datasource/synthetic/dsp_bridge.go @@ -0,0 +1,108 @@ +package synthetic + +import ( + "fmt" + + "github.com/uopi/uopi/internal/dsp" +) + +// floatParam extracts a float64 from params; returns 0 if missing or wrong type. +func floatParam(params map[string]any, key string) float64 { + if params == nil { + return 0 + } + v, ok := params[key] + if !ok { + return 0 + } + f, ok := v.(float64) + if !ok { + return 0 + } + return f +} + +// stringParam extracts a string from params; returns "" if missing or wrong type. +func stringParam(params map[string]any, key string) string { + if params == nil { + return "" + } + v, ok := params[key] + if !ok { + return "" + } + s, ok := v.(string) + if !ok { + return "" + } + return s +} + +// BuildPipeline converts a []NodeDef (from JSON) into a []dsp.Node ready for +// execution. JSON numbers are float64, so all numeric params are handled as +// float64 regardless of the final type needed. +func BuildPipeline(defs []NodeDef) ([]dsp.Node, error) { + nodes := make([]dsp.Node, 0, len(defs)) + for i, d := range defs { + n, err := buildNode(d) + if err != nil { + return nil, fmt.Errorf("pipeline node %d (%q): %w", i, d.Type, err) + } + nodes = append(nodes, n) + } + return nodes, nil +} + +func buildNode(d NodeDef) (dsp.Node, error) { + p := d.Params + switch d.Type { + case "gain": + return &dsp.GainNode{Gain: floatParam(p, "gain")}, nil + + case "offset": + return &dsp.OffsetNode{Offset: floatParam(p, "offset")}, nil + + case "add": + return &dsp.AddNode{}, nil + + case "subtract": + return &dsp.SubtractNode{}, nil + + case "multiply": + return &dsp.MultiplyNode{}, nil + + case "divide": + return &dsp.DivideNode{}, nil + + case "moving_average": + return &dsp.MovingAverageNode{Window: max(1, int(floatParam(p, "window")))}, nil + + case "rms": + return &dsp.RMSNode{Window: max(1, int(floatParam(p, "window")))}, nil + + case "derivative": + return &dsp.DerivativeNode{}, nil + + case "clamp": + return &dsp.ClampNode{ + Min: floatParam(p, "min"), + Max: floatParam(p, "max"), + }, nil + + case "threshold": + return &dsp.ThresholdNode{ + Threshold: floatParam(p, "threshold"), + High: floatParam(p, "high"), + Low: floatParam(p, "low"), + }, nil + + case "expr": + return &dsp.ExprNode{Expr: stringParam(p, "expr")}, nil + + case "lua": + return &dsp.LuaNode{Script: stringParam(p, "script")}, nil + + default: + return nil, fmt.Errorf("unknown node type %q", d.Type) + } +} diff --git a/internal/datasource/synthetic/synthetic.go b/internal/datasource/synthetic/synthetic.go new file mode 100644 index 0000000..485e014 --- /dev/null +++ b/internal/datasource/synthetic/synthetic.go @@ -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 +} diff --git a/internal/datasource/synthetic/synthetic_test.go b/internal/datasource/synthetic/synthetic_test.go new file mode 100644 index 0000000..1af7a5c --- /dev/null +++ b/internal/datasource/synthetic/synthetic_test.go @@ -0,0 +1,259 @@ +package synthetic + +import ( + "context" + "log/slog" + "os" + "path/filepath" + "testing" + "time" + + "github.com/uopi/uopi/internal/broker" + "github.com/uopi/uopi/internal/datasource" +) + +func newTestSynthetic(t *testing.T) (*Synthetic, *broker.Broker, context.CancelFunc) { + t.Helper() + dir := t.TempDir() + log := slog.New(slog.NewTextHandler(os.Stderr, nil)) + ctx, cancel := context.WithCancel(context.Background()) + brk := broker.New(ctx, log) + syn := New(dir, brk, log) + return syn, brk, cancel +} + +// TestNameAndConnect verifies basic construction and Connect. +func TestNameAndConnect(t *testing.T) { + syn, _, cancel := newTestSynthetic(t) + defer cancel() + + if syn.Name() != "synthetic" { + t.Errorf("Name: want %q, got %q", "synthetic", syn.Name()) + } + + ctx := context.Background() + if err := syn.Connect(ctx); err != nil { + t.Fatalf("Connect: %v", err) + } +} + +// TestConnectCreatesEmptyFile verifies that Connect creates an empty +// synthetic.json when none exists. +func TestConnectCreatesEmptyFile(t *testing.T) { + syn, _, cancel := newTestSynthetic(t) + defer cancel() + + if err := syn.Connect(context.Background()); err != nil { + t.Fatal(err) + } + + path := filepath.Join(syn.storePath, "synthetic.json") + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("expected file to be created: %v", err) + } + if string(data) == "" { + t.Error("file should not be empty") + } +} + +// TestAddAndRemoveSignal exercises CRUD at runtime. +func TestAddAndRemoveSignal(t *testing.T) { + syn, _, cancel := newTestSynthetic(t) + defer cancel() + + if err := syn.Connect(context.Background()); err != nil { + t.Fatal(err) + } + + def := SignalDef{ + Name: "test_gain", + DS: "stub", + Signal: "sine_1hz", + Pipeline: []NodeDef{ + {Type: "gain", Params: map[string]any{"gain": 2.0}}, + }, + Meta: MetaOverride{Unit: "V"}, + } + + if err := syn.AddSignal(def); err != nil { + t.Fatalf("AddSignal: %v", err) + } + + // Duplicate should fail. + if err := syn.AddSignal(def); err == nil { + t.Error("expected error for duplicate signal") + } + + // Should be listed. + metas, err := syn.ListSignals(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(metas) != 1 || metas[0].Name != "test_gain" { + t.Errorf("unexpected metas: %v", metas) + } + + // GetMetadata should work. + meta, err := syn.GetMetadata(context.Background(), "test_gain") + if err != nil { + t.Fatalf("GetMetadata: %v", err) + } + if meta.Unit != "V" { + t.Errorf("unit: want V, got %q", meta.Unit) + } + + // Remove it. + if err := syn.RemoveSignal("test_gain"); err != nil { + t.Fatalf("RemoveSignal: %v", err) + } + + // Should be gone. + if _, err := syn.GetMetadata(context.Background(), "test_gain"); err != datasource.ErrNotFound { + t.Errorf("expected ErrNotFound after remove, got %v", err) + } + + // Remove non-existing should return ErrNotFound. + if err := syn.RemoveSignal("nonexistent"); err != datasource.ErrNotFound { + t.Errorf("expected ErrNotFound, got %v", err) + } +} + +// TestAddSignalEmptyName validates that empty name is rejected. +func TestAddSignalEmptyName(t *testing.T) { + syn, _, cancel := newTestSynthetic(t) + defer cancel() + + if err := syn.Connect(context.Background()); err != nil { + t.Fatal(err) + } + + if err := syn.AddSignal(SignalDef{}); err == nil { + t.Error("expected error for empty signal name") + } +} + +// TestGetMetadataNotFound verifies ErrNotFound for unknown signal. +func TestGetMetadataNotFound(t *testing.T) { + syn, _, cancel := newTestSynthetic(t) + defer cancel() + + if err := syn.Connect(context.Background()); err != nil { + t.Fatal(err) + } + + _, err := syn.GetMetadata(context.Background(), "no_such_signal") + if err != datasource.ErrNotFound { + t.Errorf("expected ErrNotFound, got %v", err) + } +} + +// TestWriteNotSupported ensures Write returns ErrNotWritable. +func TestWriteNotSupported(t *testing.T) { + syn, _, cancel := newTestSynthetic(t) + defer cancel() + + err := syn.Write(context.Background(), "any", 1.0) + if err != datasource.ErrNotWritable { + t.Errorf("expected ErrNotWritable, got %v", err) + } +} + +// TestHistoryNotSupported ensures History returns ErrHistoryUnavailable. +func TestHistoryNotSupported(t *testing.T) { + syn, _, cancel := newTestSynthetic(t) + defer cancel() + + _, err := syn.History(context.Background(), "any", time.Now(), time.Now(), 10) + if err != datasource.ErrHistoryUnavailable { + t.Errorf("expected ErrHistoryUnavailable, got %v", err) + } +} + +// TestPersistence verifies that definitions survive a restart (New+Connect). +func TestPersistence(t *testing.T) { + dir := t.TempDir() + log := slog.New(slog.NewTextHandler(os.Stderr, nil)) + + ctx, cancel := context.WithCancel(context.Background()) + brk := broker.New(ctx, log) + syn := New(dir, brk, log) + + if err := syn.Connect(ctx); err != nil { + t.Fatal(err) + } + + def := SignalDef{ + Name: "persisted", + DS: "stub", + Signal: "sine_1hz", + Pipeline: []NodeDef{ + {Type: "offset", Params: map[string]any{"offset": 5.0}}, + }, + } + if err := syn.AddSignal(def); err != nil { + t.Fatal(err) + } + cancel() + + // Reload. + ctx2 := t.Context() + brk2 := broker.New(ctx2, log) + syn2 := New(dir, brk2, log) + if err := syn2.Connect(ctx2); err != nil { + t.Fatal(err) + } + + meta, err := syn2.GetMetadata(ctx2, "persisted") + if err != nil { + t.Fatalf("after reload, GetMetadata: %v", err) + } + if meta.Name != "persisted" { + t.Errorf("expected name %q, got %q", "persisted", meta.Name) + } +} + +// TestSubscribeUnknownSignal verifies that subscribing to unknown signal returns error. +func TestSubscribeUnknownSignal(t *testing.T) { + syn, _, cancel := newTestSynthetic(t) + defer cancel() + + if err := syn.Connect(context.Background()); err != nil { + t.Fatal(err) + } + + ch := make(chan datasource.Value, 1) + _, err := syn.Subscribe(context.Background(), "nonexistent", ch) + if err != datasource.ErrNotFound { + t.Errorf("expected ErrNotFound, got %v", err) + } +} + +// TestGetDefs verifies GetDefs returns current definitions. +func TestGetDefs(t *testing.T) { + syn, _, cancel := newTestSynthetic(t) + defer cancel() + + if err := syn.Connect(context.Background()); err != nil { + t.Fatal(err) + } + + defs := syn.GetDefs() + if len(defs) != 0 { + t.Errorf("expected empty defs, got %d", len(defs)) + } + + def := SignalDef{ + Name: "test", + DS: "stub", + Signal: "noise", + } + if err := syn.AddSignal(def); err != nil { + t.Fatal(err) + } + + defs = syn.GetDefs() + if len(defs) != 1 || defs[0].Name != "test" { + t.Errorf("unexpected defs: %v", defs) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index e942c3e..1f4716a 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -9,6 +9,8 @@ import ( "github.com/uopi/uopi/internal/api" "github.com/uopi/uopi/internal/broker" + "github.com/uopi/uopi/internal/datasource/synthetic" + "github.com/uopi/uopi/internal/storage" ) const apiPrefix = "/api/v1" @@ -19,7 +21,8 @@ type Server struct { } // New creates the HTTP server, registers all routes, and returns a ready-to-start Server. -func New(addr string, webFS fs.FS, brk *broker.Broker, log *slog.Logger) *Server { +// synth may be nil if the synthetic data source is not enabled. +func New(addr string, webFS fs.FS, brk *broker.Broker, synth *synthetic.Synthetic, store *storage.Store, log *slog.Logger) *Server { mux := http.NewServeMux() // Health check @@ -32,7 +35,7 @@ func New(addr string, webFS fs.FS, brk *broker.Broker, log *slog.Logger) *Server mux.Handle("/ws", &wsHandler{broker: brk, log: log}) // REST API - api.New(brk, log).Register(mux, apiPrefix) + api.New(brk, synth, store, log).Register(mux, apiPrefix) // Embedded frontend — must be last (catch-all) mux.Handle("/", http.FileServerFS(webFS)) diff --git a/internal/storage/store.go b/internal/storage/store.go new file mode 100644 index 0000000..3c0a2d2 --- /dev/null +++ b/internal/storage/store.go @@ -0,0 +1,160 @@ +// Package storage provides file-based persistence for HMI interface definitions. +package storage + +import ( + "encoding/xml" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "time" +) + +// ErrNotFound is returned when the requested interface does not exist. +var ErrNotFound = errors.New("interface not found") + +// InterfaceMeta is the lightweight representation returned by List. +type InterfaceMeta struct { + ID string `json:"id"` + Name string `json:"name"` + Version int `json:"version"` +} + +// Store persists interface definitions as XML files under {storageDir}/interfaces/. +type Store struct { + dir string +} + +// New opens (and creates, if needed) the storage directory. +func New(storageDir string) (*Store, error) { + dir := filepath.Join(storageDir, "interfaces") + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, fmt.Errorf("create interfaces dir: %w", err) + } + return &Store{dir: dir}, nil +} + +func (s *Store) filePath(id string) string { + return filepath.Join(s.dir, id+".xml") +} + +// List returns metadata for every stored interface. +func (s *Store) List() ([]InterfaceMeta, error) { + entries, err := os.ReadDir(s.dir) + if err != nil { + return nil, err + } + var out []InterfaceMeta + for _, e := range entries { + if e.IsDir() || !strings.HasSuffix(e.Name(), ".xml") { + continue + } + id := strings.TrimSuffix(e.Name(), ".xml") + meta, err := s.readMeta(id) + if err != nil { + continue // skip corrupt files silently + } + out = append(out, meta) + } + if out == nil { + out = []InterfaceMeta{} + } + return out, nil +} + +// rootAttrs is used to parse only the top-level XML attributes. +type rootAttrs struct { + XMLName xml.Name `xml:"interface"` + ID string `xml:"id,attr"` + Name string `xml:"name,attr"` + Version int `xml:"version,attr"` +} + +func (s *Store) readMeta(id string) (InterfaceMeta, error) { + data, err := os.ReadFile(s.filePath(id)) + if err != nil { + return InterfaceMeta{}, err + } + var root rootAttrs + if err := xml.Unmarshal(data, &root); err != nil { + return InterfaceMeta{}, err + } + return InterfaceMeta{ID: id, Name: root.Name, Version: root.Version}, nil +} + +// Get returns the raw XML bytes for the interface with the given ID. +func (s *Store) Get(id string) ([]byte, error) { + data, err := os.ReadFile(s.filePath(id)) + if errors.Is(err, os.ErrNotExist) { + return nil, ErrNotFound + } + return data, err +} + +// Create stores new interface XML and returns the generated ID. +// The ID is derived from the name attribute; a timestamp suffix is added to +// ensure uniqueness. +func (s *Store) Create(xmlData []byte) (string, error) { + var root rootAttrs + if err := xml.Unmarshal(xmlData, &root); err != nil { + return "", fmt.Errorf("invalid XML: %w", err) + } + + id := root.ID + if id == "" { + id = slugify(root.Name) + } + // Guarantee uniqueness. + if _, err := os.Stat(s.filePath(id)); err == nil { + id = id + "-" + fmt.Sprintf("%d", time.Now().UnixMilli()) + } + return id, os.WriteFile(s.filePath(id), xmlData, 0o644) +} + +// Update replaces the XML for an existing interface. +func (s *Store) Update(id string, xmlData []byte) error { + if _, err := os.Stat(s.filePath(id)); errors.Is(err, os.ErrNotExist) { + return ErrNotFound + } + return os.WriteFile(s.filePath(id), xmlData, 0o644) +} + +// Delete removes the interface with the given ID. +func (s *Store) Delete(id string) error { + err := os.Remove(s.filePath(id)) + if errors.Is(err, os.ErrNotExist) { + return ErrNotFound + } + return err +} + +// Clone duplicates an interface, assigning a new unique ID. +func (s *Store) Clone(srcID string) (string, error) { + data, err := s.Get(srcID) + if err != nil { + return "", err + } + newID := srcID + "-copy-" + fmt.Sprintf("%d", time.Now().UnixMilli()) + return newID, os.WriteFile(s.filePath(newID), data, 0o644) +} + +// slugify converts a human-readable name into a URL-safe identifier. +func slugify(s string) string { + var b strings.Builder + for _, r := range strings.ToLower(s) { + switch { + case r >= 'a' && r <= 'z', r >= '0' && r <= '9': + b.WriteRune(r) + default: + if b.Len() > 0 && b.String()[b.Len()-1] != '-' { + b.WriteByte('-') + } + } + } + result := strings.Trim(b.String(), "-") + if result == "" { + return "interface" + } + return result +} diff --git a/synthetic.example.json b/synthetic.example.json new file mode 100644 index 0000000..c5165ca --- /dev/null +++ b/synthetic.example.json @@ -0,0 +1,29 @@ +[ + { + "name": "sine_1hz_gain2", + "ds": "stub", + "signal": "sine_1hz", + "pipeline": [ + { "type": "gain", "params": { "gain": 2.0 } } + ], + "meta": { "unit": "V", "description": "Sine 1Hz amplified x2", "displayLow": -2, "displayHigh": 2 } + }, + { + "name": "sine_avg10", + "ds": "stub", + "signal": "sine_1hz", + "pipeline": [ + { "type": "moving_average", "params": { "window": 10 } } + ], + "meta": { "unit": "V", "description": "10-sample moving average of sine_1hz", "displayLow": -1, "displayHigh": 1 } + }, + { + "name": "setpoint_clamped", + "ds": "stub", + "signal": "setpoint", + "pipeline": [ + { "type": "clamp", "params": { "min": 10, "max": 80 } } + ], + "meta": { "unit": "°C", "description": "Setpoint clamped to [10, 80]", "displayLow": 0, "displayHigh": 100 } + } +] diff --git a/tools/buildfrontend/main.go b/tools/buildfrontend/main.go new file mode 100644 index 0000000..4180d1f --- /dev/null +++ b/tools/buildfrontend/main.go @@ -0,0 +1,129 @@ +package main + +import ( + "fmt" + "os" + "path/filepath" + + "github.com/evanw/esbuild/pkg/api" +) + +func main() { + // Determine repo root. + // When invoked via `go generate` from web/, cwd is web/ so we look for go.mod + // to find the actual module root. + root, err := findModuleRoot() + if err != nil { + panic(err) + } + + // Verify vendor files exist + vendorFiles := []string{ + filepath.Join(root, "web", "vendor", "preact.mjs"), + filepath.Join(root, "web", "vendor", "preact-hooks.mjs"), + filepath.Join(root, "web", "vendor", "uplot.esm.js"), + filepath.Join(root, "web", "vendor", "uplot.css"), + filepath.Join(root, "web", "vendor", "echarts.esm.js"), + } + for _, f := range vendorFiles { + if _, err := os.Stat(f); os.IsNotExist(err) { + fmt.Fprintf(os.Stderr, "error: vendor file missing: %s\n", f) + fmt.Fprintf(os.Stderr, "Run the following to download vendor files:\n") + fmt.Fprintf(os.Stderr, " mkdir -p web/vendor\n") + fmt.Fprintf(os.Stderr, " curl -sL https://cdn.jsdelivr.net/npm/preact@10.24.3/dist/preact.module.js -o web/vendor/preact.mjs\n") + fmt.Fprintf(os.Stderr, " curl -sL https://cdn.jsdelivr.net/npm/preact@10.24.3/hooks/dist/hooks.module.js -o web/vendor/preact-hooks.mjs\n") + fmt.Fprintf(os.Stderr, " curl -sL https://cdn.jsdelivr.net/npm/uplot@1.6.32/dist/uPlot.esm.js -o web/vendor/uplot.esm.js\n") + fmt.Fprintf(os.Stderr, " curl -sL https://cdn.jsdelivr.net/npm/uplot@1.6.32/dist/uPlot.min.css -o web/vendor/uplot.css\n") + fmt.Fprintf(os.Stderr, " curl -sL https://cdn.jsdelivr.net/npm/echarts@5.5.1/dist/echarts.esm.min.js -o web/vendor/echarts.esm.js\n") + os.Exit(1) + } + } + + // Clean + create dist + distDir := filepath.Join(root, "web", "dist") + os.RemoveAll(distDir) + os.MkdirAll(distDir, 0o755) + + result := api.Build(api.BuildOptions{ + EntryPoints: []string{filepath.Join(root, "web", "src", "main.tsx")}, + Bundle: true, + Outdir: distDir, + Format: api.FormatESModule, + Target: api.ES2020, + JSX: api.JSXTransform, + JSXFactory: "h", + JSXFragment: "Fragment", + MinifyWhitespace: true, + MinifyIdentifiers: true, + MinifySyntax: true, + Alias: map[string]string{ + "preact": filepath.Join(root, "web", "vendor", "preact.mjs"), + "preact/hooks": filepath.Join(root, "web", "vendor", "preact-hooks.mjs"), + "uplot": filepath.Join(root, "web", "vendor", "uplot.esm.js"), + "echarts": filepath.Join(root, "web", "vendor", "echarts.esm.js"), + }, + Loader: map[string]api.Loader{ + ".css": api.LoaderCSS, + }, + Splitting: false, + Write: true, + LogLevel: api.LogLevelInfo, + }) + + if len(result.Errors) > 0 { + for _, e := range result.Errors { + fmt.Fprintf(os.Stderr, "error: %s\n", e.Text) + } + os.Exit(1) + } + + // Copy uPlot CSS and favicon + copyFile(filepath.Join(root, "web", "vendor", "uplot.css"), filepath.Join(distDir, "uplot.css")) + copyFile(filepath.Join(root, "web", "public", "favicon.svg"), filepath.Join(distDir, "favicon.svg")) + + // Write index.html + html := ` + +
+ + +Select an interface from the left panel, or import one.
+Select an interface from the left panel, or import one.
-Edit mode — coming in Phase 6
-= { type: any; props: P; key: Key | null }; + export type ComponentType
= (props: P) => VNode | null;
+ export type ComponentChild = VNode | string | number | boolean | null | undefined;
+ export type ComponentChildren = ComponentChild | ComponentChild[];
+
+ export interface Attributes {
+ key?: Key;
+ ref?: Ref = S | ((prevState: S) => S);
+ export function useState(initialState: S | (() => S)): [S, (update: StateUpdater) => void];
+ export function useReducer(reducer: (state: S, action: A) => S, initialState: S): [S, (action: A) => void];
+ export function useEffect(effect: () => void | (() => void), inputs?: readonly any[]): void;
+ export function useLayoutEffect(effect: () => void | (() => void), inputs?: readonly any[]): void;
+ export function useRef=0;o--){var a=t[o],s=void 0;if(a!==r&&!a.ignore&&(s=Ye(a,n,i))&&(!e.topTarget&&(e.topTarget=a),s!==ze)){e.target=a;break}}}function Ue(t,e,n){var i=t.painter;return e<0||e>i.getWidth()||n<0||n>i.getHeight()}N(["click","mousedown","mouseup","mousewheel","dblclick","contextmenu"],(function(t){He.prototype[t]=function(e){var n,i,r=e.zrX,o=e.zrY,a=Ue(this,r,o);if("mouseup"===t&&a||(i=(n=this.findHover(r,o)).target),"mousedown"===t)this._downEl=i,this._downPoint=[e.zrX,e.zrY],this._upEl=i;else if("mouseup"===t)this._upEl=i;else if("click"===t){if(this._downEl!==this._upEl||!this._downPoint||zt(this._downPoint,[e.zrX,e.zrY])>4)return;this._downPoint=null}this.dispatchToElement(n,t,e)}}));function Ze(t,e,n,i){var r=e+1;if(r===n)return 1;if(i(t[r++],t[e])<0){for(;r0;)a=l,(l=1+(l<<1))<=0&&(l=s);l>s&&(l=s),a+=r,l+=r}else{for(s=r+1;ls&&(l=s);var u=a;a=r-l,l=r-u}for(a++;as&&(l=s);var u=a;a=r-l,l=r-u}else{for(s=i-r;l=0;)a=l,(l=1+(l<<1))<=0&&(l=s);l>s&&(l=s),a+=r,l+=r}for(a++;a=0;l--)t[d+l]=t[p+l];return void(t[c]=a[h])}var f=r;for(;;){var g=0,y=0,v=!1;do{if(e(a[h],t[u])<0){if(t[c--]=t[u--],g++,y=0,0==--i){v=!0;break}}else if(t[c--]=a[h--],y++,g=0,1==--s){v=!0;break}}while((g|y)1;){var t=o-2;if(t>=1&&i[t-1]<=i[t]+i[t+1]||t>=2&&i[t-2]<=i[t]+i[t-1])i[t-1]i[t+1])break;s(t)}},forceMergeRuns:function(){for(;o>1;){var t=o-2;t>0&&i[t-1]=32;)e|=1&t,t>>=1;return t+e}(r);do{if((o=Ze(t,n,i,e))s&&(l=s),je(t,n,n+l,n+o,e),o=l}a.pushRun(n,o),a.mergeRuns(),r-=o,n+=o}while(0!==r);a.forceMergeRuns()}}}var Qe=!1;function tn(){Qe||(Qe=!0,console.warn("z / z2 / zlevel of displayable is invalid, which may cause unexpected errors"))}function en(t,e){return t.zlevel===e.zlevel?t.z===e.z?t.z2-e.z2:t.z-e.z:t.zlevel-e.zlevel}var nn=function(){function t(){this._roots=[],this._displayList=[],this._displayListLen=0,this.displayableSortFunc=en}return t.prototype.traverse=function(t,e){for(var n=0;n-1e-4}function pi(t){return li(1e3*t)/1e3}function di(t){return li(1e4*t)/1e4}var fi={left:"start",right:"end",center:"middle",middle:"middle"};function gi(t){return t&&!!t.image}function yi(t){return gi(t)||function(t){return t&&!!t.svgElement}(t)}function vi(t){return"linear"===t.type}function mi(t){return"radial"===t.type}function xi(t){return t&&("linear"===t.type||"radial"===t.type)}function _i(t){return"url(#"+t+")"}function bi(t){var e=t.getGlobalScale(),n=Math.max(e[0],e[1]);return Math.max(Math.ceil(Math.log(n)/Math.log(10)),1)}function wi(t){var e=t.x||0,n=t.y||0,i=(t.rotation||0)*bt,r=it(t.scaleX,1),o=it(t.scaleY,1),a=t.skewX||0,s=t.skewY||0,l=[];return(e||n)&&l.push("translate("+e+"px,"+n+"px)"),i&&l.push("rotate("+i+")"),1===r&&1===o||l.push("scale("+r+","+o+")"),(a||s)&&l.push("skew("+li(a*bt)+"deg, "+li(s*bt)+"deg)"),l.join(" ")}var Si=i.hasGlobalWindow&&Y(window.btoa)?function(t){return window.btoa(unescape(encodeURIComponent(t)))}:"undefined"!=typeof Buffer?function(t){return Buffer.from(t).toString("base64")}:function(t){return null},Mi=Array.prototype.slice;function Ii(t,e,n){return(e-t)*n+t}function Ti(t,e,n,i){for(var r=e.length,o=0;o i?e:t,o=Math.min(n,i),a=r[o-1]||{color:[0,0,0,0],offset:0},s=o;s
=a)}}for(var h=this.__startIndex;h