this is a doozy - device auto discovery and handling

The main goal was to decouple more things We got devices being looked for in the background and the stream view is like a device now too
2026-09-17 23:01:50 +01:00
parent 1321977f18
commit a6ab64117c
17 changed files with 327 additions and 111 deletions
@@ -2,23 +2,32 @@ package services

 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
+	"sync"
 	"sync/atomic"
+	"time"

 	"esdi/devices"
-	"esdi/devices/cdashdisplay"
+	"esdi/peripheral"
 	"esdi/telemetry"
 )

+var ErrPeripheralAlreadyRegistered = errors.New("peripheral is already registered")
+
 // DeviceService will handle sending the data from the telemetry service to the
 // actual devices
 // NOTE: create a virtual device and make it be the output window or something so
 // we can just add it as a device or whatever instead of being a custom made thing
 // that would be pretty cool I think
 type DeviceService struct {
-	Logger  *slog.Logger
-	Devices map[string]devices.Device
+	Logger *slog.Logger
+	// Device discovery
+	mu                 sync.RWMutex
+	ctxDiscovery       context.Context
+	ctxDiscoveryCancel context.CancelFunc
+	Devices            map[string]peripheral.Peripheral
 	// Strem handling
 	streamCancel context.CancelFunc
 	TelemCh      <-chan telemetry.TelemetryData
@@ -28,41 +37,75 @@ type DeviceService struct {

 func NewDeviceService(logger *slog.Logger) *DeviceService {
 	sharedChannel := make(chan string, 10)
-	return &DeviceService{
-		Devices:  make(map[string]devices.Device),
+
+	dev := &DeviceService{
+		Devices:  make(map[string]peripheral.Peripheral),
 		Logger:   logger,
 		Messages: sharedChannel,
 	}
+
+	// Start the routine that looks for devices - should always be running in the background
+	// Create a routine to poll this provider while we wait to start the stream or pause it
+	dev.ctxDiscovery, dev.ctxDiscoveryCancel = context.WithCancel(context.Background())
+	go dev.FindDevices()
+
+	return dev
 }

 func (ds *DeviceService) FindDevices() {
 	// Need to define a list of devices to search for
 	// For now lets just try to find our cdashdisplay - will think about the rest later
+	ticker := time.NewTicker(2 * time.Second)
+	defer ticker.Stop()

-	// Find CDashDisplay
-	{
-		ds.Messages <- "looking for " + cdashdisplay.Name + "...\n"
-		ds.Logger.Info("Looking for " + cdashdisplay.Name)
-
-		cdashdisplay.SetLogger(ds.Logger.With("[device]", cdashdisplay.Name))
-
-		// Create a cdashdisplay
-		display, err := cdashdisplay.Discover()
-		if err == nil {
-			ds.Devices[cdashdisplay.Name] = display
-			ds.Logger.Info("found " + cdashdisplay.Name + " on: " + display.WT.Cfg.Name)
-			ds.Messages <- "found " + cdashdisplay.Name + " on: " + display.WT.Cfg.Name + "\n"
+	for {
+		select {
+		case <-ds.ctxDiscovery.Done():
+			// If requested to cancel we cancel background discovery
 			return
-		}
+		case <-ticker.C:
+			for pName, peripheral := range devices.List {
+				if ds.DeviceExists(pName) {
+					// We already discovered this device
+					continue
+				}

-		ds.Logger.Info("didn't find " + cdashdisplay.Name)
-		ds.Messages <- "didn't find " + cdashdisplay.Name + "\n"
-		// No CDashDisplay available for one reason or another, so we don't set the
-		// key
+				ds.Logger.Debug("looking for device", "name", pName)
+				dev, err := peripheral.Discover()
+				if err != nil {
+					ds.Logger.Debug("didn't find device", "name", pName)
+					continue
+				}
+
+				// Register the device we just found
+				ds.RegisterDevice(dev)
+			}
+		}
 	}
 }

-func (ds *DeviceService) GetDevice(name string) (devices.Device, error) {
+// func (ds *DeviceService) SubscribeFields() error {
+// 	for _, dev := range ds.Devices {
+// 		fields := dev.RequiredFields()
+// 	}
+//
+// 	return nil
+// }
+
+func (ds *DeviceService) RegisterDevice(dev peripheral.Peripheral) error {
+	ds.mu.Lock()
+	defer ds.mu.Unlock()
+
+	if ds.DeviceExists(dev.Name()) {
+		return ErrPeripheralAlreadyRegistered
+	}
+
+	ds.Devices[dev.Name()] = dev
+
+	return nil
+}
+
+func (ds *DeviceService) GetDevice(name string) (peripheral.Peripheral, error) {
 	val, ok := ds.Devices[name]
 	if !ok {
 		return nil, fmt.Errorf("device `%s` couldn't be found", name)
@@ -122,9 +165,12 @@ func (ds *DeviceService) transmit(ctx context.Context) {

 			// TODO: make a copy of the data and send that copy instead of keeping
 			// the data locked
+			ds.mu.RLock()
 			for _, dev := range ds.Devices {
 				dev.SendData(&data)
 			}
+			ds.mu.RUnlock()
+
 			isSending.Store(false)
 		}
 	}
@@ -54,9 +54,9 @@ func (t *TelemetryService) ProviderMonitor(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			slog.Debug("checking if provider is still running")
+			slog.Info("checking if provider is still running")
 			if !t.activeProvider.IsAlive(500 * time.Millisecond) {
-				slog.Debug("provider healthcheck failed")
+				slog.Warn("provider healthcheck failed")
 				t.dropActiveProvider()
 				t.onProviderHealthCheckFailed()
 				return
@@ -70,10 +70,14 @@ func (t *TelemetryService) onProviderHealthCheckFailed() {
 	go t.FindProvider(t.CtxMonitor)
 }

-func (t *TelemetryService) onFindProvider(prov providers.Provider) {
+func (t *TelemetryService) onFindProvider(prov telem.TelemetryProvider) {
 	// Attach to the provider
-	t.logger.Info("found provider for " + prov.Name)
-	t.SwitchProvider(prov.NewProvider(t.logger))
+	t.logger.Info("found provider for " + prov.Name())
+	err := t.SwitchProvider(prov)
+	if err != nil {
+		t.logger.Error("failed to switch to provider onFindProvider", "err", err)
+		return
+	}

 	// Create a routine to poll this provider while we wait to start the stream or pause it
 	t.CtxHealthcheck, t.healthCheckCancel = context.WithCancel(context.Background())
@@ -102,9 +106,15 @@ func (t *TelemetryService) FindProvider(ctx context.Context) {
 			return
 		case <-ticker.C:
 			for _, prov := range providers.Providers {
-				t.logger.Debug("checking provider: " + prov.Name)
-				if prov.IsRunning() {
-					t.onFindProvider(prov)
+				// t.logger.Debug("checking provider: " + prov.Name)
+				provider, err := prov.NewProvider(t.logger)
+				if err != nil {
+					// Its not running
+					continue
+				}
+
+				if provider.IsAlive(500 * time.Millisecond) {
+					t.onFindProvider(provider)
 					return
 				}
 			}
@@ -141,6 +151,7 @@ func (t *TelemetryService) multiplexData(ctx context.Context, dataCh <-chan tele

 			t.mut.RLock()
 			for _, ch := range t.listeners {
+				// t.logger.Debug("sending data to listener", "listener", key, "data", data)
 				select {
 				case ch <- data:
 					// Sends data to the subscriber
@@ -191,8 +202,14 @@ func (t *TelemetryService) UnsubscribeListener(id string) {
 	}
 }

-func (t *TelemetryService) SubscribeToFields(fields map[int16]telem.FieldID) {
-	t.activeProvider.Subscribe(fields)
+func (t *TelemetryService) SubscribeToFields() {
+	// _ = t.devService.SubscribeFields()
+	// TODO: we need to find a way of requesting devices to send all subscribed fields
+	// instead of going through the devices on DevService here
+	for _, dev := range t.devService.Devices {
+		fields := dev.RequiredFields()
+		t.activeProvider.Subscribe(fields)
+	}
 }

 func (t *TelemetryService) StartStream() {