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 c688821..8bb2c49 100644 --- a/services/devices.go +++ b/services/devices.go @@ -2,6 +2,7 @@ package services import ( "context" + "fmt" "log/slog" "sync/atomic" @@ -24,7 +25,8 @@ type DeviceService struct { Messages chan string // Callbacks // Telemetry service data fetchers - telemetryProvider func() (string, error) + telemetryProvider func() (string, error) + triggerFieldSubscription func([]telemetry.FieldID) []string } func NewDeviceService(logger *slog.Logger, msg chan string) *DeviceService { @@ -43,6 +45,7 @@ func NewDeviceService(logger *slog.Logger, msg chan string) *DeviceService { // Set the callbacks for PSS dev.PSS.telemetryProvider = dev.getTelemetryProvider + dev.PSS.onPeripheralConfigured = dev.peripheralConfigured return dev } @@ -59,7 +62,7 @@ func (ds *DeviceService) GetDevices() []peripheral.Peripheral { peripherals := make([]peripheral.Peripheral, 0, len(snapshot)) for _, state := range snapshot { - if state.State < DeviceIsConnected { + if state.State < DeviceIsConfigured { continue } peripherals = append(peripherals, state.Peripheral) @@ -95,6 +98,8 @@ func (ds *DeviceService) StopStream() { return } + ds.PSS.OnStopStream() + ds.streamCancel() ds.streamCancel = nil } @@ -145,5 +150,10 @@ func (ds *DeviceService) transmit(ctx context.Context) { } // Callbacks [START] ----------------------------------------------------------- +func (ds *DeviceService) peripheralConfigured(pname string, fields []telemetry.FieldID) { + // We need to retrigger field subscription here + 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 499b8ea..788e553 100644 --- a/services/devices_lookup.go +++ b/services/devices_lookup.go @@ -10,6 +10,7 @@ import ( "esdi/devices" "esdi/peripheral" + "esdi/telemetry" ) var ( @@ -85,7 +86,8 @@ type PeripheralStateStore struct { // Messaging for UI and stuff Messages chan string // Callbacks - telemetryProvider func() (string, error) + telemetryProvider func() (string, error) + onPeripheralConfigured func(string, []telemetry.FieldID) // Internal State isStreaming bool } @@ -183,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) @@ -204,6 +215,12 @@ func (pss *PeripheralStateStore) OnStopStream() { pss.mu.Lock() pss.isStreaming = false pss.mu.Unlock() + + for _, state := range pss.GetStates() { + if state.State == DeviceIsStreaming { + pss.setDeviceConfigured(state.device.Name) + } + } } // "Events" [END] -------------------------------------------------------------- @@ -270,8 +287,10 @@ func (pss *PeripheralStateStore) setDeviceConfigured(pname string) { pss.Logger.Info("device is configured and ready for data", "device", pname) pss.mu.Lock() - defer pss.mu.Unlock() pss.store[pname].State = DeviceIsConfigured + pss.mu.Unlock() + + pss.onPeripheralConfigured(pname, pss.GetPeripheralFields(pname)) } func (pss *PeripheralStateStore) setDeviceIsStreaming(pname string) { @@ -338,7 +357,6 @@ func (pss *PeripheralStateStore) configurePeripheral( } onSuccess(pname) - pss.setDeviceConfigured(pname) }() } diff --git a/services/services.go b/services/services.go index 18aff69..f8960fc 100644 --- a/services/services.go +++ b/services/services.go @@ -26,9 +26,10 @@ func NewOrchestrator(logger *slog.Logger) (*Orchestrator, error) { // Setup device service callbacks devService.telemetryProvider = telemService.GetTelemetryProviderName + devService.triggerFieldSubscription = telemService.SubscribeToFields // Setup telemetry service callbacks - telemService.peripheralProvider = devService.GetDevices + // telemService.getRequiredFields = devService.GetRequiredFields return &Orchestrator{ DeviceService: devService, diff --git a/services/telemetry.go b/services/telemetry.go index 7ca2112..e05e177 100644 --- a/services/telemetry.go +++ b/services/telemetry.go @@ -7,7 +7,6 @@ import ( "sync" "time" - "esdi/peripheral" "esdi/providers" "esdi/telemetry" telem "esdi/telemetry" @@ -37,7 +36,8 @@ type TelemetryService struct { healthCheckCancel context.CancelFunc // Callbacks // Devices data request - peripheralProvider func() []peripheral.Peripheral + // peripheralProvider func() []peripheral.Peripheral + // getRequiredFields func() []telemetry.FieldID } func NewTelemetryService( @@ -114,23 +114,11 @@ func (t *TelemetryService) UnsubscribeListener(id string) { } } -func (t *TelemetryService) SubscribeToFields() []telem.FieldID { - seen := make(map[telemetry.FieldID]struct{}) - var allFields []telemetry.FieldID +func (t *TelemetryService) SubscribeToFields(fields []telemetry.FieldID) []string { + t.logger.Debug("requested fields", "fields", fields) + subscribed := t.activeProvider.Subscribe(fields) - for _, dev := range t.peripheralProvider() { - for _, field := range dev.RequiredFields() { - if _, exists := seen[field]; !exists { - seen[field] = struct{}{} - allFields = append(allFields, field) - } - } - } - - t.logger.Debug("requested fields", "fields", allFields) - t.activeProvider.Subscribe(allFields) - - return allFields + 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 +} diff --git a/tui/internal/controllers/device.go b/tui/internal/controllers/device.go index 9d3d653..86f67c7 100644 --- a/tui/internal/controllers/device.go +++ b/tui/internal/controllers/device.go @@ -93,7 +93,7 @@ func (mc *DeviceController) AddDeviceAPIListItems() { mc.StreamCtrl.StreamView.Flex, ) - mc.StreamCtrl.SetInternalState() + // mc.StreamCtrl.SetInternalState() mc.App.SetFocus(mc.StreamCtrl.StreamView.Options.Form) }) diff --git a/tui/internal/controllers/streaming.go b/tui/internal/controllers/streaming.go index 4fd57c1..8f40005 100644 --- a/tui/internal/controllers/streaming.go +++ b/tui/internal/controllers/streaming.go @@ -154,8 +154,8 @@ func (sc *StreamingCtrl) updateStream() { // Performance reasoning: this is not used during the high frequency data transmission // so we can get away with using a map for convenience here func (sc *StreamingCtrl) SetInternalState() { - fields := sc.TelemServ.SubscribeToFields() - sc.Messages <- fmt.Sprintf("Subscribed to fields: %+v\n", fields) + // fields := sc.TelemServ.SubscribeToAllFields() + // sc.Messages <- fmt.Sprintf("Subscribed to fields: %+v\n", fields) } func (sc *StreamingCtrl) listenToUIStream() {