starting the foundation for multiple types of telemetry data
This commit is contained in:
@@ -0,0 +1,45 @@
|
||||
package mockserver
|
||||
|
||||
import (
|
||||
"io"
|
||||
"unsafe"
|
||||
|
||||
sdk "github.com/ESilva15/gobngsdk"
|
||||
)
|
||||
|
||||
type GobReader struct {
|
||||
TotalRead int64
|
||||
File io.ReadSeeker
|
||||
Buf []byte
|
||||
}
|
||||
|
||||
func NewGobReader(r io.ReadSeeker) *GobReader {
|
||||
return &GobReader{
|
||||
TotalRead: 0,
|
||||
File: r,
|
||||
Buf: make([]byte, unsafe.Sizeof(sdk.Outgauge{})),
|
||||
}
|
||||
}
|
||||
|
||||
func (g *GobReader) Reset() error {
|
||||
_, err := g.File.Seek(0, io.SeekStart)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
g.TotalRead = 0
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (g *GobReader) Next(buffer []byte) error {
|
||||
_, err := io.ReadFull(g.File, buffer)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
pos, _ := g.File.Seek(0, io.SeekCurrent)
|
||||
g.TotalRead = pos
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
package mockserver
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
bngsdk "github.com/ESilva15/gobngsdk"
|
||||
)
|
||||
|
||||
type recorderViewData struct {
|
||||
TotalBytes int
|
||||
SDK *bngsdk.BeamNGSDK
|
||||
}
|
||||
|
||||
type Recorder struct {
|
||||
SDK bngsdk.BeamNGSDK
|
||||
OutputFile *os.File
|
||||
TotalBytes int
|
||||
// Views
|
||||
mut sync.RWMutex
|
||||
viewDataMut sync.RWMutex
|
||||
viewData recorderViewData
|
||||
viewCh chan *recorderViewData
|
||||
recorderCh chan []byte
|
||||
}
|
||||
|
||||
func NewRecorder(fp string, address string, port int) (*Recorder, error) {
|
||||
var recorder Recorder
|
||||
var err error
|
||||
|
||||
recorder.SDK, err = bngsdk.Init(address, port)
|
||||
if err != nil {
|
||||
return &Recorder{}, err
|
||||
}
|
||||
|
||||
recorder.OutputFile, err = os.Create(fp)
|
||||
if err != nil {
|
||||
return &Recorder{}, err
|
||||
}
|
||||
|
||||
recorder.viewData = recorderViewData{}
|
||||
recorder.viewCh = make(chan *recorderViewData, 1)
|
||||
recorder.recorderCh = make(chan []byte, 1)
|
||||
|
||||
return &recorder, nil
|
||||
}
|
||||
|
||||
func (r *Recorder) Close() {
|
||||
r.SDK.Close()
|
||||
|
||||
if r.OutputFile != nil {
|
||||
r.OutputFile.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Recorder) record(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case data := <-r.recorderCh:
|
||||
r.mut.Lock()
|
||||
err := binary.Write(r.OutputFile, binary.LittleEndian, r.SDK.Data)
|
||||
r.TotalBytes += len(data)
|
||||
r.mut.Unlock()
|
||||
|
||||
if err != nil {
|
||||
// NOTE: find a way of logging this somehow
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Recorder) view(ctx context.Context) {
|
||||
var buf bytes.Buffer
|
||||
var nBytes int
|
||||
|
||||
buf.Grow(2048)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case viewData := <-r.viewCh:
|
||||
buf.Reset()
|
||||
buf.WriteString("\x1b[2J\x1b[H")
|
||||
fmt.Fprintf(&buf, "\x1b]0;%s - Recording ", ProgramName)
|
||||
stringifyRecordingProgress(&buf, nBytes)
|
||||
fmt.Fprintf(&buf, "\x07")
|
||||
|
||||
stringifyRecordingProgress(&buf, nBytes)
|
||||
fmt.Fprintf(&buf, "\n\n")
|
||||
|
||||
r.viewDataMut.RLock()
|
||||
nBytes = viewData.TotalBytes
|
||||
stringifyOutgaugeData(&buf, &r.SDK)
|
||||
r.viewDataMut.RUnlock()
|
||||
|
||||
_, _ = buf.WriteTo(os.Stdout)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Record records data from the UDP connection created by address and port
|
||||
func (r *Recorder) Record(ctx context.Context) error {
|
||||
ticker := time.NewTicker(time.Second / 60)
|
||||
defer ticker.Stop()
|
||||
|
||||
go r.record(ctx)
|
||||
go r.view(ctx)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case <-ticker.C:
|
||||
err := r.SDK.ReadData()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
r.viewData.TotalBytes = r.TotalBytes
|
||||
|
||||
// Send the data to the view
|
||||
select {
|
||||
case r.viewCh <- &r.viewData:
|
||||
// Sent the data
|
||||
default:
|
||||
// Dropped the frame!
|
||||
}
|
||||
|
||||
// Write the data to the file
|
||||
select {
|
||||
case r.recorderCh <- r.SDK.Buffer:
|
||||
// Sent the data
|
||||
default:
|
||||
// Dropped the frame!
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,188 @@
|
||||
// Package mockserver is the core of this program and it will have the API
|
||||
// to record binary data from the UDP server and then be able to mock it
|
||||
package mockserver
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
"unsafe"
|
||||
|
||||
bngsdk "github.com/ESilva15/gobngsdk"
|
||||
)
|
||||
|
||||
type ViewData struct {
|
||||
SDK *bngsdk.BeamNGSDK
|
||||
SizeRead int64
|
||||
}
|
||||
|
||||
// Replayer does the replaying
|
||||
// Should we make a "player" struct that can record and replay?
|
||||
type Replayer struct {
|
||||
SDK bngsdk.BeamNGSDK
|
||||
DataSourcePath string
|
||||
Socket *UDPTransport
|
||||
|
||||
// Streams
|
||||
dataViewCh chan ViewData
|
||||
socketCh chan []byte
|
||||
|
||||
// Mut
|
||||
mut sync.RWMutex
|
||||
|
||||
// View
|
||||
viewData ViewData
|
||||
}
|
||||
|
||||
func NewReplayer(address string, port int, fp string) (*Replayer, error) {
|
||||
udp, err := NewUDPTransport(address, port)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
replayer := &Replayer{
|
||||
DataSourcePath: fp,
|
||||
SDK: bngsdk.BeamNGSDK{
|
||||
Data: bngsdk.Outgauge{},
|
||||
Buffer: make([]byte, unsafe.Sizeof(bngsdk.Outgauge{})),
|
||||
},
|
||||
Socket: udp,
|
||||
viewData: ViewData{},
|
||||
dataViewCh: make(chan ViewData, 1),
|
||||
socketCh: make(chan []byte, 1),
|
||||
}
|
||||
|
||||
return replayer, nil
|
||||
}
|
||||
|
||||
// renderToTerminal will render the data for the users viewing pleasure
|
||||
func (r *Replayer) renderToTerminal(ctx context.Context) {
|
||||
fileInfo, err := os.Stat(r.DataSourcePath)
|
||||
if err != nil {
|
||||
// NOTE: learn how to handle this error
|
||||
// return fmt.Errorf("error stating file: %v", err)
|
||||
}
|
||||
|
||||
var buf bytes.Buffer
|
||||
var bytesReader bytes.Reader
|
||||
|
||||
buf.Grow(2048)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case data := <-r.dataViewCh:
|
||||
// Reset to the start of the terminal
|
||||
percent := int(float64(data.SizeRead) / float64(fileInfo.Size()) * 100)
|
||||
|
||||
buf.Reset()
|
||||
buf.WriteString("\x1b[2J\x1b[H")
|
||||
fmt.Fprintf(&buf, "\x1b]0;%s - Replaying %d%%\x07", ProgramName, percent)
|
||||
|
||||
bytesReader.Reset(data.SDK.Buffer)
|
||||
err := binary.Read(&bytesReader, binary.LittleEndian, &r.SDK.Data)
|
||||
if err != nil {
|
||||
fmt.Fprintf(&buf, "FAILED TO PARSE DATA\nError: %+v", err)
|
||||
_, _ = buf.WriteTo(os.Stdout)
|
||||
continue
|
||||
}
|
||||
|
||||
fmt.Fprintf(&buf, "Replayed: %d%%\n", percent)
|
||||
|
||||
stringifyOutgaugeData(&buf, data.SDK)
|
||||
|
||||
buf.WriteTo(os.Stdout)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// writeToUDPSocket will write the telemetry data to the UDP socket
|
||||
func (r *Replayer) writeToUDPSocket(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case data := <-r.socketCh:
|
||||
r.mut.RLock()
|
||||
_, err := r.Socket.Send(data)
|
||||
r.mut.RUnlock()
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("error writing buffer to socket: %+v", err))
|
||||
continue
|
||||
// NOTE: log the error somewhere maybe
|
||||
// return err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Replay replays a given file <fp> in a UDP server <addr>:<port>
|
||||
func (r *Replayer) Replay(ctx context.Context, loop bool) error {
|
||||
bin, err := os.Open(r.DataSourcePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error opening file: %v", err)
|
||||
}
|
||||
|
||||
reader := NewGobReader(bin)
|
||||
|
||||
go r.renderToTerminal(ctx)
|
||||
go r.writeToUDPSocket(ctx)
|
||||
|
||||
ticker := time.NewTicker(time.Second / 60)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
|
||||
case <-ticker.C:
|
||||
r.mut.Lock()
|
||||
err := reader.Next(r.SDK.Buffer)
|
||||
r.mut.Unlock()
|
||||
|
||||
if err == io.EOF {
|
||||
if !loop {
|
||||
return r.Socket.Close()
|
||||
}
|
||||
|
||||
err = reader.Reset()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// NOTE: Really like this???
|
||||
r.viewData.SDK = &r.SDK
|
||||
r.viewData.SizeRead = reader.TotalRead
|
||||
|
||||
// Send the data to the view
|
||||
select {
|
||||
case r.dataViewCh <- r.viewData:
|
||||
// Sent the data
|
||||
default:
|
||||
// Dropped the frame!
|
||||
}
|
||||
|
||||
// Send the data to the UDP socket
|
||||
select {
|
||||
case r.socketCh <- r.SDK.Buffer:
|
||||
// Sent the data
|
||||
default:
|
||||
// Dropped the frame!
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
package mockserver
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"os"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func BenchmarkReplayAsync(b *testing.B) {
|
||||
// Dummy socket
|
||||
addr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
b.Fatalf("failed to resolve UDP address: %v", err)
|
||||
}
|
||||
|
||||
listener, err := net.ListenUDP("udp", addr)
|
||||
if err != nil {
|
||||
b.Fatalf("failed to start background UDP listener: %v", err)
|
||||
}
|
||||
defer listener.Close()
|
||||
|
||||
// Drain the socket
|
||||
go func() {
|
||||
buf := make([]byte, 65535)
|
||||
for {
|
||||
_, _, err := listener.ReadFrom(buf)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// Extract the random port assigned by the OS
|
||||
assignedAddr := listener.LocalAddr().(*net.UDPAddr)
|
||||
|
||||
// Get the file path
|
||||
filePath := "sunburstManual.bin"
|
||||
if _, err := os.Stat(filePath); os.IsNotExist(err) {
|
||||
filePath = "../" + filePath
|
||||
}
|
||||
|
||||
// Reset the timer
|
||||
b.ResetTimer()
|
||||
|
||||
// Benchmark loop
|
||||
for i := 0; i < b.N; i++ {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
replayer, err := NewReplayer(assignedAddr.IP.String(), assignedAddr.Port, filePath)
|
||||
if err != nil {
|
||||
cancel()
|
||||
b.Fatalf("Error setting up replayer: %+v", err)
|
||||
}
|
||||
|
||||
err = replayer.Replay(ctx, false)
|
||||
if err != nil {
|
||||
cancel()
|
||||
b.Fatalf("Replay failed during benchmark run: %+v", err)
|
||||
}
|
||||
|
||||
cancel()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package mockserver
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
)
|
||||
|
||||
const ProgramName = "TelemetryMockerserver"
|
||||
|
||||
type Transport interface {
|
||||
Send(data []byte) error
|
||||
Close() error
|
||||
}
|
||||
|
||||
type UDPTransport struct {
|
||||
Conn *net.UDPConn
|
||||
Addr *net.UDPAddr
|
||||
}
|
||||
|
||||
func NewUDPTransport(address string, port int) (*UDPTransport, error) {
|
||||
addr, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", address, port))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
conn, err := net.ListenUDP("udp", nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &UDPTransport{Conn: conn, Addr: addr}, nil
|
||||
}
|
||||
|
||||
// Send will send a byte array of data trough the UDP server
|
||||
func (u *UDPTransport) Send(data []byte) (int, error) {
|
||||
return u.Conn.WriteToUDP(data, u.Addr)
|
||||
}
|
||||
|
||||
func (u *UDPTransport) Close() error {
|
||||
return u.Conn.Close()
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package mockserver
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
|
||||
bngsdk "github.com/ESilva15/gobngsdk"
|
||||
)
|
||||
|
||||
const (
|
||||
_ = iota
|
||||
KiB = 1 << (10 * iota)
|
||||
MiB
|
||||
GiB
|
||||
)
|
||||
|
||||
func stringifyRecordingProgress(s *bytes.Buffer, nBytes int) {
|
||||
if nBytes < KiB {
|
||||
fmt.Fprintf(s, "%d B", nBytes)
|
||||
} else if nBytes < MiB {
|
||||
fmt.Fprintf(s, "%.2f KiB", float64(nBytes)/float64(KiB))
|
||||
} else if nBytes < GiB {
|
||||
fmt.Fprintf(s, "%.2f MiB", float64(nBytes)/float64(MiB))
|
||||
} else {
|
||||
fmt.Fprintf(s, "%.2f GiB", float64(nBytes)/float64(GiB))
|
||||
}
|
||||
}
|
||||
|
||||
func stringifyOutgaugeData(s *bytes.Buffer, sdk *bngsdk.BeamNGSDK) {
|
||||
// NOTE: write a string serialization function on the SDK itself
|
||||
fmt.Fprint(s, "Outgauge {\n")
|
||||
fmt.Fprintf(s, " Time: %d ms\n", sdk.Data.Time)
|
||||
fmt.Fprintf(s, " Car: %s\n", sdk.Data.Car)
|
||||
fmt.Fprintf(s, " Flags: %b\n", sdk.Data.Flags)
|
||||
fmt.Fprintf(s, " Gear: %d\n", sdk.Data.Gear)
|
||||
fmt.Fprintf(s, " Plid: %d\n", sdk.Data.Plid)
|
||||
fmt.Fprintf(s, " Speed: %f m/s\n", sdk.Data.Speed)
|
||||
fmt.Fprintf(s, " RPM: %f RPM\n", sdk.Data.RPM)
|
||||
fmt.Fprintf(s, " Turbo: %f Bar\n", sdk.Data.Turbo)
|
||||
fmt.Fprintf(s, " EngTemp: %f °C\n", sdk.Data.EngTemp)
|
||||
fmt.Fprintf(s, " Fuel: %f\n", sdk.Data.Fuel)
|
||||
fmt.Fprintf(s, " OilPressure: %f Bar\n", sdk.Data.OilPressure)
|
||||
fmt.Fprintf(s, " OilTemp: %f °C\n", sdk.Data.OilTemp)
|
||||
fmt.Fprintf(s, " DashLights: %b\n", sdk.Data.DashLights)
|
||||
fmt.Fprintf(s, " ShowLights: %b\n", sdk.Data.ShowLights)
|
||||
fmt.Fprintf(s, " Throttle: %f\n", sdk.Data.Throttle)
|
||||
fmt.Fprintf(s, " Brakes: %f\n", sdk.Data.Brake)
|
||||
fmt.Fprintf(s, " Clutch: %f\n", sdk.Data.Clutch)
|
||||
fmt.Fprintf(s, " Display1: %s\n", sdk.Data.Display1)
|
||||
fmt.Fprintf(s, " Display2: %s\n", sdk.Data.Display2)
|
||||
fmt.Fprintf(s, " ID: %d\n", sdk.Data.ID)
|
||||
fmt.Fprint(s, "}\n\n")
|
||||
|
||||
fmt.Fprint(s, "DashLights {\n")
|
||||
fmt.Fprintf(s, " DL_SHIFT: %t\n", sdk.HasShiftLight())
|
||||
fmt.Fprintf(s, " DL_FULLBEAM: %t\n", sdk.HasHighBeamLight())
|
||||
fmt.Fprintf(s, " DL_HANDBRAKE: %t\n", sdk.HasHandbrakeLight())
|
||||
fmt.Fprintf(s, " DL_PITSPEED: %t\n", sdk.HasPitspeed())
|
||||
fmt.Fprintf(s, " DL_TC: %t\n", sdk.HasTractionControlLight())
|
||||
fmt.Fprintf(s, " DL_SIGNAL_L: %t\n", sdk.HasLeftIndicatorLight())
|
||||
fmt.Fprintf(s, " DL_SIGNAL_R: %t\n", sdk.HasRightIndicatorLight())
|
||||
fmt.Fprintf(s, " DL_SIGNAL_ANY: %t\n", sdk.HasAnyIndicatorLight())
|
||||
fmt.Fprintf(s, " DL_OILWARN: %t\n", sdk.HasOilLight())
|
||||
fmt.Fprintf(s, " DL_BATTERY: %t\n", sdk.HasBatteryLight())
|
||||
fmt.Fprintf(s, " DL_ABS: %t\n", sdk.HasABSLight())
|
||||
fmt.Fprintf(s, " DL_SPARE: %t\n", sdk.Data.DashLights&bngsdk.DL_SPARE != 0)
|
||||
fmt.Fprint(s, "}\n\n")
|
||||
|
||||
fmt.Fprint(s, "ShowLights {\n") // Fixed typo "ShowLigths"
|
||||
fmt.Fprintf(s, " DL_SHIFT: %t\n", sdk.ShiftLight())
|
||||
fmt.Fprintf(s, " DL_FULLBEAM: %t\n", sdk.HighBeam())
|
||||
fmt.Fprintf(s, " DL_HANDBRAKE: %t\n", sdk.Handbrake())
|
||||
fmt.Fprintf(s, " DL_PITSPEED: %t\n", sdk.Pitspeed())
|
||||
fmt.Fprintf(s, " DL_TC: %t\n", sdk.TractionControl())
|
||||
fmt.Fprintf(s, " DL_SIGNAL_L: %t\n", sdk.LeftIndicator())
|
||||
fmt.Fprintf(s, " DL_SIGNAL_R: %t\n", sdk.RightIndicator())
|
||||
fmt.Fprintf(s, " DL_SIGNAL_ANY: %t\n", sdk.AnyIndicator())
|
||||
fmt.Fprintf(s, " DL_OILWARN: %t\n", sdk.OilLight())
|
||||
fmt.Fprintf(s, " DL_BATTERY: %t\n", sdk.BatteryLight())
|
||||
fmt.Fprintf(s, " DL_ABS: %t\n", sdk.ABS())
|
||||
fmt.Fprintf(s, " DL_SPARE: %t\n", sdk.Data.ShowLights&bngsdk.DL_SPARE != 0)
|
||||
fmt.Fprint(s, "}\n\n")
|
||||
|
||||
fmt.Fprint(s, "Flags {\n")
|
||||
fmt.Fprintf(s, " OG_TURBO (Has Turbo): %t\n", sdk.HasTurbo())
|
||||
fmt.Fprintf(s, " OG_KM (Is Metric): %t\n", sdk.PrefersKm())
|
||||
fmt.Fprintf(s, " OG_BAR (Pressure): %t\n", sdk.PrefersBAR())
|
||||
fmt.Fprint(s, "}")
|
||||
}
|
||||
Reference in New Issue
Block a user