From cf9da3df0adab847a1b36e561d39e40809edf65e Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Wed, 24 Jun 2026 06:12:52 +0200 Subject: [PATCH] Implemented new datasources (modbus,scpi) --- cmd/uopi/main.go | 31 ++ internal/config/config.go | 5 + internal/datasource/modbus/client.go | 216 +++++++++++++ internal/datasource/modbus/config.go | 152 +++++++++ internal/datasource/modbus/decode.go | 89 ++++++ internal/datasource/modbus/decode_test.go | 115 +++++++ internal/datasource/modbus/modbus.go | 258 +++++++++++++++ internal/datasource/modbus/modbus_test.go | 364 ++++++++++++++++++++++ internal/datasource/scpi/config.go | 75 +++++ internal/datasource/scpi/scpi.go | 237 ++++++++++++++ internal/datasource/scpi/scpi_test.go | 267 ++++++++++++++++ internal/datasource/scpi/transport.go | 107 +++++++ uopi.example.toml | 51 +++ 13 files changed, 1967 insertions(+) create mode 100644 internal/datasource/modbus/client.go create mode 100644 internal/datasource/modbus/config.go create mode 100644 internal/datasource/modbus/decode.go create mode 100644 internal/datasource/modbus/decode_test.go create mode 100644 internal/datasource/modbus/modbus.go create mode 100644 internal/datasource/modbus/modbus_test.go create mode 100644 internal/datasource/scpi/config.go create mode 100644 internal/datasource/scpi/scpi.go create mode 100644 internal/datasource/scpi/scpi_test.go create mode 100644 internal/datasource/scpi/transport.go diff --git a/cmd/uopi/main.go b/cmd/uopi/main.go index 8221dd7..dd163e0 100644 --- a/cmd/uopi/main.go +++ b/cmd/uopi/main.go @@ -21,7 +21,9 @@ import ( "github.com/uopi/uopi/internal/confmgr" "github.com/uopi/uopi/internal/controllogic" "github.com/uopi/uopi/internal/datasource/epics" + "github.com/uopi/uopi/internal/datasource/modbus" "github.com/uopi/uopi/internal/datasource/pva" + "github.com/uopi/uopi/internal/datasource/scpi" "github.com/uopi/uopi/internal/datasource/servervar" "github.com/uopi/uopi/internal/datasource/stub" "github.com/uopi/uopi/internal/datasource/synthetic" @@ -146,6 +148,35 @@ func main() { } } + // Modbus TCP data source: polls configured holding/input/coil/discrete + // registers on one or more devices. Disabled unless [datasource.modbus] is + // configured with devices. + if cfg.Datasource.Modbus.Enabled { + mbDS, err := modbus.New(cfg.Datasource.Modbus) + if err != nil { + log.Error("modbus init", "err", err) + } else if err := mbDS.Connect(ctx); err != nil { + log.Error("modbus connect", "err", err) + } else { + brk.Register(mbDS) + context.AfterFunc(ctx, mbDS.Close) + } + } + + // SCPI data source: polls instrument channels over raw TCP sockets. + // Disabled unless [datasource.scpi] is configured with instruments. + if cfg.Datasource.SCPI.Enabled { + scpiDS, err := scpi.New(cfg.Datasource.SCPI) + if err != nil { + log.Error("scpi init", "err", err) + } else if err := scpiDS.Connect(ctx); err != nil { + log.Error("scpi connect", "err", err) + } else { + brk.Register(scpiDS) + context.AfterFunc(ctx, scpiDS.Close) + } + } + // Server variables: a small persistent key/value source ("srv") that the // control-logic engine writes (e.g. sequence state) and panels can read. srvVars, err := servervar.New(cfg.Server.StorageDir) diff --git a/internal/config/config.go b/internal/config/config.go index cc34a68..b103ba4 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -7,6 +7,9 @@ import ( "strings" "github.com/BurntSushi/toml" + + "github.com/uopi/uopi/internal/datasource/modbus" + "github.com/uopi/uopi/internal/datasource/scpi" ) type Config struct { @@ -194,6 +197,8 @@ type DatasourceConfig struct { EPICS EPICSConfig `toml:"epics"` PVA PVAConfig `toml:"pva"` Synthetic SyntheticConfig `toml:"synthetic"` + Modbus modbus.Config `toml:"modbus"` + SCPI scpi.Config `toml:"scpi"` } type StubConfig struct { diff --git a/internal/datasource/modbus/client.go b/internal/datasource/modbus/client.go new file mode 100644 index 0000000..50e8de4 --- /dev/null +++ b/internal/datasource/modbus/client.go @@ -0,0 +1,216 @@ +package modbus + +import ( + "encoding/binary" + "fmt" + "io" + "net" + "sync" + "time" +) + +// Modbus function codes. +const ( + fcReadCoils = 0x01 + fcReadDiscrete = 0x02 + fcReadHolding = 0x03 + fcReadInput = 0x04 + fcWriteSingleCoil = 0x05 + fcWriteSingleReg = 0x06 + fcWriteMultipleRegs = 0x10 +) + +// client is a minimal Modbus TCP master for a single device address. Requests +// are serialised by mu (Modbus TCP is request/response and the connection is +// shared by all of a device's polled registers). The connection is dialled +// lazily and dropped on any I/O error so the next request reconnects. +type client struct { + addr string + timeout time.Duration + + mu sync.Mutex + conn net.Conn + txID uint16 +} + +func newClient(addr string, timeout time.Duration) *client { + if timeout <= 0 { + timeout = 3 * time.Second + } + return &client{addr: addr, timeout: timeout} +} + +func (c *client) close() { + c.mu.Lock() + c.closeLocked() + c.mu.Unlock() +} + +func (c *client) closeLocked() { + if c.conn != nil { + _ = c.conn.Close() + c.conn = nil + } +} + +// request sends a PDU to unitID and returns the response PDU (function code + +// data). It dials on demand and tears the connection down on error. +func (c *client) request(unitID byte, pdu []byte) ([]byte, error) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.conn == nil { + conn, err := net.DialTimeout("tcp", c.addr, c.timeout) + if err != nil { + return nil, fmt.Errorf("modbus: dial %s: %w", c.addr, err) + } + c.conn = conn + } + + resp, err := c.transact(unitID, pdu) + if err != nil { + c.closeLocked() + return nil, err + } + return resp, nil +} + +// transact performs one MBAP-framed exchange. Caller holds mu. +func (c *client) transact(unitID byte, pdu []byte) ([]byte, error) { + c.txID++ + tx := c.txID + + frame := make([]byte, 7+len(pdu)) + binary.BigEndian.PutUint16(frame[0:], tx) // transaction id + binary.BigEndian.PutUint16(frame[2:], 0) // protocol id (0 = Modbus) + binary.BigEndian.PutUint16(frame[4:], uint16(1+len(pdu))) // length: unit id + PDU + frame[6] = unitID + copy(frame[7:], pdu) + + _ = c.conn.SetDeadline(time.Now().Add(c.timeout)) + if _, err := c.conn.Write(frame); err != nil { + return nil, fmt.Errorf("modbus: write: %w", err) + } + + head := make([]byte, 7) + if _, err := io.ReadFull(c.conn, head); err != nil { + return nil, fmt.Errorf("modbus: read header: %w", err) + } + if binary.BigEndian.Uint16(head[0:]) != tx { + return nil, fmt.Errorf("modbus: transaction id mismatch") + } + length := binary.BigEndian.Uint16(head[4:]) + if length < 2 { // unit id + at least a function code + return nil, fmt.Errorf("modbus: short frame length %d", length) + } + body := make([]byte, length-1) // header already consumed the unit id + if _, err := io.ReadFull(c.conn, body); err != nil { + return nil, fmt.Errorf("modbus: read body: %w", err) + } + + fc := body[0] + if fc&0x80 != 0 { // exception response + var ex byte + if len(body) >= 2 { + ex = body[1] + } + return nil, fmt.Errorf("modbus: exception 0x%02x (%s)", ex, exceptionText(ex)) + } + return body, nil +} + +// readRegisters reads `quantity` 16-bit registers via fc (holding or input). +func (c *client) readRegisters(unitID, fc byte, addr, quantity uint16) ([]uint16, error) { + pdu := []byte{fc, byte(addr >> 8), byte(addr), byte(quantity >> 8), byte(quantity)} + resp, err := c.request(unitID, pdu) + if err != nil { + return nil, err + } + if len(resp) < 2 { + return nil, fmt.Errorf("modbus: short register response") + } + byteCount := int(resp[1]) + if byteCount != int(quantity)*2 || len(resp) < 2+byteCount { + return nil, fmt.Errorf("modbus: register byte count %d (want %d)", byteCount, quantity*2) + } + regs := make([]uint16, quantity) + for i := range regs { + regs[i] = binary.BigEndian.Uint16(resp[2+i*2:]) + } + return regs, nil +} + +// readBits reads `quantity` bits via fc (coils or discrete inputs). +func (c *client) readBits(unitID, fc byte, addr, quantity uint16) ([]bool, error) { + pdu := []byte{fc, byte(addr >> 8), byte(addr), byte(quantity >> 8), byte(quantity)} + resp, err := c.request(unitID, pdu) + if err != nil { + return nil, err + } + if len(resp) < 2 { + return nil, fmt.Errorf("modbus: short bit response") + } + byteCount := int(resp[1]) + if len(resp) < 2+byteCount { + return nil, fmt.Errorf("modbus: bit byte count %d exceeds frame", byteCount) + } + bits := make([]bool, quantity) + for i := range bits { + idx := 2 + i/8 + if idx >= len(resp) { + break + } + bits[i] = resp[idx]&(1<<(uint(i)%8)) != 0 + } + return bits, nil +} + +func (c *client) writeSingleRegister(unitID byte, addr, value uint16) error { + pdu := []byte{fcWriteSingleReg, byte(addr >> 8), byte(addr), byte(value >> 8), byte(value)} + _, err := c.request(unitID, pdu) + return err +} + +func (c *client) writeMultipleRegisters(unitID byte, addr uint16, values []uint16) error { + pdu := make([]byte, 6+len(values)*2) + pdu[0] = fcWriteMultipleRegs + binary.BigEndian.PutUint16(pdu[1:], addr) + binary.BigEndian.PutUint16(pdu[3:], uint16(len(values))) + pdu[5] = byte(len(values) * 2) + for i, v := range values { + binary.BigEndian.PutUint16(pdu[6+i*2:], v) + } + _, err := c.request(unitID, pdu) + return err +} + +func (c *client) writeSingleCoil(unitID byte, addr uint16, on bool) error { + var v uint16 + if on { + v = 0xFF00 + } + pdu := []byte{fcWriteSingleCoil, byte(addr >> 8), byte(addr), byte(v >> 8), byte(v)} + _, err := c.request(unitID, pdu) + return err +} + +func exceptionText(code byte) string { + switch code { + case 0x01: + return "illegal function" + case 0x02: + return "illegal data address" + case 0x03: + return "illegal data value" + case 0x04: + return "server device failure" + case 0x05: + return "acknowledge" + case 0x06: + return "server device busy" + case 0x0B: + return "gateway target failed to respond" + default: + return "unknown" + } +} diff --git a/internal/datasource/modbus/config.go b/internal/datasource/modbus/config.go new file mode 100644 index 0000000..aaad022 --- /dev/null +++ b/internal/datasource/modbus/config.go @@ -0,0 +1,152 @@ +package modbus + +import ( + "fmt" + "strings" + + "github.com/uopi/uopi/internal/datasource" +) + +// Config is the [datasource.modbus] section. Devices share no connection state; +// each is polled independently over its own TCP socket. +type Config struct { + Enabled bool `toml:"enabled"` + // PollIntervalMs is the default polling period for every register that does + // not override it. Zero → 1000 ms. + PollIntervalMs int `toml:"poll_interval_ms"` + Devices []Device `toml:"devices"` +} + +// Device is one Modbus TCP slave. Address is "host:port" (default port 502 is +// appended if absent). UnitID is the Modbus unit/slave identifier (0–255). +type Device struct { + Name string `toml:"name"` + Address string `toml:"address"` + UnitID uint8 `toml:"unit_id"` + TimeoutMs int `toml:"timeout_ms"` + Registers []Register `toml:"registers"` +} + +// Register describes one logical signal mapped onto a Modbus address. +// +// - Kind selects the address space / function code: +// "holding" (FC03/06/10), "input" (FC04, read-only), +// "coil" (FC01/05, bool), "discrete" (FC02, read-only bool). +// - Encoding selects how holding/input words are decoded: +// "uint16", "int16", "uint32", "int32", "float32", "float64". +// Ignored for coil/discrete (always bool). +// - WordOrder is "big" (default, high word first) or "little" for the +// multi-word encodings. +// - Scale/Offset transform the raw numeric value: value*Scale + Offset. +// Scale 0 is treated as 1. +type Register struct { + Name string `toml:"name"` + Kind string `toml:"kind"` + Address uint16 `toml:"address"` + Encoding string `toml:"encoding"` + WordOrder string `toml:"word_order"` + Unit string `toml:"unit"` + Scale float64 `toml:"scale"` + Offset float64 `toml:"offset"` + Min float64 `toml:"min"` + Max float64 `toml:"max"` + Writable bool `toml:"writable"` + Description string `toml:"description"` +} + +// register kinds. +const ( + kindHolding = "holding" + kindInput = "input" + kindCoil = "coil" + kindDiscrete = "discrete" +) + +// isBool reports whether the register addresses a single-bit space. +func (r Register) isBool() bool { + return r.kind() == kindCoil || r.kind() == kindDiscrete +} + +func (r Register) kind() string { + if r.Kind == "" { + return kindHolding + } + return strings.ToLower(r.Kind) +} + +func (r Register) encoding() string { + if r.Encoding == "" { + return "uint16" + } + return strings.ToLower(r.Encoding) +} + +func (r Register) littleWordOrder() bool { + return strings.ToLower(r.WordOrder) == "little" +} + +func (r Register) scale() float64 { + if r.Scale == 0 { + return 1 + } + return r.Scale +} + +// wordCount returns the number of 16-bit registers the encoding occupies. +func (r Register) wordCount() (int, error) { + switch r.encoding() { + case "uint16", "int16": + return 1, nil + case "uint32", "int32", "float32": + return 2, nil + case "float64": + return 4, nil + default: + return 0, fmt.Errorf("modbus: unknown encoding %q", r.Encoding) + } +} + +// dataType maps the register to a datasource value type. +func (r Register) dataType() datasource.DataType { + if r.isBool() { + return datasource.TypeBool + } + // Integer encodings with no fractional scaling stay integers; anything + // scaled, offset, or float-encoded becomes a float64. + switch r.encoding() { + case "float32", "float64": + return datasource.TypeFloat64 + default: + if r.scale() == 1 && r.Offset == 0 { + return datasource.TypeInt64 + } + return datasource.TypeFloat64 + } +} + +// writable reports whether writes are permitted. Input registers and discrete +// inputs are read-only regardless of the Writable flag. +func (r Register) writable() bool { + switch r.kind() { + case kindInput, kindDiscrete: + return false + default: + return r.Writable + } +} + +// metadata builds the datasource.Metadata for this register under signal name +// "device:register". +func (r Register) metadata(device string) datasource.Metadata { + return datasource.Metadata{ + Name: device + ":" + r.Name, + Type: r.dataType(), + Unit: r.Unit, + Description: r.Description, + DisplayLow: r.Min, + DisplayHigh: r.Max, + DriveLow: r.Min, + DriveHigh: r.Max, + Writable: r.writable(), + } +} diff --git a/internal/datasource/modbus/decode.go b/internal/datasource/modbus/decode.go new file mode 100644 index 0000000..f894659 --- /dev/null +++ b/internal/datasource/modbus/decode.go @@ -0,0 +1,89 @@ +package modbus + +import ( + "fmt" + "math" +) + +// orderWords returns the registers in big-word-first order, reversing them when +// the register declares little word order. The slice is copied so the caller's +// data is left untouched. +func (r Register) orderWords(words []uint16) []uint16 { + if !r.littleWordOrder() { + return words + } + out := make([]uint16, len(words)) + for i, w := range words { + out[len(words)-1-i] = w + } + return out +} + +// decode converts the raw registers into the register's numeric value and +// applies scale/offset. The result type is int64 or float64 per dataType. +func (r Register) decode(words []uint16) (any, error) { + w := r.orderWords(words) + var raw float64 + var rawInt int64 + switch r.encoding() { + case "uint16": + rawInt = int64(w[0]) + raw = float64(w[0]) + case "int16": + rawInt = int64(int16(w[0])) + raw = float64(int16(w[0])) + case "uint32": + u := uint32(w[0])<<16 | uint32(w[1]) + rawInt = int64(u) + raw = float64(u) + case "int32": + u := uint32(w[0])<<16 | uint32(w[1]) + rawInt = int64(int32(u)) + raw = float64(int32(u)) + case "float32": + u := uint32(w[0])<<16 | uint32(w[1]) + raw = float64(math.Float32frombits(u)) + case "float64": + u := uint64(w[0])<<48 | uint64(w[1])<<32 | uint64(w[2])<<16 | uint64(w[3]) + raw = math.Float64frombits(u) + default: + return nil, fmt.Errorf("modbus: unknown encoding %q", r.Encoding) + } + + // Integer fast-path: no scaling/offset, integer encoding. + if r.scale() == 1 && r.Offset == 0 { + switch r.encoding() { + case "uint16", "int16", "uint32", "int32": + return rawInt, nil + } + } + return raw*r.scale() + r.Offset, nil +} + +// encode converts a value destined for a Write back into raw registers, +// inverting scale/offset. Only the holding-register encodings are writable. +func (r Register) encode(value float64) ([]uint16, error) { + v := (value - r.Offset) / r.scale() + var words []uint16 + switch r.encoding() { + case "uint16": + words = []uint16{uint16(int64(math.Round(v)))} + case "int16": + words = []uint16{uint16(int16(int64(math.Round(v))))} + case "uint32": + u := uint32(int64(math.Round(v))) + words = []uint16{uint16(u >> 16), uint16(u)} + case "int32": + u := uint32(int32(int64(math.Round(v)))) + words = []uint16{uint16(u >> 16), uint16(u)} + case "float32": + u := math.Float32bits(float32(v)) + words = []uint16{uint16(u >> 16), uint16(u)} + case "float64": + u := math.Float64bits(v) + words = []uint16{uint16(u >> 48), uint16(u >> 32), uint16(u >> 16), uint16(u)} + default: + return nil, fmt.Errorf("modbus: unknown encoding %q", r.Encoding) + } + return r.orderWords(words), nil +} diff --git a/internal/datasource/modbus/decode_test.go b/internal/datasource/modbus/decode_test.go new file mode 100644 index 0000000..df30ac1 --- /dev/null +++ b/internal/datasource/modbus/decode_test.go @@ -0,0 +1,115 @@ +package modbus + +import ( + "math" + "testing" + + "github.com/uopi/uopi/internal/datasource" +) + +func TestDecodeEncodings(t *testing.T) { + cases := []struct { + name string + reg Register + words []uint16 + want any + }{ + {"uint16", Register{Encoding: "uint16"}, []uint16{42}, int64(42)}, + {"int16 neg", Register{Encoding: "int16"}, []uint16{0xFFFF}, int64(-1)}, + {"uint32", Register{Encoding: "uint32"}, []uint16{0x0001, 0x0000}, int64(65536)}, + {"int32 neg", Register{Encoding: "int32"}, []uint16{0xFFFF, 0xFFFF}, int64(-1)}, + {"scaled", Register{Encoding: "int16", Scale: 0.1}, []uint16{235}, 23.5}, + {"float32", Register{Encoding: "float32"}, f32Words(3.5), 3.5}, + {"float64", Register{Encoding: "float64"}, f64Words(2.25), 2.25}, + } + for _, tc := range cases { + got, err := tc.reg.decode(tc.words) + if err != nil { + t.Errorf("%s: decode error %v", tc.name, err) + continue + } + switch w := tc.want.(type) { + case int64: + if got != w { + t.Errorf("%s: got %v (%T), want %v", tc.name, got, got, w) + } + case float64: + gf, ok := got.(float64) + if !ok || math.Abs(gf-w) > 1e-6 { + t.Errorf("%s: got %v (%T), want %v", tc.name, got, got, w) + } + } + } +} + +func TestWordOrder(t *testing.T) { + big := Register{Encoding: "uint32", WordOrder: "big"} + little := Register{Encoding: "uint32", WordOrder: "little"} + words := []uint16{0x0001, 0x0002} // big: 0x00010002, little reverses to 0x00020001 + gb, _ := big.decode(words) + gl, _ := little.decode(words) + if gb != int64(0x00010002) { + t.Errorf("big = %v, want %d", gb, 0x00010002) + } + if gl != int64(0x00020001) { + t.Errorf("little = %v, want %d", gl, 0x00020001) + } +} + +func TestEncodeRoundTrip(t *testing.T) { + for _, enc := range []string{"uint16", "int16", "uint32", "int32", "float32", "float64"} { + reg := Register{Encoding: enc} + words, err := reg.encode(123) + if err != nil { + t.Fatalf("%s encode: %v", enc, err) + } + got, err := reg.decode(words) + if err != nil { + t.Fatalf("%s decode: %v", enc, err) + } + var f float64 + switch v := got.(type) { + case int64: + f = float64(v) + case float64: + f = v + } + if math.Abs(f-123) > 1e-3 { + t.Errorf("%s round-trip = %v, want 123", enc, f) + } + } +} + +func TestDataTypeAndWritable(t *testing.T) { + if got := (Register{Kind: "coil"}).dataType(); got != datasource.TypeBool { + t.Errorf("coil type = %v, want bool", got) + } + if got := (Register{Encoding: "float32"}).dataType(); got != datasource.TypeFloat64 { + t.Errorf("float type = %v, want float64", got) + } + if got := (Register{Encoding: "uint16", Scale: 0.5}).dataType(); got != datasource.TypeFloat64 { + t.Errorf("scaled int type = %v, want float64", got) + } + if got := (Register{Encoding: "uint16"}).dataType(); got != datasource.TypeInt64 { + t.Errorf("plain int type = %v, want int64", got) + } + if (Register{Kind: "input", Writable: true}).writable() { + t.Error("input register must be read-only") + } + if (Register{Kind: "discrete", Writable: true}).writable() { + t.Error("discrete input must be read-only") + } + if !(Register{Kind: "holding", Writable: true}).writable() { + t.Error("writable holding should be writable") + } +} + +func f32Words(v float32) []uint16 { + u := math.Float32bits(v) + return []uint16{uint16(u >> 16), uint16(u)} +} + +func f64Words(v float64) []uint16 { + u := math.Float64bits(v) + return []uint16{uint16(u >> 48), uint16(u >> 32), uint16(u >> 16), uint16(u)} +} diff --git a/internal/datasource/modbus/modbus.go b/internal/datasource/modbus/modbus.go new file mode 100644 index 0000000..2cca308 --- /dev/null +++ b/internal/datasource/modbus/modbus.go @@ -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) + } +} diff --git a/internal/datasource/modbus/modbus_test.go b/internal/datasource/modbus/modbus_test.go new file mode 100644 index 0000000..f3d8361 --- /dev/null +++ b/internal/datasource/modbus/modbus_test.go @@ -0,0 +1,364 @@ +package modbus + +import ( + "context" + "encoding/binary" + "io" + "net" + "sync" + "testing" + "time" + + "github.com/uopi/uopi/internal/datasource" +) + +// mockServer is an in-process Modbus TCP slave for tests. It serves a small +// register/coil store and records writes. Only the function codes exercised by +// the data source are implemented. +type mockServer struct { + ln net.Listener + + mu sync.Mutex + holding map[uint16]uint16 + input map[uint16]uint16 + coils map[uint16]bool + discrete map[uint16]bool + lastWrite []uint16 // registers from the most recent write +} + +func newMockServer(t *testing.T) *mockServer { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + s := &mockServer{ + ln: ln, + holding: map[uint16]uint16{}, + input: map[uint16]uint16{}, + coils: map[uint16]bool{}, + discrete: map[uint16]bool{}, + } + go s.serve() + t.Cleanup(func() { ln.Close() }) + return s +} + +func (s *mockServer) addr() string { return s.ln.Addr().String() } + +func (s *mockServer) serve() { + for { + conn, err := s.ln.Accept() + if err != nil { + return + } + go s.handle(conn) + } +} + +func (s *mockServer) handle(conn net.Conn) { + defer conn.Close() + for { + head := make([]byte, 7) + if _, err := io.ReadFull(conn, head); err != nil { + return + } + tx := binary.BigEndian.Uint16(head[0:]) + length := binary.BigEndian.Uint16(head[4:]) + body := make([]byte, length-1) + if _, err := io.ReadFull(conn, body); err != nil { + return + } + resp := s.respond(body) + out := make([]byte, 7+len(resp)) + binary.BigEndian.PutUint16(out[0:], tx) + binary.BigEndian.PutUint16(out[2:], 0) + binary.BigEndian.PutUint16(out[4:], uint16(1+len(resp))) + out[6] = head[6] + copy(out[7:], resp) + if _, err := conn.Write(out); err != nil { + return + } + } +} + +func (s *mockServer) respond(pdu []byte) []byte { + s.mu.Lock() + defer s.mu.Unlock() + fc := pdu[0] + switch fc { + case fcReadHolding, fcReadInput: + addr := binary.BigEndian.Uint16(pdu[1:]) + qty := binary.BigEndian.Uint16(pdu[3:]) + out := []byte{fc, byte(qty * 2)} + src := s.holding + if fc == fcReadInput { + src = s.input + } + for i := uint16(0); i < qty; i++ { + out = binary.BigEndian.AppendUint16(out, src[addr+i]) + } + return out + case fcReadCoils, fcReadDiscrete: + addr := binary.BigEndian.Uint16(pdu[1:]) + qty := binary.BigEndian.Uint16(pdu[3:]) + nbytes := (int(qty) + 7) / 8 + out := []byte{fc, byte(nbytes)} + bits := make([]byte, nbytes) + src := s.coils + if fc == fcReadDiscrete { + src = s.discrete + } + for i := uint16(0); i < qty; i++ { + if src[addr+i] { + bits[i/8] |= 1 << (i % 8) + } + } + return append(out, bits...) + case fcWriteSingleReg: + addr := binary.BigEndian.Uint16(pdu[1:]) + val := binary.BigEndian.Uint16(pdu[3:]) + s.holding[addr] = val + s.lastWrite = []uint16{val} + return pdu // echo + case fcWriteSingleCoil: + addr := binary.BigEndian.Uint16(pdu[1:]) + s.coils[addr] = binary.BigEndian.Uint16(pdu[3:]) == 0xFF00 + return pdu + case fcWriteMultipleRegs: + addr := binary.BigEndian.Uint16(pdu[1:]) + qty := binary.BigEndian.Uint16(pdu[3:]) + s.lastWrite = nil + for i := uint16(0); i < qty; i++ { + v := binary.BigEndian.Uint16(pdu[6+i*2:]) + s.holding[addr+i] = v + s.lastWrite = append(s.lastWrite, v) + } + return append([]byte{fc}, pdu[1:5]...) + default: + return []byte{fc | 0x80, 0x01} + } +} + +func (s *mockServer) setHolding(addr, val uint16) { + s.mu.Lock() + s.holding[addr] = val + s.mu.Unlock() +} + +func (s *mockServer) setInput(addr, val uint16) { + s.mu.Lock() + s.input[addr] = val + s.mu.Unlock() +} + +func (s *mockServer) setDiscrete(addr uint16, on bool) { + s.mu.Lock() + s.discrete[addr] = on + s.mu.Unlock() +} + +func testConfig(addr string) Config { + return Config{ + Enabled: true, + PollIntervalMs: 20, + Devices: []Device{{ + Name: "dev", + Address: addr, + UnitID: 1, + Registers: []Register{ + {Name: "temp", Kind: "holding", Address: 10, Encoding: "int16", Scale: 0.1, Unit: "C", Writable: true}, + {Name: "count", Kind: "holding", Address: 20, Encoding: "uint16"}, + {Name: "big", Kind: "input", Address: 30, Encoding: "uint32"}, + {Name: "flag", Kind: "discrete", Address: 5}, + {Name: "relay", Kind: "coil", Address: 6, Writable: true}, + {Name: "sp", Kind: "holding", Address: 40, Encoding: "float32", Writable: true}, + }, + }}, + } +} + +func TestReadRegisters(t *testing.T) { + srv := newMockServer(t) + srv.setHolding(10, 235) // int16, scale 0.1 → 23.5 + srv.setHolding(20, 7) + srv.setInput(30, 0) + srv.setInput(31, 1000) // uint32 big-word-first: low word at 31 + srv.setDiscrete(5, true) + + m, err := New(testConfig(srv.addr())) + if err != nil { + t.Fatalf("New: %v", err) + } + defer m.Close() + + temp, err := m.readSignal(m.signals["dev:temp"]) + if err != nil { + t.Fatalf("read temp: %v", err) + } + if f, ok := temp.Data.(float64); !ok || f < 23.49 || f > 23.51 { + t.Errorf("temp = %v (%T), want 23.5", temp.Data, temp.Data) + } + + count, err := m.readSignal(m.signals["dev:count"]) + if err != nil { + t.Fatalf("read count: %v", err) + } + if v, ok := count.Data.(int64); !ok || v != 7 { + t.Errorf("count = %v (%T), want int64 7", count.Data, count.Data) + } + + big, err := m.readSignal(m.signals["dev:big"]) + if err != nil { + t.Fatalf("read big: %v", err) + } + if v, ok := big.Data.(int64); !ok || v != 1000 { + t.Errorf("big = %v (%T), want int64 1000", big.Data, big.Data) + } + + flag, err := m.readSignal(m.signals["dev:flag"]) + if err != nil { + t.Fatalf("read flag: %v", err) + } + if b, ok := flag.Data.(bool); !ok || !b { + t.Errorf("flag = %v, want true", flag.Data) + } +} + +func TestWriteRoundTrip(t *testing.T) { + srv := newMockServer(t) + m, err := New(testConfig(srv.addr())) + if err != nil { + t.Fatalf("New: %v", err) + } + defer m.Close() + ctx := context.Background() + + // Scaled int16: writing 23.5 with scale 0.1 should store raw 235. + if err := m.Write(ctx, "dev:temp", 23.5); err != nil { + t.Fatalf("write temp: %v", err) + } + srv.mu.Lock() + raw := srv.holding[10] + srv.mu.Unlock() + if raw != 235 { + t.Errorf("holding[10] = %d, want 235", raw) + } + + // Coil write. + if err := m.Write(ctx, "dev:relay", true); err != nil { + t.Fatalf("write relay: %v", err) + } + srv.mu.Lock() + on := srv.coils[6] + srv.mu.Unlock() + if !on { + t.Error("coil 6 not set") + } + + // float32 multi-register write. + if err := m.Write(ctx, "dev:sp", 12.5); err != nil { + t.Fatalf("write sp: %v", err) + } + got, err := m.readSignal(m.signals["dev:sp"]) + if err != nil { + t.Fatalf("read sp: %v", err) + } + if f, ok := got.Data.(float64); !ok || f < 12.49 || f > 12.51 { + t.Errorf("sp = %v, want 12.5", got.Data) + } +} + +func TestWriteErrors(t *testing.T) { + srv := newMockServer(t) + m, _ := New(testConfig(srv.addr())) + defer m.Close() + ctx := context.Background() + + if err := m.Write(ctx, "dev:missing", 1); err != datasource.ErrNotFound { + t.Errorf("missing write err = %v, want ErrNotFound", err) + } + // Input register is read-only. + if err := m.Write(ctx, "dev:big", 1); err != datasource.ErrNotWritable { + t.Errorf("input write err = %v, want ErrNotWritable", err) + } +} + +func TestSubscribe(t *testing.T) { + srv := newMockServer(t) + srv.setHolding(20, 42) + m, _ := New(testConfig(srv.addr())) + defer m.Close() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + ch := make(chan datasource.Value, 4) + stop, err := m.Subscribe(ctx, "dev:count", ch) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + defer stop() + + select { + case v := <-ch: + if v.Quality != datasource.QualityGood { + t.Errorf("quality = %v, want good", v.Quality) + } + if iv, ok := v.Data.(int64); !ok || iv != 42 { + t.Errorf("first value = %v, want 42", v.Data) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for first value") + } +} + +func TestSubscribeBadQualityOnError(t *testing.T) { + // Point at a closed port so reads fail; expect QualityBad, not a hang. + cfg := testConfig("127.0.0.1:1") // port 1: connection refused + m, _ := New(cfg) + defer m.Close() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + ch := make(chan datasource.Value, 1) + stop, err := m.Subscribe(ctx, "dev:count", ch) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + defer stop() + + select { + case v := <-ch: + if v.Quality != datasource.QualityBad { + t.Errorf("quality = %v, want bad", v.Quality) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out") + } +} + +func TestNewValidation(t *testing.T) { + if _, err := New(Config{Devices: []Device{{Name: "", Address: "x"}}}); err == nil { + t.Error("expected error for missing device name") + } + if _, err := New(Config{Devices: []Device{{Name: "a", Address: "x"}, {Name: "a", Address: "y"}}}); err == nil { + t.Error("expected error for duplicate device") + } + if _, err := New(Config{Devices: []Device{{Name: "a", Address: "x", Registers: []Register{{Name: "r", Encoding: "bogus"}}}}}); err == nil { + t.Error("expected error for bad encoding") + } + if _, err := New(Config{Devices: []Device{{Name: "a", Address: "x", Registers: []Register{{Name: "r"}, {Name: "r"}}}}}); err == nil { + t.Error("expected error for duplicate register") + } +} + +func TestSubscribeUnknownSignal(t *testing.T) { + m, _ := New(testConfig("127.0.0.1:502")) + defer m.Close() + if _, err := m.Subscribe(context.Background(), "nope", nil); err != datasource.ErrNotFound { + t.Errorf("err = %v, want ErrNotFound", err) + } + if _, err := m.GetMetadata(context.Background(), "nope"); err != datasource.ErrNotFound { + t.Errorf("meta err = %v, want ErrNotFound", err) + } +} diff --git a/internal/datasource/scpi/config.go b/internal/datasource/scpi/config.go new file mode 100644 index 0000000..c2aa378 --- /dev/null +++ b/internal/datasource/scpi/config.go @@ -0,0 +1,75 @@ +package scpi + +import "github.com/uopi/uopi/internal/datasource" + +// Config is the [datasource.scpi] section. Each instrument is polled +// independently over its own TCP socket. +type Config struct { + Enabled bool `toml:"enabled"` + // PollIntervalMs is the default polling period for channels that do not + // override it. Zero → 1000 ms. + PollIntervalMs int `toml:"poll_interval_ms"` + Instruments []Instrument `toml:"instruments"` +} + +// Instrument is one SCPI device reachable over a raw TCP socket. Address is +// "host:port"; the conventional SCPI-raw port 5025 is appended if absent. +type Instrument struct { + Name string `toml:"name"` + // Transport selects the link type. "raw" (default) is line-based SCPI over + // TCP. Reserved: "vxi11" (not yet implemented). + Transport string `toml:"transport"` + Address string `toml:"address"` + TimeoutMs int `toml:"timeout_ms"` + // Terminator is appended to every command. Empty → "\n". + Terminator string `toml:"terminator"` + Channels []Channel `toml:"channels"` +} + +// Channel maps a SCPI query/command pair onto a signal named +// "instrument:channel". +type Channel struct { + Name string `toml:"name"` + // Query is the SCPI command whose response is the channel value, + // e.g. "MEAS:VOLT?". Required. + Query string `toml:"query"` + // WriteCmd is a printf-style template used by Write; "%v" is replaced with + // the value, e.g. "VOLT %v". Empty → channel is read-only. + WriteCmd string `toml:"write_cmd"` + // Type is the value type: "float" (default), "string", "int", or "bool". + Type string `toml:"type"` + Unit string `toml:"unit"` + Min float64 `toml:"min"` + Max float64 `toml:"max"` + PollIntervalMs int `toml:"poll_interval_ms"` + Description string `toml:"description"` +} + +func (c Channel) dataType() datasource.DataType { + switch c.Type { + case "string": + return datasource.TypeString + case "int": + return datasource.TypeInt64 + case "bool": + return datasource.TypeBool + default: + return datasource.TypeFloat64 + } +} + +func (c Channel) writable() bool { return c.WriteCmd != "" } + +func (c Channel) metadata(instrument string) datasource.Metadata { + return datasource.Metadata{ + Name: instrument + ":" + c.Name, + Type: c.dataType(), + Unit: c.Unit, + Description: c.Description, + DisplayLow: c.Min, + DisplayHigh: c.Max, + DriveLow: c.Min, + DriveHigh: c.Max, + Writable: c.writable(), + } +} diff --git a/internal/datasource/scpi/scpi.go b/internal/datasource/scpi/scpi.go new file mode 100644 index 0000000..91f676b --- /dev/null +++ b/internal/datasource/scpi/scpi.go @@ -0,0 +1,237 @@ +// 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 + } +} diff --git a/internal/datasource/scpi/scpi_test.go b/internal/datasource/scpi/scpi_test.go new file mode 100644 index 0000000..ee2de64 --- /dev/null +++ b/internal/datasource/scpi/scpi_test.go @@ -0,0 +1,267 @@ +package scpi + +import ( + "bufio" + "context" + "net" + "strings" + "sync" + "testing" + "time" + + "github.com/uopi/uopi/internal/datasource" +) + +// mockInstrument is an in-process line-based SCPI server. It answers queries +// from a fixed table and records commands that produce no response (writes). +type mockInstrument struct { + ln net.Listener + + mu sync.Mutex + answers map[string]string // query → response + writes []string +} + +func newMockInstrument(t *testing.T) *mockInstrument { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + m := &mockInstrument{ln: ln, answers: map[string]string{}} + go m.serve() + t.Cleanup(func() { ln.Close() }) + return m +} + +func (m *mockInstrument) addr() string { return m.ln.Addr().String() } + +func (m *mockInstrument) setAnswer(q, a string) { + m.mu.Lock() + m.answers[q] = a + m.mu.Unlock() +} + +func (m *mockInstrument) writeLog() []string { + m.mu.Lock() + defer m.mu.Unlock() + return append([]string(nil), m.writes...) +} + +func (m *mockInstrument) serve() { + for { + conn, err := m.ln.Accept() + if err != nil { + return + } + go m.handle(conn) + } +} + +func (m *mockInstrument) handle(conn net.Conn) { + defer conn.Close() + br := bufio.NewReader(conn) + for { + line, err := br.ReadString('\n') + if err != nil { + return + } + cmd := strings.TrimRight(line, "\r\n") + m.mu.Lock() + if strings.HasSuffix(cmd, "?") { + resp, ok := m.answers[cmd] + if !ok { + resp = "0" + } + m.mu.Unlock() + conn.Write([]byte(resp + "\n")) + continue + } + m.writes = append(m.writes, cmd) + m.mu.Unlock() + } +} + +func testCfg(addr string) Config { + return Config{ + Enabled: true, + PollIntervalMs: 20, + Instruments: []Instrument{{ + Name: "dmm", + Address: addr, + Channels: []Channel{ + {Name: "volt", Query: "MEAS:VOLT?", WriteCmd: "VOLT %v", Type: "float", Unit: "V"}, + {Name: "id", Query: "*IDN?", Type: "string"}, + {Name: "n", Query: "COUNT?", Type: "int"}, + {Name: "out", Query: "OUTP?", WriteCmd: "OUTP", Type: "bool"}, + }, + }}, + } +} + +func TestQueryTypes(t *testing.T) { + srv := newMockInstrument(t) + srv.setAnswer("MEAS:VOLT?", "12.34") + srv.setAnswer("*IDN?", "ACME,DMM,1,2.0") + srv.setAnswer("COUNT?", "7") + srv.setAnswer("OUTP?", "ON") + + s, err := New(testCfg(srv.addr())) + if err != nil { + t.Fatalf("New: %v", err) + } + defer s.Close() + + v, err := s.readSignal(s.signals["dmm:volt"]) + if err != nil { + t.Fatalf("read volt: %v", err) + } + if f, ok := v.Data.(float64); !ok || f < 12.33 || f > 12.35 { + t.Errorf("volt = %v (%T), want 12.34", v.Data, v.Data) + } + + id, _ := s.readSignal(s.signals["dmm:id"]) + if id.Data != "ACME,DMM,1,2.0" { + t.Errorf("id = %v", id.Data) + } + + n, _ := s.readSignal(s.signals["dmm:n"]) + if iv, ok := n.Data.(int64); !ok || iv != 7 { + t.Errorf("n = %v (%T), want 7", n.Data, n.Data) + } + + out, _ := s.readSignal(s.signals["dmm:out"]) + if b, ok := out.Data.(bool); !ok || !b { + t.Errorf("out = %v, want true", out.Data) + } +} + +func TestWrite(t *testing.T) { + srv := newMockInstrument(t) + s, _ := New(testCfg(srv.addr())) + defer s.Close() + ctx := context.Background() + + if err := s.Write(ctx, "dmm:volt", 3.3); err != nil { + t.Fatalf("write volt: %v", err) + } + // bool write uses a template with no verb → "OUTP ". + if err := s.Write(ctx, "dmm:out", true); err != nil { + t.Fatalf("write out: %v", err) + } + + // Give the server a moment to record both writes. + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + if len(srv.writeLog()) >= 2 { + break + } + time.Sleep(5 * time.Millisecond) + } + got := srv.writeLog() + if len(got) != 2 || got[0] != "VOLT 3.3" || got[1] != "OUTP true" { + t.Errorf("writes = %v, want [VOLT 3.3, OUTP true]", got) + } +} + +func TestWriteErrors(t *testing.T) { + srv := newMockInstrument(t) + s, _ := New(testCfg(srv.addr())) + defer s.Close() + ctx := context.Background() + + if err := s.Write(ctx, "dmm:missing", 1); err != datasource.ErrNotFound { + t.Errorf("missing = %v, want ErrNotFound", err) + } + if err := s.Write(ctx, "dmm:id", 1); err != datasource.ErrNotWritable { + t.Errorf("read-only = %v, want ErrNotWritable", err) + } +} + +func TestSubscribe(t *testing.T) { + srv := newMockInstrument(t) + srv.setAnswer("MEAS:VOLT?", "5.0") + s, _ := New(testCfg(srv.addr())) + defer s.Close() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + ch := make(chan datasource.Value, 4) + stop, err := s.Subscribe(ctx, "dmm:volt", ch) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + defer stop() + + select { + case v := <-ch: + if v.Quality != datasource.QualityGood { + t.Errorf("quality = %v", v.Quality) + } + if f, ok := v.Data.(float64); !ok || f != 5.0 { + t.Errorf("value = %v, want 5.0", v.Data) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out") + } +} + +func TestSubscribeBadQuality(t *testing.T) { + cfg := testCfg("127.0.0.1:1") // refused + s, _ := New(cfg) + defer s.Close() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + ch := make(chan datasource.Value, 1) + stop, _ := s.Subscribe(ctx, "dmm:volt", ch) + defer stop() + + select { + case v := <-ch: + if v.Quality != datasource.QualityBad { + t.Errorf("quality = %v, want bad", v.Quality) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out") + } +} + +func TestValidation(t *testing.T) { + if _, err := New(Config{Instruments: []Instrument{{Name: "", Address: "x"}}}); err == nil { + t.Error("want error for missing name") + } + if _, err := New(Config{Instruments: []Instrument{{Name: "a", Address: "x"}, {Name: "a", Address: "y"}}}); err == nil { + t.Error("want error for duplicate instrument") + } + if _, err := New(Config{Instruments: []Instrument{{Name: "a", Address: "x", Transport: "vxi11"}}}); err == nil { + t.Error("want error for unimplemented vxi11 transport") + } + if _, err := New(Config{Instruments: []Instrument{{Name: "a", Address: "x", Transport: "bogus"}}}); err == nil { + t.Error("want error for unknown transport") + } + if _, err := New(Config{Instruments: []Instrument{{Name: "a", Address: "x", Channels: []Channel{{Name: "c"}}}}}); err == nil { + t.Error("want error for channel without query") + } +} + +func TestFormatWrite(t *testing.T) { + if got := formatWrite("VOLT %v", 3.3); got != "VOLT 3.3" { + t.Errorf("verb template = %q", got) + } + if got := formatWrite("OUTP", true); got != "OUTP true" { + t.Errorf("plain template = %q", got) + } +} + +func TestParseValue(t *testing.T) { + if v, _ := parseValue(datasource.TypeBool, "OFF"); v != false { + t.Errorf("OFF = %v, want false", v) + } + if v, _ := parseValue(datasource.TypeBool, "2.0"); v != true { + t.Errorf("2.0 bool = %v, want true", v) + } + if _, err := parseValue(datasource.TypeFloat64, "notnum"); err == nil { + t.Error("want parse error for non-numeric float") + } +} diff --git a/internal/datasource/scpi/transport.go b/internal/datasource/scpi/transport.go new file mode 100644 index 0000000..0669618 --- /dev/null +++ b/internal/datasource/scpi/transport.go @@ -0,0 +1,107 @@ +package scpi + +import ( + "bufio" + "fmt" + "net" + "strings" + "sync" + "time" +) + +// transport is the request/response channel to an instrument. The raw-socket +// implementation below speaks line-oriented SCPI over TCP (the common +// "SCPI raw" / port 5025 convention). The interface is kept narrow so a VXI-11 +// (ONC-RPC) transport can be added later without touching the data source. +type transport interface { + // query sends cmd and returns the instrument's single-line response. + query(cmd string) (string, error) + // write sends cmd and does not wait for a response. + write(cmd string) error + close() +} + +// rawSocket is a line-based SCPI transport over a single TCP connection. The +// connection is dialled lazily and dropped on any I/O error so the next call +// reconnects. Calls are serialised by mu because SCPI is request/response. +type rawSocket struct { + addr string + timeout time.Duration + terminator string + + mu sync.Mutex + conn net.Conn + br *bufio.Reader +} + +func newRawSocket(addr string, timeout time.Duration, terminator string) *rawSocket { + if timeout <= 0 { + timeout = 3 * time.Second + } + if terminator == "" { + terminator = "\n" + } + return &rawSocket{addr: addr, timeout: timeout, terminator: terminator} +} + +func (s *rawSocket) dialLocked() error { + if s.conn != nil { + return nil + } + conn, err := net.DialTimeout("tcp", s.addr, s.timeout) + if err != nil { + return fmt.Errorf("scpi: dial %s: %w", s.addr, err) + } + s.conn = conn + s.br = bufio.NewReader(conn) + return nil +} + +func (s *rawSocket) closeLocked() { + if s.conn != nil { + _ = s.conn.Close() + s.conn = nil + s.br = nil + } +} + +func (s *rawSocket) close() { + s.mu.Lock() + s.closeLocked() + s.mu.Unlock() +} + +func (s *rawSocket) sendLocked(cmd string) error { + _ = s.conn.SetDeadline(time.Now().Add(s.timeout)) + if _, err := s.conn.Write([]byte(cmd + s.terminator)); err != nil { + s.closeLocked() + return fmt.Errorf("scpi: write: %w", err) + } + return nil +} + +func (s *rawSocket) write(cmd string) error { + s.mu.Lock() + defer s.mu.Unlock() + if err := s.dialLocked(); err != nil { + return err + } + return s.sendLocked(cmd) +} + +func (s *rawSocket) query(cmd string) (string, error) { + s.mu.Lock() + defer s.mu.Unlock() + if err := s.dialLocked(); err != nil { + return "", err + } + if err := s.sendLocked(cmd); err != nil { + return "", err + } + line, err := s.br.ReadString('\n') + if err != nil { + s.closeLocked() + return "", fmt.Errorf("scpi: read: %w", err) + } + return strings.TrimRight(line, "\r\n"), nil +} diff --git a/uopi.example.toml b/uopi.example.toml index 035c7ce..09b81dc 100644 --- a/uopi.example.toml +++ b/uopi.example.toml @@ -129,3 +129,54 @@ auto_sync_filter = "" # e.g. area=StorageRing or tags=production [datasource.synthetic] enabled = true + +# Modbus TCP. Each device is polled over its own connection; registers become +# signals named "device:register". Disabled unless enabled = true and at least +# one device is declared. +[datasource.modbus] +enabled = false +poll_interval_ms = 1000 # default poll period for every register + +# [[datasource.modbus.devices]] +# name = "plc1" +# address = "192.168.1.50:502" # ":502" appended if no port given +# unit_id = 1 +# timeout_ms = 3000 +# +# [[datasource.modbus.devices.registers]] +# name = "temperature" +# kind = "holding" # holding | input | coil | discrete +# address = 100 +# encoding = "int16" # uint16|int16|uint32|int32|float32|float64 +# word_order = "big" # big (default) | little, for multi-word encodings +# unit = "C" +# scale = 0.1 # value = raw*scale + offset (scale 0 → 1) +# offset = 0.0 +# min = -50 +# max = 200 +# writable = false # input/discrete are always read-only +# description = "Process temperature" + +# SCPI instruments over raw TCP sockets (line-based "SCPI raw", port 5025). +# Channels become signals named "instrument:channel". A vxi11 transport is +# reserved for a future release. +[datasource.scpi] +enabled = false +poll_interval_ms = 1000 + +# [[datasource.scpi.instruments]] +# name = "dmm1" +# transport = "raw" # raw (default); vxi11 reserved +# address = "192.168.1.60:5025" # ":5025" appended if no port given +# timeout_ms = 3000 +# terminator = "\n" +# +# [[datasource.scpi.instruments.channels]] +# name = "voltage" +# query = "MEAS:VOLT?" # SCPI query whose response is the value +# write_cmd = "VOLT %v" # printf template; omit for read-only +# type = "float" # float (default) | int | string | bool +# unit = "V" +# min = 0 +# max = 30 +# description = "DC voltage"