ESilvaLabs/esdi
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

telemetry service and telemetry agents interface

Browse Source
This commit is contained in:
esilva
2026-03-05 16:26:30 +00:00
parent c9086cae56
commit e65358ea1d
16 changed files with 388 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files
conversions/conversions.go
+6
View File
@@ -0,0 +1,6 @@
// Package conversions will host all of our unit conversion functions
package conversions
func MsToKph(v float32) int {
return int((3600 * v) / 1000)
}
helpers/helper.go
+22 -3
View File
@@ -11,9 +11,9 @@ type Vector struct {
DY uint16
}
type MultiError struct {
Errors []error
}
// type MultiError struct {
// Errors []error
// }
func B32(s string) [32]byte {
var b [32]byte
@@ -30,3 +30,22 @@ func StructToBytes(s any) ([]byte, error) {
return buf.Bytes(), nil
}
// func CopyBytes(dest []byte, destSize int, src string) {
// copy(dest[:], []byte(src))
// dest[min(destSize-1, len(src))] = '\x00'
// }
func CopyBytes(dest []byte, src string) {
// 1. Clear the destination (optional but safer for fixed-width telemetry)
for i := range dest {
dest[i] = 0
}
// 2. Copy as much as fits, leaving at least 1 byte for a null terminator
// We limit the copy to len(dest) - 1
n := copy(dest[:len(dest)-1], src)
// 3. Explicitly null terminate after the last written byte
dest[n] = '\x00'
}
oldEsdi/esdi.go
+11 -22
View File
@@ -2,9 +2,7 @@
package esdi
import (
"esdi/sources/iracing"
"log"
"os"
"time"
"github.com/ESilva15/goirsdk"
@@ -67,26 +65,17 @@ func RunLiveTelemetry(port string, output string, session string) {
}
func RunOfflineTelemetry(port string, input string, output string, session string) {
esdi, err := ESDIInit(port, 115200)
if err != nil {
log.Fatalf("Failed to get Desktop Interface: %v", err)
}
file, err := os.Open(input)
if err != nil {
log.Fatalf("Failed to open IBT file: %v", err)
}
irsdk, err := iracing.Init(file, output, session)
if err != nil {
log.Fatalf("Failed to create iRacing interface: %v", err)
}
// irsdk, err := goirsdk.Init(file, outFile, sessionFile)
// esdi, err := ESDIInit(port, 115200)
// if err != nil {
// log.Fatalf("Failed to create irsdk instance: %v\n", err)
// log.Fatalf("Failed to get Desktop Interface: %v", err)
// }
esdi.irsdk = irsdk.SDK
esdi.telemetry()
//
// irsdk, err := iracing.Init(input, output, session)
// if err != nil {
// log.Fatalf("Failed to create iRacing interface: %v", err)
// }
//
// esdi.irsdk = irsdk.SDK
//
// esdi.telemetry()
}
sources/beamng/beamngSink.go → providers/beamng/beamng.go
View File
providers/iracing/iracing.go
+126
View File
@@ -0,0 +1,126 @@
package iracing
import (
"context"
"fmt"
"log"
"os"
"sync"
"time"
conv "esdi/conversions"
helper "esdi/helpers"
telem "esdi/telemetry"
"github.com/ESilva15/goirsdk"
)
// IRacing is our iRacing telemetry data provider - its a TelemetryProvider interface
type IRacing struct {
SDK *goirsdk.IBT
// Data Handling
mut sync.Mutex
data *telem.TelemetryData
// Timing information
initialTime time.Time
lastMessageTime time.Time
ticker time.Ticker // ticker will keep polling intervals constant
// Stream
streamCh chan telem.TelemetryData
streamCancel context.CancelFunc
// isRunning bool
}
func NewIRacingProvider(source string, telemOut string, yamlOut string) (*IRacing, error) {
var err error
// Open the input file if provided
var file *os.File = nil
if source != "" {
file, err = os.Open(source)
if err != nil {
log.Fatalf("Failed to open IBT file: %v", err)
}
}
sdk, err := goirsdk.Init(file, telemOut, yamlOut)
if err != nil {
return &IRacing{}, err
}
return &IRacing{
SDK: sdk,
data: telem.NewTelemetryData(),
streamCh: make(chan telem.TelemetryData),
ticker: *time.NewTicker(time.Second / 60),
}, nil
}
func (i *IRacing) stream(ctx context.Context) {
i.initialTime = time.Now()
go func() {
for {
select {
case <-ctx.Done():
return
case <-i.ticker.C:
i.ReadData()
// Publish data
i.streamCh <- *i.data
}
}
}()
}
func (i *IRacing) ReadData() {
i.mut.Lock()
defer i.mut.Unlock()
var err error
_, err = i.SDK.Update(time.Millisecond * 100)
if err != nil {
return
}
i.readVehicleData()
i.lastMessageTime = time.Now()
}
func (i *IRacing) readVehicleData() {
curGear := i.SDK.Vars.Vars["Gear"].Value
curRPM := i.SDK.Vars.Vars["RPM"].Value
curSpeed := i.SDK.Vars.Vars["Speed"].Value
speed := fmt.Sprintf("%3d", int32(conv.MsToKph(curSpeed.(float32))))
gear := fmt.Sprintf("%2d", int32(curGear.(int)))
rpm := fmt.Sprintf("%3d", int32(curRPM.(float32)))
speedArray := i.data.Values["Speed"]
helper.CopyBytes(speedArray[:], speed)
i.data.Values["Speed"] = speedArray
gearArray := i.data.Values["Gear"]
helper.CopyBytes(gearArray[:], gear)
i.data.Values["Gear"] = gearArray
rpmArray := i.data.Values["RPM"]
helper.CopyBytes(rpmArray[:], rpm)
i.data.Values["RPM"] = rpmArray
}
func (i *IRacing) Stream() (<-chan telem.TelemetryData, error) {
var ctx context.Context
ctx, i.streamCancel = context.WithCancel(context.Background())
// Start the stream
i.stream(ctx)
return i.streamCh, nil
}
sources/iracing/iracingSink.go
-45
View File
@@ -1,45 +0,0 @@
package iracing
import (
"fmt"
"time"
"github.com/ESilva15/goirsdk"
)
// This will implement the GameSink interface from the main package
type IRacing struct {
SDK *goirsdk.IBT
}
const (
NAME = "iRacing"
)
func Init(f goirsdk.Reader, telemOut string, yamlOut string) (IRacing, error) {
var err error
sdk, err := goirsdk.Init(f, telemOut, yamlOut)
if err != nil {
return IRacing{}, err
}
return IRacing{SDK: sdk}, nil
}
func (i *IRacing) GetData(fieldName string) (interface{}, error) {
if val, ok := i.SDK.Vars.Vars[fieldName]; ok {
return val.Value, nil
}
return nil, fmt.Errorf("key `%s` doesn't exist", fieldName)
}
func (i *IRacing) UpdateData() error {
_, err := i.SDK.Update(100 * time.Millisecond)
return err
}
func (i *IRacing) GetSessionInfo() (interface{}, error) {
return i.SDK.SessionInfo, nil
}
telemetry/data.go
+22
View File
@@ -0,0 +1,22 @@
package telemetry
type TelemetryField struct {
Parser func()
Value any
}
// NOTE: Replace values with a more appropriate custom field approach where
// every custom field only takes as many bytes as required
// NOTE: Add the timing fields here to count frames of data gathering and whatnot
// remember to do the same to whoever is sending data
type TelemetryData struct {
// Values map[string]*TelemetryField
Values map[string][32]byte
}
func NewTelemetryData() *TelemetryData {
return &TelemetryData{
Values: make(map[string][32]byte),
}
}
telemetry/provider.go
+6
View File
@@ -0,0 +1,6 @@
// Package telemetry is our interface with our data sources
package telemetry
type TelemetryProvider interface {
Stream() (<-chan TelemetryData, error)
}
tui/internal/controllers/device.go
+6 -10
View File
@@ -17,22 +17,18 @@ type DeviceController struct {
DevService *serv.CDashService
}
func NewDeviceController(base *Controller, devService *serv.CDashService) *DeviceController {
func NewDeviceController(
base *Controller,
devService *serv.CDashService,
telemService *serv.TelemetryService,
) *DeviceController {
mc := &DeviceController{
Controller: base,
LayoutCtrl: NewLayoutController(base, devService),
DevService: devService,
StreamCtrl: NewStreamingCtrl(base, devService),
StreamCtrl: NewStreamingCtrl(base, devService, telemService),
}
// mc.Bus.On(ui.StartStreamingReqEv{}, func(e any) {
// mc.StreamStrl.Start(mc.Bus)
// })
// mc.Bus.On(ui.StopStreamingReqEv{}, func(e any) {
// mc.StreamStrl.Stop(mc.Bus)
// })
return mc
}
tui/internal/controllers/streaming.go
+10 -13
View File
@@ -16,12 +16,18 @@ type StreamingCtrl struct {
Run bool
OnExit func()
isRunning bool
TelemServ *services.TelemetryService
}
func NewStreamingCtrl(base *Controller, ser *services.CDashService) *StreamingCtrl {
func NewStreamingCtrl(
base *Controller,
serCDash *services.CDashService,
serTelem *services.TelemetryService,
) *StreamingCtrl {
ctrl := &StreamingCtrl{
Controller: base,
Service: ser,
Service: serCDash,
TelemServ: serTelem,
Messages: make(chan string, 10),
Internal: make(chan string, 10),
Run: false,
@@ -47,27 +53,18 @@ func (sc *StreamingCtrl) registerHooks() {
sc.Start()
case 'p':
// Pause
sc.Stop()
// sc.Stop()
}
return ev
})
}
func (sc *StreamingCtrl) Stop() {
sc.Service.StopStream()
}
func (sc *StreamingCtrl) Start() {
sc.Service.StartStream()
sc.Messages <- "started stream\n"
stream := sc.Service.GetStream()
sc.Messages <- "got stream\n"
stream := sc.TelemServ.StartStream()
go func() {
for msg := range stream {
sc.Messages <- "received message\n"
sc.App.QueueUpdateDraw(func() {
sc.StreamView.Update(msg)
})
tui/internal/services/cdashdisplay.go
+40 -20
View File
@@ -9,21 +9,21 @@ import (
)
type CDashService struct {
Logger *slog.Logger
CDash *cdashdisplay.CDashDisplay
iRacingTelemetry *IRacingService
DevClerk *peripheral.PeripheralDeviceClerk
Messages chan string
Logger *slog.Logger
CDash *cdashdisplay.CDashDisplay
// iRacingTelemetry *IRacingService
DevClerk *peripheral.PeripheralDeviceClerk
Messages chan string
}
func NewCDashService(logger *slog.Logger) *CDashService {
sharedChannel := make(chan string, 10)
return &CDashService{
Logger: logger,
CDash: nil,
DevClerk: peripheral.NewPeripheralDeviceClerk(),
Messages: sharedChannel,
iRacingTelemetry: NewIRacingService(sharedChannel),
Logger: logger,
CDash: nil,
DevClerk: peripheral.NewPeripheralDeviceClerk(),
Messages: sharedChannel,
// iRacingTelemetry: NewIRacingService(sharedChannel),
}
}
@@ -94,14 +94,34 @@ func (cds *CDashService) MoveWindow(win *models.UIWindow, vec *helper.Vector) er
return nil
}
func (cds *CDashService) StartStream() {
cds.iRacingTelemetry.StartStream()
}
// func (cds *CDashService) StartStream() {
// // We need to get a channel from the telemetry agent to listen to and
// // propogate the data to the clients
// dataStream := cds.iRacingTelemetry.StartStream()
//
// // Custom types to be able to actually send data to the device
// type CDashDisplayUIWindowStreamData struct {
// IDX int16
// Data [32]byte
// }
//
// type CDashDisplayStreamData struct {
// Data []CDashDisplayUIWindowStreamData
// }
//
// // We will need a mechanism to kill this channel too I reckon
// go func() {
// for data := range dataStream {
// // We have to send this data to the device
// cds.Messages <- fmt.Sprintf("%d %d %d\n", data.Speed[:], data.Gear[:], data.RPM[:])
// }
// }()
// }
func (cds *CDashService) GetStream() <-chan string {
return cds.iRacingTelemetry.GetStream()
}
func (cds *CDashService) StopStream() {
cds.iRacingTelemetry.StopStream()
}
// func (cds *CDashService) GetStream() <-chan string {
// return cds.iRacingTelemetry.GetStream()
// }
//
// func (cds *CDashService) StopStream() {
// cds.iRacingTelemetry.StopStream()
// }
tui/internal/services/device.go
+9
View File
@@ -0,0 +1,9 @@
package services
// DeviceService will handle sending the data from the telemetry service to the
// actual devices
// NOTE: create a virtual device and make it be the output window or something so
// we can just add it as a device or whatever instead of being a custom made thing
// that would be pretty cool I think
type DeviceService struct {
}
tui/internal/services/iracingTelemetry.go
+65 -190
View File
@@ -4,19 +4,6 @@ package services
// iracing support only. But I should have a generic service that can gather
// telemetry from multiples sims and just publish it in an internal format
import (
"context"
esdi "esdi/oldEsdi"
"fmt"
"log"
"os"
"strings"
"sync"
"time"
"github.com/ESilva15/goirsdk"
)
// Car data lengths
const (
SpeedLen = 5
@@ -33,183 +20,71 @@ const (
StreamStateOff StreamState = 2
)
type IRacingService struct {
Message chan string
// Timers
LastMessageTime time.Time
InitialTime time.Time
LastTime time.Time
ticker *time.Ticker
// Data vessels
data *esdi.SimulationData
dataView *esdi.DataPacket
// Data access control
Mut sync.Mutex
// Source
Irsdk *goirsdk.IBT
// Stream control
isRunning bool
Stream chan string
StreamCancel context.CancelFunc
}
// type IRacingService struct {
// Message chan string
// // Timers
// LastMessageTime time.Time
// InitialTime time.Time
// LastTime time.Time
// ticker *time.Ticker
// // Data vessels
// data *esdi.SimulationData
// dataView *esdi.DataPacket
// // Data access control
// Mut sync.Mutex
// // Source
// Irsdk *goirsdk.IBT
// // Stream control
// isRunning bool
// UIStream chan string
// DataStream chan *esdi.DataPacket
// StreamCancel context.CancelFunc
// }
func NewIRacingService(msg chan string) *IRacingService {
// Open the telemetry file
file, err := os.Open("/home/esilva/Desktop/projetos/simracing_peripherals/testTelemetry/supercars_indianapolis.ibt")
if err != nil {
log.Fatalf("Failed to open IBT file: %v", err)
}
// func NewIRacingService(msg chan string) *IRacingService {
// // Open the telemetry file
// file, err := os.Open("/home/esilva/Desktop/projetos/simracing_peripherals/testTelemetry/supercars_indianapolis.ibt")
// if err != nil {
// log.Fatalf("Failed to open IBT file: %v", err)
// }
//
// irsdk, err := goirsdk.Init(file, "./out.ibt", "./out.yaml")
// if err != nil {
// log.Fatalf("Failed to load iRacing data")
// }
//
// return &IRacingService{
// Message: msg,
// ticker: time.NewTicker(time.Second / 60),
// Irsdk: irsdk,
// UIStream: make(chan string, 10),
// DataStream: make(chan *esdi.DataPacket),
// data: &esdi.SimulationData{},
// dataView: &esdi.DataPacket{},
// isRunning: false,
// }
// }
irsdk, err := goirsdk.Init(file, "./out.ibt", "./out.yaml")
if err != nil {
log.Fatalf("Failed to load iRacing data")
}
// func (irs *IRacingService) GetStream() <-chan string {
// return irs.UIStream
// }
//
// func (irs *IRacingService) StartStream() <-chan *esdi.DataPacket {
// if irs.isRunning {
// return nil
// }
//
// var ctx context.Context
// ctx, irs.StreamCancel = context.WithCancel(context.Background())
//
// irs.startStream(ctx)
// irs.isRunning = true
//
// return irs.DataStream
// }
return &IRacingService{
Message: msg,
ticker: time.NewTicker(time.Second / 60),
Irsdk: irsdk,
Stream: make(chan string, 10),
data: &esdi.SimulationData{},
dataView: &esdi.DataPacket{},
isRunning: false,
}
}
func (irs *IRacingService) GetStream() <-chan string {
return irs.Stream
}
func (irs *IRacingService) StartStream() {
if irs.isRunning {
return
}
var ctx context.Context
ctx, irs.StreamCancel = context.WithCancel(context.Background())
irs.startStream(ctx)
irs.isRunning = true
}
func (irs *IRacingService) StopStream() {
if irs.StreamCancel != nil {
irs.StreamCancel()
}
}
func (irs *IRacingService) startStream(ctx context.Context) {
irs.Message <- "STARTING THIS\n"
irs.InitialTime = time.Now()
go func() {
for {
select {
case <-ctx.Done():
irs.Message <- "CONTEXT SAID WE ARE DONE\n"
return
case <-irs.ticker.C:
irs.Message <- "GOT A TICK\n"
irs.ReadData(ctx)
irs.Stream <- irs.Stringified()
}
irs.Message <- "we are we going???\n"
}
}()
}
func (irs *IRacingService) ReadData(ctx context.Context) {
irs.Mut.Lock()
defer irs.Mut.Unlock()
var err error
_, err = irs.Irsdk.Update(time.Millisecond * 100)
if err != nil {
return
}
irs.getVehicleData()
// Test the actual dataPacket we are sending over the wire
copyBytes(irs.dataView.Speed[:], SpeedLen, fmt.Sprintf("%3d", irs.data.Speed))
copyBytes(irs.dataView.Gear[:], GearLen, fmt.Sprintf("%2d", irs.data.Gear))
copyBytes(irs.dataView.RPM[:], RpmLen, fmt.Sprintf("%3d", irs.data.RPM))
irs.LastMessageTime = time.Now()
}
func (irs *IRacingService) getVehicleData() {
curGear := irs.Irsdk.Vars.Vars["Gear"].Value
curRPM := irs.Irsdk.Vars.Vars["RPM"].Value
curSpeed := irs.Irsdk.Vars.Vars["Speed"].Value
irs.data.Gear = int32(curGear.(int))
irs.data.RPM = int32(curRPM.(float32))
irs.data.Speed = int32(msToKph(curSpeed.(float32)))
}
func copyBytes(dest []byte, destSize int, src string) {
copy(dest[:], []byte(src))
dest[min(destSize-1, len(src))] = '\x00'
}
func msToKph(v float32) int {
return int((3600 * v) / 1000)
}
func (irs *IRacingService) Stringified() string {
var buffer strings.Builder
irs.Mut.Lock()
sessionTimeR := irs.Irsdk.Vars.Vars["SessionTime"].Value
sessionTime := float64(sessionTimeR.(float64))
currTime := time.Now()
delta := currTime.Sub(irs.LastTime)
buffer.WriteString(fmt.Sprintf("[%s]\n", currTime.Format("2006/01/02 15:04:05.000")))
buffer.WriteString(fmt.Sprintf("Delta: %d [%f]\n\n", delta.Milliseconds(), 1000.0/60.0))
irs.LastTime = currTime
elapsed := currTime.Sub(irs.InitialTime)
softwareElapsed := time.Unix(0, 0).Add(elapsed).Format("04:05.000")
sessionElapsed := time.Unix(0, 0).
Add(time.Duration(sessionTime * float64(time.Second))).
Format("04:05.000")
buffer.WriteString(fmt.Sprintf("Elapsed (software): %s\n",
softwareElapsed))
buffer.WriteString(fmt.Sprintf("Elapsed (session): %s\n\n",
sessionElapsed))
// buffer.WriteString("Car data:\n")
buffer.WriteString(fmt.Sprintf("Gear: %d, RPM: %d, Speed: %d\n\n",
irs.data.Gear, irs.data.RPM, irs.data.Speed))
// buffer.WriteString("Fuel data:\n")
// buffer.WriteString(fmt.Sprintf("Fuel Est: %s\n\n", e.dataPacket.FuelEst))
// buffer.WriteString("Lap data:\n")
// buffer.WriteString(fmt.Sprintf("Delta: [%s] [%f] [%s]\n", e.dataPacket.DeltaToBestLap,
// e.data.LapDeltaFloat, lapTimeDeltaRepresentation(e.data.LapDeltaFloat)))
// buffer.WriteString(fmt.Sprintf("LapTime: %s\n", e.dataPacket.CurrLapTime))
// buffer.WriteString(fmt.Sprintf("Best Lap Time: %s\n", e.dataPacket.BestLapTime))
// buffer.WriteString(fmt.Sprintf("Last Lap Time: %s\n", e.dataPacket.LastLapTime))
// buffer.WriteString(fmt.Sprintf("LapBestNLapTi: %f\n\n", e.data.LapBestNLapTime))
// buffer.WriteString("Position data:\n")
// buffer.WriteString(fmt.Sprintf("Pos: %d\n", e.dataPacket.Position))
// for p, v := range e.dataPacket.Standings {
// s := fmt.Sprintf("[%2d] %s %-16s %-16s\n",
// p+1, v.Lap, string(bytes.Trim(v.DriverName[:], "\x00")), v.TimeBehindString)
// buffer.WriteString(s)
// }
// buffer.WriteString(fmt.Sprintf("Size: %v\n", binary.Size(DataPacket{})))
// buffer.WriteString(fmt.Sprintf("Recv: %d\n", e.data.Recv))
// buffer.WriteString(fmt.Sprintf("Recv Err: %v\n", e.data.ReadError))
irs.Mut.Unlock()
return buffer.String()
}
// func (irs *IRacingService) StopStream() {
// if irs.StreamCancel != nil {
// irs.StreamCancel()
// }
// }
tui/internal/services/telemetry.go
+41 -2
View File
@@ -1,6 +1,45 @@
package services
// Telemetry will be our base struct to handle telemetry data
import (
providerir "esdi/providers/iracing"
telemetry "esdi/telemetry"
)
// 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 Telemetry struct {
type TelemetryService struct {
ActiveProvider telemetry.TelemetryProvider
}
func NewTelemetryService() *TelemetryService {
return &TelemetryService{}
}
func (t *TelemetryService) setIRacingProvider() {
path := "/home/esilva/Desktop/projetos/simracing_peripherals/testTelemetry/supercars_indianapolis.ibt"
provider, _ := providerir.NewIRacingProvider(path, "", "")
t.ActiveProvider = provider
}
func (t *TelemetryService) SetProvider(provider string) *TelemetryService {
if t.ActiveProvider != nil {
// Gotta do something here to clean up before switching
}
switch provider {
case "iRacing":
// Set up iRacing
t.setIRacingProvider()
default:
return nil
}
return t
}
func (t *TelemetryService) StartStream() <-chan telemetry.TelemetryData {
stream, _ := t.ActiveProvider.Stream()
return stream
}
tui/internal/views/stream.go
+18 -7
View File
@@ -1,6 +1,12 @@
package views
import "github.com/rivo/tview"
import (
"esdi/telemetry"
"fmt"
"strings"
"github.com/rivo/tview"
)
// Car data lengths
const (
@@ -10,10 +16,6 @@ const (
BrakeBiasLen = 6
)
const (
streamingBoxID = "streaming-box"
)
type StreamView struct {
TextView *tview.TextView
}
@@ -27,6 +29,15 @@ func NewStreamView() *StreamView {
}
}
func (sv *StreamView) Update(str string) {
sv.TextView.SetText(str)
func (sv *StreamView) Update(data telemetry.TelemetryData) {
sv.TextView.SetText(stringify(data))
}
func stringify(data telemetry.TelemetryData) string {
var buffer strings.Builder
buffer.WriteString(fmt.Sprintf("Gear: %d, RPM: %d, Speed: %d\n\n",
data.Values["Gear"], data.Values["RPM"], data.Values["Speed"]))
return buffer.String()
}
tui/tui.go
+6 -2
View File
@@ -21,11 +21,15 @@ func NewControlPanel(logger *slog.Logger) *ControlPanel {
App: tview.NewApplication(),
}
deviceService := services.NewCDashService(logger)
devService := services.NewCDashService(logger)
telemService := services.NewTelemetryService().SetProvider("iRacing")
if telemService == nil {
panic("failed to create the telemetry service")
}
return &ControlPanel{
Controller: baseController,
DeviceController: controllers.NewDeviceController(baseController, deviceService),
DeviceController: controllers.NewDeviceController(baseController, devService, telemService),
}
}