diff --git a/constants/constants.go b/constants/constants.go new file mode 100644 index 0000000..b5d3305 --- /dev/null +++ b/constants/constants.go @@ -0,0 +1,6 @@ +package constants + +const ( + IRacingProviderName = "iRacing" + BeamNGProviderName = "BeamNG.drive" +) 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 267c86d..51b84a0 100644 --- a/devices/cdashdisplay/display.go +++ b/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) diff --git a/devices/cdashdisplay/events.go b/devices/cdashdisplay/events.go new file mode 100644 index 0000000..365e32b --- /dev/null +++ b/devices/cdashdisplay/events.go @@ -0,0 +1,9 @@ +package cdashdisplay + +func (cds *CDashDisplay) OnLoad() error { + return nil +} + +func (cds *CDashDisplay) OnTelemetryProviderFound() error { + return nil +} 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..74117f4 --- /dev/null +++ b/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) + } +} diff --git a/devices/device.go b/devices/device.go index 84ab486..33db928 100644 --- a/devices/device.go +++ b/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 +// } diff --git a/devices/uidevice/device.go b/devices/uidevice/device.go index 763f759..c32032d 100644 --- a/devices/uidevice/device.go +++ b/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 +} diff --git a/devices/uidevice/events.go b/devices/uidevice/events.go new file mode 100644 index 0000000..f43b2a4 --- /dev/null +++ b/devices/uidevice/events.go @@ -0,0 +1,9 @@ +package uidevice + +func (uid *UIDevice) OnLoad() error { + return nil +} + +func (uid *UIDevice) OnTelemetryProviderFound() error { + return nil +} 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..a751f73 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 } @@ -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, diff --git a/peripheral/peripheral.go b/peripheral/peripheral.go index ed667a7..2ea1170 100644 --- a/peripheral/peripheral.go +++ b/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 } diff --git a/providers/beamng/beamng.go b/providers/beamng/beamng.go index f07b933..7e371cc 100644 --- a/providers/beamng/beamng.go +++ b/providers/beamng/beamng.go @@ -8,6 +8,7 @@ import ( "sync" "time" + "esdi/constants" "esdi/telemetry" bngsdk "github.com/ESilva15/gobngsdk" @@ -36,7 +37,7 @@ type BeamNG struct { } const ( - NAME = "BeamNG.drive" + NAME = constants.BeamNGProviderName ) func NewBeamNGProvider(logger *slog.Logger, opts *bngsdk.Options) (*BeamNG, error) { diff --git a/providers/iracing/iracing.go b/providers/iracing/iracing.go index 9701f6f..d271ea6 100644 --- a/providers/iracing/iracing.go +++ b/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 diff --git a/services/devices.go b/services/devices.go index 214208a..c688821 100644 --- a/services/devices.go +++ b/services/devices.go @@ -22,15 +22,18 @@ type DeviceService struct { TelemCh <-chan telemetry.TelemetryData // Output Messages chan string + // Callbacks + // Telemetry service data fetchers + telemetryProvider func() (string, error) } -func NewDeviceService(logger *slog.Logger) *DeviceService { - sharedChannel := make(chan string, 10) - +func NewDeviceService(logger *slog.Logger, msg chan string) *DeviceService { 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 +41,25 @@ 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 + 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 < DeviceIsConnected { continue } peripherals = append(peripherals, state.Peripheral) @@ -73,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) } @@ -114,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 != DeviceIsConnected { + if dev.State != DeviceIsStreaming { continue } @@ -129,3 +143,7 @@ func (ds *DeviceService) transmit(ctx context.Context) { } } } + +// Callbacks [START] ----------------------------------------------------------- + +// Callbacks [END] ------------------------------------------------------------- diff --git a/services/devices_events.go b/services/devices_events.go new file mode 100644 index 0000000..5e568ea --- /dev/null +++ b/services/devices_events.go @@ -0,0 +1 @@ +package services diff --git a/services/devices_lookup.go b/services/devices_lookup.go index 98e7655..499b8ea 100644 --- a/services/devices_lookup.go +++ b/services/devices_lookup.go @@ -2,6 +2,7 @@ package services import ( "errors" + "fmt" "log/slog" "maps" "sync" @@ -15,17 +16,48 @@ 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 +82,23 @@ 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) + // 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), + Logger: nLogger, + store: make(map[string]*PeripheralState), + Messages: msg, } for _, dev := range devList { @@ -91,7 +131,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 +177,59 @@ func (pss *PeripheralStateStore) DeleteDevice(pname string) error { return nil } +func (pss *PeripheralStateStore) GetStreamingState() bool { + pss.mu.RLock() + defer pss.mu.RUnlock() + return pss.isStreaming +} + // "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() +} + // "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,33 +250,214 @@ 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() + defer pss.mu.Unlock() + 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(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) + pss.setDeviceConfigured(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) - dev, err := pState.device.Discover() - if err != nil { - continue - } - - // Register the device we just found - pss.setDeviceConnected(pName, dev) + 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) + 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) + } + } } @@ -208,7 +466,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 bb02e66..18aff69 100644 --- a/services/services.go +++ b/services/services.go @@ -1,2 +1,38 @@ // 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 + + // Setup telemetry service callbacks + telemService.peripheralProvider = devService.GetDevices + + return &Orchestrator{ + DeviceService: devService, + TelemetryService: telemService, + Messages: msg, + }, nil +} diff --git a/services/telemetry.go b/services/telemetry.go index 674be83..7ca2112 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 @@ -32,16 +35,20 @@ type TelemetryService struct { cancelMonitor context.CancelFunc CtxHealthcheck context.Context healthCheckCancel context.CancelFunc + // Callbacks + // Devices data request + peripheralProvider func() []peripheral.Peripheral } -func NewTelemetryService(logger *slog.Logger, devServo *DeviceService) *TelemetryService { - sharedChannel := make(chan string, 10) +func NewTelemetryService( + 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 { @@ -102,7 +118,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{}{} @@ -154,34 +170,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 +273,7 @@ func (t *TelemetryService) IsStreaming() bool { } // Streaming Control [END] ----------------------------------------------------- + +// Callbacks [START] ----------------------------------------------------------- + +// Callbacks [END] ------------------------------------------------------------- diff --git a/services/telemetry_events.go b/services/telemetry_events.go new file mode 100644 index 0000000..6774a63 --- /dev/null +++ b/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) +} diff --git a/tui/internal/controllers/cdashdisplay_layout.go b/tui/internal/controllers/cdashdisplay_layout.go index f90f6a7..a1a12b6 100644 --- a/tui/internal/controllers/cdashdisplay_layout.go +++ b/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() diff --git a/tui/internal/controllers/device.go b/tui/internal/controllers/device.go index f0f86a3..9d3d653 100644 --- a/tui/internal/controllers/device.go +++ b/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, - telemService *serv.TelemetryService, + orchestrator *serv.Orchestrator, ) *DeviceController { mc := &DeviceController{ - Controller: base, - LayoutCtrl: NewLayoutController(base, devService), - DevService: devService, - StreamCtrl: NewStreamingCtrl(base, devService, telemService), + Controller: base, + LayoutCtrl: NewLayoutController(base, orchestrator.DeviceService), + Orchestrator: orchestrator, + 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) { - mc.DevService.Messages <- "CDashDisplay it not loaded yet\n" + if !mc.Orchestrator.DeviceService.PeripheralExists(cdashdisplay.NAME) { + mc.Orchestrator.DeviceService.Messages <- "CDashDisplay it not loaded yet\n" return } @@ -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) } }() diff --git a/tui/internal/controllers/streaming.go b/tui/internal/controllers/streaming.go index 9778602..4fd57c1 100644 --- a/tui/internal/controllers/streaming.go +++ b/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() { @@ -160,10 +155,7 @@ func (sc *StreamingCtrl) updateStream() { // so we can get away with using a map for convenience here func (sc *StreamingCtrl) SetInternalState() { fields := sc.TelemServ.SubscribeToFields() - - // 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) + sc.Messages <- fmt.Sprintf("Subscribed to fields: %+v\n", fields) } func (sc *StreamingCtrl) listenToUIStream() { diff --git a/tui/tui.go b/tui/tui.go index 4add4bb..a1c1f3b 100644 --- a/tui/tui.go +++ b/tui/tui.go @@ -23,19 +23,15 @@ func NewControlPanel(logger *slog.Logger) *ControlPanel { App: tview.NewApplication(), } - // NOTE: create our device service here - devService := services.NewDeviceService(logger.With("service", "DeviceService")) - - telemService := services.NewTelemetryService(logger, devService) - if telemService == nil { - panic("failed to create the telemetry service") + orchestrator, err := services.NewOrchestrator(logger) + if err != nil { + // TODO: no panic here + panic("failed to create services orchestrator") } - go telemService.FindProvider(telemService.CtxMonitor) - return &ControlPanel{ Controller: baseController, - DeviceController: controllers.NewDeviceController(baseController, devService, telemService), + DeviceController: controllers.NewDeviceController(baseController, orchestrator), } }