From 82972c96651e412af24110cc48d52480045ddc3d Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Tue, 22 Sep 2026 16:21:50 +0100 Subject: [PATCH] Added healthchecks, and maybe got a bit lost --- devices/cdashdisplay/connect.go | 2 +- devices/cdashdisplay/display.go | 34 +-- devices/cdashdisplay/info.go | 12 ++ devices/cdashdisplay/setup.go | 36 ++++ devices/uidevice/device.go | 4 + peripheral/communication/communication.go | 1 + .../packets/newWindowIDPakcket.go | 19 ++ peripheral/communication/walkieTalkie.go | 9 +- peripheral/peripheral.go | 1 + services/devices.go | 4 +- services/devices_lookup.go | 203 ++++++++++++++---- 11 files changed, 241 insertions(+), 84 deletions(-) create mode 100644 devices/cdashdisplay/setup.go diff --git a/devices/cdashdisplay/connect.go b/devices/cdashdisplay/connect.go index fe65496..bdec694 100644 --- a/devices/cdashdisplay/connect.go +++ b/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) } diff --git a/devices/cdashdisplay/display.go b/devices/cdashdisplay/display.go index b8e2d07..4314ec1 100644 --- a/devices/cdashdisplay/display.go +++ b/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) - } -} diff --git a/devices/cdashdisplay/info.go b/devices/cdashdisplay/info.go index 1dfb8d2..68b7bb1 100644 --- a/devices/cdashdisplay/info.go +++ b/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 +} diff --git a/devices/cdashdisplay/setup.go b/devices/cdashdisplay/setup.go new file mode 100644 index 0000000..0afd001 --- /dev/null +++ b/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) + } +} diff --git a/devices/uidevice/device.go b/devices/uidevice/device.go index c97d493..c32032d 100644 --- a/devices/uidevice/device.go +++ b/devices/uidevice/device.go @@ -58,3 +58,7 @@ func (uid *UIDevice) RequiredFields() []telemetry.FieldID { telemetry.RPM, } } + +func (uid *UIDevice) HealthCheck() bool { + return true +} diff --git a/peripheral/communication/communication.go b/peripheral/communication/communication.go index bf7393b..729f95e 100644 --- a/peripheral/communication/communication.go +++ b/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{ diff --git a/peripheral/communication/packets/newWindowIDPakcket.go b/peripheral/communication/packets/newWindowIDPakcket.go index baef6c1..346b0c9 100644 --- a/peripheral/communication/packets/newWindowIDPakcket.go +++ b/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 +} diff --git a/peripheral/communication/walkieTalkie.go b/peripheral/communication/walkieTalkie.go index fde5936..e2b4c0a 100644 --- a/peripheral/communication/walkieTalkie.go +++ b/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 err + return fmt.Errorf("error reading incoming: %+v, err:", b, 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 } diff --git a/peripheral/peripheral.go b/peripheral/peripheral.go index 56cf08d..2ea1170 100644 --- a/peripheral/peripheral.go +++ b/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 diff --git a/services/devices.go b/services/devices.go index 8b52d2d..c688821 100644 --- a/services/devices.go +++ b/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 } diff --git a/services/devices_lookup.go b/services/devices_lookup.go index 99cf69b..3203dbc 100644 --- a/services/devices_lookup.go +++ b/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), -) error { - state, err := pss.GetState(pname) - if err != nil { - pss.Logger.Error("Can't reconnect device", "device", pname, "error", err) - return err - } + onDiscovery func(string, peripheral.Peripheral), + onFailure func(string), +) { + pss.Logger.Debug("looking for device", "name", pname) + pss.setDeviceIsDiscovering(pname) + pss.Messages <- "Discovering " + pname + "\n" - dev, err := state.device.Discover() - if err != nil { - return err - } + go func() { + state, err := pss.GetState(pname) + if err != nil { + pss.Logger.Error("Can't reconnect device", "device", pname, "error", err) + onFailure(pname) + return + } - // Register the device we just found - onDiscovery(pname, dev) + dev, err := state.device.Discover() + if err != nil { + onFailure(pname) + return + } - return nil + // 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) + pss.setDeviceConfigured(pname) + }() } func (pss *PeripheralStateStore) handleDeviceTimedOut(pname string) error { - err := pss.discoverPeripheral(pname, pss.setDeviceReconnected) - if err != nil { - // Log something - return err - } - + pss.discoverPeripheral(pname, pss.setDeviceReconnected, pss.setDeviceTimedOut) 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() - if err != nil { - return err - } - - err = state.Peripheral.Setup(provider) - if err != nil { - return err - } - - pss.setDeviceConfigured(pname) + pss.setDeviceIsConfiguring(pname) + pss.configurePeripheral(pname, pss.setDeviceConfigured, pss.setDeviceUnconfigured) 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) - err := pss.discoverPeripheral(pName, pss.setDeviceConnected) - if err != nil { - // Log something - continue - } + pss.discoverPeripheral(pName, pss.setDeviceConnected, pss.setDeviceDisconnected) + case DeviceIsDiscovering: + // We need to set a device into discovery mode so we won't retrigger discoveries + // and pool them up 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) - if err != nil { - // Something is not adding up, it should be logged somewhere... SYKE - continue - } + pss.handleDeviceIsUnconfigured(pName) + case DeviceIsConfiguring: + // Do nothing configuration is happening in the background 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) + } + } }