Implemented new datasources (modbus,scpi)

2026-06-24 06:12:52 +02:00
parent 999a1510d4
commit cf9da3df0a
13 changed files with 1967 additions and 0 deletions
@@ -0,0 +1,258 @@
+// Package modbus implements a Modbus TCP data source. Each configured device is
+// polled over its own TCP connection; registers are exposed as signals named
+// "device:register". Reads use the holding/input/coil/discrete function codes;
+// writable holding registers and coils accept Write.
+package modbus
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/uopi/uopi/internal/datasource"
+)
+
+const defaultPollInterval = time.Second
+
+// deviceClient bundles a parsed device with its wire client and register lookup.
+type deviceClient struct {
+	dev    Device
+	cli    *client
+	byName map[string]Register
+}
+
+// Modbus is a datasource.DataSource backed by one or more Modbus TCP devices.
+type Modbus struct {
+	pollInterval time.Duration
+	devices      map[string]*deviceClient // device name → client
+	signals      map[string]signalRef     // "device:register" → ref
+}
+
+type signalRef struct {
+	device string
+	reg    Register
+}
+
+// New builds a Modbus source from config. It does not dial; connections are
+// established lazily on first poll/write.
+func New(cfg Config) (*Modbus, error) {
+	poll := time.Duration(cfg.PollIntervalMs) * time.Millisecond
+	if poll <= 0 {
+		poll = defaultPollInterval
+	}
+	m := &Modbus{
+		pollInterval: poll,
+		devices:      make(map[string]*deviceClient),
+		signals:      make(map[string]signalRef),
+	}
+	for _, dev := range cfg.Devices {
+		if dev.Name == "" || dev.Address == "" {
+			return nil, fmt.Errorf("modbus: device needs name and address")
+		}
+		if _, dup := m.devices[dev.Name]; dup {
+			return nil, fmt.Errorf("modbus: duplicate device %q", dev.Name)
+		}
+		addr := dev.Address
+		if !strings.Contains(addr, ":") {
+			addr += ":502"
+		}
+		dc := &deviceClient{
+			dev:    dev,
+			cli:    newClient(addr, time.Duration(dev.TimeoutMs)*time.Millisecond),
+			byName: make(map[string]Register),
+		}
+		for _, reg := range dev.Registers {
+			if reg.Name == "" {
+				return nil, fmt.Errorf("modbus: device %q has a register with no name", dev.Name)
+			}
+			if _, err := reg.wordCount(); !reg.isBool() && err != nil {
+				return nil, err
+			}
+			if _, dup := dc.byName[reg.Name]; dup {
+				return nil, fmt.Errorf("modbus: device %q duplicate register %q", dev.Name, reg.Name)
+			}
+			dc.byName[reg.Name] = reg
+			m.signals[dev.Name+":"+reg.Name] = signalRef{device: dev.Name, reg: reg}
+		}
+		m.devices[dev.Name] = dc
+	}
+	return m, nil
+}
+
+// Name implements datasource.DataSource.
+func (m *Modbus) Name() string { return "modbus" }
+
+// Connect is a no-op; TCP connections are dialled lazily per device.
+func (m *Modbus) Connect(_ context.Context) error { return nil }
+
+// ListSignals returns metadata for every configured register.
+func (m *Modbus) ListSignals(_ context.Context) ([]datasource.Metadata, error) {
+	out := make([]datasource.Metadata, 0, len(m.signals))
+	for _, ref := range m.signals {
+		out = append(out, ref.reg.metadata(ref.device))
+	}
+	return out, nil
+}
+
+// GetMetadata returns metadata for one signal.
+func (m *Modbus) GetMetadata(_ context.Context, signal string) (datasource.Metadata, error) {
+	ref, ok := m.signals[signal]
+	if !ok {
+		return datasource.Metadata{}, datasource.ErrNotFound
+	}
+	return ref.reg.metadata(ref.device), nil
+}
+
+// readSignal performs one synchronous read of a register's current value.
+func (m *Modbus) readSignal(ref signalRef) (datasource.Value, error) {
+	dc := m.devices[ref.device]
+	reg := ref.reg
+	now := time.Now()
+
+	if reg.isBool() {
+		fc := byte(fcReadCoils)
+		if reg.kind() == kindDiscrete {
+			fc = fcReadDiscrete
+		}
+		bits, err := dc.cli.readBits(dc.dev.UnitID, fc, reg.Address, 1)
+		if err != nil {
+			return datasource.Value{}, err
+		}
+		return datasource.Value{Timestamp: now, Data: bits[0], Quality: datasource.QualityGood}, nil
+	}
+
+	count, err := reg.wordCount()
+	if err != nil {
+		return datasource.Value{}, err
+	}
+	fc := byte(fcReadHolding)
+	if reg.kind() == kindInput {
+		fc = fcReadInput
+	}
+	words, err := dc.cli.readRegisters(dc.dev.UnitID, fc, reg.Address, uint16(count))
+	if err != nil {
+		return datasource.Value{}, err
+	}
+	data, err := reg.decode(words)
+	if err != nil {
+		return datasource.Value{}, err
+	}
+	return datasource.Value{Timestamp: now, Data: data, Quality: datasource.QualityGood}, nil
+}
+
+// Subscribe polls the register at the configured interval and pushes values
+// into ch. On a read error a QualityBad value is emitted and polling continues.
+func (m *Modbus) Subscribe(ctx context.Context, signal string, ch chan<- datasource.Value) (datasource.CancelFunc, error) {
+	ref, ok := m.signals[signal]
+	if !ok {
+		return nil, datasource.ErrNotFound
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+	go func() {
+		ticker := time.NewTicker(m.pollInterval)
+		defer ticker.Stop()
+		emit := func() {
+			v, err := m.readSignal(ref)
+			if err != nil {
+				v = datasource.Value{Timestamp: time.Now(), Quality: datasource.QualityBad}
+			}
+			select {
+			case ch <- v:
+			case <-ctx.Done():
+			}
+		}
+		emit() // first reading without waiting a full interval
+		for {
+			select {
+			case <-ticker.C:
+				emit()
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	return datasource.CancelFunc(cancel), nil
+}
+
+// Write sets a writable holding register or coil.
+func (m *Modbus) Write(_ context.Context, signal string, value any) error {
+	ref, ok := m.signals[signal]
+	if !ok {
+		return datasource.ErrNotFound
+	}
+	reg := ref.reg
+	if !reg.writable() {
+		return datasource.ErrNotWritable
+	}
+	dc := m.devices[ref.device]
+
+	if reg.isBool() {
+		on, err := toBool(value)
+		if err != nil {
+			return err
+		}
+		return dc.cli.writeSingleCoil(dc.dev.UnitID, reg.Address, on)
+	}
+
+	f, err := toFloat(value)
+	if err != nil {
+		return err
+	}
+	words, err := reg.encode(f)
+	if err != nil {
+		return err
+	}
+	if len(words) == 1 {
+		return dc.cli.writeSingleRegister(dc.dev.UnitID, reg.Address, words[0])
+	}
+	return dc.cli.writeMultipleRegisters(dc.dev.UnitID, reg.Address, words)
+}
+
+// History is unavailable for Modbus devices.
+func (m *Modbus) History(_ context.Context, _ string, _, _ time.Time, _ int) ([]datasource.Value, error) {
+	return nil, datasource.ErrHistoryUnavailable
+}
+
+// Close tears down every device connection.
+func (m *Modbus) Close() {
+	for _, dc := range m.devices {
+		dc.cli.close()
+	}
+}
+
+func toFloat(v any) (float64, error) {
+	switch x := v.(type) {
+	case float64:
+		return x, nil
+	case float32:
+		return float64(x), nil
+	case int:
+		return float64(x), nil
+	case int64:
+		return float64(x), nil
+	case bool:
+		if x {
+			return 1, nil
+		}
+		return 0, nil
+	default:
+		return 0, fmt.Errorf("modbus: cannot write value of type %T", v)
+	}
+}
+
+func toBool(v any) (bool, error) {
+	switch x := v.(type) {
+	case bool:
+		return x, nil
+	case float64:
+		return x != 0, nil
+	case int:
+		return x != 0, nil
+	case int64:
+		return x != 0, nil
+	default:
+		return false, fmt.Errorf("modbus: cannot write bool value of type %T", v)
+	}
+}