partial subscriptions

not really partial. they just happen per peripheral now

constants/constants.go

@@ -0,0 +1,13 @@
+package constants
+
+// Provider Names
+const (
+	IRacingProviderName = "iRacing"
+	BeamNGProviderName  = "BeamNG.drive"
+)
+
+// Virtual Field Names
+const (
+	FuelCalculatorName = "FuelCalculator"
+	RPMLightsName      = "RPMLights"
+)

devices/cdashdisplay/connect.go

@@ -73,7 +73,7 @@ func findDisplayPort() (*communication.WalkieTalkie, error) {
 		select {
 		case err = <-probeResult:
 			// Probe completed normally (could be success or error)
-		case <-time.After(2 * time.Second):
+		case <-time.After(1000 * time.Millisecond):
 			// Hard timeout reached
 			err = fmt.Errorf("probe completely hung/timed out: %s", port)
 		}

devices/cdashdisplay/display.go

@@ -33,6 +33,8 @@ const (
 	updateWindowCMDID     types.Command = 6 // Change this to a move cmd instead
 	sendDataCMDID         types.Command = 7
 	newLayoutCMDID        types.Command = 8
+	healthCheckCMDID      types.Command = 9
+	resetCMDID            types.Command = 10
 )
 
 const (
@@ -148,9 +150,6 @@ func (cds *CDashDisplay) Close() error {
 	return nil
 }
 
-func (d *CDashDisplay) SendCommand() {
-}
-
 func (d *CDashDisplay) RegisterFieldMapping(fieldID telemetry.FieldID, winID int16) {
 	d.fieldToWindows[fieldID] = append(d.fieldToWindows[fieldID], winID)
 }
@@ -406,6 +405,17 @@ func (d *CDashDisplay) UnloadLayout() error {
 	return nil
 }
 
+func (cds *CDashDisplay) reset() error {
+	err := cds.WT.SendCommand(resetCMDID, []byte{0x01, 0x02, 0x03, 0x04}, nil)
+	if err != nil {
+		return err
+	}
+
+	time.Sleep(3000 * time.Millisecond)
+
+	return nil
+}
+
 func (d *CDashDisplay) SendData(data *telemetry.TelemetryData) error {
 	packet := d.encodePacket(data)
 

devices/cdashdisplay/events.go

@@ -0,0 +1,9 @@
+package cdashdisplay
+
+func (cds *CDashDisplay) OnLoad() error {
+	return nil
+}
+
+func (cds *CDashDisplay) OnTelemetryProviderFound() error {
+	return nil
+}

devices/cdashdisplay/info.go

@@ -1,6 +1,7 @@
 package cdashdisplay
 
 import (
+	"esdi/peripheral/communication/packets"
 	"esdi/peripheral/devices"
 	"esdi/telemetry"
 )
@@ -26,3 +27,14 @@ func (cds *CDashDisplay) RequiredFields() []telemetry.FieldID {
 
 	return fields
 }
+
+func (cds *CDashDisplay) HealthCheck() bool {
+	// Send the command
+	var health packets.HealthCheck
+	err := cds.WT.SendCommand(healthCheckCMDID, []byte{0x01, 0x02, 0x03, 0x04}, &health)
+	if err != nil {
+		return false
+	}
+
+	return true
+}

devices/cdashdisplay/setup.go

@@ -0,0 +1,44 @@
+package cdashdisplay
+
+import (
+	"fmt"
+
+	"esdi/constants"
+)
+
+func (cds *CDashDisplay) setupForIracing() error {
+	err := cds.LoadLayout("layout.yaml")
+	if err != nil {
+		return err
+	}
+
+	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 {
+	// Doesn't matter which one we are picking, we need to reset the CDashDisplay
+	// first - we will improve this setup behaviour later on with an Update to
+	// change it without dropping connection or some shit
+	err := cds.reset()
+	if err != nil {
+		return err
+	}
+
+	switch provider {
+	case constants.IRacingProviderName:
+		return cds.setupForIracing()
+	case constants.BeamNGProviderName:
+		return cds.setupForBeamNG()
+	default:
+		return fmt.Errorf("unknown provider: %s", provider)
+	}
+}

devices/device.go

@@ -2,28 +2,33 @@
 package devices
 
 import (
+	"errors"
+
 	"esdi/devices/cdashdisplay"
 	"esdi/devices/uidevice"
 	"esdi/peripheral"
 )
 
+var ErrInvalidDevice = errors.New("invalid device")
+
 type Device struct {
 	Name     string
 	Discover func() (peripheral.Peripheral, error)
+	// DefaultSetup func(peripheral.Peripheral) error
 }
 
 var List map[string]*Device = map[string]*Device{
 	uidevice.NAME: {
 		Name:     uidevice.NAME,
-		Discover: DiscoverUIDevice,
+		Discover: UIDeviceDiscover,
 	},
 	cdashdisplay.NAME: {
 		Name:     cdashdisplay.NAME,
-		Discover: DiscoverCDashDisplay,
+		Discover: CDashDisplayDiscover,
 	},
 }
 
-func DiscoverUIDevice() (peripheral.Peripheral, error) {
+func UIDeviceDiscover() (peripheral.Peripheral, error) {
 	uidev, err := uidevice.NewUIDevice()
 	if err != nil {
 		return nil, err
@@ -32,7 +37,11 @@ func DiscoverUIDevice() (peripheral.Peripheral, error) {
 	return uidev, nil
 }
 
-func DiscoverCDashDisplay() (peripheral.Peripheral, error) {
+// func UIDeviceSetup(peripheral peripheral.Peripheral) error {
+// 	return nil
+// }
+
+func CDashDisplayDiscover() (peripheral.Peripheral, error) {
 	// Create a cdashdisplay
 	display, err := cdashdisplay.NewCDashDisplay()
 	if err != nil {
@@ -41,3 +50,17 @@ func DiscoverCDashDisplay() (peripheral.Peripheral, error) {
 
 	return display, nil
 }
+
+// func CDashDisplaySetup(peripheral peripheral.Peripheral) error {
+// 	cdash, ok := peripheral.(*cdashdisplay.CDashDisplay)
+// 	if !ok {
+// 		return ErrInvalidDevice
+// 	}
+//
+// 	err := cdash.LoadLayout("layout.yaml")
+// 	if err != nil {
+// 		return err
+// 	}
+//
+// 	return nil
+// }

devices/uidevice/device.go

@@ -25,6 +25,10 @@ func (uid *UIDevice) Close() error {
 	return nil
 }
 
+func (uid *UIDevice) Setup(provider string) error {
+	return nil
+}
+
 func (uid *UIDevice) SendData(data *telemetry.TelemetryData) error {
 	if data == nil {
 		return peripheral.ErrInvalidData
@@ -54,3 +58,7 @@ func (uid *UIDevice) RequiredFields() []telemetry.FieldID {
 		telemetry.RPM,
 	}
 }
+
+func (uid *UIDevice) HealthCheck() bool {
+	return true
+}

devices/uidevice/events.go

@@ -0,0 +1,9 @@
+package uidevice
+
+func (uid *UIDevice) OnLoad() error {
+	return nil
+}
+
+func (uid *UIDevice) OnTelemetryProviderFound() error {
+	return nil
+}

peripheral/communication/communication.go

@@ -17,6 +17,7 @@ const (
 	CmdAckID         types.Command = 2
 	CmdCreateWindow  types.Command = 3
 	CmdDestroyWindow types.Command = 4
+	CmdHealthCheck   types.Command = 9
 )
 
 var crc8Table = [256]byte{

peripheral/communication/packets/newWindowIDPakcket.go

@@ -18,3 +18,22 @@ func (pkt *NewWindowID) Validate() bool {
 
 	return true
 }
+
+type HealthCheck struct {
+	StartMarker byte
+	Response    byte
+	EndMarker   byte
+}
+
+func (pkt *HealthCheck) Validate() bool {
+	if pkt.StartMarker != constvar.StartOfText ||
+		pkt.EndMarker != constvar.EndOfText {
+		return false
+	}
+
+	if pkt.Response != 0x06 {
+		return false
+	}
+
+	return true
+}

peripheral/communication/walkieTalkie.go

@@ -26,8 +26,7 @@ func (wt *WalkieTalkie) ReadFramedData(size int, packet any) error {
 		b := make([]byte, 1)
 		_, err := wt.Serial.Read(b)
 		if err != nil {
-			// fmt.Fprintf(os.Stderr, "dev read: %s\n", err.Error())
+			return fmt.Errorf("error reading incoming: %+v, err:", b, err)
-			return err
 		}
 
 		if b[0] == constvar.StartOfText {
@@ -44,9 +43,11 @@ func (wt *WalkieTalkie) ReadFramedData(size int, packet any) error {
 	reader := bytes.NewReader(buf)
 	err = binary.Read(reader, binary.LittleEndian, packet)
 	if err != nil {
-		return err
+		return fmt.Errorf("error parsing incoming: %+v, err:", buf, err)
 	}
 
+	wt.Serial.Flush()
+
 	return nil
 }
 
@@ -137,6 +138,8 @@ func (wt *WalkieTalkie) sendPacket(cmd types.Command, data any) error {
 		return err
 	}
 
+	wt.Serial.Flush()
+
 	return nil
 }
 
@@ -158,35 +161,6 @@ func (wt *WalkieTalkie) readPacket(resp packets.Packet) error {
 	return nil
 }
 
-// func (wt *WalkieTalkie) sendHeader(h *header) error {
-// 	err := wt.sendPacket(h)
-// 	if err != nil {
-// 		return err
-// 	}
-//
-// 	// var ack packets.AckPacket
-// 	// err = wt.readPacket(&ack)
-// 	// if err != nil {
-// 	// 	return err
-// 	// }
-//
-// 	return nil
-// }
-
-// func (wt *WalkieTalkie) sendBody(payload any, resp packets.Packet) error {
-// 	err := wt.sendPacket(payload)
-// 	if err != nil {
-// 		return err
-// 	}
-//
-// 	// err = wt.readPacket(resp)
-// 	// if err != nil {
-// 	// 	return err
-// 	// }
-//
-// 	return nil
-// }
-
 func (wt *WalkieTalkie) SendCommand(
 	cmd types.Command,
 	payload any,

peripheral/peripheral.go

@@ -17,8 +17,12 @@ const (
 
 type Peripheral interface {
 	Name() string
+	Setup(string) error
+	HealthCheck() bool
 	SendData(*telemetry.TelemetryData) error
 	RequiredFields() []telemetry.FieldID
+	OnLoad() error
+	OnTelemetryProviderFound() error
 	Close() error
 }
 

providers/beamng/beamng.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 	"time"
 
+	"esdi/constants"
 	"esdi/telemetry"
 
 	bngsdk "github.com/ESilva15/gobngsdk"
@@ -27,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
@@ -36,7 +40,7 @@ type BeamNG struct {
 }
 
 const (
-	NAME = "BeamNG.drive"
+	NAME = constants.BeamNGProviderName
 )
 
 func NewBeamNGProvider(logger *slog.Logger, opts *bngsdk.Options) (*BeamNG, error) {
@@ -50,6 +54,8 @@ func NewBeamNGProvider(logger *slog.Logger, opts *bngsdk.Options) (*BeamNG, erro
 		data:   telemetry.NewTelemetryData(),
 		SDK:    beam,
 		og:     &bngsdk.Outgauge{},
+		// Field subscription management
+		boundFields: make(map[telemetry.FieldID]bool, telemetry.MaxFields),
 		ticker:      time.NewTicker(time.Second / 60),
 	}
 
@@ -114,47 +120,68 @@ func (b *BeamNG) Stream() (<-chan telemetry.TelemetryData, error) {
 	return ch, nil
 }
 
-func (b *BeamNG) Subscribe(requestFields []telemetry.FieldID) {
+// TODO: this function is exactly the same in BeamNG drive now, and I reckon it will be the same
-	// NOTE: document how the Subscribe funtion works
+// In plenty other things. I should make it TelemetryData method
-	slog.Debug(fmt.Sprintf("Len Req: %d\n", len(requestFields)))
+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
+	for _, id := range fields {
-	// we will add their dependencies and the primitives to a slice
+		if i.boundFields[id] {
-	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] {
 			continue
 		}
 
-		binding := telemetry.BoundField{
+		i.data.ActiveBinds[id] = telemetry.BoundField{
 			ID: id,
 		}
-
+		i.boundFields[id] = true
-		b.data.ActiveBinds = append(b.data.ActiveBinds, binding)
+		newSubscriptions = append(newSubscriptions, telemetry.FieldNames[id])
-		boundCheck[id] = true
 	}
 
-	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

providers/iracing/iracing.go

@@ -10,13 +10,14 @@ import (
 	"sync"
 	"time"
 
+	"esdi/constants"
 	"esdi/telemetry"
 
 	"github.com/ESilva15/goirsdk"
 )
 
 const (
-	NAME = "iRacing"
+	NAME = constants.IRacingProviderName
 )
 
 // IRacing is our iRacing telemetry data provider - its a TelemetryProvider interface
@@ -28,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
 	streamCancel context.CancelFunc
 }
 
@@ -54,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),
 	}
@@ -224,46 +228,67 @@ func (i *IRacing) StopStream() {
 	i.streamCancel = nil
 }
 
-func (i *IRacing) Subscribe(requestFields []telemetry.FieldID) {
+// TODO: this function is exactly the same in BeamNG drive now, and I reckon it will be the same
-	i.logger.Debug(fmt.Sprintf("Len Req: %d\n", len(requestFields)))
+// 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
+	for _, id := range fields {
-	// we will add their dependencies and the primitives to a slice
+		if i.boundFields[id] {
-	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] {
 			continue
 		}
 
-		binding := telemetry.BoundField{
+		i.data.ActiveBinds[id] = telemetry.BoundField{
 			ID: id,
 		}
-
+		i.boundFields[id] = true
-		i.data.ActiveBinds = append(i.data.ActiveBinds, binding)
+		newSubscriptions = append(newSubscriptions, telemetry.FieldNames[id])
-		boundCheck[id] = true
 	}
 
+	// 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 {

services/devices.go

@@ -2,6 +2,7 @@ package services
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
 	"sync/atomic"
 
@@ -22,15 +23,19 @@ type DeviceService struct {
 	TelemCh      <-chan telemetry.TelemetryData
 	// Output
 	Messages chan string
+	// Callbacks
+	// Telemetry service data fetchers
+	telemetryProvider        func() (string, error)
+	triggerFieldSubscription func([]telemetry.FieldID) []string
 }
 
-func NewDeviceService(logger *slog.Logger) *DeviceService {
+func NewDeviceService(logger *slog.Logger, msg chan string) *DeviceService {
-	sharedChannel := make(chan string, 10)
-
 	dev := &DeviceService{
-		PSS:      NewPeripheralStateStore(logger.With("Service", "PeripheralStateStore"), devices.List),
+		PSS: NewPeripheralStateStore(
+			logger.With("Service", "PeripheralStateStore"), devices.List, msg,
+		),
 		Logger:   logger,
-		Messages: sharedChannel,
+		Messages: msg,
 	}
 
 	// Start the routine that looks for devices - should always be running in the background
@@ -38,16 +43,26 @@ func NewDeviceService(logger *slog.Logger) *DeviceService {
 	dev.ctxDiscovery, dev.ctxDiscoveryCancel = context.WithCancel(context.Background())
 	go dev.FindDevices()
 
+	// Set the callbacks for PSS
+	dev.PSS.telemetryProvider = dev.getTelemetryProvider
+	dev.PSS.onPeripheralConfigured = dev.peripheralConfigured
+
 	return dev
 }
 
 // Getters [START] -------------------------------------------------------------
+// This function is currently only being used by PSS, we may have to find a better
+// pattern for this
+func (ds *DeviceService) getTelemetryProvider() (string, error) {
+	return ds.telemetryProvider()
+}
+
 func (ds *DeviceService) GetDevices() []peripheral.Peripheral {
 	snapshot := ds.PSS.GetStates()
 	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)
@@ -73,6 +88,8 @@ func (ds *DeviceService) StartStream() {
 	var ctx context.Context
 	ctx, ds.streamCancel = context.WithCancel(context.Background())
 
+	ds.PSS.OnStartStream()
+
 	go ds.transmit(ctx)
 }
 
@@ -81,6 +98,8 @@ func (ds *DeviceService) StopStream() {
 		return
 	}
 
+	ds.PSS.OnStopStream()
+
 	ds.streamCancel()
 	ds.streamCancel = nil
 }
@@ -114,7 +133,7 @@ func (ds *DeviceService) transmit(ctx context.Context) {
 			// TODO: make a copy of the data and send that copy instead of keeping
 			// the data locked
 			for _, dev := range ds.PSS.GetStates() {
-				if dev.State != DeviceIsConnected {
+				if dev.State != DeviceIsStreaming {
 					continue
 				}
 
@@ -129,3 +148,12 @@ 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] -------------------------------------------------------------

services/devices_events.go

@@ -0,0 +1 @@
+package services

services/devices_lookup.go

@@ -2,6 +2,7 @@ package services
 
 import (
 	"errors"
+	"fmt"
 	"log/slog"
 	"maps"
 	"sync"
@@ -9,23 +10,55 @@ import (
 
 	"esdi/devices"
 	"esdi/peripheral"
+	"esdi/telemetry"
 )
 
 var (
 	ErrPeripheralAlreadyRegistered = errors.New("peripheral is already registered")
 	ErrNoSuchDevice                = errors.New("device doesn't exist")
 	ErrDeviceIsNotConnected        = errors.New("device isn't connected")
+	ErrFailedToSetupPeripheral     = errors.New("peripheral setup failed")
 )
 
 type DeviceState = uint8
 
 const (
 	DeviceTimedOut uint8 = iota
-	DeviceIsConnected
 	DeviceIsDisconnected
 	DeviceReconnected
+	DeviceIsDiscovering
+	DeviceIsConnected
+	DeviceIsUnconfigured
+	DeviceIsConfiguring
+	DeviceIsConfigured
+	DeviceIsStreaming
 )
 
+func DeviceStateToStr(state DeviceState) string {
+	switch state {
+	case DeviceTimedOut:
+		return "DeviceTimedOut"
+	case DeviceIsDisconnected:
+		return "DeviceIsDisconnected"
+	case DeviceReconnected:
+		return "DeviceReconnected"
+	case DeviceIsDiscovering:
+		return "DeviceIsDiscovering"
+	case DeviceIsConnected:
+		return "DeviceIsConnected"
+	case DeviceIsUnconfigured:
+		return "DeviceIsUnconfigured"
+	case DeviceIsConfiguring:
+		return "DeviceIsConfiguring"
+	case DeviceIsConfigured:
+		return "DeviceIsConfigured"
+	case DeviceIsStreaming:
+		return "DeviceIsStreaming"
+	default:
+		return "UnknownState"
+	}
+}
+
 type PeripheralState struct {
 	device     *devices.Device
 	Peripheral peripheral.Peripheral
@@ -50,15 +83,24 @@ type PeripheralStateStore struct {
 	Logger *slog.Logger
 	mu     sync.RWMutex
 	store  map[string]*PeripheralState
+	// Messaging for UI and stuff
+	Messages chan string
+	// Callbacks
+	telemetryProvider      func() (string, error)
+	onPeripheralConfigured func(string, []telemetry.FieldID)
+	// Internal State
+	isStreaming bool
 }
 
 func NewPeripheralStateStore(
 	nLogger *slog.Logger,
 	devList map[string]*devices.Device,
+	msg chan string,
 ) *PeripheralStateStore {
 	store := PeripheralStateStore{
 		Logger:   nLogger,
 		store:    make(map[string]*PeripheralState),
+		Messages: msg,
 	}
 
 	for _, dev := range devList {
@@ -91,7 +133,7 @@ func (pss *PeripheralStateStore) GetPeripheral(pname string) (peripheral.Periphe
 		return nil, err
 	}
 
-	if state.State != DeviceIsConnected {
+	if state.State < DeviceIsConnected {
 		return nil, ErrDeviceIsNotConnected
 	}
 
@@ -137,22 +179,74 @@ func (pss *PeripheralStateStore) DeleteDevice(pname string) error {
 	return nil
 }
 
+func (pss *PeripheralStateStore) GetStreamingState() bool {
+	pss.mu.RLock()
+	defer pss.mu.RUnlock()
+	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) {
-	// We need to deregister the device
 	ds.PSS.setDeviceTimedOut(pname)
 }
 
+func (pss *PeripheralStateStore) OnStartStream() {
+	pss.mu.Lock()
+	pss.isStreaming = true
+	pss.mu.Unlock()
+
+	for _, state := range pss.GetStates() {
+		if state.State == DeviceIsConfigured {
+			pss.setDeviceIsStreaming(state.device.Name)
+		}
+	}
+}
+
+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] --------------------------------------------------------------
 
 // Device State Handling [START] -----------------------------------------------
+func (pss *PeripheralStateStore) setDeviceDisconnected(pname string) {
+	pss.mu.Lock()
+	defer pss.mu.Unlock()
+	pss.store[pname].State = DeviceIsDisconnected
+}
+
+func (pss *PeripheralStateStore) setDeviceIsDiscovering(pname string) {
+	pss.mu.Lock()
+	defer pss.mu.Unlock()
+	pss.store[pname].State = DeviceIsDiscovering
+}
+
 func (pss *PeripheralStateStore) setDeviceConnected(pname string, per peripheral.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)
 }
 
 func (pss *PeripheralStateStore) setDeviceTimedOut(pname string) {
@@ -173,34 +267,216 @@ func (pss *PeripheralStateStore) setDeviceReconnected(pname string, per peripher
 	pss.store[pname].State = DeviceReconnected
 }
 
+func (pss *PeripheralStateStore) setDeviceUnconfigured(pname string) {
+	pss.Logger.Info("device is connected but not configured", "device", pname)
+
+	pss.mu.Lock()
+	defer pss.mu.Unlock()
+	pss.store[pname].State = DeviceIsUnconfigured
+}
+
+func (pss *PeripheralStateStore) setDeviceIsConfiguring(pname string) {
+	pss.Logger.Info("device is configuring for new provider", "device", pname)
+
+	pss.mu.Lock()
+	defer pss.mu.Unlock()
+	pss.store[pname].State = DeviceIsConfiguring
+}
+
+func (pss *PeripheralStateStore) setDeviceConfigured(pname string) {
+	pss.Logger.Info("device is configured and ready for data", "device", pname)
+
+	pss.mu.Lock()
+	pss.store[pname].State = DeviceIsConfigured
+	pss.mu.Unlock()
+
+	pss.onPeripheralConfigured(pname, pss.GetPeripheralFields(pname))
+}
+
+func (pss *PeripheralStateStore) setDeviceIsStreaming(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 = DeviceIsStreaming
+}
+
 // Device State Handling [END] -------------------------------------------------
 
 // Device Handling [START] -----------------------------------------------------
+func (pss *PeripheralStateStore) discoverPeripheral(
+	pname string,
+	onDiscovery func(string, peripheral.Peripheral),
+	onFailure func(string),
+) {
+	pss.Logger.Debug("looking for device", "name", pname)
+	pss.setDeviceIsDiscovering(pname)
+	pss.Messages <- "Discovering " + pname + "\n"
+
+	go func() {
+		state, err := pss.GetState(pname)
+		if err != nil {
+			pss.Logger.Error("Can't reconnect device", "device", pname, "error", err)
+			onFailure(pname)
+			return
+		}
+
+		dev, err := state.device.Discover()
+		if err != nil {
+			onFailure(pname)
+			return
+		}
+
+		// Register the device we just found
+		onDiscovery(pname, dev)
+	}()
+}
+
+func (pss *PeripheralStateStore) configurePeripheral(
+	pname string,
+	onSuccess func(string),
+	onFailure func(string),
+) {
+	go func() {
+		state, err := pss.GetState(pname)
+		if err != nil {
+			onFailure(pname)
+			return
+		}
+
+		provider, err := pss.telemetryProvider()
+		if err != nil {
+			onFailure(pname)
+			return
+		}
+
+		err = state.Peripheral.Setup(provider)
+		if err != nil {
+			onFailure(pname)
+			return
+		}
+
+		onSuccess(pname)
+	}()
+}
+
+func (pss *PeripheralStateStore) handleDeviceTimedOut(pname string) error {
+	pss.Messages <- "device " + pname + " timed out\n"
+	pss.discoverPeripheral(pname, pss.setDeviceReconnected, pss.setDeviceTimedOut)
+	return nil
+}
+
+func (pss *PeripheralStateStore) handleDeviceReconnected(pname string) error {
+	// The device has reconnected, but we must set its state again
+	state, err := pss.GetState(pname)
+	if err != nil {
+		return err
+	}
+
+	// Update the peripheral state
+	pss.setDeviceConnected(pname, state.Peripheral)
+
+	return nil
+}
+
+// handleDeviceConnected will handle the device setup after it connects
+// NOTE: should this be a state after Connected?
+// Connected -> Unconfigured -> Configured I believe this would work nicely
+// THIS IS A TODO ↑↑↑↑↑↑
+func (pss *PeripheralStateStore) handleDeviceConnected(pname string) error {
+	pss.setDeviceUnconfigured(pname)
+	return nil
+}
+
+func (pss *PeripheralStateStore) handleDeviceIsUnconfigured(pname string) error {
+	// Here we need to configure our device. If no error occurs its configured!
+	_, err := pss.telemetryProvider()
+	if err != nil {
+		return err
+	}
+
+	pss.setDeviceIsConfiguring(pname)
+	pss.configurePeripheral(pname, pss.setDeviceConfigured, pss.setDeviceUnconfigured)
+
+	return nil
+}
+
+func (pss *PeripheralStateStore) handleDeviceIsConfigured(pname string) error {
+	// Here we have to check wheter we are streaming or not. If we aren't streaming
+	// then we ought to do a healthcheck on the peripheral
+	if pss.GetStreamingState() {
+		pss.Messages <- "returning device " + pname + " into streaming\n"
+		pss.setDeviceIsStreaming(pname)
+	}
+
+	return nil
+}
+
+func (pss *PeripheralStateStore) performHealthCheck(pname string, state *PeripheralState) bool {
+	healthStatus := state.Peripheral.HealthCheck()
+	if !healthStatus {
+		pss.Messages <- "peripheral " + pname + " failed healthcheck"
+		pss.setDeviceTimedOut(pname)
+		return false
+	}
+
+	return true
+}
+
 func (pss *PeripheralStateStore) HandleDeviceState() {
 	peripherals := pss.GetStates()
 
 	for pName, pState := range peripherals {
 		switch pState.State {
 		case DeviceIsDisconnected:
-			pss.Logger.Debug("looking for device", "name", pName)
+			pss.discoverPeripheral(pName, pss.setDeviceConnected, pss.setDeviceDisconnected)
-			dev, err := pState.device.Discover()
+		case DeviceIsDiscovering:
-			if err != nil {
+			// We need to set a device into discovery mode so we won't retrigger discoveries
-				continue
+			// and pool them up
-			}
-
-			// Register the device we just found
-			pss.setDeviceConnected(pName, dev)
 		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 DeviceIsUnconfigured:
+			pss.Logger.Debug("Device is still being configured.", "device", pName)
+			pss.handleDeviceIsUnconfigured(pName)
+		case DeviceIsConfiguring:
+			// Do nothing configuration is happening in the background
+		case DeviceIsConfigured:
+			// Nothing to do here
+			pss.handleDeviceIsConfigured(pName)
+		case DeviceIsStreaming:
+			//
 		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)
+			err := pss.handleDeviceReconnected(pName)
+			if err != nil {
+				pss.Logger.Error("device reconnection handler failed", "error", err)
+				continue
+			}
 		case DeviceTimedOut:
 			// If the device has timed out we need to re-discover it or something
 			pss.Logger.Debug("Device is timed out. Attempting to recconect", "device", pName)
+			err := pss.handleDeviceTimedOut(pName)
+			if err != nil {
+				pss.Logger.Error("device timing out handler failed", "error", err)
+				continue
 			}
 		}
+
+		updatedState, err := pss.GetState(pName)
+		if err != nil {
+			// TODO: log something useful here
+			continue
+		}
+		if updatedState.State == DeviceIsConnected ||
+			updatedState.State == DeviceIsUnconfigured ||
+			updatedState.State == DeviceIsConfigured {
+			pss.performHealthCheck(pName, updatedState)
+		}
+
+	}
 }
 
 // Device Handling [END] -------------------------------------------------------
@@ -208,7 +484,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 {

services/services.go

@@ -1,2 +1,39 @@
 // Package services interacts with the other libraries required for this UI
 package services
+
+import (
+	"errors"
+	"log/slog"
+)
+
+type Orchestrator struct {
+	DeviceService    *DeviceService
+	TelemetryService *TelemetryService
+	Messages         chan string
+}
+
+func NewOrchestrator(logger *slog.Logger) (*Orchestrator, error) {
+	msg := make(chan string, 10)
+
+	devService := NewDeviceService(logger.With("service", "DeviceService"), 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)
+
+	// Setup device service callbacks
+	devService.telemetryProvider = telemService.GetTelemetryProviderName
+	devService.triggerFieldSubscription = telemService.SubscribeToFields
+
+	// Setup telemetry service callbacks
+	// telemService.getRequiredFields = devService.GetRequiredFields
+
+	return &Orchestrator{
+		DeviceService:    devService,
+		TelemetryService: telemService,
+		Messages:         msg,
+	}, nil
+}

services/telemetry.go

@@ -2,6 +2,7 @@ package services
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"sync"
 	"time"
@@ -11,11 +12,12 @@ import (
 	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
 	// Streaming
 	isStreaming bool
 	// Concurrency protection
@@ -32,16 +34,21 @@ type TelemetryService struct {
 	cancelMonitor     context.CancelFunc
 	CtxHealthcheck    context.Context
 	healthCheckCancel context.CancelFunc
+	// Callbacks
+	// Devices data request
+	// peripheralProvider func() []peripheral.Peripheral
+	// getRequiredFields func() []telemetry.FieldID
 }
 
-func NewTelemetryService(logger *slog.Logger, devServo *DeviceService) *TelemetryService {
+func NewTelemetryService(
-	sharedChannel := make(chan string, 10)
+	logger *slog.Logger,
+	msg chan string,
+) *TelemetryService {
 	newService := &TelemetryService{
 		logger:      logger,
 		isConnected: false,
-		devService:  devServo,
 		listeners:   make(map[string]chan telem.TelemetryData),
-		Messages:    sharedChannel,
+		Messages:    msg,
 	}
 	newService.CtxMonitor, newService.cancelMonitor = context.WithCancel(context.Background())
 
@@ -59,6 +66,7 @@ func (t *TelemetryService) ProviderMonitor(ctx context.Context) {
 		case <-ticker.C:
 			slog.Info("checking if provider is still running")
 			if !t.activeProvider.IsAlive(500 * time.Millisecond) {
+				t.Messages <- "Healthcheck on provider failing. Dropping provider.\n"
 				slog.Warn("provider healthcheck failed")
 				t.dropActiveProvider()
 				t.onProviderHealthCheckFailed()
@@ -68,6 +76,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 {
@@ -98,23 +114,11 @@ func (t *TelemetryService) UnsubscribeListener(id string) {
 	}
 }
 
-func (t *TelemetryService) SubscribeToFields() []telem.FieldID {
+func (t *TelemetryService) SubscribeToFields(fields []telemetry.FieldID) []string {
-	seen := make(map[telemetry.FieldID]struct{})
+	t.logger.Debug("requested fields", "fields", fields)
-	var allFields []telemetry.FieldID
+	subscribed := t.activeProvider.Subscribe(fields)
 
-	for _, dev := range t.devService.GetDevices() {
+	return subscribed
-		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
 }
 
 // Listener Control [END] ------------------------------------------------------
@@ -154,34 +158,6 @@ func (t *TelemetryService) SwitchProvider(newProvider telem.TelemetryProvider) e
 	return nil
 }
 
-func (t *TelemetryService) onProviderHealthCheckFailed() {
-	// Just restart the whole lookup process
-	go t.FindProvider(t.CtxMonitor)
-}
-
-func (t *TelemetryService) onFindProvider(prov telem.TelemetryProvider) {
-	// Attach to the provider
-	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())
-	go t.ProviderMonitor(t.CtxHealthcheck)
-}
-
-func (t *TelemetryService) onProviderStopsMidStream() {
-	// clear the current provider
-	// TODO: now we need to also clear the devices to restart everything,
-	// if the stream stopped we have to restart the devices and everything
-	t.logger.Info("cleaning dropped provider and restarting lookup service")
-	t.dropActiveProvider()
-	go t.FindProvider(t.CtxMonitor)
-}
-
 // TODO: add some way of retriggering this. Currently it should:
 // start monitoring on startup -> find provider -> stop monitoring (when game closes for example)
 func (t *TelemetryService) FindProvider(ctx context.Context) {
@@ -285,3 +261,7 @@ func (t *TelemetryService) IsStreaming() bool {
 }
 
 // Streaming Control [END] -----------------------------------------------------
+
+// Callbacks [START] -----------------------------------------------------------
+
+// Callbacks [END] -------------------------------------------------------------

services/telemetry_events.go

@@ -0,0 +1,39 @@
+package services
+
+import (
+	"context"
+	"fmt"
+
+	telem "esdi/telemetry"
+)
+
+func (t *TelemetryService) onProviderHealthCheckFailed() {
+	// Just restart the whole lookup process
+	go t.FindProvider(t.CtxMonitor)
+}
+
+func (t *TelemetryService) onFindProvider(prov telem.TelemetryProvider) {
+	// Attach to the provider
+	t.logger.Info("found provider for " + prov.Name())
+	t.Messages <- fmt.Sprintf("Found provider \"%s\"\n", prov.Name())
+	err := t.SwitchProvider(prov)
+	if err != nil {
+		t.Messages <- fmt.Sprintf("Failed to switch to provider: %+v\n", err.Error())
+		t.logger.Error("failed to switch to provider onFindProvider", "err", err)
+		return
+	}
+
+	// Start the healthcheck on our provider so we can drop it if it stops
+	t.CtxHealthcheck, t.healthCheckCancel = context.WithCancel(context.Background())
+	go t.ProviderMonitor(t.CtxHealthcheck)
+}
+
+func (t *TelemetryService) onProviderStopsMidStream() {
+	// clear the current provider
+	// TODO: now we need to also clear the devices to restart everything,
+	// if the stream stopped we have to restart the devices and everything
+	t.Messages <- "Telemetry provider stopped mid stream\n"
+	t.logger.Info("cleaning dropped provider and restarting lookup service")
+	t.dropActiveProvider()
+	go t.FindProvider(t.CtxMonitor)
+}

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
+	ActiveBinds         map[FieldID]BoundField
-	VirtualBinds        []VirtualField
+	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),
+	}
 }

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
+}

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()

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
+}

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
+}

tui/internal/controllers/cdashdisplay_layout.go

@@ -35,6 +35,7 @@ func NewLayoutController(base *Controller, service *services.DeviceService) *Lay
 		DevService:     service,
 		MoveToolState:  &windowManipState{Mode: moveMode},
 		// SelectedLayout: "beamng.yaml",
+		// TODO: this can't be here - the service/peripheral needs to know about it
 		SelectedLayout: "layout.yaml",
 	}
 
@@ -404,7 +405,8 @@ func (lc *LayoutController) loadLayout() {
 	}
 	// ---
 
-	// We would get the layout path from somewhere but for nots its layout.yaml
+	// TODO: This can happen here, but we need to address how the layout is gotten
+	// THE UI SHOULD SET STATE IN THE SERVICES ONLY
 	err = display.LoadLayout(lc.SelectedLayout)
 	if err != nil {
 		lc.Messages <- "failed to load layout: " + err.Error()

tui/internal/controllers/device.go

@@ -16,19 +16,18 @@ type DeviceController struct {
 	DeviceAPIView *views.DeviceAPIView
 	LayoutCtrl    *LayoutController
 	StreamCtrl    *StreamingCtrl
-	DevService    *serv.DeviceService
+	Orchestrator  *serv.Orchestrator
 }
 
 func NewDeviceController(
 	base *Controller,
-	devService *serv.DeviceService,
+	orchestrator *serv.Orchestrator,
-	telemService *serv.TelemetryService,
 ) *DeviceController {
 	mc := &DeviceController{
 		Controller:   base,
-		LayoutCtrl: NewLayoutController(base, devService),
+		LayoutCtrl:   NewLayoutController(base, orchestrator.DeviceService),
-		DevService: devService,
+		Orchestrator: orchestrator,
-		StreamCtrl: NewStreamingCtrl(base, devService, telemService),
+		StreamCtrl:   NewStreamingCtrl(base, orchestrator.DeviceService, orchestrator.TelemetryService),
 	}
 
 	return mc
@@ -57,7 +56,7 @@ func (mc *DeviceController) setDeviceAPIViewEvents() {
 		SetInputCapture(func(ev *tcell.EventKey) *tcell.EventKey {
 			switch ev.Rune() {
 			case 'r':
-				go mc.DevService.FindDevices()
+				go mc.Orchestrator.DeviceService.FindDevices()
 			}
 			return ev
 		})
@@ -67,8 +66,8 @@ func (mc *DeviceController) AddDeviceAPIListItems() {
 	mc.DeviceAPIView.DevAPIList.
 		AddItem("layout", "build a layout for CDashDisplay", func() {
 			// This CDashDisplay specific, only load if we have a CDashDisplay
-			if !mc.DevService.PeripheralExists(cdashdisplay.NAME) {
+			if !mc.Orchestrator.DeviceService.PeripheralExists(cdashdisplay.NAME) {
-				mc.DevService.Messages <- "CDashDisplay it not loaded yet\n"
+				mc.Orchestrator.DeviceService.Messages <- "CDashDisplay it not loaded yet\n"
 				return
 			}
 
@@ -94,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)
 		})
@@ -116,7 +115,7 @@ func (mc *DeviceController) injectControllerCallbacks() {
 
 func (mc *DeviceController) injectChannels() {
 	go func() {
-		for msg := range mc.DevService.Messages {
+		for msg := range mc.Orchestrator.Messages {
 			mc.PrintToOutputWindow(msg)
 		}
 	}()

tui/internal/controllers/streaming.go

@@ -59,11 +59,6 @@ func NewStreamingCtrl(
 	return ctrl
 }
 
-// func (sc *StreamingCtrl) subscribeListeners() {
-// 	// Here I will set a UIDevice
-// 	sc.TelemetryCh = sc.TelemServ.SubscribeListener("UI", 1)
-// }
-
 func (sc *StreamingCtrl) registerHooks() {
 	sc.StreamView.Options.Form.SetInputCapture(func(ev *tcell.EventKey) *tcell.EventKey {
 		switch ev.Key() {
@@ -159,11 +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()
+	// fields := sc.TelemServ.SubscribeToAllFields()
-
+	// sc.Messages <- fmt.Sprintf("Subscribed to fields: %+v\n", fields)
-	// sc.Messages <- fmt.Sprintf("Subscribed Fields: %+v [%d]\n", fields, len(fields))
-	// Should I update this?
-	sc.Messages <- fmt.Sprintf("Subscribed to fields: %+v", fields)
 }
 
 func (sc *StreamingCtrl) listenToUIStream() {

tui/tui.go

@@ -23,19 +23,15 @@ func NewControlPanel(logger *slog.Logger) *ControlPanel {
 		App:    tview.NewApplication(),
 	}
 
-	// NOTE: create our device service here
+	orchestrator, err := services.NewOrchestrator(logger)
-	devService := services.NewDeviceService(logger.With("service", "DeviceService"))
+	if err != nil {
-
+		// TODO: no panic here
-	telemService := services.NewTelemetryService(logger, devService)
+		panic("failed to create services orchestrator")
-	if telemService == nil {
-		panic("failed to create the telemetry service")
 	}
 
-	go telemService.FindProvider(telemService.CtxMonitor)
-
 	return &ControlPanel{
 		Controller:       baseController,
-		DeviceController: controllers.NewDeviceController(baseController, devService, telemService),
+		DeviceController: controllers.NewDeviceController(baseController, orchestrator),
 	}
 }