wshub: add per-signal calibration store and flat config-file codec

This commit is contained in:
Martino Ferrari
2026-08-16 19:16:22 +02:00
parent 6d26e8191c
commit 47f1567a26
2 changed files with 389 additions and 0 deletions
+206
View File
@@ -0,0 +1,206 @@
package wshub
import (
"encoding/json"
"log"
"math"
"sort"
"strings"
"sync"
)
// maxUnitLen bounds the calibration unit override. Mirrored by kMaxUnitLen in
// the C++ StreamHub and MAX_UNIT_LEN in the SPA's calibration.js.
const maxUnitLen = 16
// CalConfig is one per-signal affine calibration: y = raw*Scale + Offset.
//
// The key is (Source label, base Signal name). It is deliberately the source
// *label* and not the runtime id ("s1", "s2"): ids are assigned in add-order at
// startup, so a calibration keyed by id would rebind to a different source
// whenever the source list order changed.
type CalConfig struct {
Source string `json:"source"`
Signal string `json:"signal"`
Scale float64 `json:"scale"`
Offset float64 `json:"offset"`
Unit string `json:"unit,omitempty"`
}
// calKey builds the calTable map key. NUL cannot occur in either component,
// so the concatenation is unambiguous.
func calKey(source, signal string) string { return source + "\x00" + signal }
// Normalise trims and validates the entry in place, reporting whether it is
// usable. A zero or non-finite Scale is rejected because it makes the
// calibration non-invertible, which the trigger threshold path depends on.
func (c *CalConfig) Normalise() bool {
c.Source = strings.TrimSpace(c.Source)
c.Signal = strings.TrimSpace(c.Signal)
if c.Source == "" || c.Signal == "" {
return false
}
if math.IsNaN(c.Scale) || math.IsInf(c.Scale, 0) || c.Scale == 0 {
return false
}
if math.IsNaN(c.Offset) || math.IsInf(c.Offset, 0) {
return false
}
c.Unit = strings.TrimSpace(c.Unit)
if len(c.Unit) > maxUnitLen {
c.Unit = c.Unit[:maxUnitLen]
}
return true
}
// IsIdentity reports whether the entry carries no information and can be
// dropped rather than stored and persisted.
func (c CalConfig) IsIdentity() bool {
return c.Scale == 1 && c.Offset == 0 && c.Unit == ""
}
// calTable is the hub's calibration store, safe for concurrent use.
type calTable struct {
mu sync.RWMutex
entries map[string]CalConfig
}
func newCalTable() *calTable {
return &calTable{entries: make(map[string]CalConfig)}
}
// Set validates and stores one entry, reporting whether it was accepted.
// Storing an identity entry removes any existing one for that key.
func (t *calTable) Set(c CalConfig) bool {
if !c.Normalise() {
return false
}
t.mu.Lock()
defer t.mu.Unlock()
if c.IsIdentity() {
delete(t.entries, calKey(c.Source, c.Signal))
} else {
t.entries[calKey(c.Source, c.Signal)] = c
}
return true
}
// Replace swaps the whole table for the given entries, silently dropping the
// invalid and identity ones. Used by config load and reload.
func (t *calTable) Replace(list []CalConfig) {
next := make(map[string]CalConfig, len(list))
for _, c := range list {
if !c.Normalise() || c.IsIdentity() {
continue
}
next[calKey(c.Source, c.Signal)] = c
}
t.mu.Lock()
t.entries = next
t.mu.Unlock()
}
// List returns the entries sorted by source then signal, so both the wire
// message and the config file have a stable order.
func (t *calTable) List() []CalConfig {
t.mu.RLock()
out := make([]CalConfig, 0, len(t.entries))
for _, c := range t.entries {
out = append(out, c)
}
t.mu.RUnlock()
sort.Slice(out, func(i, j int) bool {
if out[i].Source != out[j].Source {
return out[i].Source < out[j].Source
}
return out[i].Signal < out[j].Signal
})
return out
}
// ─── Config file codec ────────────────────────────────────────────────────────
// configFileEntry is the union of a source block and a calibration block.
//
// The file is one FLAT array of FLAT objects — never a nested one. The C++
// StreamHub's LoadSourcesFile is a hand-rolled scanner that takes each "{" up
// to the next "}" as one object, so a nested block would truncate the parse.
// Scale and Offset are pointers so that an absent field can be told apart from
// an explicit zero and defaulted to the identity values.
type configFileEntry struct {
// Source fields.
Label string `json:"label,omitempty"`
Addr string `json:"addr,omitempty"`
MulticastGroup string `json:"multicastGroup,omitempty"`
DataPort int `json:"dataPort,omitempty"`
// Calibration fields.
Source string `json:"source,omitempty"`
Signal string `json:"signal,omitempty"`
Scale *float64 `json:"scale,omitempty"`
Offset *float64 `json:"offset,omitempty"`
Unit string `json:"unit,omitempty"`
}
// parseConfigFile splits the flat array into sources and calibration entries.
// A block with "addr" is a source, one with "signal" is a calibration; anything
// else is skipped with a warning.
func parseConfigFile(data []byte) ([]SourceConfig, []CalConfig, error) {
var raw []configFileEntry
if err := json.Unmarshal(data, &raw); err != nil {
return nil, nil, err
}
srcs := make([]SourceConfig, 0, len(raw))
cals := make([]CalConfig, 0, len(raw))
for _, e := range raw {
switch {
case e.Addr != "":
srcs = append(srcs, SourceConfig{
Label: e.Label,
Addr: e.Addr,
MulticastGroup: e.MulticastGroup,
DataPort: e.DataPort,
})
case e.Signal != "":
c := CalConfig{Source: e.Source, Signal: e.Signal, Scale: 1, Offset: 0, Unit: e.Unit}
if e.Scale != nil {
c.Scale = *e.Scale
}
if e.Offset != nil {
c.Offset = *e.Offset
}
if !c.Normalise() {
log.Printf("wshub: skipping invalid calibration entry %q/%q", e.Source, e.Signal)
continue
}
cals = append(cals, c)
default:
log.Printf("wshub: skipping unrecognised config block")
}
}
return srcs, cals, nil
}
// encodeConfigFile renders the sources followed by the calibration entries as
// one flat array, in the indented shape the existing files already use.
func encodeConfigFile(srcs []SourceConfig, cals []CalConfig) ([]byte, error) {
out := make([]configFileEntry, 0, len(srcs)+len(cals))
for _, s := range srcs {
out = append(out, configFileEntry{
Label: s.Label,
Addr: s.Addr,
MulticastGroup: s.MulticastGroup,
DataPort: s.DataPort,
})
}
for _, c := range cals {
scale, offset := c.Scale, c.Offset
out = append(out, configFileEntry{
Source: c.Source,
Signal: c.Signal,
Scale: &scale,
Offset: &offset,
Unit: c.Unit,
})
}
return json.MarshalIndent(out, "", " ")
}
+183
View File
@@ -0,0 +1,183 @@
package wshub
import (
"math"
"testing"
)
func TestCalConfigNormalise(t *testing.T) {
cases := []struct {
name string
in CalConfig
want bool
wantUnit string
}{
{"plain", CalConfig{Source: "wave", Signal: "Adc", Scale: 2, Offset: -1, Unit: "V"}, true, "V"},
{"trims", CalConfig{Source: " wave ", Signal: " Adc ", Scale: 1, Unit: " V "}, true, "V"},
{"emptySource", CalConfig{Signal: "Adc", Scale: 1}, false, ""},
{"emptySignal", CalConfig{Source: "wave", Scale: 1}, false, ""},
{"zeroScale", CalConfig{Source: "wave", Signal: "Adc", Scale: 0}, false, ""},
{"nanScale", CalConfig{Source: "wave", Signal: "Adc", Scale: math.NaN()}, false, ""},
{"infScale", CalConfig{Source: "wave", Signal: "Adc", Scale: math.Inf(1)}, false, ""},
{"nanOffset", CalConfig{Source: "wave", Signal: "Adc", Scale: 1, Offset: math.NaN()}, false, ""},
{"infOffset", CalConfig{Source: "wave", Signal: "Adc", Scale: 1, Offset: math.Inf(-1)}, false, ""},
{"negScaleOK", CalConfig{Source: "wave", Signal: "Adc", Scale: -1}, true, ""},
{"longUnit", CalConfig{Source: "wave", Signal: "Adc", Scale: 1,
Unit: "0123456789abcdefGHIJ"}, true, "0123456789abcdef"},
}
for _, c := range cases {
got := c.in
if ok := got.Normalise(); ok != c.want {
t.Errorf("%s: Normalise() = %v, want %v", c.name, ok, c.want)
continue
}
if c.want && got.Unit != c.wantUnit {
t.Errorf("%s: Unit = %q, want %q", c.name, got.Unit, c.wantUnit)
}
}
if len("0123456789abcdef") != maxUnitLen {
t.Fatalf("test assumes maxUnitLen == 16, got %d", maxUnitLen)
}
}
func TestCalTableSetListAndIdentityRemoval(t *testing.T) {
tab := newCalTable()
if !tab.Set(CalConfig{Source: "b", Signal: "Y", Scale: 3, Offset: 1, Unit: "A"}) {
t.Fatal("Set(b/Y) rejected")
}
if !tab.Set(CalConfig{Source: "a", Signal: "X", Scale: 2}) {
t.Fatal("Set(a/X) rejected")
}
if tab.Set(CalConfig{Source: "a", Signal: "Z", Scale: 0}) {
t.Error("Set with scale=0 accepted, want rejected")
}
got := tab.List()
if len(got) != 2 {
t.Fatalf("List() = %d entries, want 2", len(got))
}
// Sorted by source then signal.
if got[0].Source != "a" || got[1].Source != "b" {
t.Errorf("List() order = %q,%q, want a,b", got[0].Source, got[1].Source)
}
// An identity entry removes the stored one.
if !tab.Set(CalConfig{Source: "a", Signal: "X", Scale: 1, Offset: 0, Unit: ""}) {
t.Fatal("identity Set rejected")
}
if got := tab.List(); len(got) != 1 || got[0].Source != "b" {
t.Errorf("after identity Set, List() = %+v, want only b/Y", got)
}
}
func TestCalTableReplace(t *testing.T) {
tab := newCalTable()
tab.Set(CalConfig{Source: "old", Signal: "X", Scale: 5})
tab.Replace([]CalConfig{
{Source: "new", Signal: "Y", Scale: 2},
{Source: "bad", Signal: "Z", Scale: 0}, // invalid → dropped
{Source: "id", Signal: "W", Scale: 1, Offset: 0, Unit: ""}, // identity → dropped
})
got := tab.List()
if len(got) != 1 || got[0].Source != "new" {
t.Fatalf("List() = %+v, want only new/Y", got)
}
}
func TestParseConfigFileCurrentFormat(t *testing.T) {
// A file written by the current binaries — sources only, spaces after colons.
data := []byte(`[
{
"label": "wave",
"addr": "127.0.0.1:44500"
},
{
"label": "mc",
"addr": "127.0.0.1:44501",
"multicastGroup": "239.0.0.1",
"dataPort": 44502
}
]`)
srcs, cals, err := parseConfigFile(data)
if err != nil {
t.Fatalf("parseConfigFile: %v", err)
}
if len(srcs) != 2 || len(cals) != 0 {
t.Fatalf("got %d sources / %d cals, want 2 / 0", len(srcs), len(cals))
}
if srcs[1].MulticastGroup != "239.0.0.1" || srcs[1].DataPort != 44502 {
t.Errorf("multicast source = %+v", srcs[1])
}
}
func TestParseConfigFileMixed(t *testing.T) {
data := []byte(`[
{"label":"wave","addr":"127.0.0.1:44500"},
{"source":"wave","signal":"Adc","scale":0.00030518,"offset":-1.25,"unit":"V"},
{"source":"wave","signal":"Bare"},
{"source":"wave","signal":"Bad","scale":0},
{"nonsense":true}
]`)
srcs, cals, err := parseConfigFile(data)
if err != nil {
t.Fatalf("parseConfigFile: %v", err)
}
if len(srcs) != 1 {
t.Fatalf("got %d sources, want 1", len(srcs))
}
if len(cals) != 2 {
t.Fatalf("got %d cals, want 2 (Adc and Bare; Bad is invalid)", len(cals))
}
if cals[0].Scale != 0.00030518 || cals[0].Offset != -1.25 || cals[0].Unit != "V" {
t.Errorf("Adc = %+v", cals[0])
}
// Absent scale/offset default to the identity values, not to zero.
if cals[1].Signal != "Bare" || cals[1].Scale != 1 || cals[1].Offset != 0 {
t.Errorf("Bare = %+v, want scale 1 / offset 0", cals[1])
}
}
func TestParseConfigFileMalformed(t *testing.T) {
if _, _, err := parseConfigFile([]byte("not json")); err == nil {
t.Error("parseConfigFile(garbage) = nil error, want error")
}
}
func TestEncodeConfigFileRoundTrip(t *testing.T) {
srcs := []SourceConfig{
{Label: "wave", Addr: "127.0.0.1:44500"},
{Label: "mc", Addr: "127.0.0.1:44501", MulticastGroup: "239.0.0.1", DataPort: 44502},
}
cals := []CalConfig{
{Source: "wave", Signal: "Adc", Scale: 0.5, Offset: 0, Unit: "V"},
}
data, err := encodeConfigFile(srcs, cals)
if err != nil {
t.Fatalf("encodeConfigFile: %v", err)
}
gotSrcs, gotCals, err := parseConfigFile(data)
if err != nil {
t.Fatalf("parseConfigFile(encoded): %v\n%s", err, data)
}
if len(gotSrcs) != 2 || len(gotCals) != 1 {
t.Fatalf("round-trip gave %d sources / %d cals, want 2 / 1\n%s",
len(gotSrcs), len(gotCals), data)
}
if gotSrcs[1] != srcs[1] {
t.Errorf("source round-trip: got %+v, want %+v", gotSrcs[1], srcs[1])
}
if gotCals[0] != cals[0] {
t.Errorf("cal round-trip: got %+v, want %+v", gotCals[0], cals[0])
}
// offset 0 must survive as an explicit field, not be dropped by omitempty.
if !bytesContains(data, []byte(`"offset": 0`)) {
t.Errorf("encoded file lost the zero offset:\n%s", data)
}
}
func bytesContains(hay, needle []byte) bool {
for i := 0; i+len(needle) <= len(hay); i++ {
if string(hay[i:i+len(needle)]) == string(needle) {
return true
}
}
return false
}