Martino Ferrari
366 lines
9.4 KiB
Go
366 lines
9.4 KiB
Go
// Package pva provides an EPICS PV Access data source for uopi using the
|
|
// pure-Go gopva library.
|
|
package pva
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
gopva "github.com/uopi/gopva"
|
|
"github.com/uopi/gopva/pvdata"
|
|
"github.com/uopi/uopi/internal/datasource"
|
|
)
|
|
|
|
// PVA is the PV Access data source.
|
|
type PVA struct {
|
|
client *gopva.Client
|
|
|
|
mu sync.RWMutex
|
|
metadata map[string]datasource.Metadata
|
|
}
|
|
|
|
// New creates a new PV Access data source.
|
|
// Server addresses are read from the EPICS_PVA_ADDR_LIST environment variable;
|
|
// addrList overrides that when non-empty.
|
|
func New(addrList []string) datasource.DataSource {
|
|
if len(addrList) > 0 {
|
|
// Inject addresses into the environment so gopva.NewClient picks them up.
|
|
_ = os.Setenv("EPICS_PVA_ADDR_LIST", strings.Join(addrList, " "))
|
|
}
|
|
return &PVA{
|
|
metadata: make(map[string]datasource.Metadata),
|
|
}
|
|
}
|
|
|
|
// Name implements datasource.DataSource.
|
|
func (p *PVA) Name() string { return "pva" }
|
|
|
|
// Connect creates the PVA client.
|
|
func (p *PVA) Connect(_ context.Context) error {
|
|
cl, err := gopva.NewClient()
|
|
if err != nil {
|
|
return fmt.Errorf("pva: connect: %w", err)
|
|
}
|
|
p.client = cl
|
|
slog.Info("pva: client started")
|
|
return nil
|
|
}
|
|
|
|
// -------------------------------------------------------------------------- //
|
|
// Subscribe //
|
|
// -------------------------------------------------------------------------- //
|
|
|
|
// Subscribe registers ch to receive live PVA value updates for signal.
|
|
func (p *PVA) Subscribe(ctx context.Context, signal string, ch chan<- datasource.Value) (datasource.CancelFunc, error) {
|
|
slog.Info("pva: subscribing", "pv", signal)
|
|
|
|
monCh := p.client.Monitor(ctx, signal)
|
|
|
|
// Wait for the first event to confirm the subscription is live.
|
|
// We do this in the goroutine to avoid blocking the caller.
|
|
done := make(chan struct{})
|
|
go func() {
|
|
defer close(done)
|
|
for evt := range monCh {
|
|
if evt.Err != nil {
|
|
slog.Error("pva: monitor error", "pv", signal, "err", evt.Err)
|
|
select {
|
|
case ch <- datasource.Value{
|
|
Timestamp: time.Now(),
|
|
Data: nil,
|
|
Quality: datasource.QualityBad,
|
|
}:
|
|
default:
|
|
}
|
|
return
|
|
}
|
|
// Cache metadata from the first value.
|
|
p.updateMetadata(signal, evt.Value)
|
|
|
|
val := structToValue(evt.Value)
|
|
select {
|
|
case ch <- val:
|
|
default:
|
|
}
|
|
}
|
|
}()
|
|
|
|
cancel := func() {
|
|
// cancelling the context passed to Monitor stops it; the channel will be closed.
|
|
// The goroutine will exit when monCh is closed.
|
|
<-done
|
|
}
|
|
return datasource.CancelFunc(cancel), nil
|
|
}
|
|
|
|
// -------------------------------------------------------------------------- //
|
|
// GetMetadata //
|
|
// -------------------------------------------------------------------------- //
|
|
|
|
// GetMetadata returns metadata for signal. It performs a one-shot GET if
|
|
// metadata has not been cached yet.
|
|
func (p *PVA) GetMetadata(ctx context.Context, signal string) (datasource.Metadata, error) {
|
|
p.mu.RLock()
|
|
if m, ok := p.metadata[signal]; ok {
|
|
p.mu.RUnlock()
|
|
return m, nil
|
|
}
|
|
p.mu.RUnlock()
|
|
|
|
tctx, cancel := context.WithTimeout(ctx, 15*time.Second)
|
|
defer cancel()
|
|
|
|
sv, err := p.client.Get(tctx, signal)
|
|
if err != nil {
|
|
return datasource.Metadata{}, fmt.Errorf("pva: GetMetadata %q: %w", signal, err)
|
|
}
|
|
|
|
p.updateMetadata(signal, sv)
|
|
|
|
p.mu.RLock()
|
|
m := p.metadata[signal]
|
|
p.mu.RUnlock()
|
|
return m, nil
|
|
}
|
|
|
|
// updateMetadata derives and caches metadata from a StructValue.
|
|
func (p *PVA) updateMetadata(signal string, sv pvdata.StructValue) {
|
|
m := structToMeta(signal, sv)
|
|
p.mu.Lock()
|
|
p.metadata[signal] = m
|
|
p.mu.Unlock()
|
|
}
|
|
|
|
// -------------------------------------------------------------------------- //
|
|
// ListSignals //
|
|
// -------------------------------------------------------------------------- //
|
|
|
|
// ListSignals returns metadata for all signals with cached information.
|
|
func (p *PVA) ListSignals(_ context.Context) ([]datasource.Metadata, error) {
|
|
p.mu.RLock()
|
|
defer p.mu.RUnlock()
|
|
out := make([]datasource.Metadata, 0, len(p.metadata))
|
|
for _, m := range p.metadata {
|
|
out = append(out, m)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// -------------------------------------------------------------------------- //
|
|
// Write //
|
|
// -------------------------------------------------------------------------- //
|
|
|
|
// Write is not yet implemented for PVA.
|
|
func (p *PVA) Write(_ context.Context, _ string, _ any) error {
|
|
return fmt.Errorf("pva: write not yet implemented")
|
|
}
|
|
|
|
// -------------------------------------------------------------------------- //
|
|
// History //
|
|
// -------------------------------------------------------------------------- //
|
|
|
|
// History is not supported for PVA sources.
|
|
func (p *PVA) History(_ context.Context, _ string, _, _ time.Time, _ int) ([]datasource.Value, error) {
|
|
return nil, datasource.ErrHistoryUnavailable
|
|
}
|
|
|
|
// -------------------------------------------------------------------------- //
|
|
// Helpers //
|
|
// -------------------------------------------------------------------------- //
|
|
|
|
// structToValue converts a PVA StructValue to a datasource.Value.
|
|
// It follows NTScalar/NTScalarArray conventions: looks for a "value" field
|
|
// and an optional "alarm" struct for quality, and "timeStamp" for the timestamp.
|
|
func structToValue(sv pvdata.StructValue) datasource.Value {
|
|
val := fieldByName(sv, "value")
|
|
ts := extractTimestamp(sv)
|
|
quality := extractQuality(sv)
|
|
|
|
var data any
|
|
if val != nil {
|
|
data = normaliseValue(val)
|
|
}
|
|
|
|
return datasource.Value{
|
|
Timestamp: ts,
|
|
Data: data,
|
|
Quality: quality,
|
|
}
|
|
}
|
|
|
|
// structToMeta derives datasource.Metadata from a PVA StructValue.
|
|
func structToMeta(name string, sv pvdata.StructValue) datasource.Metadata {
|
|
m := datasource.Metadata{
|
|
Name: name,
|
|
Writable: true, // PVA channels are assumed writable until proven otherwise
|
|
}
|
|
|
|
val := fieldByName(sv, "value")
|
|
if val == nil && len(sv.Fields) > 0 {
|
|
val = sv.Fields[0].Value
|
|
}
|
|
|
|
if val != nil {
|
|
switch val.(type) {
|
|
case float64, float32:
|
|
m.Type = datasource.TypeFloat64
|
|
case int64, int32, int16, int8, uint64, uint32, uint16, uint8:
|
|
m.Type = datasource.TypeInt64
|
|
case string:
|
|
m.Type = datasource.TypeString
|
|
case bool:
|
|
m.Type = datasource.TypeBool
|
|
case []float64, []float32:
|
|
m.Type = datasource.TypeFloat64Array
|
|
default:
|
|
m.Type = datasource.TypeFloat64
|
|
}
|
|
}
|
|
|
|
// Extract display limits from NTScalar display structure.
|
|
if disp := structByName(sv, "display"); disp != nil {
|
|
if v := fieldByName(*disp, "limitLow"); v != nil {
|
|
m.DisplayLow, _ = toFloat64(v)
|
|
}
|
|
if v := fieldByName(*disp, "limitHigh"); v != nil {
|
|
m.DisplayHigh, _ = toFloat64(v)
|
|
}
|
|
if v := fieldByName(*disp, "units"); v != nil {
|
|
m.Unit, _ = v.(string)
|
|
}
|
|
}
|
|
|
|
// Extract control limits from NTScalar control structure.
|
|
if ctrl := structByName(sv, "control"); ctrl != nil {
|
|
if v := fieldByName(*ctrl, "limitLow"); v != nil {
|
|
m.DriveLow, _ = toFloat64(v)
|
|
}
|
|
if v := fieldByName(*ctrl, "limitHigh"); v != nil {
|
|
m.DriveHigh, _ = toFloat64(v)
|
|
}
|
|
}
|
|
|
|
return m
|
|
}
|
|
|
|
// normaliseValue converts any PVA scalar/array value to datasource-compatible types.
|
|
func normaliseValue(v any) any {
|
|
switch x := v.(type) {
|
|
case float64:
|
|
return x
|
|
case float32:
|
|
return float64(x)
|
|
case int64:
|
|
return x
|
|
case int32:
|
|
return int64(x)
|
|
case int16:
|
|
return int64(x)
|
|
case int8:
|
|
return int64(x)
|
|
case uint64:
|
|
return int64(x)
|
|
case uint32:
|
|
return int64(x)
|
|
case uint16:
|
|
return int64(x)
|
|
case uint8:
|
|
return int64(x)
|
|
case bool:
|
|
return x
|
|
case string:
|
|
return x
|
|
case []float64:
|
|
return x
|
|
case []float32:
|
|
out := make([]float64, len(x))
|
|
for i, f := range x {
|
|
out[i] = float64(f)
|
|
}
|
|
return out
|
|
default:
|
|
return fmt.Sprintf("%v", v)
|
|
}
|
|
}
|
|
|
|
func extractTimestamp(sv pvdata.StructValue) time.Time {
|
|
ts := structByName(sv, "timeStamp")
|
|
if ts == nil {
|
|
return time.Now()
|
|
}
|
|
sec := int64(0)
|
|
nsec := int32(0)
|
|
if v := fieldByName(*ts, "secondsPastEpoch"); v != nil {
|
|
sec, _ = v.(int64)
|
|
}
|
|
if v := fieldByName(*ts, "nanoseconds"); v != nil {
|
|
nsec, _ = v.(int32)
|
|
}
|
|
if sec == 0 {
|
|
return time.Now()
|
|
}
|
|
// EPICS epoch: 1990-01-01 00:00:00 UTC = Unix 631152000
|
|
const epicsEpoch = 631152000
|
|
return time.Unix(sec+epicsEpoch, int64(nsec)).UTC()
|
|
}
|
|
|
|
func extractQuality(sv pvdata.StructValue) datasource.Quality {
|
|
alarm := structByName(sv, "alarm")
|
|
if alarm == nil {
|
|
return datasource.QualityGood
|
|
}
|
|
sev := fieldByName(*alarm, "severity")
|
|
if sev == nil {
|
|
return datasource.QualityGood
|
|
}
|
|
s, _ := sev.(int32)
|
|
switch s {
|
|
case 1:
|
|
return datasource.QualityUncertain
|
|
case 2, 3:
|
|
return datasource.QualityBad
|
|
default:
|
|
return datasource.QualityGood
|
|
}
|
|
}
|
|
|
|
func toFloat64(v any) (float64, bool) {
|
|
switch x := v.(type) {
|
|
case float64:
|
|
return x, true
|
|
case float32:
|
|
return float64(x), true
|
|
case int64:
|
|
return float64(x), true
|
|
case int32:
|
|
return float64(x), true
|
|
default:
|
|
return 0, false
|
|
}
|
|
}
|
|
|
|
func fieldByName(sv pvdata.StructValue, name string) any {
|
|
for _, f := range sv.Fields {
|
|
if f.Name == name {
|
|
return f.Value
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func structByName(sv pvdata.StructValue, name string) *pvdata.StructValue {
|
|
for _, f := range sv.Fields {
|
|
if f.Name == name {
|
|
if s, ok := f.Value.(pvdata.StructValue); ok {
|
|
return &s
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|