From 2952528a605408fb91927e0cc370fbec561d6408 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Wed, 23 Sep 2026 17:41:06 +0100 Subject: [PATCH] partial subscriptions not really partial. they just happen per peripheral now --- constants/constants.go | 7 +++ providers/beamng/beamng.go | 92 +++++++++++++++++++++++------------- providers/iracing/iracing.go | 88 +++++++++++++++++++++------------- services/devices.go | 26 ++-------- services/devices_lookup.go | 15 ++++-- services/services.go | 4 +- services/telemetry.go | 10 ++-- telemetry/data.go | 17 +++---- telemetry/fuel_calculator.go | 6 +++ telemetry/provider.go | 2 +- telemetry/rpm_lights.go | 6 +++ telemetry/virtual_field.go | 9 ++++ 12 files changed, 172 insertions(+), 110 deletions(-) create mode 100644 telemetry/virtual_field.go diff --git a/constants/constants.go b/constants/constants.go index b5d3305..7eb152c 100644 --- a/constants/constants.go +++ b/constants/constants.go @@ -1,6 +1,13 @@ package constants +// Provider Names const ( IRacingProviderName = "iRacing" BeamNGProviderName = "BeamNG.drive" ) + +// Virtual Field Names +const ( + FuelCalculatorName = "FuelCalculator" + RPMLightsName = "RPMLights" +) diff --git a/providers/beamng/beamng.go b/providers/beamng/beamng.go index 7e371cc..f8abeb2 100644 --- a/providers/beamng/beamng.go +++ b/providers/beamng/beamng.go @@ -28,6 +28,9 @@ type BeamNG struct { data *telemetry.TelemetryData updaters [telemetry.MaxFields]func(*telemetry.TelemetryField) + // Field Subscription management + boundFields map[telemetry.FieldID]bool + // stream control wg sync.WaitGroup streamCancel context.CancelFunc @@ -51,7 +54,9 @@ func NewBeamNGProvider(logger *slog.Logger, opts *bngsdk.Options) (*BeamNG, erro data: telemetry.NewTelemetryData(), SDK: beam, og: &bngsdk.Outgauge{}, - ticker: time.NewTicker(time.Second / 60), + // Field subscription management + boundFields: make(map[telemetry.FieldID]bool, telemetry.MaxFields), + ticker: time.NewTicker(time.Second / 60), } provider.updaters = [telemetry.MaxFields]func(*telemetry.TelemetryField){ @@ -115,47 +120,68 @@ func (b *BeamNG) Stream() (<-chan telemetry.TelemetryData, error) { return ch, nil } -func (b *BeamNG) Subscribe(requestFields []telemetry.FieldID) { - // NOTE: document how the Subscribe funtion works - slog.Debug(fmt.Sprintf("Len Req: %d\n", len(requestFields))) +// TODO: this function is exactly the same in BeamNG drive now, and I reckon it will be the same +// In plenty other things. I should make it TelemetryData method +func (i *BeamNG) subscribe(fields []telemetry.FieldID) []string { + newSubscriptions := make([]string, 0, telemetry.MaxFields) - b.data.ActiveBinds = make([]telemetry.BoundField, 0, len(requestFields)) + i.mut.Lock() + defer i.mut.Unlock() - // First we must add the virtual fields - // we will add their dependencies and the primitives to a slice - pendingBinds := make([]telemetry.FieldID, telemetry.MaxFields) - - for _, id := range requestFields { - switch id { - case telemetry.RPMStateColour: - b.data.VirtualBinds = append(b.data.VirtualBinds, telemetry.NewRPMLights()) - case telemetry.FCCurrentLap: - b.data.VirtualBinds = append(b.data.VirtualBinds, - telemetry.NewFuelCalculator(slog.Default().WithGroup("FUEL CALC"))) - default: - // primitive telemetry field - pendingBinds = append(pendingBinds, id) - } - } - - boundCheck := make(map[telemetry.FieldID]bool) - - // Now that we know all the fields we need to bind we follow the binding procedure - for _, id := range pendingBinds { - // Check if we already bound this FieldID - if boundCheck[id] { + for _, id := range fields { + if i.boundFields[id] { continue } - binding := telemetry.BoundField{ + i.data.ActiveBinds[id] = telemetry.BoundField{ ID: id, } - - b.data.ActiveBinds = append(b.data.ActiveBinds, binding) - boundCheck[id] = true + i.boundFields[id] = true + newSubscriptions = append(newSubscriptions, telemetry.FieldNames[id]) } - slog.Debug(fmt.Sprintf("Subscribed: %+v\n", b.data.ActiveBinds)) + // unsubscribe from fields we many not need anymore + for key, bound := range i.boundFields { + if _, ok := i.data.ActiveBinds[key]; bound && !ok { + delete(i.data.ActiveBinds, key) + i.boundFields[key] = false + } + } + + return newSubscriptions +} + +func (b *BeamNG) Subscribe(requestFields []telemetry.FieldID) []string { + // NOTE: document how the Subscribe funtion works + slog.Debug(fmt.Sprintf("Len Req: %d\n", len(requestFields))) + + // First we must add the virtual fields + // we will add their dependencies and the primitives to a slice + toBind := make([]telemetry.FieldID, telemetry.MaxFields) + + b.mut.Lock() + for _, id := range requestFields { + switch id { + case telemetry.RPMStateColour: + rpmLights := telemetry.NewRPMLights() + b.data.VirtualBinds[rpmLights.Name()] = rpmLights + toBind = append(toBind, rpmLights.EnsureSubscribed()...) + case telemetry.FCCurrentLap: + fuelCalc := telemetry.NewFuelCalculator(b.logger.WithGroup("FUEL CALC")) + b.data.VirtualBinds[fuelCalc.Name()] = fuelCalc + toBind = append(toBind, fuelCalc.EnsureSubscribed()...) + default: + // primitive telemetry field + toBind = append(toBind, id) + } + } + b.mut.Unlock() + + newSubs := b.subscribe(toBind) + + slog.Debug(fmt.Sprintf("Subscribed: %+v\n", toBind)) + + return newSubs } // Internal diff --git a/providers/iracing/iracing.go b/providers/iracing/iracing.go index d271ea6..bcf9c93 100644 --- a/providers/iracing/iracing.go +++ b/providers/iracing/iracing.go @@ -29,13 +29,14 @@ type IRacing struct { mut sync.Mutex data *telemetry.TelemetryData updaters [telemetry.MaxFields]func(*telemetry.TelemetryField) + // Field Subscription management + boundFields map[telemetry.FieldID]bool // Timing information ticker *time.Ticker // ticker will keep polling intervals constant // Stream - wg sync.WaitGroup - // streamCh chan telemetry.TelemetryData + wg sync.WaitGroup streamCancel context.CancelFunc } @@ -55,6 +56,8 @@ func NewIRacingProvider( logger: logger, SDK: sdk, data: telemetry.NewTelemetryData(), + // Field subscription management + boundFields: make(map[telemetry.FieldID]bool, telemetry.MaxFields), // TODO: make this configurable from the user side ticker: time.NewTicker(time.Second / 60), } @@ -225,46 +228,67 @@ func (i *IRacing) StopStream() { i.streamCancel = nil } -func (i *IRacing) Subscribe(requestFields []telemetry.FieldID) { - i.logger.Debug(fmt.Sprintf("Len Req: %d\n", len(requestFields))) +// TODO: this function is exactly the same in BeamNG drive now, and I reckon it will be the same +// In plenty other things. I should make it TelemetryData method +func (i *IRacing) subscribe(fields []telemetry.FieldID) []string { + newSubscriptions := make([]string, 0, telemetry.MaxFields) - i.data.ActiveBinds = make([]telemetry.BoundField, 0, len(requestFields)) + i.mut.Lock() + defer i.mut.Unlock() - // First we must add the virtual fields - // we will add their dependencies and the primitives to a slice - pendingBinds := make([]telemetry.FieldID, 0, telemetry.MaxFields) - - for _, id := range requestFields { - switch id { - case telemetry.RPMStateColour: - i.data.VirtualBinds = append(i.data.VirtualBinds, telemetry.NewRPMLights()) - case telemetry.FCCurrentLap: - i.data.VirtualBinds = append(i.data.VirtualBinds, - telemetry.NewFuelCalculator(i.logger.WithGroup("FUEL CALC"))) - default: - // primitive telemetry field - pendingBinds = append(pendingBinds, id) - } - } - - boundCheck := make(map[telemetry.FieldID]bool) - - // Now that we know all the fields we need to bind we follow the binding procedure - for _, id := range pendingBinds { - // Check if we already bound this FieldID - if boundCheck[id] { + for _, id := range fields { + if i.boundFields[id] { continue } - binding := telemetry.BoundField{ + i.data.ActiveBinds[id] = telemetry.BoundField{ ID: id, } - - i.data.ActiveBinds = append(i.data.ActiveBinds, binding) - boundCheck[id] = true + i.boundFields[id] = true + newSubscriptions = append(newSubscriptions, telemetry.FieldNames[id]) } + // unsubscribe from fields we many not need anymore + for key, bound := range i.boundFields { + if _, ok := i.data.ActiveBinds[key]; bound && !ok { + delete(i.data.ActiveBinds, key) + i.boundFields[key] = false + } + } + + return newSubscriptions +} + +func (i *IRacing) Subscribe(requestFields []telemetry.FieldID) []string { + i.logger.Debug(fmt.Sprintf("Len Req: %d\n", len(requestFields))) + + // First we must add the virtual fields + // we will add their dependencies and the primitives to a slice + toBind := make([]telemetry.FieldID, 0, telemetry.MaxFields) + + i.mut.Lock() + for _, id := range requestFields { + switch id { + case telemetry.RPMStateColour: + rpmLights := telemetry.NewRPMLights() + i.data.VirtualBinds[rpmLights.Name()] = rpmLights + toBind = append(toBind, rpmLights.EnsureSubscribed()...) + case telemetry.FCCurrentLap: + fuelCalc := telemetry.NewFuelCalculator(i.logger.WithGroup("FUEL CALC")) + i.data.VirtualBinds[fuelCalc.Name()] = fuelCalc + toBind = append(toBind, fuelCalc.EnsureSubscribed()...) + default: + // primitive telemetry field + toBind = append(toBind, id) + } + } + i.mut.Unlock() + + newSubs := i.subscribe(toBind) + i.logger.Debug(fmt.Sprintf("Subscribed: %+v\n", i.data.ActiveBinds)) + + return newSubs } func (i *IRacing) Name() string { diff --git a/services/devices.go b/services/devices.go index 8448d90..8bb2c49 100644 --- a/services/devices.go +++ b/services/devices.go @@ -26,7 +26,7 @@ type DeviceService struct { // Callbacks // Telemetry service data fetchers telemetryProvider func() (string, error) - triggerFieldSubscription func() []telemetry.FieldID + triggerFieldSubscription func([]telemetry.FieldID) []string } func NewDeviceService(logger *slog.Logger, msg chan string) *DeviceService { @@ -80,24 +80,6 @@ func (ds *DeviceService) PeripheralExists(pname string) bool { return err == nil } -func (ds *DeviceService) GetRequiredFields() []telemetry.FieldID { - // NOTE: this can be optimized, not that it matters at this stage, but if - // it runs while telemetry is running we want it optimized I guess - seen := make(map[telemetry.FieldID]struct{}) - var allFields []telemetry.FieldID - - for _, dev := range ds.GetDevices() { - for _, field := range dev.RequiredFields() { - if _, exists := seen[field]; !exists { - seen[field] = struct{}{} - allFields = append(allFields, field) - } - } - } - - return allFields -} - // Getters [END] --------------------------------------------------------------- // Actions [START] ------------------------------------------------------------- @@ -168,10 +150,10 @@ func (ds *DeviceService) transmit(ctx context.Context) { } // Callbacks [START] ----------------------------------------------------------- -func (ds *DeviceService) peripheralConfigured(pname string) { +func (ds *DeviceService) peripheralConfigured(pname string, fields []telemetry.FieldID) { // We need to retrigger field subscription here - fields := ds.triggerFieldSubscription() - ds.Messages <- fmt.Sprintf("subscribed to fields: %+v\n", fields) + subscribedTo := ds.triggerFieldSubscription(fields) + ds.Messages <- fmt.Sprintf("subscribed to fields: %q\n", subscribedTo) } // Callbacks [END] ------------------------------------------------------------- diff --git a/services/devices_lookup.go b/services/devices_lookup.go index 3c46eeb..788e553 100644 --- a/services/devices_lookup.go +++ b/services/devices_lookup.go @@ -10,6 +10,7 @@ import ( "esdi/devices" "esdi/peripheral" + "esdi/telemetry" ) var ( @@ -86,7 +87,7 @@ type PeripheralStateStore struct { Messages chan string // Callbacks telemetryProvider func() (string, error) - onPeripheralConfigured func(string) + onPeripheralConfigured func(string, []telemetry.FieldID) // Internal State isStreaming bool } @@ -184,6 +185,15 @@ func (pss *PeripheralStateStore) GetStreamingState() bool { return pss.isStreaming } +func (pss *PeripheralStateStore) GetPeripheralFields(pname string) []telemetry.FieldID { + per, err := pss.GetState(pname) + if err != nil { + return nil + } + + return per.Peripheral.RequiredFields() +} + // "Events" [START] ------------------------------------------------------------ func (ds *DeviceService) onDeviceTimedOut(pname string) { ds.PSS.setDeviceTimedOut(pname) @@ -280,7 +290,7 @@ func (pss *PeripheralStateStore) setDeviceConfigured(pname string) { pss.store[pname].State = DeviceIsConfigured pss.mu.Unlock() - pss.onPeripheralConfigured(pname) + pss.onPeripheralConfigured(pname, pss.GetPeripheralFields(pname)) } func (pss *PeripheralStateStore) setDeviceIsStreaming(pname string) { @@ -347,7 +357,6 @@ func (pss *PeripheralStateStore) configurePeripheral( } onSuccess(pname) - pss.setDeviceConfigured(pname) }() } diff --git a/services/services.go b/services/services.go index d8f1ec7..f8960fc 100644 --- a/services/services.go +++ b/services/services.go @@ -26,10 +26,10 @@ func NewOrchestrator(logger *slog.Logger) (*Orchestrator, error) { // Setup device service callbacks devService.telemetryProvider = telemService.GetTelemetryProviderName - devService.triggerFieldSubscription = telemService.SubscribeToAllFields + devService.triggerFieldSubscription = telemService.SubscribeToFields // Setup telemetry service callbacks - telemService.getRequiredFields = devService.GetRequiredFields + // telemService.getRequiredFields = devService.GetRequiredFields return &Orchestrator{ DeviceService: devService, diff --git a/services/telemetry.go b/services/telemetry.go index 2f9983d..e05e177 100644 --- a/services/telemetry.go +++ b/services/telemetry.go @@ -37,7 +37,7 @@ type TelemetryService struct { // Callbacks // Devices data request // peripheralProvider func() []peripheral.Peripheral - getRequiredFields func() []telemetry.FieldID + // getRequiredFields func() []telemetry.FieldID } func NewTelemetryService( @@ -114,13 +114,11 @@ func (t *TelemetryService) UnsubscribeListener(id string) { } } -func (t *TelemetryService) SubscribeToAllFields() []telem.FieldID { - fields := t.getRequiredFields() - +func (t *TelemetryService) SubscribeToFields(fields []telemetry.FieldID) []string { t.logger.Debug("requested fields", "fields", fields) - t.activeProvider.Subscribe(fields) + subscribed := t.activeProvider.Subscribe(fields) - return fields + return subscribed } // Listener Control [END] ------------------------------------------------------ diff --git a/telemetry/data.go b/telemetry/data.go index 166b17c..52d8dda 100644 --- a/telemetry/data.go +++ b/telemetry/data.go @@ -23,13 +23,6 @@ type FieldMapper struct { Transform func(any) uint64 } -// VirtualField will derive data from telemetry primitives -// So fuel per lap predictions, compound gauge lights and so on -type VirtualField interface { - Process(td *TelemetryData) - EnsureSubscribed() []FieldID -} - // NOTE: Update the iracing SDK to write data to the same map ALWAYS, then // I can bind that address and read directly from there on the transform @@ -228,16 +221,18 @@ func GetFieldID(name string) (FieldID, bool) { } // TelemetryData is -// I need to find a way of having the values be per window or some other type TelemetryData struct { Values [MaxFields]TelemetryField - ActiveBinds []BoundField - VirtualBinds []VirtualField + ActiveBinds map[FieldID]BoundField + VirtualBinds map[string]VirtualField InitialTime time.Time PenultimateDataPoll time.Time LastDataPoll time.Time } func NewTelemetryData() *TelemetryData { - return &TelemetryData{} + return &TelemetryData{ + ActiveBinds: make(map[FieldID]BoundField, MaxFields), + VirtualBinds: make(map[string]VirtualField), + } } diff --git a/telemetry/fuel_calculator.go b/telemetry/fuel_calculator.go index 548bbc8..f7a8b83 100644 --- a/telemetry/fuel_calculator.go +++ b/telemetry/fuel_calculator.go @@ -5,6 +5,8 @@ import ( "log/slog" "strconv" "sync" + + "esdi/constants" ) // NOTE: for managing fuel consumption and predictions we need to filter out @@ -264,3 +266,7 @@ func (fc *FuelCalculator) resetHistory() { func (fc *FuelCalculator) EnsureSubscribed() []FieldID { return []FieldID{FuelLevel, LapNumber} } + +func (fc *FuelCalculator) Name() string { + return constants.FuelCalculatorName +} diff --git a/telemetry/provider.go b/telemetry/provider.go index 43c8370..3d86c4c 100644 --- a/telemetry/provider.go +++ b/telemetry/provider.go @@ -6,7 +6,7 @@ import "time" type TelemetryProvider interface { StopStream() Stream() (<-chan TelemetryData, error) - Subscribe([]FieldID) + Subscribe([]FieldID) []string IsAlive(time.Duration) bool Name() string Close() diff --git a/telemetry/rpm_lights.go b/telemetry/rpm_lights.go index 16d4ae1..efca00e 100644 --- a/telemetry/rpm_lights.go +++ b/telemetry/rpm_lights.go @@ -1,5 +1,7 @@ package telemetry +import "esdi/constants" + type RPMLights struct { State string } @@ -28,3 +30,7 @@ func (rl *RPMLights) Process(td *TelemetryData) { func (rl *RPMLights) EnsureSubscribed() []FieldID { return []FieldID{RPM} } + +func (rl *RPMLights) Name() string { + return constants.RPMLightsName +} diff --git a/telemetry/virtual_field.go b/telemetry/virtual_field.go new file mode 100644 index 0000000..cfde6f7 --- /dev/null +++ b/telemetry/virtual_field.go @@ -0,0 +1,9 @@ +package telemetry + +// VirtualField will derive data from telemetry primitives +// So fuel per lap predictions, compound gauge lights and so on +type VirtualField interface { + Name() string + Process(td *TelemetryData) + EnsureSubscribed() []FieldID +}