Compare commits

..

3 commits

Author SHA1 Message Date
saji d5381a3c33 add comments
All checks were successful
Go / build (1.21) (push) Successful in 1m6s
Go / build (1.22) (push) Successful in 1m4s
2024-03-07 13:30:32 -06:00
saji 0b5a917e40 add JSON line parser, rename logparsers 2024-03-07 13:30:15 -06:00
saji 1ff4adf5e4 add generator test for skylab 2024-03-07 13:29:50 -06:00
8 changed files with 2351 additions and 560 deletions

View file

@ -9,6 +9,8 @@ import (
"github.com/kschamplin/gotelem/skylab"
)
// Broker is a Bus Event broadcast system. You can subscribe to events,
// and send events.
type Broker struct {
subs map[string]chan skylab.BusEvent // contains the channel for each subsciber
@ -17,6 +19,7 @@ type Broker struct {
bufsize int // size of chan buffer in elements.
}
// NewBroker creates a new broker with a given logger.
func NewBroker(bufsize int, logger *slog.Logger) *Broker {
return &Broker{
subs: make(map[string]chan skylab.BusEvent),
@ -25,6 +28,7 @@ func NewBroker(bufsize int, logger *slog.Logger) *Broker {
}
}
// Subscribe joins the broker with the given name. The name must be unique.
func (b *Broker) Subscribe(name string) (ch chan skylab.BusEvent, err error) {
// get rw lock.
b.lock.Lock()
@ -40,6 +44,9 @@ func (b *Broker) Subscribe(name string) (ch chan skylab.BusEvent, err error) {
return
}
// Unsubscribe removes a subscriber matching the name. It doesn't do anything
// if there's nobody subscribed with that name
func (b *Broker) Unsubscribe(name string) {
// remove the channel from the map. We don't need to close it.
b.lock.Lock()
@ -47,6 +54,8 @@ func (b *Broker) Unsubscribe(name string) {
delete(b.subs, name)
}
// Publish sends a bus event to all subscribers. It includes a sender
// string which prevents loopback.
func (b *Broker) Publish(sender string, message skylab.BusEvent) {
b.lock.RLock()
defer b.lock.RUnlock()

View file

@ -92,7 +92,7 @@ func run(ctx *cli.Context) (err error) {
fileReader := bufio.NewReader(istream)
var pfun logparsers.BusParserFunc
var pfun logparsers.BusEventParser
pfun, ok := logparsers.ParsersMap[ctx.String("format")]
if !ok {

View file

@ -2,6 +2,7 @@ package logparsers
import (
"encoding/hex"
"encoding/json"
"fmt"
"regexp"
"strconv"
@ -35,10 +36,10 @@ func NewFormatError(msg string, err error) error {
return &FormatError{msg: msg, err: err}
}
// type LineParserFunc is a function that takes a string
// type CanFrameParser is a function that takes a string
// and returns a can frame. This is useful for common
// can dump formats.
type LineParserFunc func(string) (can.Frame, time.Time, error)
type CanFrameParser func(string) (can.Frame, time.Time, error)
var candumpRegex = regexp.MustCompile(`^\((\d+)\.(\d{6})\) \w+ (\w+)#(\w+)$`)
@ -157,20 +158,27 @@ func parseTelemLogLine(line string) (frame can.Frame, ts time.Time, err error) {
}
// BusParserFunc is a function that takes a string and returns a busevent.
type BusParserFunc func(string) (skylab.BusEvent, error)
// BusEventParser is a function that takes a string and returns a busevent.
type BusEventParser func(string) (skylab.BusEvent, error)
// parserBusEventMapper takes a line parser (that returns a can frame)
// skylabify JSON parser.
func parseSkylabifyLogLine(input string) (skylab.BusEvent, error) {
var b = skylab.BusEvent{}
err := json.Unmarshal([]byte(input), &b)
return b, err
}
// frameParseToBusEvent takes a line parser (that returns a can frame)
// and makes it return a busEvent instead.
func parserBusEventMapper(f LineParserFunc) BusParserFunc {
func frameParseToBusEvent(fun CanFrameParser) BusEventParser {
return func(s string) (skylab.BusEvent, error) {
var b = skylab.BusEvent{}
f, ts, err := f(s)
frame, ts, err := fun(s)
if err != nil {
return b, err
}
b.Timestamp = ts
b.Data, err = skylab.FromCanFrame(f)
b.Data, err = skylab.FromCanFrame(frame)
if err != nil {
return b, err
}
@ -179,7 +187,8 @@ func parserBusEventMapper(f LineParserFunc) BusParserFunc {
}
}
var ParsersMap = map[string]BusParserFunc{
"telem": parserBusEventMapper(parseTelemLogLine),
"candump": parserBusEventMapper(parseCanDumpLine),
var ParsersMap = map[string]BusEventParser{
"telem": frameParseToBusEvent(parseTelemLogLine),
"candump": frameParseToBusEvent(parseCanDumpLine),
"json": parseSkylabifyLogLine,
}

View file

@ -6,6 +6,7 @@ import (
"time"
"github.com/kschamplin/gotelem/internal/can"
"github.com/kschamplin/gotelem/skylab"
)
func Test_parseCanDumpLine(t *testing.T) {
@ -170,3 +171,42 @@ func Test_parseTelemLogLine_errors(t *testing.T) {
})
}
}
func Test_parseSkylabifyLogLine(t *testing.T) {
type args struct {
input string
}
tests := []struct {
name string
args args
want skylab.BusEvent
wantErr bool
}{
{
name: "basic test",
args: args{
input: `{"ts":1685141873612,"id":259,"name":"wsl_velocity","data":{"motor_velocity":89.97547,"vehicle_velocity":2.38853}}`},
want: skylab.BusEvent{
Timestamp: time.UnixMilli(1685141873612),
Name: "wsl_velocity",
Data: &skylab.WslVelocity{
MotorVelocity: 89.97547,
VehicleVelocity: 2.38853,
},
},
wantErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got, err := parseSkylabifyLogLine(tt.args.input)
if (err != nil) != tt.wantErr {
t.Errorf("parseSkylabifyLogLine() error = %v, wantErr %v", err, tt.wantErr)
return
}
if !reflect.DeepEqual(got, tt.want) {
t.Errorf("parseSkylabifyLogLine() = %v, want %v", got, tt.want)
}
})
}
}

File diff suppressed because one or more lines are too long

File diff suppressed because it is too large Load diff

View file

@ -3,6 +3,7 @@ package skylab
import (
"testing"
"reflect"
"encoding/json"
)
@ -44,4 +45,20 @@ func TestJSON{{$structName}}(t *testing.T) {
}
func TestCanFrame{{$structName}}(t *testing.T) {
v := &{{$structName}}{}
frame, err := ToCanFrame(v)
if err != nil {
t.Fatalf("expected no error, got %v", err)
}
retpkt, err := FromCanFrame(frame)
if err != nil {
t.Fatalf("expected no error, got %v", err)
}
if !reflect.DeepEqual(v, retpkt) {
t.Fatalf("decoded packet did not match sent %v got %v", v, retpkt)
}
}
{{- end }}

View file

@ -29,7 +29,7 @@ func TestCanSocket(t *testing.T) {
}
})
t.Run("test name", func(t *testing.T) {
t.Run("test interface name", func(t *testing.T) {
sock, _ := NewCanSocket("vcan0")
defer sock.Close()