Added healthchecks, and maybe got a bit lost

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

@@ -11,7 +11,6 @@ import (
 	"sync"
 	"time"
 
-	"esdi/constants"
 	helper "esdi/helpers"
 	"esdi/peripheral"
 	"esdi/peripheral/communication"
@@ -34,6 +33,7 @@ const (
 	updateWindowCMDID     types.Command = 6 // Change this to a move cmd instead
 	sendDataCMDID         types.Command = 7
 	newLayoutCMDID        types.Command = 8
+	healthCheckCMDID      types.Command = 9
 )
 
 const (
@@ -149,9 +149,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)
 }
@@ -438,32 +435,3 @@ func (d *CDashDisplay) SendData(data *telemetry.TelemetryData) error {
 
 	return nil
 }
-
-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 {
-	switch provider {
-	case constants.IRacingProviderName:
-		return cds.setupForIracing()
-	case constants.BeamNGProviderName:
-		return cds.setupForBeamNG()
-	default:
-		return fmt.Errorf("unknown provider: %s", provider)
-	}
-}

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,36 @@
+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 {
+	switch provider {
+	case constants.IRacingProviderName:
+		return cds.setupForIracing()
+	case constants.BeamNGProviderName:
+		return cds.setupForBeamNG()
+	default:
+		return fmt.Errorf("unknown provider: %s", provider)
+	}
+}

devices/uidevice/device.go

@@ -58,3 +58,7 @@ func (uid *UIDevice) RequiredFields() []telemetry.FieldID {
 		telemetry.RPM,
 	}
 }
+
+func (uid *UIDevice) HealthCheck() bool {
+	return true
+}

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
 }
 

peripheral/peripheral.go

@@ -18,6 +18,7 @@ const (
 type Peripheral interface {
 	Name() string
 	Setup(string) error
+	HealthCheck() bool
 	SendData(*telemetry.TelemetryData) error
 	RequiredFields() []telemetry.FieldID
 	OnLoad() error

services/devices.go

@@ -85,6 +85,8 @@ func (ds *DeviceService) StartStream() {
 	var ctx context.Context
 	ctx, ds.streamCancel = context.WithCancel(context.Background())
 
+	ds.PSS.OnStartStream()
+
 	go ds.transmit(ctx)
 }
 
@@ -126,7 +128,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 != DeviceIsConfigured {
+				if dev.State != DeviceIsStreaming {
 					continue
 				}
 

services/devices_lookup.go

@@ -25,11 +25,39 @@ const (
 	DeviceTimedOut uint8 = iota
 	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
@@ -149,13 +177,32 @@ func (pss *PeripheralStateStore) DeleteDevice(pname string) error {
 
 // "Events" [START] ------------------------------------------------------------
 func (ds *DeviceService) onDeviceTimedOut(pname string) {
-	// We need to deregister the device
 	ds.PSS.setDeviceTimedOut(pname)
 }
 
+func (pss *PeripheralStateStore) OnStartStream() {
+	for _, state := range pss.GetStates() {
+		if state.State == DeviceIsConfigured {
+			pss.setDeviceIsStreaming(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)
 
@@ -193,6 +240,14 @@ func (pss *PeripheralStateStore) setDeviceUnconfigured(pname string) {
 	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)
 
@@ -201,37 +256,76 @@ func (pss *PeripheralStateStore) setDeviceConfigured(pname string) {
 	pss.store[pname].State = DeviceIsConfigured
 }
 
+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(pName string, peripheral peripheral.Peripheral),
+	onDiscovery func(string, peripheral.Peripheral),
-) error {
+	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)
-		return err
+			onFailure(pname)
+			return
 		}
 
 		dev, err := state.device.Discover()
 		if err != nil {
-		return err
+			onFailure(pname)
+			return
 		}
 
 		// Register the device we just found
 		onDiscovery(pname, dev)
+	}()
+}
 
-	return nil
+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)
+		pss.setDeviceConfigured(pname)
+	}()
 }
 
 func (pss *PeripheralStateStore) handleDeviceTimedOut(pname string) error {
-	err := pss.discoverPeripheral(pname, pss.setDeviceReconnected)
+	pss.discoverPeripheral(pname, pss.setDeviceReconnected, pss.setDeviceTimedOut)
-	if err != nil {
-		// Log something
-		return err
-	}
-
 	return nil
 }
 
@@ -259,57 +353,58 @@ func (pss *PeripheralStateStore) handleDeviceConnected(pname string) error {
 
 func (pss *PeripheralStateStore) handleDeviceIsUnconfigured(pname string) error {
 	// Here we need to configure our device. If no error occurs its configured!
-	state, err := pss.GetState(pname)
+	_, err := pss.telemetryProvider()
 	if err != nil {
 		return err
 	}
 
-	provider, err := pss.telemetryProvider()
+	pss.setDeviceIsConfiguring(pname)
-	if err != nil {
+	pss.configurePeripheral(pname, pss.setDeviceConfigured, pss.setDeviceUnconfigured)
-		return err
-	}
-
-	err = state.Peripheral.Setup(provider)
-	if err != nil {
-		return err
-	}
-
-	pss.setDeviceConfigured(pname)
 
 	return nil
 }
 
 func (pss *PeripheralStateStore) handleDeviceIsConfigured(pname string) error {
-	// Nothing to do - this method shouldn't even exist then
+	// 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
 	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)
-			err := pss.discoverPeripheral(pName, pss.setDeviceConnected)
+		case DeviceIsDiscovering:
-			if err != nil {
+			// We need to set a device into discovery mode so we won't retrigger discoveries
-				// Log something
+			// and pool them up
-				continue
-			}
 		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)
-			err := pss.handleDeviceIsUnconfigured(pName)
+			pss.handleDeviceIsUnconfigured(pName)
-			if err != nil {
+		case DeviceIsConfiguring:
-				// Something is not adding up, it should be logged somewhere... SYKE
+			// Do nothing configuration is happening in the background
-				continue
-			}
 		case DeviceIsConfigured:
 			// Nothing to do here
-			continue
+			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)
@@ -327,6 +422,22 @@ func (pss *PeripheralStateStore) HandleDeviceState() {
 				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.Messages <- fmt.Sprintf(
+				"Performing healthcheck. STATE: %s\n",
+				DeviceStateToStr(updatedState.State),
+			)
+			pss.performHealthCheck(pName, updatedState)
+		}
+
 	}
 }