Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
009609da93 | ||
|
|
91f33c79ed | ||
|
|
d10c6997f4 | ||
|
|
b364931c80 | ||
|
|
6ff4cc3014 | ||
|
|
263d3c29a3 | ||
|
|
5d5b7bfdc3 | ||
|
|
dcba0774c2 |
@@ -0,0 +1,6 @@
|
||||
package constants
|
||||
|
||||
const (
|
||||
IRacingProviderName = "iRacing"
|
||||
BeamNGProviderName = "BeamNG.drive"
|
||||
)
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"esdi/constants"
|
||||
helper "esdi/helpers"
|
||||
"esdi/peripheral"
|
||||
"esdi/peripheral/communication"
|
||||
@@ -437,3 +438,32 @@ 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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
package cdashdisplay
|
||||
|
||||
func (cds *CDashDisplay) OnLoad() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cds *CDashDisplay) OnTelemetryProviderFound() error {
|
||||
return nil
|
||||
}
|
||||
+27
-4
@@ -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
|
||||
// }
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
package uidevice
|
||||
|
||||
func (uid *UIDevice) OnLoad() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (uid *UIDevice) OnTelemetryProviderFound() error {
|
||||
return nil
|
||||
}
|
||||
@@ -17,8 +17,11 @@ const (
|
||||
|
||||
type Peripheral interface {
|
||||
Name() string
|
||||
Setup(string) error
|
||||
SendData(*telemetry.TelemetryData) error
|
||||
RequiredFields() []telemetry.FieldID
|
||||
OnLoad() error
|
||||
OnTelemetryProviderFound() error
|
||||
Close() error
|
||||
}
|
||||
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
|
||||
+23
-7
@@ -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)
|
||||
@@ -114,7 +126,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 != DeviceIsConfigured {
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -129,3 +141,7 @@ func (ds *DeviceService) transmit(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Callbacks [START] -----------------------------------------------------------
|
||||
|
||||
// Callbacks [END] -------------------------------------------------------------
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
package services
|
||||
+137
-10
@@ -2,6 +2,7 @@ package services
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"maps"
|
||||
"sync"
|
||||
@@ -15,15 +16,18 @@ 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
|
||||
DeviceIsConnected
|
||||
DeviceIsUnconfigured
|
||||
DeviceIsConfigured
|
||||
)
|
||||
|
||||
type PeripheralState struct {
|
||||
@@ -50,15 +54,21 @@ 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)
|
||||
}
|
||||
|
||||
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 +101,7 @@ func (pss *PeripheralStateStore) GetPeripheral(pname string) (peripheral.Periphe
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if state.State != DeviceIsConnected {
|
||||
if state.State < DeviceIsConnected {
|
||||
return nil, ErrDeviceIsNotConnected
|
||||
}
|
||||
|
||||
@@ -150,9 +160,11 @@ func (pss *PeripheralStateStore) setDeviceConnected(pname string, per 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,9 +185,105 @@ 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) 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
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
dev, err := state.device.Discover()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Register the device we just found
|
||||
onDiscovery(pname, dev)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (pss *PeripheralStateStore) handleDeviceTimedOut(pname string) error {
|
||||
err := pss.discoverPeripheral(pname, pss.setDeviceReconnected)
|
||||
if err != nil {
|
||||
// Log something
|
||||
return err
|
||||
}
|
||||
|
||||
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!
|
||||
state, err := pss.GetState(pname)
|
||||
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)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (pss *PeripheralStateStore) handleDeviceIsConfigured(pname string) error {
|
||||
// Nothing to do - this method shouldn't even exist then
|
||||
return nil
|
||||
}
|
||||
|
||||
func (pss *PeripheralStateStore) HandleDeviceState() {
|
||||
peripherals := pss.GetStates()
|
||||
|
||||
@@ -183,22 +291,41 @@ func (pss *PeripheralStateStore) HandleDeviceState() {
|
||||
switch pState.State {
|
||||
case DeviceIsDisconnected:
|
||||
pss.Logger.Debug("looking for device", "name", pName)
|
||||
dev, err := pState.device.Discover()
|
||||
err := pss.discoverPeripheral(pName, pss.setDeviceConnected)
|
||||
if err != nil {
|
||||
// Log something
|
||||
continue
|
||||
}
|
||||
|
||||
// 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)
|
||||
err := pss.handleDeviceIsUnconfigured(pName)
|
||||
if err != nil {
|
||||
// Something is not adding up, it should be logged somewhere... SYKE
|
||||
continue
|
||||
}
|
||||
case DeviceIsConfigured:
|
||||
// Nothing to do here
|
||||
continue
|
||||
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
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -208,7 +335,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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
+27
-35
@@ -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] -------------------------------------------------------------
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -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() {
|
||||
|
||||
+5
-9
@@ -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),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user