diff --git a/Makefile b/Makefile index d5c04a6..e2c203a 100644 --- a/Makefile +++ b/Makefile @@ -42,13 +42,46 @@ $(BINARY): $(GO_SRCS) $(FRONTEND_OUT) 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" +# +# Auto-detection: if EPICS_BASE is set in the environment, CGO flags are +# derived automatically. Override any variable on the command line: +# +# make backend-epics EPICS_BASE=/path/to/epics/base EPICS_ARCH=linux-x86_64 +# +# If EPICS_BASE is not set, set it to the directory that contains +# include/cadef.h and lib//libca.so. + +EPICS_BASE ?= $(EPICS_BASE) + +# Detect host arch from EPICS startup script if not provided. +EPICS_ARCH ?= $(shell \ + if [ -x "$(EPICS_BASE)/startup/EpicsHostArch" ]; then \ + "$(EPICS_BASE)/startup/EpicsHostArch"; \ + elif [ -x "$(EPICS_BASE)/lib" ]; then \ + ls "$(EPICS_BASE)/lib" | grep linux | head -1; \ + else \ + echo "linux-x86_64"; \ + fi) + +EPICS_CGO_CFLAGS ?= -I$(EPICS_BASE)/include -I$(EPICS_BASE)/include/os/Linux -I$(EPICS_BASE)/include/compiler/gcc +EPICS_CGO_LDFLAGS ?= -L$(EPICS_BASE)/lib/$(EPICS_ARCH) -lca -lCom -Wl,-rpath,$(EPICS_BASE)/lib/$(EPICS_ARCH) + backend-epics: $(FRONTEND_OUT) + @if [ -z "$(EPICS_BASE)" ]; then \ + echo "ERROR: EPICS_BASE is not set."; \ + echo " export EPICS_BASE=/path/to/epics/base"; \ + echo " then re-run: make backend-epics"; \ + exit 1; \ + fi + @echo "Building with EPICS support" + @echo " EPICS_BASE = $(EPICS_BASE)" + @echo " EPICS_ARCH = $(EPICS_ARCH)" @mkdir -p dist - CGO_ENABLED=1 go build -tags epics -ldflags="-s -w" -o $(BINARY) ./cmd/uopi + CGO_ENABLED=1 \ + CGO_CFLAGS="$(EPICS_CGO_CFLAGS)" \ + CGO_LDFLAGS="$(EPICS_CGO_LDFLAGS)" \ + go build -tags epics -ldflags="-s -w" -o $(BINARY) ./cmd/uopi + @echo "Built: $(BINARY) (EPICS CA enabled)" # Unstripped binary for debugging backend-debug: $(FRONTEND_OUT) diff --git a/cmd/uopi/main.go b/cmd/uopi/main.go index b68d706..661a271 100644 --- a/cmd/uopi/main.go +++ b/cmd/uopi/main.go @@ -59,19 +59,21 @@ func main() { brk := broker.New(ctx, log) - // 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) + // Stub data source: built-in simulated signals (sine, ramp, noise, setpoint). + // Enabled by default; disable via [datasource.stub] enabled = false in config. + if cfg.Datasource.Stub.Enabled { + stubDS := stub.New() + if err := stubDS.Connect(ctx); err != nil { + log.Error("stub connect", "err", err) + os.Exit(1) + } + brk.Register(stubDS) } - 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 { + if cfg.Datasource.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) @@ -83,8 +85,8 @@ func main() { // 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 cfg.Datasource.EPICS.Enabled && epics.Available() { + ds := epics.New(cfg.Datasource.EPICS.CAAddrList, cfg.Datasource.EPICS.ArchiveURL) if err := ds.Connect(ctx); err != nil { log.Error("epics connect", "err", err) } else { @@ -92,7 +94,7 @@ func main() { } } - srv := server.New(cfg.Server.Listen, webFS, brk, synthDS, store, cfg.EPICS.ChannelFinderURL, log) + srv := server.New(cfg.Server.Listen, webFS, brk, synthDS, store, cfg.Datasource.EPICS.ChannelFinderURL, log) if err := srv.Start(ctx); err != nil { fmt.Fprintf(os.Stderr, "server error: %v\n", err) diff --git a/internal/broker/broker.go b/internal/broker/broker.go index 231dd28..7cb5b18 100644 --- a/internal/broker/broker.go +++ b/internal/broker/broker.go @@ -86,23 +86,47 @@ func (b *Broker) Source(name string) (datasource.DataSource, bool) { // If this is the first subscriber for the signal, the upstream DS subscription // is started. Subsequent calls for the same signal share the existing one. // The returned func must be called when the client no longer needs the signal. +// +// The broker mutex is NOT held while the upstream ds.Subscribe call is in +// progress — some data sources (e.g. EPICS) block until the channel connects +// or times out. Releasing the lock allows other goroutines to proceed. func (b *Broker) Subscribe(ref SignalRef, ch chan<- Update) (func(), error) { b.mu.Lock() - defer b.mu.Unlock() sub, exists := b.subs[ref] - if !exists { - ds, found := b.sources[ref.DS] - if !found { - return nil, fmt.Errorf("unknown data source %q", ref.DS) - } + if exists { + // Fast path: already subscribed upstream; just add this client. + sub.mu.Lock() + sub.clients[ch] = struct{}{} + sub.mu.Unlock() + b.mu.Unlock() + return func() { b.unsubscribe(ref, ch) }, nil + } - rawCh := make(chan datasource.Value, 32) - dsCancel, err := ds.Subscribe(b.ctx, ref.Name, rawCh) - if err != nil { - return nil, fmt.Errorf("subscribe %s/%s: %w", ref.DS, ref.Name, err) - } + // Slow path: need to start an upstream subscription. + // Release the lock before the potentially-blocking ds.Subscribe call. + ds, found := b.sources[ref.DS] + b.mu.Unlock() + if !found { + return nil, fmt.Errorf("unknown data source %q", ref.DS) + } + + rawCh := make(chan datasource.Value, 32) + dsCancel, err := ds.Subscribe(b.ctx, ref.Name, rawCh) + if err != nil { + return nil, fmt.Errorf("subscribe %s/%s: %w", ref.DS, ref.Name, err) + } + + // Re-acquire the lock to register the new subscription. + // Another goroutine might have registered the same ref while the lock + // was released; if so, use that subscription and discard ours. + b.mu.Lock() + if existing, ok := b.subs[ref]; ok { + // Race: someone else subscribed first. Discard duplicate upstream. + dsCancel() + sub = existing + } else { sub = &signalSub{ clients: make(map[chan<- Update]struct{}), done: make(chan struct{}), @@ -116,6 +140,7 @@ func (b *Broker) Subscribe(ref SignalRef, ch chan<- Update) (func(), error) { sub.mu.Lock() sub.clients[ch] = struct{}{} sub.mu.Unlock() + b.mu.Unlock() return func() { b.unsubscribe(ref, ch) }, nil } diff --git a/internal/config/config.go b/internal/config/config.go index 68ceaa6..93379d5 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -10,8 +10,7 @@ import ( type Config struct { Server ServerConfig `toml:"server"` - EPICS EPICSConfig `toml:"datasource.epics"` - Synthetic SyntheticConfig `toml:"datasource.synthetic"` + Datasource DatasourceConfig `toml:"datasource"` } type ServerConfig struct { @@ -19,6 +18,16 @@ type ServerConfig struct { StorageDir string `toml:"storage_dir"` } +type DatasourceConfig struct { + Stub StubConfig `toml:"stub"` + EPICS EPICSConfig `toml:"epics"` + Synthetic SyntheticConfig `toml:"synthetic"` +} + +type StubConfig struct { + Enabled bool `toml:"enabled"` +} + type EPICSConfig struct { Enabled bool `toml:"enabled"` CAAddrList string `toml:"ca_addr_list"` @@ -27,8 +36,7 @@ type EPICSConfig struct { } type SyntheticConfig struct { - Enabled bool `toml:"enabled"` - DefinitionsFile string `toml:"definitions_file"` + Enabled bool `toml:"enabled"` } func Default() Config { @@ -37,12 +45,10 @@ func Default() Config { Listen: ":8080", StorageDir: "./interfaces", }, - EPICS: EPICSConfig{ - Enabled: true, - }, - Synthetic: SyntheticConfig{ - Enabled: true, - DefinitionsFile: "./synthetic.json", + Datasource: DatasourceConfig{ + Stub: StubConfig{Enabled: true}, + EPICS: EPICSConfig{Enabled: true}, + Synthetic: SyntheticConfig{Enabled: true}, }, } } @@ -71,16 +77,13 @@ func applyEnv(cfg *Config) { cfg.Server.StorageDir = v } if v := env("UOPI_EPICS_CA_ADDR_LIST"); v != "" { - cfg.EPICS.CAAddrList = v + cfg.Datasource.EPICS.CAAddrList = v } if v := env("UOPI_EPICS_ARCHIVE_URL"); v != "" { - cfg.EPICS.ArchiveURL = v + cfg.Datasource.EPICS.ArchiveURL = v } if v := env("UOPI_EPICS_CHANNEL_FINDER_URL"); v != "" { - cfg.EPICS.ChannelFinderURL = v - } - if v := env("UOPI_SYNTHETIC_DEFINITIONS_FILE"); v != "" { - cfg.Synthetic.DefinitionsFile = v + cfg.Datasource.EPICS.ChannelFinderURL = v } } diff --git a/internal/datasource/epics/ca_callback.go b/internal/datasource/epics/ca_callback.go index cde92c7..88c25a8 100644 --- a/internal/datasource/epics/ca_callback.go +++ b/internal/datasource/epics/ca_callback.go @@ -3,6 +3,7 @@ package epics /* +#cgo CFLAGS: -Wno-unused-function #include "ca_wrapper.h" */ import "C" diff --git a/internal/datasource/epics/ca_wrapper.h b/internal/datasource/epics/ca_wrapper.h index 3c309de..830870e 100644 --- a/internal/datasource/epics/ca_wrapper.h +++ b/internal/datasource/epics/ca_wrapper.h @@ -13,7 +13,7 @@ * goCAConnectionCallback is called when a channel connects or disconnects. */ extern void goCAMonitorCallback(uintptr_t handle, int dbrType, long count, - const void *dbr, int severity, + void *dbr, int severity, double epicsTimeSecs); extern void goCAConnectionCallback(uintptr_t handle, int connected); @@ -78,7 +78,7 @@ static void caMonitorCallbackShim(struct event_handler_args args) { } goCAMonitorCallback((uintptr_t)args.usr, (int)args.type, (long)args.count, - args.dbr, severity, timeSecs); + (void *)args.dbr, severity, timeSecs); } /* @@ -115,4 +115,29 @@ static int caAddMonitor(chid ch, short dbrType, uintptr_t userHandle, (void *)userHandle, pEvid); } +/* Wrappers for ca_get / ca_put, which are macros in cadef.h that expand to + * ca_array_get / ca_array_put with count=1. CGo cannot call macros directly, + * so we provide thin static-inline shims here. */ + +static int caGet(chtype type, chid chan, void *pValue) { + return ca_array_get(type, 1u, chan, pValue); +} + +static int caPut(chtype type, chid chan, const void *pValue) { + return ca_array_put(type, 1u, chan, pValue); +} + +/* Retrieve the current thread's CA context as an opaque pointer. + * Store the result and pass it to caAttachContext() from other threads. */ +static void *caCurrentContext(void) { + return (void *)ca_current_context(); +} + +/* Attach an existing CA context to the calling thread. Must be called at the + * start of every goroutine (OS thread) that makes CA calls, if that thread did + * not call ca_context_create() itself. Returns ECA_NORMAL on success. */ +static int caAttachContext(void *ctx) { + return ca_attach_context((struct ca_client_context *)ctx); +} + #endif /* CA_WRAPPER_H */ diff --git a/internal/datasource/epics/epics.go b/internal/datasource/epics/epics.go index 023c903..ae4ebb2 100644 --- a/internal/datasource/epics/epics.go +++ b/internal/datasource/epics/epics.go @@ -17,7 +17,7 @@ package epics /* -#cgo CFLAGS: -Wall +#cgo CFLAGS: -Wall -Wno-unused-function #include "ca_wrapper.h" #include */ @@ -26,6 +26,7 @@ import "C" import ( "context" "fmt" + "os" "sync" "sync/atomic" "time" @@ -68,6 +69,12 @@ type EPICS struct { caAddrList string archiveURL string + // caCtx is the CA context created in Connect(). Stored as unsafe.Pointer + // because the C type (ca_client_context *) is opaque. Every goroutine + // that calls CA functions must call caAttachContext(caCtx) first, because + // Go goroutines can run on any OS thread and CA contexts are thread-local. + caCtx unsafe.Pointer + mu sync.Mutex channels map[string]*caChannel // PV name → CA channel metadata map[string]datasource.Metadata @@ -98,27 +105,43 @@ func (e *EPICS) Name() string { return "epics" } // callbacks are delivered on CA's background threads without needing a // ca_pend_event polling loop. func (e *EPICS) Connect(_ context.Context) error { - // Optionally override EPICS_CA_ADDR_LIST at runtime. + // Apply CA address list overrides before creating the context so the + // library picks them up during initialisation. + // os.Setenv is used instead of C.putenv: putenv takes ownership of the + // string pointer, which would race with Go's garbage collector. if e.caAddrList != "" { - cs := C.CString(e.caAddrList) - defer C.free(unsafe.Pointer(cs)) - // ca_setenv is not available in all EPICS versions; use putenv via C. - envStr := C.CString("EPICS_CA_ADDR_LIST=" + e.caAddrList) - defer C.free(unsafe.Pointer(envStr)) - C.putenv(envStr) + os.Setenv("EPICS_CA_ADDR_LIST", e.caAddrList) + os.Setenv("EPICS_CA_AUTO_ADDR_LIST", "NO") } status := C.ca_context_create(C.ca_enable_preemptive_callback) if status != C.ECA_NORMAL { return fmt.Errorf("epics: ca_context_create failed: status %d", int(status)) } + // Save the context so goroutines on other OS threads can attach to it. + e.caCtx = C.caCurrentContext() return nil } -// Subscribe connects to the named PV (if not already connected), installs a -// monitor, and streams value updates into ch. The returned CancelFunc removes -// the monitor and clears the channel. +// attachCAContext attaches the saved CA context to the calling OS thread. +// Must be called at the top of every goroutine that uses CA functions, +// because Go goroutines can be scheduled onto any OS thread. +// Safe to call redundantly (ECA_ISATTACHED is not an error). +func (e *EPICS) attachCAContext() { + if e.caCtx != nil { + C.caAttachContext(e.caCtx) + } +} + +// Subscribe creates a CA channel for signal and returns immediately. +// Connection and monitor setup happen in a background goroutine so that +// the caller is never blocked waiting for the IOC to respond. +// A QualityBad value is sent to ch if the channel cannot connect within +// 60 seconds. The returned CancelFunc tears down the subscription. func (e *EPICS) Subscribe(ctx context.Context, signal string, ch chan<- datasource.Value) (datasource.CancelFunc, error) { + // Attach the CA context to this OS thread before any CA calls. + e.attachCAContext() + handle := nextHandle() // Register the handle → channel mapping before creating the CA channel so @@ -129,7 +152,8 @@ func (e *EPICS) Subscribe(ctx context.Context, signal string, ch chan<- datasour connTable[handle] = entry handleMu.Unlock() - // Create the CA channel. + // Create the CA channel — this call returns immediately; the connection + // callback fires asynchronously on CA's internal thread. pvName := C.CString(signal) defer C.free(unsafe.Pointer(pvName)) @@ -142,54 +166,78 @@ func (e *EPICS) Subscribe(ctx context.Context, signal string, ch chan<- datasour handleMu.Unlock() return nil, fmt.Errorf("epics: ca_create_channel(%q) failed: status %d", signal, int(status)) } - - // Flush the create request and wait for the connection callback. C.ca_flush_io() - select { - case <-entry.connCh: - // Connected. - case <-time.After(5 * time.Second): - // Timeout; clean up and report error. - C.ca_clear_channel(chid) - handleMu.Lock() - delete(handleTable, handle) - delete(connTable, handle) - handleMu.Unlock() - return nil, fmt.Errorf("epics: timeout waiting for channel %q to connect", signal) - case <-ctx.Done(): - C.ca_clear_channel(chid) - handleMu.Lock() - delete(handleTable, handle) - delete(connTable, handle) - handleMu.Unlock() - return nil, ctx.Err() - } - - // Determine the native DBR_TIME_* type for the channel. - dbrType := nativeTimeType(chid) - - // Add the value monitor. - var evid C.evid - status = C.caAddMonitor(chid, C.short(dbrType), C.uintptr_t(handle), &evid) - if status != C.ECA_NORMAL { - C.ca_clear_channel(chid) - handleMu.Lock() - delete(handleTable, handle) - delete(connTable, handle) - handleMu.Unlock() - return nil, fmt.Errorf("epics: ca_add_event(%q) failed: status %d", signal, int(status)) - } - C.ca_flush_io() - - caCh := &caChannel{chid: chid, evid: evid, handle: handle} - + // Track the channel immediately so Write can find the chid. e.mu.Lock() - e.channels[signal] = caCh + e.channels[signal] = &caChannel{chid: chid, handle: handle} e.mu.Unlock() - cancel := datasource.CancelFunc(func() { - if evid != nil { + // Use a done channel to signal the goroutine to stop. + done := make(chan struct{}) + var closeOnce sync.Once + stopFn := func() { closeOnce.Do(func() { close(done) }) } + + // Background goroutine: wait for connection, then install the monitor. + go func() { + // Attach CA context to this OS thread before making any CA calls. + e.attachCAContext() + + var evid C.evid + connected := false + + select { + case <-entry.connCh: + connected = true + case <-time.After(60 * time.Second): + // IOC unreachable; report bad quality so the widget shows an error state. + select { + case ch <- datasource.Value{Timestamp: time.Now(), Data: float64(0), Quality: datasource.QualityBad}: + default: + } + case <-ctx.Done(): + case <-done: + } + + if connected { + // Refresh metadata now that the channel is connected so type, unit, + // and display limits are accurate (the earlier sendMeta call may have + // run before the channel was up, leaving an incomplete cache entry). + meta := e.fetchMetadata(chid, signal) + e.mu.Lock() + e.metadata[signal] = meta + e.mu.Unlock() + + // Signal to the broker/dispatcher that metadata has been refreshed. + select { + case ch <- datasource.Value{MetaUpdate: true}: + default: + } + + dbrType := nativeTimeType(chid) + st := C.caAddMonitor(chid, C.short(dbrType), C.uintptr_t(handle), &evid) + if st != C.ECA_NORMAL { + select { + case ch <- datasource.Value{Timestamp: time.Now(), Data: float64(0), Quality: datasource.QualityBad}: + default: + } + connected = false + } else { + C.ca_flush_io() + e.mu.Lock() + e.channels[signal] = &caChannel{chid: chid, evid: evid, handle: handle} + e.mu.Unlock() + } + } + + // Wait until the subscription is cancelled. + select { + case <-done: + case <-ctx.Done(): + } + + // Cleanup. + if connected && evid != nil { C.ca_clear_event(evid) } C.ca_clear_channel(chid) @@ -203,15 +251,18 @@ func (e *EPICS) Subscribe(ctx context.Context, signal string, ch chan<- datasour e.mu.Lock() delete(e.channels, signal) e.mu.Unlock() - }) + }() - return cancel, nil + return datasource.CancelFunc(stopFn), nil } // GetMetadata performs a synchronous ca_get for DBR_CTRL_DOUBLE (or the // appropriate control type) to retrieve units, display limits, enum strings, // and writability information for the named signal. func (e *EPICS) GetMetadata(ctx context.Context, signal string) (datasource.Metadata, error) { + // Attach the CA context to this OS thread before any CA calls. + e.attachCAContext() + // Check cache first. e.mu.Lock() if m, ok := e.metadata[signal]; ok { @@ -289,7 +340,10 @@ func (e *EPICS) GetMetadata(ctx context.Context, signal string) (datasource.Meta // fetchMetadata retrieves metadata from a connected chid using ca_get. func (e *EPICS) fetchMetadata(chid C.chid, name string) datasource.Metadata { - meta := datasource.Metadata{Name: name} + // In EPICS, write access is governed by CA security (host/user rules), not + // by the field type. Default to writable=true; the IOC will reject puts + // that violate its security policy. + meta := datasource.Metadata{Name: name, Writable: true} fieldType := C.ca_field_type(chid) count := C.ca_element_count(chid) @@ -326,7 +380,7 @@ func (e *EPICS) fetchMetadata(chid C.chid, name string) datasource.Metadata { switch ctrlType { case C.DBR_CTRL_DOUBLE: var buf C.struct_dbr_ctrl_double - status := C.ca_get(C.DBR_CTRL_DOUBLE, chid, unsafe.Pointer(&buf)) + status := C.caGet(C.DBR_CTRL_DOUBLE, chid, unsafe.Pointer(&buf)) if status == C.ECA_NORMAL { C.ca_pend_io(3.0) meta.Unit = C.GoString((*C.char)(unsafe.Pointer(&buf.units[0]))) @@ -334,14 +388,11 @@ func (e *EPICS) fetchMetadata(chid C.chid, name string) datasource.Metadata { meta.DisplayHigh = float64(buf.upper_disp_limit) meta.DriveLow = float64(buf.lower_ctrl_limit) meta.DriveHigh = float64(buf.upper_ctrl_limit) - // CA does not expose a writable flag directly; assume writable unless - // it is a read-only field type. Conservatively mark as writable. - meta.Writable = true } case C.DBR_CTRL_LONG: var buf C.struct_dbr_ctrl_long - status := C.ca_get(C.DBR_CTRL_LONG, chid, unsafe.Pointer(&buf)) + status := C.caGet(C.DBR_CTRL_LONG, chid, unsafe.Pointer(&buf)) if status == C.ECA_NORMAL { C.ca_pend_io(3.0) meta.Unit = C.GoString((*C.char)(unsafe.Pointer(&buf.units[0]))) @@ -349,12 +400,11 @@ func (e *EPICS) fetchMetadata(chid C.chid, name string) datasource.Metadata { meta.DisplayHigh = float64(buf.upper_disp_limit) meta.DriveLow = float64(buf.lower_ctrl_limit) meta.DriveHigh = float64(buf.upper_ctrl_limit) - meta.Writable = true } case C.DBR_CTRL_ENUM: var buf C.struct_dbr_ctrl_enum - status := C.ca_get(C.DBR_CTRL_ENUM, chid, unsafe.Pointer(&buf)) + status := C.caGet(C.DBR_CTRL_ENUM, chid, unsafe.Pointer(&buf)) if status == C.ECA_NORMAL { C.ca_pend_io(3.0) n := int(buf.no_str) @@ -363,28 +413,39 @@ func (e *EPICS) fetchMetadata(chid C.chid, name string) datasource.Metadata { strs[i] = C.GoString((*C.char)(unsafe.Pointer(&buf.strs[i][0]))) } meta.EnumStrings = strs - meta.Writable = true } } return meta } -// ListSignals returns cached metadata for all currently connected channels. -// Full enumeration of all available PVs is deferred to Phase 9. +// ListSignals returns metadata for all currently tracked channels. +// Channels with cached metadata return full info; others return a stub entry +// with the PV name so the signal tree can display them immediately after +// a manual add, before GetMetadata has been called. func (e *EPICS) ListSignals(_ context.Context) ([]datasource.Metadata, error) { e.mu.Lock() defer e.mu.Unlock() - out := make([]datasource.Metadata, 0, len(e.metadata)) - for _, m := range e.metadata { + // Start with all fully-fetched metadata entries. + out := make([]datasource.Metadata, 0, len(e.channels)) + seen := make(map[string]bool, len(e.metadata)) + for name, m := range e.metadata { out = append(out, m) + seen[name] = true + } + // Add stub entries for channels that are tracked but not yet fetched. + for name := range e.channels { + if !seen[name] { + out = append(out, datasource.Metadata{Name: name, Type: datasource.TypeFloat64}) + } } return out, nil } // Write puts a new value onto a CA channel. func (e *EPICS) Write(_ context.Context, signal string, value any) error { + e.attachCAContext() e.mu.Lock() caCh, ok := e.channels[signal] e.mu.Unlock() @@ -396,20 +457,20 @@ func (e *EPICS) Write(_ context.Context, signal string, value any) error { switch v := value.(type) { case float64: cv := C.double(v) - status = C.ca_put(C.DBR_DOUBLE, caCh.chid, unsafe.Pointer(&cv)) + status = C.caPut(C.DBR_DOUBLE, caCh.chid, unsafe.Pointer(&cv)) case int64: cv := C.long(v) - status = C.ca_put(C.DBR_LONG, caCh.chid, unsafe.Pointer(&cv)) + status = C.caPut(C.DBR_LONG, caCh.chid, unsafe.Pointer(&cv)) case string: cs := C.CString(v) defer C.free(unsafe.Pointer(cs)) - status = C.ca_put(C.DBR_STRING, caCh.chid, unsafe.Pointer(cs)) + status = C.caPut(C.DBR_STRING, caCh.chid, unsafe.Pointer(cs)) case bool: var iv C.short if v { iv = 1 } - status = C.ca_put(C.DBR_SHORT, caCh.chid, unsafe.Pointer(&iv)) + status = C.caPut(C.DBR_SHORT, caCh.chid, unsafe.Pointer(&iv)) default: return fmt.Errorf("epics: unsupported value type %T for Write", value) } diff --git a/internal/datasource/iface.go b/internal/datasource/iface.go index 00e46ce..e321b3c 100644 --- a/internal/datasource/iface.go +++ b/internal/datasource/iface.go @@ -42,9 +42,10 @@ const ( // Value is a timestamped signal reading. type Value struct { - Timestamp time.Time - Data any // float64 | []float64 | string | int64 | bool - Quality Quality + Timestamp time.Time + Data any // float64 | []float64 | string | int64 | bool + Quality Quality + MetaUpdate bool // if true, metadata was refreshed — dispatcher should re-send meta } // Metadata describes the static properties of a signal. diff --git a/internal/server/ws.go b/internal/server/ws.go index 67a8b34..51a0ea0 100644 --- a/internal/server/ws.go +++ b/internal/server/ws.go @@ -163,6 +163,12 @@ func (c *wsClient) dispatchLoop(ctx context.Context) { for { select { case u := <-c.updateCh: + if u.Value.MetaUpdate { + // Datasource refreshed metadata (e.g. after channel connected). + // Re-send the meta message so the client gets accurate type/unit info. + c.sendMeta(ctx, u.Ref) + continue + } msg := outMsg{ Type: "update", DS: u.Ref.DS, @@ -265,6 +271,7 @@ func (c *wsClient) handleUnsubscribe(refs []sigRef) { func (c *wsClient) handleWrite(ctx context.Context, msg inMsg) { ds, ok := c.broker.Source(msg.DS) if !ok { + c.log.Warn("write: unknown data source", "ds", msg.DS, "name", msg.Name) c.sendError(ctx, "NOT_FOUND", "unknown data source: "+msg.DS) return } @@ -272,10 +279,12 @@ func (c *wsClient) handleWrite(ctx context.Context, msg inMsg) { // Enforce write permission before touching the value payload. meta, err := ds.GetMetadata(ctx, msg.Name) if err != nil { + c.log.Warn("write: GetMetadata failed", "ds", msg.DS, "name", msg.Name, "err", err) c.sendError(ctx, "NOT_FOUND", "unknown signal: "+msg.Name) return } if !meta.Writable { + c.log.Warn("write: signal not writable", "ds", msg.DS, "name", msg.Name) c.sendError(ctx, "NOT_WRITABLE", "signal is read-only: "+msg.Name) return } @@ -283,12 +292,15 @@ func (c *wsClient) handleWrite(ctx context.Context, msg inMsg) { // Decode the JSON value as the appropriate Go type. var value any if err := json.Unmarshal(msg.Value, &value); err != nil { + c.log.Warn("write: parse error", "ds", msg.DS, "name", msg.Name, "err", err) c.sendError(ctx, "PARSE_ERROR", "invalid value: "+err.Error()) return } + c.log.Info("write", "ds", msg.DS, "name", msg.Name, "value", value) metrics.IncWrites() if err := ds.Write(ctx, msg.Name, value); err != nil { + c.log.Warn("write: ds.Write failed", "ds", msg.DS, "name", msg.Name, "err", err) c.sendError(ctx, "WRITE_ERROR", err.Error()) } } diff --git a/uopi b/uopi index 4d84378..4908105 100755 Binary files a/uopi and b/uopi differ diff --git a/web/src/App.tsx b/web/src/App.tsx index a61141d..2863804 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -3,10 +3,14 @@ import { useState, useEffect } from 'preact/hooks'; import { wsClient } from './lib/ws'; import ViewMode from './ViewMode'; import EditMode from './EditMode'; +import FullscreenMode from './FullscreenMode'; import type { Interface } from './lib/types'; type AppMode = 'view' | 'edit'; +// If URL has ?fs=, render bare fullscreen canvas +const fsParam = new URLSearchParams(window.location.search).get('fs'); + export default function App() { const [mode, setMode] = useState('view'); const [wsStatus, setWsStatus] = useState('connecting'); @@ -20,9 +24,13 @@ export default function App() { useEffect(() => { const dpr = window.devicePixelRatio ?? 1; document.documentElement.style.setProperty('--dpr', String(dpr)); - wsClient.connect('/ws'); + if (!fsParam) wsClient.connect('/ws'); }, []); + if (fsParam) { + return ; + } + function enterEdit(iface?: Interface) { setEditTarget(iface ?? null); setMode('edit'); diff --git a/web/src/EditCanvas.tsx b/web/src/EditCanvas.tsx index 6a41417..eedf8d9 100644 --- a/web/src/EditCanvas.tsx +++ b/web/src/EditCanvas.tsx @@ -37,6 +37,11 @@ export const DEFAULT_SIZES: Record = { link: [140, 50], }; +// Widget types that support multiple signals (signal can be appended) +const MULTI_SIGNAL_TYPES = new Set(['plot', 'multiled']); +// Widget types that have exactly one signal (signal can be replaced) +const SINGLE_SIGNAL_TYPES = new Set(['textview', 'gauge', 'barh', 'barv', 'led', 'setvalue', 'button']); + interface PickerState { visible: boolean; x: number; @@ -46,6 +51,14 @@ interface PickerState { signal: SignalRef | null; } +interface SignalDropMenuState { + visible: boolean; + x: number; + y: number; + signal: SignalRef; + targetWidget: Widget; +} + interface DragState { startMouseX: number; startMouseY: number; @@ -65,9 +78,9 @@ interface ResizeState { interface RubberBand { active: boolean; - startX: number; startY: number; // canvas-relative + startX: number; startY: number; curX: number; curY: number; - clientStartX: number; clientStartY: number; // for scroll adjustment + clientStartX: number; clientStartY: number; } interface Props { @@ -75,7 +88,7 @@ interface Props { selectedIds: string[]; onSelect: (ids: string[]) => void; onChange: (widgets: Widget[]) => void; - snapGrid: number; // 0 = off + snapGrid: number; } let _widgetSeq = Date.now(); @@ -95,7 +108,6 @@ function rectsIntersect( export default function EditCanvas({ iface, selectedIds, onSelect, onChange, snapGrid }: Props) { const containerRef = useRef(null); - // Refs: always hold current values so event handlers don't go stale const ifaceRef = useRef(iface); ifaceRef.current = iface; const onChangeRef = useRef(onChange); @@ -111,14 +123,14 @@ export default function EditCanvas({ iface, selectedIds, onSelect, onChange, sna const resizeRef = useRef(null); const rubberRef = useRef({ active: false, startX: 0, startY: 0, curX: 0, curY: 0, clientStartX: 0, clientStartY: 0 }); - // Rubber-band visual state (drives re-renders during drag) const [rubberVis, setRubberVis] = useState<{ x: number; y: number; w: number; h: number } | null>(null); const [picker, setPicker] = useState({ visible: false, x: 0, y: 0, canvasX: 0, canvasY: 0, signal: null, }); - // Set up document-level mouse handlers once + const [signalDropMenu, setSignalDropMenu] = useState(null); + useEffect(() => { function onMouseMove(e: MouseEvent) { const grid = snapGridRef.current; @@ -213,7 +225,7 @@ export default function EditCanvas({ iface, selectedIds, onSelect, onChange, sna function handleCanvasMouseDown(e: MouseEvent) { if ((e.target as Element).closest('.widget-overlay')) return; if ((e.target as Element).closest('.widget-picker-backdrop')) return; - // Start rubber-band on empty canvas + if ((e.target as Element).closest('.signal-drop-backdrop')) return; const { cx, cy } = getCanvasPos(e); rubberRef.current = { active: true, startX: cx, startY: cy, curX: cx, curY: cy, clientStartX: e.clientX, clientStartY: e.clientY }; if (!e.ctrlKey && !e.metaKey) { @@ -232,10 +244,50 @@ export default function EditCanvas({ iface, selectedIds, onSelect, onChange, sna if (!json) return; let sig: SignalRef; try { sig = JSON.parse(json); } catch { return; } + const { cx, cy } = getCanvasPos(e); + + // Check if drop lands on an existing widget + const hit = iface.widgets.find(w => + cx >= w.x && cx <= w.x + w.w && cy >= w.y && cy <= w.y + w.h + ); + + if (hit && (MULTI_SIGNAL_TYPES.has(hit.type) || SINGLE_SIGNAL_TYPES.has(hit.type))) { + // Offer add/replace vs create-new + setSignalDropMenu({ visible: true, x: e.clientX, y: e.clientY, signal: sig, targetWidget: hit }); + return; + } + + // No existing widget hit — open type picker setPicker({ visible: true, x: e.clientX, y: e.clientY, canvasX: cx, canvasY: cy, signal: sig }); } + function handleAddToWidget(widget: Widget, sig: SignalRef) { + setSignalDropMenu(null); + let updated: Widget; + if (MULTI_SIGNAL_TYPES.has(widget.type)) { + // Append signal (avoid duplicates) + const already = widget.signals.some(s => s.ds === sig.ds && s.name === sig.name); + if (already) return; + updated = { ...widget, signals: [...widget.signals, sig] }; + } else { + // Replace first signal + updated = { ...widget, signals: [sig, ...widget.signals.slice(1)] }; + } + onChange(iface.widgets.map(w => w.id === updated.id ? updated : w)); + onSelect([updated.id]); + } + + function handleSignalDropCreateNew(sig: SignalRef, cx: number, cy: number) { + setSignalDropMenu(null); + // We don't have canvas coords in the menu state, approximate from screen coords + const el = containerRef.current; + const rect = el ? el.getBoundingClientRect() : { left: 0, top: 0 }; + const canvasX = cx - rect.left + (el?.scrollLeft ?? 0); + const canvasY = cy - rect.top + (el?.scrollTop ?? 0); + setPicker({ visible: true, x: cx, y: cy, canvasX, canvasY, signal: sig }); + } + function handlePick(type: string) { if (!picker.signal) return; const [defW, defH] = DEFAULT_SIZES[type] ?? [150, 60]; @@ -255,10 +307,6 @@ export default function EditCanvas({ iface, selectedIds, onSelect, onChange, sna setPicker(p => ({ ...p, visible: false })); } - // Called from EditMode when inserting a static widget (no signal) - // This is triggered via the parent placing a widget directly. - // (handled via onInsert prop in extended usage) - function handleOverlayMouseDown(e: MouseEvent, widget: Widget) { e.stopPropagation(); const ctrl = e.ctrlKey || e.metaKey; @@ -274,7 +322,6 @@ export default function EditCanvas({ iface, selectedIds, onSelect, onChange, sna } onSelect(nextIds); - // Don't start drag on Ctrl+click deselect if (ctrl && isSelected) return; const movers = nextIds.includes(widget.id) ? nextIds : [widget.id]; @@ -317,7 +364,6 @@ export default function EditCanvas({ iface, selectedIds, onSelect, onChange, sna >
- {/* Live widget previews */} {iface.widgets.map(widget => { const Comp = COMPONENTS[widget.type]; return Comp @@ -330,7 +376,6 @@ export default function EditCanvas({ iface, selectedIds, onSelect, onChange, sna ); })} - {/* Interaction overlays */} {iface.widgets.map(widget => { const isSelected = selectedIds.includes(widget.id); return ( @@ -361,7 +406,6 @@ export default function EditCanvas({ iface, selectedIds, onSelect, onChange, sna ); })} - {/* Rubber-band selection rectangle */} {rubberVis && (
setPicker(p => ({ ...p, visible: false }))} /> )} + + {/* Signal drop menu: appears when dropping onto an existing widget */} + {signalDropMenu && ( +
setSignalDropMenu(null)}> +
e.stopPropagation()} + > +
+ Drop signal {signalDropMenu.signal.name} +
+ {MULTI_SIGNAL_TYPES.has(signalDropMenu.targetWidget.type) && ( + + )} + {SINGLE_SIGNAL_TYPES.has(signalDropMenu.targetWidget.type) && ( + + )} + + +
+
+ )}
); } diff --git a/web/src/FullscreenMode.tsx b/web/src/FullscreenMode.tsx new file mode 100644 index 0000000..c29f6be --- /dev/null +++ b/web/src/FullscreenMode.tsx @@ -0,0 +1,46 @@ +import { h } from 'preact'; +import { useState, useEffect } from 'preact/hooks'; +import { wsClient } from './lib/ws'; +import { parseInterface } from './lib/xml'; +import type { Interface } from './lib/types'; +import Canvas from './Canvas'; + +interface Props { + interfaceId: string; +} + +export default function FullscreenMode({ interfaceId }: Props) { + const [iface, setIface] = useState(null); + const [error, setError] = useState(null); + + useEffect(() => { + wsClient.connect('/ws'); + fetch(`/api/v1/interfaces/${encodeURIComponent(interfaceId)}`) + .then(r => { + if (!r.ok) throw new Error(`HTTP ${r.status}`); + return r.text(); + }) + .then(xml => setIface(parseInterface(xml))) + .catch(err => setError(err instanceof Error ? err.message : String(err))); + }, [interfaceId]); + + if (error) { + return ( +
+

Failed to load interface: {error}

+
+ ); + } + + return ( +
+ { + try { + const res = await fetch(`/api/v1/interfaces/${encodeURIComponent(id)}`); + if (!res.ok) throw new Error(`HTTP ${res.status}`); + setIface(parseInterface(await res.text())); + } catch { /* ignore */ } + }} timeRange={null} /> +
+ ); +} diff --git a/web/src/InterfaceList.tsx b/web/src/InterfaceList.tsx index 8ad2ca6..5374c94 100644 --- a/web/src/InterfaceList.tsx +++ b/web/src/InterfaceList.tsx @@ -12,9 +12,10 @@ interface Props { onLoad: (xml: string) => void; onSelect?: (id: string) => void; onEdit?: (iface?: Interface) => void; + onEditId?: (id: string) => void; } -export default function InterfaceList({ onLoad, onSelect, onEdit }: Props) { +export default function InterfaceList({ onLoad, onSelect, onEdit, onEditId }: Props) { const [collapsed, setCollapsed] = useState(false); const [interfaces, setInterfaces] = useState([]); const [loading, setLoading] = useState(true); @@ -60,6 +61,10 @@ export default function InterfaceList({ onLoad, onSelect, onEdit }: Props) { if (res.ok) refresh(); } + function handleFullscreen(id: string) { + window.open(`?fs=${encodeURIComponent(id)}`, '_blank'); + } + return (