From 47f1567a2670287d4e24b051581db872a6a2e2ac Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Sun, 16 Aug 2026 19:16:22 +0200 Subject: [PATCH] wshub: add per-signal calibration store and flat config-file codec --- Common/Client/go/wshub/calibration.go | 206 +++++++++++++++++++++ Common/Client/go/wshub/calibration_test.go | 183 ++++++++++++++++++ 2 files changed, 389 insertions(+) create mode 100644 Common/Client/go/wshub/calibration.go create mode 100644 Common/Client/go/wshub/calibration_test.go diff --git a/Common/Client/go/wshub/calibration.go b/Common/Client/go/wshub/calibration.go new file mode 100644 index 0000000..981afce --- /dev/null +++ b/Common/Client/go/wshub/calibration.go @@ -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, "", " ") +} diff --git a/Common/Client/go/wshub/calibration_test.go b/Common/Client/go/wshub/calibration_test.go new file mode 100644 index 0000000..5ac3afe --- /dev/null +++ b/Common/Client/go/wshub/calibration_test.go @@ -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 +}