From b364931c807577afc7d75cc5dfe9c33503af6e49 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 21 Sep 2026 15:27:42 +0100 Subject: [PATCH] automatic setup for devices Its not close to being done and still need to work out the flows due to who arrives first: telemetry or the peripheral? --- devices/cdashdisplay/display.go | 23 ++++++++++++++- devices/uidevice/device.go | 2 +- peripheral/peripheral.go | 2 +- services/devices.go | 51 ++++++++++++++++++++++++++++++++- services/devices_lookup.go | 29 +++++++++++++++++-- services/services.go | 10 +++++-- services/telemetry.go | 31 ++++++++++++++++---- services/telemetry_events.go | 1 + 8 files changed, 134 insertions(+), 15 deletions(-) diff --git a/devices/cdashdisplay/display.go b/devices/cdashdisplay/display.go index 7c2bd94..b8e2d07 100644 --- a/devices/cdashdisplay/display.go +++ b/devices/cdashdisplay/display.go @@ -11,6 +11,7 @@ import ( "sync" "time" + "esdi/constants" helper "esdi/helpers" "esdi/peripheral" "esdi/peripheral/communication" @@ -438,7 +439,7 @@ func (d *CDashDisplay) SendData(data *telemetry.TelemetryData) error { return nil } -func (cds *CDashDisplay) Setup() error { +func (cds *CDashDisplay) setupForIracing() error { err := cds.LoadLayout("layout.yaml") if err != nil { return err @@ -446,3 +447,23 @@ func (cds *CDashDisplay) Setup() error { return nil } + +func (cds *CDashDisplay) setupForBeamNG() error { + err := cds.LoadLayout("beamng.yaml") + if err != nil { + return err + } + + return nil +} + +func (cds *CDashDisplay) Setup(provider string) error { + switch provider { + case constants.IRacingProviderName: + return cds.setupForIracing() + case constants.BeamNGProviderName: + return cds.setupForBeamNG() + default: + return fmt.Errorf("unknown provider: %s", provider) + } +} diff --git a/devices/uidevice/device.go b/devices/uidevice/device.go index 4b0b0cd..c97d493 100644 --- a/devices/uidevice/device.go +++ b/devices/uidevice/device.go @@ -25,7 +25,7 @@ func (uid *UIDevice) Close() error { return nil } -func (uid *UIDevice) Setup() error { +func (uid *UIDevice) Setup(provider string) error { return nil } diff --git a/peripheral/peripheral.go b/peripheral/peripheral.go index fc918ee..56cf08d 100644 --- a/peripheral/peripheral.go +++ b/peripheral/peripheral.go @@ -17,7 +17,7 @@ const ( type Peripheral interface { Name() string - Setup() error + Setup(string) error SendData(*telemetry.TelemetryData) error RequiredFields() []telemetry.FieldID OnLoad() error diff --git a/services/devices.go b/services/devices.go index 8219254..4513d01 100644 --- a/services/devices.go +++ b/services/devices.go @@ -2,6 +2,7 @@ package services import ( "context" + "fmt" "log/slog" "sync/atomic" @@ -23,7 +24,9 @@ type DeviceService struct { // Output Messages chan string // Callbacks - OnTelemetryProviderDiscovered func() + OnPeripheralFound func(string) + // Telemetry service data fetchers + telemetryProvider func() (string, error) } func NewDeviceService(logger *slog.Logger, msg chan string) *DeviceService { @@ -40,6 +43,9 @@ func NewDeviceService(logger *slog.Logger, msg chan string) *DeviceService { dev.ctxDiscovery, dev.ctxDiscoveryCancel = context.WithCancel(context.Background()) go dev.FindDevices() + // Set the callbacks for PSS + dev.PSS.OnDeviceFound = dev.deviceFound + return dev } @@ -131,3 +137,46 @@ func (ds *DeviceService) transmit(ctx context.Context) { } } } + +func (ds *DeviceService) deviceFound(pname string) { + ds.OnPeripheralFound(pname) +} + +// Callbacks [START] ----------------------------------------------------------- + +// ProviderFoundCallback should be called once the telemetry service finds a provider +// Here we need to setup our devices. Some devices might have different settings for +// different sims +func (ds *DeviceService) ProviderFoundCallback(name string) { + ds.Messages <- "Device services got triggered by a provider being found\n" + for _, peripheral := range ds.PSS.GetStates() { + // ds.Messages <- fmt.Sprintf("dev: %s, SETUP: %t, STATE: %d\n", + // peripheral.device.Name, peripheral.Setup, peripheral.State) + if peripheral.State != DeviceIsConnected || peripheral.Setup { + continue + } + + // The device is connected and still needs to run the setup + ds.Messages <- fmt.Sprintf("device '%s' needs to be setup\n", peripheral.device.Name) + + provider, err := ds.telemetryProvider() + if err != nil { + // Can't setup anything + continue + } + + err = peripheral.Peripheral.Setup(provider) + if err != nil { + // TODO: log do something + continue + } + + err = ds.PSS.UpdatePeripheralSetupState(peripheral.device.Name, true) + if err != nil { + // TODO: log do something + continue + } + } +} + +// Callbacks [END] ------------------------------------------------------------- diff --git a/services/devices_lookup.go b/services/devices_lookup.go index febabd1..1b8442a 100644 --- a/services/devices_lookup.go +++ b/services/devices_lookup.go @@ -32,6 +32,7 @@ type PeripheralState struct { device *devices.Device Peripheral peripheral.Peripheral State DeviceState + Setup bool } func NewPeripheralState( @@ -43,6 +44,7 @@ func NewPeripheralState( device: dev, Peripheral: peripheral, State: state, + Setup: false, } return &perState @@ -54,6 +56,8 @@ type PeripheralStateStore struct { store map[string]*PeripheralState // Messaging for UI and stuff Messages chan string + // Callbacks + OnDeviceFound func(string) } func NewPeripheralStateStore( @@ -117,6 +121,18 @@ func (pss *PeripheralStateStore) AddDevice(dev *devices.Device) error { return nil } +func (pss *PeripheralStateStore) UpdatePeripheralSetupState(pname string, nState bool) error { + if !pss.DeviceExists(pname) { + return ErrNoSuchDevice + } + + pss.mu.Lock() + defer pss.mu.Unlock() + pss.store[pname].Setup = nState + + return nil +} + // DeviceExists returns whether the store is already tracking `pname` func (pss *PeripheralStateStore) DeviceExists(pname string) bool { pss.mu.RLock() @@ -156,11 +172,12 @@ func (pss *PeripheralStateStore) setDeviceConnected(pname string, per peripheral pss.Logger.Info("found device", "device", pname) pss.mu.Lock() - defer pss.mu.Unlock() pss.store[pname].Peripheral = per pss.store[pname].State = DeviceIsConnected + pss.mu.Unlock() pss.Messages <- fmt.Sprintf("Device successfuly connected: %s\n", pname) + pss.OnDeviceFound(pname) } func (pss *PeripheralStateStore) setDeviceTimedOut(pname string) { @@ -222,7 +239,6 @@ func (pss *PeripheralStateStore) handleDeviceReconnected(pname string) error { return err } - err = state.Peripheral.Setup() if err != nil { pss.Logger.Error("failed to setup peripheral", "peripheral", pname, "error", err) return ErrFailedToSetupPeripheral @@ -234,6 +250,12 @@ func (pss *PeripheralStateStore) handleDeviceReconnected(pname string) error { return nil } +// handleDeviceConnected will handle the device setup after it connects +func (pss *PeripheralStateStore) handleDeviceConnected(pname string) error { + // We need to query wheter we have a telemetry provider running or not + return nil +} + func (pss *PeripheralStateStore) HandleDeviceState() { peripherals := pss.GetStates() @@ -249,6 +271,7 @@ func (pss *PeripheralStateStore) HandleDeviceState() { case DeviceIsConnected: // Need to check if its streaming, if its not streaming than we have to do a healthcheck pss.Logger.Debug("Device is connected. Normal", "device", pName) + pss.handleDeviceConnected(pName) case DeviceReconnected: // If the device has reconnected we need to reset the device and then set it as connected pss.Logger.Debug("Device has reconnected. Clearing up state", "device", pName) @@ -274,7 +297,7 @@ func (pss *PeripheralStateStore) HandleDeviceState() { // FindDevices is a routine that goes over the devices in the PeripheralStateStore // and handles their state accordingly func (ds *DeviceService) FindDevices() { - ticker := time.NewTicker(2 * time.Second) + ticker := time.NewTicker(1 * time.Second) defer ticker.Stop() for { diff --git a/services/services.go b/services/services.go index d76cb62..5abcf8f 100644 --- a/services/services.go +++ b/services/services.go @@ -17,14 +17,20 @@ func NewOrchestrator(logger *slog.Logger) (*Orchestrator, error) { devService := NewDeviceService(logger.With("service", "DeviceService"), msg) - telemService := NewTelemetryService(logger.With("service", "TelemetryService"), devService, msg) + telemService := NewTelemetryService(logger.With("service", "TelemetryService"), msg) if telemService == nil { return nil, errors.New("failed to create telemetry service") } go telemService.FindProvider(telemService.CtxMonitor) - // Need to setup the callbacks on the services + // Setup device service callbacks + devService.OnPeripheralFound = telemService.PeripheralFoundCallback + devService.telemetryProvider = telemService.GetTelemetryProviderName + + // Setup telemetry service callbacks + telemService.OnProviderFound = devService.ProviderFoundCallback + telemService.peripheralProvider = devService.GetDevices return &Orchestrator{ DeviceService: devService, diff --git a/services/telemetry.go b/services/telemetry.go index 679a85b..fa50c3b 100644 --- a/services/telemetry.go +++ b/services/telemetry.go @@ -2,20 +2,23 @@ package services import ( "context" + "errors" "log/slog" "sync" "time" + "esdi/peripheral" "esdi/providers" "esdi/telemetry" telem "esdi/telemetry" ) +var ErrNoActiveProviderAvailable = errors.New("no active provider available") + // TelemetryService will be our base struct to handle telemetry data // It should hook to a data sink and handle it like iRacing, BeamNG, AC and so on type TelemetryService struct { - logger *slog.Logger - devService *DeviceService + logger *slog.Logger // Streaming isStreaming bool // Concurrency protection @@ -33,18 +36,18 @@ type TelemetryService struct { CtxHealthcheck context.Context healthCheckCancel context.CancelFunc // Callbacks - OnDevicesDiscovered func() + OnProviderFound func(string) + // Devices data request + peripheralProvider func() []peripheral.Peripheral } func NewTelemetryService( logger *slog.Logger, - devServo *DeviceService, msg chan string, ) *TelemetryService { newService := &TelemetryService{ logger: logger, isConnected: false, - devService: devServo, listeners: make(map[string]chan telem.TelemetryData), Messages: msg, } @@ -74,6 +77,14 @@ func (t *TelemetryService) ProviderMonitor(ctx context.Context) { } } +func (t *TelemetryService) GetTelemetryProviderName() (string, error) { + if t.activeProvider == nil { + return "", ErrNoActiveProviderAvailable + } + + return t.activeProvider.Name(), nil +} + // Listener Control [START] ---------------------------------------------------- func (t *TelemetryService) SubscribeListener(id string, bufferSize int) <-chan telem.TelemetryData { @@ -108,7 +119,7 @@ func (t *TelemetryService) SubscribeToFields() []telem.FieldID { seen := make(map[telemetry.FieldID]struct{}) var allFields []telemetry.FieldID - for _, dev := range t.devService.GetDevices() { + for _, dev := range t.peripheralProvider() { for _, field := range dev.RequiredFields() { if _, exists := seen[field]; !exists { seen[field] = struct{}{} @@ -263,3 +274,11 @@ func (t *TelemetryService) IsStreaming() bool { } // Streaming Control [END] ----------------------------------------------------- + +// Callbacks [START] ----------------------------------------------------------- + +func (t *TelemetryService) PeripheralFoundCallback(pname string) { + t.Messages <- "Telemetry service callback for peripheral found called\n" +} + +// Callbacks [END] ------------------------------------------------------------- diff --git a/services/telemetry_events.go b/services/telemetry_events.go index 0899855..f42b3ab 100644 --- a/services/telemetry_events.go +++ b/services/telemetry_events.go @@ -28,6 +28,7 @@ func (t *TelemetryService) onFindProvider(prov telem.TelemetryProvider) { go t.ProviderMonitor(t.CtxHealthcheck) // Tell the devices service we got a provider + t.OnProviderFound(prov.Name()) } func (t *TelemetryService) onProviderStopsMidStream() {