martino/uopi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Wworking on improving the tool

Browse Source
This commit is contained in:
Martino Ferrari
2026-05-21 07:41:56 +02:00
parent 6ff8fb5c25
commit 71430bc3b0
30 changed files with 1820 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/api.go
+71 -10
View File
@@ -7,6 +7,7 @@ import (
"io"
"log/slog"
"net/http"
"net/url"
"strings"
"github.com/uopi/uopi/internal/broker"
@@ -21,12 +22,20 @@ type Handler struct {
synthetic *synthetic.Synthetic // nil if not enabled
store *storage.Store
channelFinderURL string // empty if not configured
archiverURL string // empty if not configured
log *slog.Logger
}
// 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, channelFinderURL string, log *slog.Logger) *Handler {
return &Handler{broker: b, synthetic: synth, store: store, channelFinderURL: channelFinderURL, log: log}
func New(b *broker.Broker, synth *synthetic.Synthetic, store *storage.Store, channelFinderURL, archiverURL string, log *slog.Logger) *Handler {
return &Handler{
broker: b,
synthetic: synth,
store: store,
channelFinderURL: channelFinderURL,
archiverURL: archiverURL,
log: log,
}
}
// Register mounts all API routes onto mux under the given prefix.
@@ -36,6 +45,7 @@ func (h *Handler) Register(mux *http.ServeMux, prefix string) {
mux.HandleFunc("GET "+prefix+"/signals", h.listSignals)
mux.HandleFunc("GET "+prefix+"/signals/search", h.searchSignals)
mux.HandleFunc("GET "+prefix+"/channel-finder", h.channelFinder)
mux.HandleFunc("GET "+prefix+"/archiver/search", h.archiverSearch)
mux.HandleFunc("GET "+prefix+"/interfaces", h.listInterfaces)
mux.HandleFunc("POST "+prefix+"/interfaces", h.createInterface)
mux.HandleFunc("GET "+prefix+"/interfaces/{id}", h.getInterface)
@@ -72,14 +82,16 @@ func (h *Handler) listDataSources(w http.ResponseWriter, r *http.Request) {
// ── /signals ──────────────────────────────────────────────────────────────────
type signalInfo struct {
Name string `json:"name"`
Type string `json:"type"`
Unit string `json:"unit,omitempty"`
Description string `json:"description,omitempty"`
DisplayLow float64 `json:"displayLow"`
DisplayHigh float64 `json:"displayHigh"`
Writable bool `json:"writable"`
EnumStrings []string `json:"enumStrings,omitempty"`
Name string `json:"name"`
Type string `json:"type"`
Unit string `json:"unit,omitempty"`
Description string `json:"description,omitempty"`
DisplayLow float64 `json:"displayLow"`
DisplayHigh float64 `json:"displayHigh"`
Writable bool `json:"writable"`
EnumStrings []string `json:"enumStrings,omitempty"`
Properties map[string]string `json:"properties,omitempty"`
Tags []string `json:"tags,omitempty"`
}
func (h *Handler) listSignals(w http.ResponseWriter, r *http.Request) {
@@ -160,6 +172,12 @@ func (h *Handler) channelFinder(w http.ResponseWriter, r *http.Request) {
cfURL := strings.TrimRight(h.channelFinderURL, "/") + "/resources/channels"
if q != "" {
cfURL += "?~name=*" + q + "*"
} else if r.URL.RawQuery != "" && r.URL.RawQuery != "q=" {
cfURL += "?" + r.URL.RawQuery
} else {
// Empty query check: return empty results to signal CFS is configured.
jsonOK(w, []struct{}{})
return
}
resp, err := http.Get(cfURL) //nolint:noctx
@@ -187,6 +205,47 @@ func (h *Handler) channelFinder(w http.ResponseWriter, r *http.Request) {
jsonOK(w, names)
}
// ── /archiver/search ──────────────────────────────────────────────────────────
// archiverSearch proxies a search query to the EPICS Archive Appliance
// management API when configured. Returns 501 otherwise.
func (h *Handler) archiverSearch(w http.ResponseWriter, r *http.Request) {
if h.archiverURL == "" {
jsonError(w, http.StatusNotImplemented, "Archiver not configured (set archive_url in config)")
return
}
pattern := r.URL.Query().Get("q")
if pattern == "" {
pattern = "*"
}
// AA expects glob patterns. If no glob characters, wrap in stars.
if !strings.ContainsAny(pattern, "*?[]") {
pattern = "*" + pattern + "*"
}
searchURL := strings.TrimRight(h.archiverURL, "/") + "/mgmt/bpl/getAllPVs?pv=" + url.QueryEscape(pattern)
resp, err := http.Get(searchURL) //nolint:noctx
if err != nil {
jsonError(w, http.StatusBadGateway, "Archiver request failed: "+err.Error())
return
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
jsonError(w, http.StatusBadGateway, "Archiver returned status "+resp.Status)
return
}
var pvs []string
if err := json.NewDecoder(resp.Body).Decode(&pvs); err != nil {
jsonError(w, http.StatusBadGateway, "Archiver response parse error: "+err.Error())
return
}
jsonOK(w, pvs)
}
// ── /groups ───────────────────────────────────────────────────────────────────
// getGroups returns the workspace group tree as a JSON array.
@@ -416,6 +475,8 @@ func metaToSignalInfo(m datasource.Metadata) signalInfo {
DisplayHigh: m.DisplayHigh,
Writable: m.Writable,
EnumStrings: m.EnumStrings,
Properties: m.Properties,
Tags: m.Tags,
}
}
internal/api/api_test.go
+1 -1
View File
@@ -40,7 +40,7 @@ func setup(t *testing.T) (*httptest.Server, func()) {
}
mux := http.NewServeMux()
api.New(brk, nil, store, "", log).Register(mux, "/api/v1")
api.New(brk, nil, store, "", "", log).Register(mux, "/api/v1")
srv := httptest.NewServer(mux)
return srv, func() {
internal/config/config.go
+9 -1
View File
@@ -36,7 +36,9 @@ type EPICSConfig struct {
CAAddrList string `toml:"ca_addr_list"`
ArchiveURL string `toml:"archive_url"`
ChannelFinderURL string `toml:"channel_finder_url"`
PVNames []string `toml:"pv_names"`
AutoSyncFilter string `toml:"auto_sync_filter"`
AutoSyncFromArchiver bool `toml:"auto_sync_from_archiver"`
PVNames []string `toml:"pv_names"`
}
type PVAConfig struct {
@@ -100,6 +102,12 @@ func applyEnv(cfg *Config) {
if v := env("UOPI_EPICS_CHANNEL_FINDER_URL"); v != "" {
cfg.Datasource.EPICS.ChannelFinderURL = v
}
if v := env("UOPI_EPICS_AUTO_SYNC_FILTER"); v != "" {
cfg.Datasource.EPICS.AutoSyncFilter = v
}
if v := env("UOPI_EPICS_AUTO_SYNC_FROM_ARCHIVER"); v != "" {
cfg.Datasource.EPICS.AutoSyncFromArchiver = (v == "true" || v == "YES")
}
if v := env("EPICS_PVA_ADDR_LIST"); v != "" {
cfg.Datasource.PVA.AddrList = strings.Fields(v)
}
internal/datasource/epics/archive.go
+50
View File
@@ -7,6 +7,7 @@ import (
"net/http"
"net/url"
"strconv"
"strings"
"time"
"github.com/uopi/uopi/internal/datasource"
@@ -117,6 +118,55 @@ func fetchArchiveHistory(ctx context.Context, archiveURL, signal string, start,
return out, nil
}
// searchArchivePVs queries the Archive Appliance management API for PV names
// matching the given pattern (glob style, e.g. "PV:*").
func searchArchivePVs(ctx context.Context, archiveURL, pattern string) ([]string, error) {
if archiveURL == "" {
return nil, nil
}
// Build the request URL.
// Format: GET {archiveURL}/mgmt/bpl/getAllPVs?pv={pattern}
reqURL, err := url.Parse(archiveURL)
if err != nil {
return nil, fmt.Errorf("epics archive: invalid archive URL %q: %w", archiveURL, err)
}
reqURL = reqURL.JoinPath("mgmt", "bpl", "getAllPVs")
q := reqURL.Query()
if pattern == "" {
pattern = "*"
}
// AA expects glob patterns. If no glob characters, wrap in stars.
if !strings.ContainsAny(pattern, "*?[]") {
pattern = "*" + pattern + "*"
}
q.Set("pv", pattern)
reqURL.RawQuery = q.Encode()
req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil)
if err != nil {
return nil, fmt.Errorf("epics archive: building request: %w", err)
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, fmt.Errorf("epics archive: HTTP request failed: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("epics archive: unexpected status %d for search %q", resp.StatusCode, pattern)
}
var pvs []string
if err := json.NewDecoder(resp.Body).Decode(&pvs); err != nil {
return nil, fmt.Errorf("epics archive: decoding response: %w", err)
}
return pvs, nil
}
// coerceArchiveValue converts the raw JSON value (which json.Unmarshal decodes
// as float64, string, bool, []interface{}, or nil) into one of the types
// accepted by datasource.Value.Data.
internal/datasource/epics/cf_client.go
+67
View File
@@ -0,0 +1,67 @@
package epics
import (
"encoding/json"
"fmt"
"log/slog"
"net/http"
"strings"
"time"
)
// cfChannel is the JSON structure returned by the Channel Finder Service.
type cfChannel struct {
Name string `json:"name"`
Owner string `json:"owner"`
Properties []cfProperty `json:"properties"`
Tags []cfTag `json:"tags"`
}
type cfProperty struct {
Name string `json:"name"`
Value string `json:"value"`
Owner string `json:"owner"`
}
type cfTag struct {
Name string `json:"name"`
Owner string `json:"owner"`
}
// queryChannelFinder fetches channels from CFS matching the given query string.
func queryChannelFinder(cfURL, query string) ([]cfChannel, error) {
if cfURL == "" {
return nil, nil
}
url := strings.TrimRight(cfURL, "/") + "/resources/channels"
if query != "" {
if strings.Contains(query, "=") || strings.Contains(query, "~") {
// Query is already a structured filter (e.g. area=L1)
url += "?" + query
} else {
// Query is just a PV name pattern
url += "?~name=*" + query + "*"
}
}
client := &http.Client{Timeout: 10 * time.Second}
slog.Info("CFS query", "url", url)
resp, err := client.Get(url)
if err != nil {
return nil, fmt.Errorf("CFS request failed: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("CFS returned status %d", resp.StatusCode)
}
var channels []cfChannel
if err := json.NewDecoder(resp.Body).Decode(&channels); err != nil {
return nil, fmt.Errorf("CFS response parse error: %w", err)
}
slog.Info("CFS results", "count", len(channels))
return channels, nil
}
internal/datasource/epics/discovery_test.go
+82
View File
@@ -0,0 +1,82 @@
package epics_test
import (
"context"
"os"
"testing"
"time"
"github.com/uopi/goca/proto"
"github.com/uopi/goca/testca"
"github.com/uopi/uopi/internal/datasource/epics"
)
func TestDiscovery_EnvironmentVariables(t *testing.T) {
// 1. Setup a fake CA server.
srv, err := testca.New([]testca.PVSpec{
{Name: "DISCO:PV", DBFType: proto.DBFDouble, Count: 1, Value: 123.45},
})
if err != nil {
t.Fatalf("testca.New: %v", err)
}
defer srv.Close()
// 2. Set the environment variable to point to this server.
os.Setenv("EPICS_CA_ADDR_LIST", srv.UDPAddr())
os.Setenv("EPICS_CA_AUTO_ADDR_LIST", "NO")
defer os.Unsetenv("EPICS_CA_ADDR_LIST")
defer os.Unsetenv("EPICS_CA_AUTO_ADDR_LIST")
// 3. Create datasource with NO explicit address (should pick up from env).
ds := epics.New("", "", "", "", false, nil)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := ds.Connect(ctx); err != nil {
t.Fatalf("Connect: %v", err)
}
// 4. Verify discovery works.
meta, err := ds.GetMetadata(ctx, "DISCO:PV")
if err != nil {
t.Fatalf("GetMetadata failed (discovery failed): %v", err)
}
if meta.Name != "DISCO:PV" {
t.Errorf("got %q, want DISCO:PV", meta.Name)
}
}
func TestDiscovery_ConfigOverride(t *testing.T) {
// 1. Setup TWO fake CA servers.
srv1, _ := testca.New([]testca.PVSpec{{Name: "PV:1", DBFType: proto.DBFDouble, Value: 1.0}})
defer srv1.Close()
srv2, _ := testca.New([]testca.PVSpec{{Name: "PV:2", DBFType: proto.DBFDouble, Value: 2.0}})
defer srv2.Close()
// 2. Point env to srv1.
os.Setenv("EPICS_CA_ADDR_LIST", srv1.UDPAddr())
os.Setenv("EPICS_CA_AUTO_ADDR_LIST", "NO")
defer os.Unsetenv("EPICS_CA_ADDR_LIST")
defer os.Unsetenv("EPICS_CA_AUTO_ADDR_LIST")
// 3. Create datasource pointing explicitly to srv2 (should OVERRIDE env).
ds := epics.New(srv2.UDPAddr(), "", "", "", false, nil)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := ds.Connect(ctx); err != nil {
t.Fatalf("Connect: %v", err)
}
// 4. PV from srv2 should be found.
if _, err := ds.GetMetadata(ctx, "PV:2"); err != nil {
t.Errorf("PV:2 from config address not found: %v", err)
}
// 5. PV from srv1 should NOT be found (since AutoAddrList is disabled by config override).
ctx2, cancel2 := context.WithTimeout(context.Background(), 500*time.Millisecond)
defer cancel2()
if _, err := ds.GetMetadata(ctx2, "PV:1"); err == nil {
t.Error("PV:1 from env address was found, but config should have overridden it")
}
}
internal/datasource/epics/epics.go
+95 -11
View File
@@ -26,6 +26,7 @@ import "C"
import (
"context"
"fmt"
"log/slog"
"os"
"sync"
"sync/atomic"
@@ -66,10 +67,14 @@ type caChannel struct {
// EPICS is the Channel Access data source.
type EPICS struct {
caAddrList string
archiveURL string
caAddrList string
archiveURL string
cfURL string
autoSyncFilter string
autoSyncFromArchiver bool
// caCtx is the CA context created in Connect(). Stored as unsafe.Pointer
// 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.
@@ -87,12 +92,15 @@ type EPICS struct {
// Appliance instance for history queries (may be empty).
// pvNames is accepted for API compatibility with the pure-Go build but ignored
// in the CGo build (the CGo implementation populates ListSignals via callbacks).
func New(caAddrList, archiveURL string, pvNames []string) datasource.DataSource {
func New(caAddrList, archiveURL, cfURL, autoSyncFilter string, autoSyncFromArchiver bool, pvNames []string) datasource.DataSource {
return &EPICS{
caAddrList: caAddrList,
archiveURL: archiveURL,
channels: make(map[string]*caChannel),
metadata: make(map[string]datasource.Metadata),
caAddrList: caAddrList,
archiveURL: archiveURL,
cfURL: cfURL,
autoSyncFilter: autoSyncFilter,
autoSyncFromArchiver: autoSyncFromArchiver,
channels: make(map[string]*caChannel),
metadata: make(map[string]datasource.Metadata),
}
}
@@ -122,9 +130,62 @@ func (e *EPICS) Connect(_ context.Context) error {
}
// Save the context so goroutines on other OS threads can attach to it.
e.caCtx = C.caCurrentContext()
// Perform background sync from Channel Finder if configured.
if e.cfURL != "" && e.autoSyncFilter != "" {
go e.syncFromChannelFinder(context.Background())
}
// Perform background sync from Archive Appliance if configured.
if e.archiveURL != "" && e.autoSyncFromArchiver {
go e.syncFromArchiver(context.Background())
}
return nil
}
func (e *EPICS) syncFromChannelFinder(ctx context.Context) {
time.Sleep(500 * time.Millisecond)
slog.Info("epics: syncing signals from Channel Finder", "url", e.cfURL, "filter", e.autoSyncFilter)
channels, err := queryChannelFinder(e.cfURL, e.autoSyncFilter)
if err != nil {
slog.Error("epics: Channel Finder sync failed", "err", err)
return
}
for _, ch := range channels {
// Convert CFS properties and tags
props := make(map[string]string)
for _, p := range ch.Properties {
props[p.Name] = p.Value
}
tags := make([]string, len(ch.Tags))
for i, t := range ch.Tags {
tags[i] = t.Name
}
e.mu.Lock()
if _, exists := e.metadata[ch.Name]; !exists {
e.metadata[ch.Name] = datasource.Metadata{
Name: ch.Name,
Properties: props,
Tags: tags,
}
}
e.mu.Unlock()
// Background pre-fetch of actual record info
go e.prefetchPV(ctx, ch.Name)
}
slog.Info("epics: synced signals from Channel Finder", "count", len(channels))
}
func (e *EPICS) prefetchPV(ctx context.Context, pv string) {
mctx, cancel := context.WithTimeout(ctx, 15*time.Second)
defer cancel()
_, _ = e.GetMetadata(mctx, pv)
}
// 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.
@@ -422,13 +483,12 @@ func (e *EPICS) fetchMetadata(chid C.chid, name string) datasource.Metadata {
}
// 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()
slog.Debug("epics: listing signals", "count", len(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))
@@ -583,3 +643,27 @@ func nativeTimeType(chid C.chid) C.chtype {
return C.DBR_TIME_DOUBLE
}
}
func (e *EPICS) syncFromArchiver(ctx context.Context) {
time.Sleep(1 * time.Second) // wait for network
slog.Info("epics: syncing signals from Archiver", "url", e.archiveURL)
pvs, err := searchArchivePVs(ctx, e.archiveURL, "*")
if err != nil {
slog.Error("epics: Archiver sync failed", "err", err)
return
}
for _, name := range pvs {
if ctx.Err() != nil {
return
}
e.mu.Lock()
if _, exists := e.metadata[name]; !exists {
e.metadata[name] = datasource.Metadata{Name: name}
}
e.mu.Unlock()
// Background pre-fetch of record info
go e.prefetchPV(ctx, name)
}
slog.Info("epics: synced signals from Archiver", "count", len(pvs))
}
internal/datasource/epics/epics_purego_test.go
+1 -1
View File
@@ -26,7 +26,7 @@ func newTestDS(t *testing.T, pvs []testca.PVSpec) (datasource.DataSource, *testc
}
t.Cleanup(srv.Close)
ds := epics.New(srv.UDPAddr(), "", nil)
ds := epics.New(srv.UDPAddr(), "", "", "", false, nil)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
t.Cleanup(cancel)
if err := ds.Connect(ctx); err != nil {
internal/datasource/epics/noop.go
+107 -25
View File
@@ -28,9 +28,12 @@ func Available() bool { return true }
// EPICS is the pure-Go Channel Access data source.
type EPICS struct {
caAddrList string
archiveURL string
pvNames []string // pre-fetched at connect time for ListSignals
caAddrList string
archiveURL string
cfURL string
autoSyncFilter string
autoSyncFromArchiver bool
pvNames []string // pre-fetched at connect time for ListSignals
client *ca.Client
@@ -47,12 +50,15 @@ type EPICS struct {
// pvNames is an optional list of PV names whose metadata will be pre-fetched
// at connect time so they appear in ListSignals immediately (useful for the
// edit-mode signal tree).
func New(caAddrList, archiveURL string, pvNames []string) datasource.DataSource {
func New(caAddrList, archiveURL, cfURL, autoSyncFilter string, autoSyncFromArchiver bool, pvNames []string) datasource.DataSource {
return &EPICS{
caAddrList: caAddrList,
archiveURL: archiveURL,
pvNames: pvNames,
metadata: make(map[string]datasource.Metadata),
caAddrList: caAddrList,
archiveURL: archiveURL,
cfURL: cfURL,
autoSyncFilter: autoSyncFilter,
autoSyncFromArchiver: autoSyncFromArchiver,
pvNames: pvNames,
metadata: make(map[string]datasource.Metadata),
}
}
@@ -75,29 +81,24 @@ func (e *EPICS) Connect(ctx context.Context) error {
e.client = cli
slog.Info("epics: CA client started", "addrs", cfg.AddrList)
// Pre-fetch metadata for any configured PV names so they appear in
// ListSignals as soon as the datasource is ready. We open a short-lived
// subscription (which drives channel connection) and then fetch CTRL info.
// Perform background sync from Channel Finder if configured.
if e.cfURL != "" && e.autoSyncFilter != "" {
go e.syncFromChannelFinder(ctx)
}
// Perform background sync from Archive Appliance if configured.
if e.archiveURL != "" && e.autoSyncFromArchiver {
go e.syncFromArchiver(ctx)
}
// Pre-fetch metadata for any configured PV names.
if len(e.pvNames) > 0 {
go func() {
for _, pv := range e.pvNames {
if ctx.Err() != nil {
return
}
func() {
tvCh := make(chan proto.TimeValue, 1)
subCtx, subCancel := context.WithTimeout(ctx, 15*time.Second)
defer subCancel()
cancel, err := e.client.Subscribe(subCtx, pv, tvCh)
if err != nil {
return
}
defer cancel()
// Channel is now connected; fetch CTRL metadata.
mctx, mcancel := context.WithTimeout(ctx, 5*time.Second)
defer mcancel()
_, _ = e.GetMetadata(mctx, pv)
}()
e.prefetchPV(ctx, pv)
}
}()
}
@@ -105,6 +106,62 @@ func (e *EPICS) Connect(ctx context.Context) error {
return nil
}
func (e *EPICS) syncFromChannelFinder(ctx context.Context) {
time.Sleep(500 * time.Millisecond)
slog.Info("epics: syncing signals from Channel Finder", "url", e.cfURL, "filter", e.autoSyncFilter)
channels, err := queryChannelFinder(e.cfURL, e.autoSyncFilter)
if err != nil {
slog.Error("epics: Channel Finder sync failed", "err", err)
return
}
for _, ch := range channels {
if ctx.Err() != nil {
return
}
// Convert CFS properties and tags to datasource.Metadata
props := make(map[string]string)
for _, p := range ch.Properties {
props[p.Name] = p.Value
}
tags := make([]string, len(ch.Tags))
for i, t := range ch.Tags {
tags[i] = t.Name
}
// Initialize metadata with CFS info. Full metadata (type, units, etc.)
// will be populated on first access via DBR_CTRL GET.
e.mu.Lock()
if _, exists := e.metadata[ch.Name]; !exists {
e.metadata[ch.Name] = datasource.Metadata{
Name: ch.Name,
Properties: props,
Tags: tags,
}
}
e.mu.Unlock()
// Background pre-fetch of actual record info (type, etc.)
go e.prefetchPV(ctx, ch.Name)
}
slog.Info("epics: synced signals from Channel Finder", "count", len(channels))
}
func (e *EPICS) prefetchPV(ctx context.Context, pv string) {
tvCh := make(chan proto.TimeValue, 1)
subCtx, subCancel := context.WithTimeout(ctx, 15*time.Second)
defer subCancel()
cancel, err := e.client.Subscribe(subCtx, pv, tvCh)
if err != nil {
return
}
defer cancel()
// Channel is now connected; fetch CTRL metadata.
mctx, mcancel := context.WithTimeout(ctx, 5*time.Second)
defer mcancel()
_, _ = e.GetMetadata(mctx, pv)
}
// -------------------------------------------------------------------------- //
// Subscribe //
// -------------------------------------------------------------------------- //
@@ -273,6 +330,7 @@ func (e *EPICS) ctrlToMeta(name string, ci ca.CtrlInfo) datasource.Metadata {
func (e *EPICS) ListSignals(_ context.Context) ([]datasource.Metadata, error) {
e.mu.RLock()
defer e.mu.RUnlock()
slog.Debug("epics: listing signals", "count", len(e.metadata))
out := make([]datasource.Metadata, 0, len(e.metadata))
for _, m := range e.metadata {
out = append(out, m)
@@ -305,3 +363,27 @@ func (e *EPICS) History(ctx context.Context, signal string, start, end time.Time
}
return fetchArchiveHistory(ctx, e.archiveURL, signal, start, end, maxPoints)
}
func (e *EPICS) syncFromArchiver(ctx context.Context) {
time.Sleep(1 * time.Second) // wait for network
slog.Info("epics: syncing signals from Archiver", "url", e.archiveURL)
pvs, err := searchArchivePVs(ctx, e.archiveURL, "*")
if err != nil {
slog.Error("epics: Archiver sync failed", "err", err)
return
}
for _, name := range pvs {
if ctx.Err() != nil {
return
}
e.mu.Lock()
if _, exists := e.metadata[name]; !exists {
e.metadata[name] = datasource.Metadata{Name: name}
}
e.mu.Unlock()
// Background pre-fetch of record info
go e.prefetchPV(ctx, name)
}
slog.Info("epics: synced signals from Archiver", "count", len(pvs))
}
internal/datasource/iface.go
+2
View File
@@ -60,6 +60,8 @@ type Metadata struct {
DriveHigh float64
EnumStrings []string // non-nil only for TypeEnum
Writable bool
Properties map[string]string // Channel Finder properties
Tags []string // Channel Finder tags
}
// CancelFunc cancels a subscription started by DataSource.Subscribe.
internal/server/server.go
+2 -2
View File
@@ -23,7 +23,7 @@ type Server struct {
// New creates the HTTP server, registers all routes, and returns a ready-to-start 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, channelFinderURL string, log *slog.Logger) *Server {
func New(addr string, webFS fs.FS, brk *broker.Broker, synth *synthetic.Synthetic, store *storage.Store, channelFinderURL, archiverURL string, log *slog.Logger) *Server {
mux := http.NewServeMux()
// Health check
@@ -39,7 +39,7 @@ func New(addr string, webFS fs.FS, brk *broker.Broker, synth *synthetic.Syntheti
mux.HandleFunc("/metrics", metrics.Handler(brk.ActiveSubscriptions))
// REST API
api.New(brk, synth, store, channelFinderURL, log).Register(mux, apiPrefix)
api.New(brk, synth, store, channelFinderURL, archiverURL, log).Register(mux, apiPrefix)
// Embedded frontend — must be last (catch-all)
mux.Handle("/", http.FileServerFS(webFS))
internal/server/ws.go
+13 -9
View File
@@ -64,15 +64,17 @@ type outMsg struct {
}
type metaPayload struct {
Type string `json:"type"`
Unit string `json:"unit,omitempty"`
Description string `json:"description,omitempty"`
DisplayLow float64 `json:"displayLow"`
DisplayHigh float64 `json:"displayHigh"`
DriveLow float64 `json:"driveLow,omitempty"`
DriveHigh float64 `json:"driveHigh,omitempty"`
EnumStrings []string `json:"enumStrings,omitempty"`
Writable bool `json:"writable"`
Type string `json:"type"`
Unit string `json:"unit,omitempty"`
Description string `json:"description,omitempty"`
DisplayLow float64 `json:"displayLow"`
DisplayHigh float64 `json:"displayHigh"`
DriveLow float64 `json:"driveLow,omitempty"`
DriveHigh float64 `json:"driveHigh,omitempty"`
EnumStrings []string `json:"enumStrings,omitempty"`
Writable bool `json:"writable"`
Properties map[string]string `json:"properties,omitempty"`
Tags []string `json:"tags,omitempty"`
}
type histPoint struct {
@@ -396,6 +398,8 @@ func metadataToPayload(m datasource.Metadata) *metaPayload {
DriveHigh: m.DriveHigh,
EnumStrings: m.EnumStrings,
Writable: m.Writable,
Properties: m.Properties,
Tags: m.Tags,
}
}
internal/storage/store.go
+97 -7
View File
@@ -7,6 +7,7 @@ import (
"fmt"
"os"
"path/filepath"
"strconv"
"strings"
"time"
"unicode"
@@ -80,10 +81,16 @@ func (s *Store) List() ([]InterfaceMeta, error) {
}
var out []InterfaceMeta
for _, e := range entries {
if e.IsDir() || !strings.HasSuffix(e.Name(), ".xml") {
name := e.Name()
if e.IsDir() || !strings.HasSuffix(strings.ToLower(name), ".xml") {
continue
}
id := strings.TrimSuffix(e.Name(), ".xml")
// Skip versioned backups: id.v1.xml, id.v2.xml, etc.
if isVersioned(name) {
continue
}
id := strings.TrimSuffix(name, filepath.Ext(name))
meta, err := s.readMeta(id)
if err != nil {
continue // skip corrupt files silently
@@ -96,6 +103,20 @@ func (s *Store) List() ([]InterfaceMeta, error) {
return out, nil
}
func isVersioned(name string) bool {
// Format: <id>.v<number>.xml
parts := strings.Split(name, ".")
if len(parts) != 3 {
return false
}
vPart := parts[1]
if !strings.HasPrefix(vPart, "v") {
return false
}
_, err := strconv.Atoi(vPart[1:])
return err == nil
}
// rootAttrs is used to parse only the top-level XML attributes.
type rootAttrs struct {
XMLName xml.Name `xml:"interface"`
@@ -145,18 +166,87 @@ func (s *Store) Create(xmlData []byte) (string, error) {
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)
// Initialize version to 1.
updatedData, err := setXMLAttribute(xmlData, "version", "1")
if err != nil {
return "", fmt.Errorf("initialize version attribute: %w", err)
}
return id, os.WriteFile(s.filePath(id), updatedData, 0o644)
}
// Update replaces the XML for an existing interface.
// Update replaces the XML for an existing interface, preserving the previous
// version as a separate file.
func (s *Store) Update(id string, xmlData []byte) error {
if err := validateID(id); err != nil {
return ErrNotFound
}
if _, err := os.Stat(s.filePath(id)); errors.Is(err, os.ErrNotExist) {
return ErrNotFound
oldPath := s.filePath(id)
oldData, err := os.ReadFile(oldPath)
if err != nil {
if errors.Is(err, os.ErrNotExist) {
return ErrNotFound
}
return err
}
return os.WriteFile(s.filePath(id), xmlData, 0o644)
// Parse current version to name the backup file.
var oldRoot rootAttrs
if err := xml.Unmarshal(oldData, &oldRoot); err != nil {
return fmt.Errorf("parse existing XML: %w", err)
}
if oldRoot.Version < 1 {
oldRoot.Version = 1
}
// Move old file to backup path: id.vN.xml
backupPath := filepath.Join(s.dir, fmt.Sprintf("%s.v%d.xml", id, oldRoot.Version))
if err := os.WriteFile(backupPath, oldData, 0o644); err != nil {
return fmt.Errorf("create backup: %w", err)
}
// Update version in new data.
newVersion := oldRoot.Version + 1
updatedData, err := setXMLAttribute(xmlData, "version", fmt.Sprintf("%d", newVersion))
if err != nil {
return fmt.Errorf("update version attribute: %w", err)
}
return os.WriteFile(oldPath, updatedData, 0o644)
}
// setXMLAttribute is a primitive helper to update a top-level attribute in an
// XML blob without full unmarshal/marshal (which might mess up formatting).
func setXMLAttribute(data []byte, attr, value string) ([]byte, error) {
// Very basic implementation: look for the attribute in the first 512 bytes.
// A proper implementation would use an XML encoder or a better regex.
s := string(data)
tagEnd := strings.Index(s, ">")
if tagEnd == -1 {
return nil, errors.New("malformed XML: no root tag end")
}
rootTag := s[:tagEnd]
attrSearch := " " + attr + "=\""
idx := strings.Index(rootTag, attrSearch)
if idx == -1 {
// Attribute not found, insert it before the tag ends.
insertIdx := tagEnd
if s[tagEnd-1] == '/' {
insertIdx-- // handle <tag />
}
return []byte(s[:insertIdx] + fmt.Sprintf(" %s=\"%s\"", attr, value) + s[insertIdx:]), nil
}
// Attribute found, replace its value.
valStart := idx + len(attrSearch)
valEnd := strings.Index(s[valStart:], "\"")
if valEnd == -1 {
return nil, errors.New("malformed XML: attribute value not closed")
}
return []byte(s[:valStart] + value + s[valStart+valEnd:]), nil
}
// Delete removes the interface with the given ID.