// Package scpi implements a SCPI instrument data source. Each configured // instrument is reached over a raw TCP socket (line-based SCPI, the "SCPI raw" / // port 5025 convention); a VXI-11 transport is reserved for later. Channels are // exposed as signals named "instrument:channel" and polled at a configurable // interval. Channels with a write_cmd template accept Write. package scpi import ( "context" "fmt" "strconv" "strings" "time" "github.com/uopi/uopi/internal/datasource" ) const defaultPollInterval = time.Second type instrumentConn struct { inst Instrument tr transport } // Scpi is a datasource.DataSource backed by one or more SCPI instruments. type Scpi struct { pollInterval time.Duration instruments map[string]*instrumentConn signals map[string]signalRef // "instrument:channel" → ref } type signalRef struct { instrument string ch Channel } // New builds a SCPI source from config. It does not dial; connections are // established lazily on first poll/write. func New(cfg Config) (*Scpi, error) { poll := time.Duration(cfg.PollIntervalMs) * time.Millisecond if poll <= 0 { poll = defaultPollInterval } s := &Scpi{ pollInterval: poll, instruments: make(map[string]*instrumentConn), signals: make(map[string]signalRef), } for _, inst := range cfg.Instruments { if inst.Name == "" || inst.Address == "" { return nil, fmt.Errorf("scpi: instrument needs name and address") } if _, dup := s.instruments[inst.Name]; dup { return nil, fmt.Errorf("scpi: duplicate instrument %q", inst.Name) } tr, err := newTransport(inst) if err != nil { return nil, err } ic := &instrumentConn{inst: inst, tr: tr} for _, ch := range inst.Channels { if ch.Name == "" || ch.Query == "" { return nil, fmt.Errorf("scpi: instrument %q channel needs name and query", inst.Name) } key := inst.Name + ":" + ch.Name if _, dup := s.signals[key]; dup { return nil, fmt.Errorf("scpi: instrument %q duplicate channel %q", inst.Name, ch.Name) } s.signals[key] = signalRef{instrument: inst.Name, ch: ch} } s.instruments[inst.Name] = ic } return s, nil } // newTransport selects the transport implementation for an instrument. func newTransport(inst Instrument) (transport, error) { addr := inst.Address switch strings.ToLower(inst.Transport) { case "", "raw": if !strings.Contains(addr, ":") { addr += ":5025" } return newRawSocket(addr, time.Duration(inst.TimeoutMs)*time.Millisecond, inst.Terminator), nil case "vxi11": return nil, fmt.Errorf("scpi: vxi11 transport not yet implemented") default: return nil, fmt.Errorf("scpi: unknown transport %q", inst.Transport) } } // Name implements datasource.DataSource. func (s *Scpi) Name() string { return "scpi" } // Connect is a no-op; connections are dialled lazily per instrument. func (s *Scpi) Connect(_ context.Context) error { return nil } // ListSignals returns metadata for every configured channel. func (s *Scpi) ListSignals(_ context.Context) ([]datasource.Metadata, error) { out := make([]datasource.Metadata, 0, len(s.signals)) for _, ref := range s.signals { out = append(out, ref.ch.metadata(ref.instrument)) } return out, nil } // GetMetadata returns metadata for one signal. func (s *Scpi) GetMetadata(_ context.Context, signal string) (datasource.Metadata, error) { ref, ok := s.signals[signal] if !ok { return datasource.Metadata{}, datasource.ErrNotFound } return ref.ch.metadata(ref.instrument), nil } // readSignal performs one synchronous query of a channel. func (s *Scpi) readSignal(ref signalRef) (datasource.Value, error) { ic := s.instruments[ref.instrument] resp, err := ic.tr.query(ref.ch.Query) if err != nil { return datasource.Value{}, err } data, err := parseValue(ref.ch.dataType(), resp) if err != nil { return datasource.Value{}, err } return datasource.Value{Timestamp: time.Now(), Data: data, Quality: datasource.QualityGood}, nil } // Subscribe polls the channel at its configured interval (falling back to the // source default) and pushes values into ch. A query error emits QualityBad and // polling continues. func (s *Scpi) Subscribe(ctx context.Context, signal string, ch chan<- datasource.Value) (datasource.CancelFunc, error) { ref, ok := s.signals[signal] if !ok { return nil, datasource.ErrNotFound } interval := s.pollInterval if ref.ch.PollIntervalMs > 0 { interval = time.Duration(ref.ch.PollIntervalMs) * time.Millisecond } ctx, cancel := context.WithCancel(ctx) go func() { ticker := time.NewTicker(interval) defer ticker.Stop() emit := func() { v, err := s.readSignal(ref) if err != nil { v = datasource.Value{Timestamp: time.Now(), Quality: datasource.QualityBad} } select { case ch <- v: case <-ctx.Done(): } } emit() for { select { case <-ticker.C: emit() case <-ctx.Done(): return } } }() return datasource.CancelFunc(cancel), nil } // Write sends the channel's write_cmd template with the value substituted. func (s *Scpi) Write(_ context.Context, signal string, value any) error { ref, ok := s.signals[signal] if !ok { return datasource.ErrNotFound } if !ref.ch.writable() { return datasource.ErrNotWritable } ic := s.instruments[ref.instrument] cmd := formatWrite(ref.ch.WriteCmd, value) return ic.tr.write(cmd) } // History is unavailable for SCPI instruments. func (s *Scpi) History(_ context.Context, _ string, _, _ time.Time, _ int) ([]datasource.Value, error) { return nil, datasource.ErrHistoryUnavailable } // Close tears down every instrument connection. func (s *Scpi) Close() { for _, ic := range s.instruments { ic.tr.close() } } // formatWrite substitutes value into the write template. A "%" in the template // is treated as a printf verb; otherwise the value is appended after a space. func formatWrite(tmpl string, value any) string { if strings.Contains(tmpl, "%") { return fmt.Sprintf(tmpl, value) } return fmt.Sprintf("%s %v", tmpl, value) } // parseValue converts a raw SCPI response string into the channel's data type. func parseValue(t datasource.DataType, resp string) (any, error) { resp = strings.TrimSpace(resp) switch t { case datasource.TypeString: return resp, nil case datasource.TypeInt64: // Accept "12", "12.0", or scientific notation by going through float. f, err := strconv.ParseFloat(resp, 64) if err != nil { return nil, fmt.Errorf("scpi: parse int %q: %w", resp, err) } return int64(f), nil case datasource.TypeBool: switch strings.ToUpper(resp) { case "1", "ON", "TRUE": return true, nil case "0", "OFF", "FALSE": return false, nil } f, err := strconv.ParseFloat(resp, 64) if err != nil { return nil, fmt.Errorf("scpi: parse bool %q: %w", resp, err) } return f != 0, nil default: f, err := strconv.ParseFloat(resp, 64) if err != nil { return nil, fmt.Errorf("scpi: parse float %q: %w", resp, err) } return f, nil } }