fixed errors
This commit is contained in:
parent
cefb0153b2
commit
e33761bc46
82
.idea/workspace.xml
generated
82
.idea/workspace.xml
generated
@ -22,25 +22,18 @@
|
||||
</component>
|
||||
<component name="ChangeListManager">
|
||||
<list default="true" id="99c957e4-aa42-481d-843d-3fbc901e0f79" name="Default Changelist" comment="">
|
||||
<change afterPath="$PROJECT_DIR$/.idea/.gitignore" afterDir="false" />
|
||||
<change afterPath="$PROJECT_DIR$/.idea/misc.xml" afterDir="false" />
|
||||
<change afterPath="$PROJECT_DIR$/cmd/tcp_only/tcp_only.go" afterDir="false" />
|
||||
<change afterPath="$PROJECT_DIR$/core/collectors.go" afterDir="false" />
|
||||
<change afterPath="$PROJECT_DIR$/core/format.go" afterDir="false" />
|
||||
<change afterPath="$PROJECT_DIR$/core/http.go" afterDir="false" />
|
||||
<change afterPath="$PROJECT_DIR$/core/pipeline.go" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/.idea/vcs.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/vcs.xml" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/cmd/serial_only/serial_only.go" beforeDir="false" afterPath="$PROJECT_DIR$/cmd/serial_only/serial_only.go" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/cmd/server/server.go" beforeDir="false" afterPath="$PROJECT_DIR$/cmd/server/server.go" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/cmd/server_only/server_only.go" beforeDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/gnet/dispatcher.go" beforeDir="false" afterPath="$PROJECT_DIR$/core/dispatcher.go" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/gnet/net.go" beforeDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/go.mod" beforeDir="false" afterPath="$PROJECT_DIR$/go.mod" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/go.sum" beforeDir="false" afterPath="$PROJECT_DIR$/go.sum" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/serial_ubx/serial.go" beforeDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/cmd/tcp_only/tcp_only.go" beforeDir="false" afterPath="$PROJECT_DIR$/cmd/tcp_only/tcp_only.go" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/core/collectors.go" beforeDir="false" afterPath="$PROJECT_DIR$/core/collectors.go" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/core/dispatcher.go" beforeDir="false" afterPath="$PROJECT_DIR$/core/dispatcher.go" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/core/format.go" beforeDir="false" afterPath="$PROJECT_DIR$/core/format.go" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/core/http.go" beforeDir="false" afterPath="$PROJECT_DIR$/core/http.go" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/core/pipeline.go" beforeDir="false" afterPath="$PROJECT_DIR$/core/pipeline.go" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/ex_websocketMessage.json" beforeDir="false" afterPath="$PROJECT_DIR$/static/ex_websocketMessage.json" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/hyperimu.json" beforeDir="false" afterPath="$PROJECT_DIR$/static/hyperimu.json" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/index.html" beforeDir="false" afterPath="$PROJECT_DIR$/static/index.html" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/ublox/decode.go" beforeDir="false" afterPath="$PROJECT_DIR$/ublox/decode.go" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/ublox/messages.go" beforeDir="false" afterPath="$PROJECT_DIR$/ublox/messages.go" afterDir="false" />
|
||||
</list>
|
||||
<option name="SHOW_DIALOG" value="false" />
|
||||
<option name="HIGHLIGHT_CONFLICTS" value="true" />
|
||||
@ -55,7 +48,7 @@
|
||||
</list>
|
||||
</option>
|
||||
</component>
|
||||
<component name="GOROOT" path="/usr/local/Cellar/go/1.15.2/libexec" />
|
||||
<component name="GOROOT" path="/usr/local/Cellar/go/1.15.5/libexec" />
|
||||
<component name="Git.Settings">
|
||||
<option name="RECENT_BRANCH_BY_REPOSITORY">
|
||||
<map>
|
||||
@ -85,14 +78,14 @@
|
||||
</component>
|
||||
<component name="RecentsManager">
|
||||
<key name="MoveFile.RECENT_KEYS">
|
||||
<recent name="$PROJECT_DIR$/static" />
|
||||
<recent name="$PROJECT_DIR$/core" />
|
||||
<recent name="$PROJECT_DIR$/net" />
|
||||
<recent name="$PROJECT_DIR$/cmd/server" />
|
||||
<recent name="$PROJECT_DIR$/serial_ubx" />
|
||||
<recent name="$PROJECT_DIR$" />
|
||||
</key>
|
||||
</component>
|
||||
<component name="RunManager">
|
||||
<component name="RunManager" selected="Go Build.go build git.timovolkmann.de/gyrogpsc/cmd/server">
|
||||
<configuration name="go build git.timovolkmann.de/gyrogpsc/cmd/serial_only" type="GoApplicationRunConfiguration" factoryName="Go Application" temporary="true" nameIsGenerated="true">
|
||||
<module name="gyrogpsc" />
|
||||
<working_directory value="$PROJECT_DIR$" />
|
||||
@ -193,7 +186,14 @@
|
||||
<option name="project" value="LOCAL" />
|
||||
<updated>1606332601566</updated>
|
||||
</task>
|
||||
<option name="localTasksCounter" value="7" />
|
||||
<task id="LOCAL-00007" summary="refactor and implemented data pipeline">
|
||||
<created>1606934501184</created>
|
||||
<option name="number" value="00007" />
|
||||
<option name="presentableId" value="LOCAL-00007" />
|
||||
<option name="project" value="LOCAL" />
|
||||
<updated>1606934501184</updated>
|
||||
</task>
|
||||
<option name="localTasksCounter" value="8" />
|
||||
<servers />
|
||||
</component>
|
||||
<component name="TypeScriptGeneratedFilesManager">
|
||||
@ -238,12 +238,17 @@
|
||||
<MESSAGE value="switched to relative sensor data (gyroscope)" />
|
||||
<MESSAGE value="changed orientation" />
|
||||
<MESSAGE value="refactored cmd & server" />
|
||||
<option name="LAST_COMMIT_MESSAGE" value="refactored cmd & server" />
|
||||
<MESSAGE value="refactor and implemented data pipeline" />
|
||||
<option name="LAST_COMMIT_MESSAGE" value="refactor and implemented data pipeline" />
|
||||
</component>
|
||||
<component name="VgoProject">
|
||||
<integration-enabled>true</integration-enabled>
|
||||
</component>
|
||||
<component name="WindowStateProjectService">
|
||||
<state x="756" y="546" key="#GOROOT" timestamp="1606935753174">
|
||||
<screen x="0" y="25" width="1920" height="1175" />
|
||||
</state>
|
||||
<state x="756" y="546" key="#GOROOT/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606935753174" />
|
||||
<state x="640" y="377" key="#Go_Modules" timestamp="1604584628285">
|
||||
<screen x="0" y="23" width="1680" height="1027" />
|
||||
</state>
|
||||
@ -256,6 +261,10 @@
|
||||
<screen x="0" y="23" width="1680" height="1027" />
|
||||
</state>
|
||||
<state x="603" y="247" key="#xdebugger.evaluate/0.23.1680.1027/0.-1417.2560.1417@0.23.1680.1027" timestamp="1604750409547" />
|
||||
<state width="595" height="466" key="DebuggerActiveHint" timestamp="1606953226425">
|
||||
<screen x="0" y="25" width="1920" height="1175" />
|
||||
</state>
|
||||
<state width="595" height="466" key="DebuggerActiveHint/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606953226425" />
|
||||
<state x="100" y="123" width="1480" height="827" key="DiffContextDialog" timestamp="1604517986662">
|
||||
<screen x="0" y="23" width="1680" height="1027" />
|
||||
</state>
|
||||
@ -280,7 +289,7 @@
|
||||
<screen x="0" y="23" width="1680" height="1027" />
|
||||
</state>
|
||||
<state width="1638" height="346" key="GridCell.Tab.-1.right/0.23.1680.1027/0.-1417.2560.1417@0.23.1680.1027" timestamp="1604750445987" />
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.bottom" timestamp="1606933146182">
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.bottom" timestamp="1606957947578">
|
||||
<screen x="0" y="25" width="1920" height="1175" />
|
||||
</state>
|
||||
<state width="1638" height="273" key="GridCell.Tab.0.bottom/0.-1417.2560.1417/0.23.1680.1027@0.-1417.2560.1417" timestamp="1604948042531" />
|
||||
@ -290,8 +299,8 @@
|
||||
<state width="1638" height="273" key="GridCell.Tab.0.bottom/0.23.1920.1116@0.23.1920.1116" timestamp="1605380018474" />
|
||||
<state width="1878" height="406" key="GridCell.Tab.0.bottom/0.23.1920.1177/0.-1417.2560.1417@0.23.1920.1177" timestamp="1606335310969" />
|
||||
<state width="1638" height="273" key="GridCell.Tab.0.bottom/0.23.1920.1177@0.23.1920.1177" timestamp="1605533225921" />
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.bottom/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606933146182" />
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.center" timestamp="1606933146180">
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.bottom/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606957947578" />
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.center" timestamp="1606957947577">
|
||||
<screen x="0" y="25" width="1920" height="1175" />
|
||||
</state>
|
||||
<state width="1638" height="273" key="GridCell.Tab.0.center/0.-1417.2560.1417/0.23.1680.1027@0.-1417.2560.1417" timestamp="1604948042530" />
|
||||
@ -301,8 +310,8 @@
|
||||
<state width="1638" height="273" key="GridCell.Tab.0.center/0.23.1920.1116@0.23.1920.1116" timestamp="1605380018473" />
|
||||
<state width="1878" height="406" key="GridCell.Tab.0.center/0.23.1920.1177/0.-1417.2560.1417@0.23.1920.1177" timestamp="1606335310968" />
|
||||
<state width="1638" height="273" key="GridCell.Tab.0.center/0.23.1920.1177@0.23.1920.1177" timestamp="1605533225920" />
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.center/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606933146180" />
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.left" timestamp="1606933146180">
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.center/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606957947577" />
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.left" timestamp="1606957947577">
|
||||
<screen x="0" y="25" width="1920" height="1175" />
|
||||
</state>
|
||||
<state width="1638" height="273" key="GridCell.Tab.0.left/0.-1417.2560.1417/0.23.1680.1027@0.-1417.2560.1417" timestamp="1604948042530" />
|
||||
@ -312,8 +321,8 @@
|
||||
<state width="1638" height="273" key="GridCell.Tab.0.left/0.23.1920.1116@0.23.1920.1116" timestamp="1605380018472" />
|
||||
<state width="1878" height="406" key="GridCell.Tab.0.left/0.23.1920.1177/0.-1417.2560.1417@0.23.1920.1177" timestamp="1606335310967" />
|
||||
<state width="1638" height="273" key="GridCell.Tab.0.left/0.23.1920.1177@0.23.1920.1177" timestamp="1605533225919" />
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.left/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606933146180" />
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.right" timestamp="1606933146181">
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.left/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606957947577" />
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.right" timestamp="1606957947578">
|
||||
<screen x="0" y="25" width="1920" height="1175" />
|
||||
</state>
|
||||
<state width="1638" height="273" key="GridCell.Tab.0.right/0.-1417.2560.1417/0.23.1680.1027@0.-1417.2560.1417" timestamp="1604948042531" />
|
||||
@ -323,7 +332,7 @@
|
||||
<state width="1638" height="273" key="GridCell.Tab.0.right/0.23.1920.1116@0.23.1920.1116" timestamp="1605380018473" />
|
||||
<state width="1878" height="406" key="GridCell.Tab.0.right/0.23.1920.1177/0.-1417.2560.1417@0.23.1920.1177" timestamp="1606335310968" />
|
||||
<state width="1638" height="273" key="GridCell.Tab.0.right/0.23.1920.1177@0.23.1920.1177" timestamp="1605533225920" />
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.right/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606933146181" />
|
||||
<state width="1878" height="402" key="GridCell.Tab.0.right/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606957947578" />
|
||||
<state width="1638" height="346" key="GridCell.Tab.1.bottom" timestamp="1604750442086">
|
||||
<screen x="0" y="23" width="1680" height="1027" />
|
||||
</state>
|
||||
@ -354,22 +363,29 @@
|
||||
<screen x="0" y="23" width="1680" height="1027" />
|
||||
</state>
|
||||
<state width="840" height="513" key="XDebugger.FullValuePopup/0.23.1680.1027/0.-1417.2560.1417@0.23.1680.1027" timestamp="1604750505120" />
|
||||
<state x="661" y="245" width="597" height="735" key="find.popup" timestamp="1606932454676">
|
||||
<state x="661" y="245" width="597" height="735" key="find.popup" timestamp="1606955246024">
|
||||
<screen x="0" y="25" width="1920" height="1175" />
|
||||
</state>
|
||||
<state x="661" y="245" width="597" height="735" key="find.popup/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606932454676" />
|
||||
<state x="661" y="245" width="597" height="735" key="find.popup/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606955246024" />
|
||||
<state x="694" y="331" key="git4idea.branch.GitSmartOperationDialog" timestamp="1606335339469">
|
||||
<screen x="0" y="23" width="1920" height="1177" />
|
||||
</state>
|
||||
<state x="694" y="331" key="git4idea.branch.GitSmartOperationDialog/0.23.1920.1177/0.-1417.2560.1417@0.23.1920.1177" timestamp="1606335339469" />
|
||||
<state x="577" y="285" width="766" height="774" key="search.everywhere.popup" timestamp="1606932588292">
|
||||
<state x="577" y="285" width="766" height="774" key="search.everywhere.popup" timestamp="1606957420261">
|
||||
<screen x="0" y="25" width="1920" height="1175" />
|
||||
</state>
|
||||
<state x="505" y="250" width="670" height="676" key="search.everywhere.popup/0.23.1680.1027/0.-1417.2560.1417@0.23.1680.1027" timestamp="1604757918385" />
|
||||
<state x="577" y="285" width="766" height="774" key="search.everywhere.popup/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606932588292" />
|
||||
<state x="577" y="285" width="766" height="774" key="search.everywhere.popup/0.25.1920.1175/0.-1415.2560.1415@0.25.1920.1175" timestamp="1606957420261" />
|
||||
<state x="596" y="393" key="vcs.readOnlyHandler.ReadOnlyStatusDialog" timestamp="1605566567276">
|
||||
<screen x="0" y="23" width="1920" height="1177" />
|
||||
</state>
|
||||
<state x="596" y="393" key="vcs.readOnlyHandler.ReadOnlyStatusDialog/0.23.1920.1177/0.-1417.2560.1417@0.23.1920.1177" timestamp="1605566567276" />
|
||||
</component>
|
||||
<component name="XDebuggerManager">
|
||||
<watches-manager>
|
||||
<configuration name="GoApplicationRunConfiguration">
|
||||
<watch expression="*(*"<-chan time.Time")(824634171712)" />
|
||||
</configuration>
|
||||
</watches-manager>
|
||||
</component>
|
||||
</project>
|
||||
@ -2,6 +2,7 @@ package main
|
||||
|
||||
import (
|
||||
"git.timovolkmann.de/gyrogpsc/core"
|
||||
"log"
|
||||
)
|
||||
|
||||
const (
|
||||
@ -11,10 +12,15 @@ const (
|
||||
)
|
||||
|
||||
func main() {
|
||||
log.Println("setup dispatcher")
|
||||
dispatcher := core.NewDispatcher()
|
||||
processor := core.NewPipeline(dispatcher, 20, 10000)
|
||||
log.Println("initialize processing pipeline")
|
||||
processor := core.NewPipeline(dispatcher, 50, 1000)
|
||||
processor.Run()
|
||||
log.Println("start data collectors")
|
||||
collectRoutines(processor)
|
||||
core.NewHttpServer(dispatcher, HTTP_PORT)
|
||||
log.Println("start http server")
|
||||
core.HttpListenAndServe(dispatcher, HTTP_PORT)
|
||||
}
|
||||
|
||||
func collectRoutines(proc core.Processor) {
|
||||
|
||||
@ -13,7 +13,7 @@ func main() {
|
||||
dispatcher := core.NewDispatcher()
|
||||
processor := core.NewPipeline(dispatcher, 20, 10000)
|
||||
collectRoutines(processor)
|
||||
core.NewHttpServer(dispatcher, HTTP_PORT)
|
||||
core.HttpListenAndServe(dispatcher, HTTP_PORT)
|
||||
}
|
||||
|
||||
func collectRoutines(proc core.Processor) {
|
||||
|
||||
@ -66,7 +66,7 @@ func SerialCollector(proc Processor, serialPort string) {
|
||||
}
|
||||
port, err := serial.Open(serialPort, mode)
|
||||
if err != nil {
|
||||
log.Fatalln(err)
|
||||
log.Fatalln(err.Error())
|
||||
}
|
||||
|
||||
decoder := ublox.NewDecoder(port)
|
||||
@ -74,20 +74,25 @@ func SerialCollector(proc Processor, serialPort string) {
|
||||
for {
|
||||
meas, err := decoder.Decode()
|
||||
if err != nil {
|
||||
if err.Error() == "NMEA not implemented" {
|
||||
continue
|
||||
}
|
||||
log.Println(err)
|
||||
continue
|
||||
}
|
||||
sd, err := ConvertUbxToSensorData(meas)
|
||||
if err != nil {
|
||||
log.Println(err)
|
||||
log.Println("convert err:", err, meas)
|
||||
continue
|
||||
}
|
||||
if sd == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
err = proc.Process(sd)
|
||||
if err != nil {
|
||||
log.Println(err)
|
||||
log.Println("process err:", err, *sd)
|
||||
continue
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@ -2,7 +2,7 @@ package core
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
)
|
||||
|
||||
type Subscriber interface {
|
||||
@ -20,7 +20,6 @@ type dispatcher struct {
|
||||
}
|
||||
|
||||
func NewDispatcher() *dispatcher {
|
||||
fmt.Println("new dispatcher")
|
||||
return &dispatcher{
|
||||
listeners: make(map[int16]chan string),
|
||||
counter: 0,
|
||||
@ -28,14 +27,13 @@ func NewDispatcher() *dispatcher {
|
||||
}
|
||||
|
||||
func (d *dispatcher) Publish(message string) {
|
||||
//fmt.Println("publish to listeners", len(d.listeners))
|
||||
log.Printf("publish to %v listeners:\n%v\n", len(d.listeners), message)
|
||||
for _, ch := range d.listeners {
|
||||
ch <- message
|
||||
}
|
||||
}
|
||||
|
||||
func (d *dispatcher) Subscribe() (id int16, receiver <-chan string) {
|
||||
fmt.Println("subscribe")
|
||||
key := d.counter
|
||||
d.counter++
|
||||
rec := make(chan string)
|
||||
@ -44,7 +42,6 @@ func (d *dispatcher) Subscribe() (id int16, receiver <-chan string) {
|
||||
}
|
||||
|
||||
func (d *dispatcher) Unsubscribe(id int16) error {
|
||||
fmt.Println("unsubscribe")
|
||||
receiver, ok := d.listeners[id]
|
||||
if !ok {
|
||||
return errors.New("no subscription with id")
|
||||
|
||||
@ -37,47 +37,74 @@ type Sensordata struct {
|
||||
Orientation [3]float64
|
||||
}
|
||||
|
||||
func (s *Sensordata) isSameEpoch(n *Sensordata) bool {
|
||||
func (s Sensordata) isSameEpoch(n Sensordata) bool {
|
||||
if n.itow == 0 {
|
||||
return false
|
||||
}
|
||||
return s.itow == n.itow
|
||||
}
|
||||
|
||||
func (s *Sensordata) Consolidate(n *Sensordata) *Sensordata {
|
||||
if s.SourceId != n.SourceId {
|
||||
func (s Sensordata) Consolidate(n Sensordata) Sensordata {
|
||||
if (s.SourceId != n.SourceId && s != Sensordata{}) {
|
||||
log.Println(s)
|
||||
log.Println(n)
|
||||
log.Fatalln("Do not consolidate Sensordata from different Sources")
|
||||
}
|
||||
if s.isSameEpoch(n) {
|
||||
// TODO: implement epoch consolidation logic
|
||||
null := Sensordata{}
|
||||
//if s.Timestamp == null.Timestamp { s.Timestamp = n.Timestamp }
|
||||
//if s.Position == null.Position { s.Position = n.Position }
|
||||
//if s.Orientation == null.Orientation { s.Orientation = n.Orientation }
|
||||
if n.Timestamp != null.Timestamp && s.Timestamp != n.Timestamp {
|
||||
s.Timestamp = n.Timestamp
|
||||
}
|
||||
if n.Position != null.Position && s.Position != n.Position {
|
||||
s.Position = n.Position
|
||||
}
|
||||
if n.Orientation != null.Orientation && s.Orientation != n.Orientation {
|
||||
s.Orientation = n.Orientation
|
||||
}
|
||||
|
||||
return s
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
var (
|
||||
errNotImplemented = errors.New("message not implemented")
|
||||
errRawMessage = errors.New("raw message")
|
||||
)
|
||||
|
||||
func ConvertUbxToSensorData(msg interface{}) (*Sensordata, error) {
|
||||
sd := &Sensordata{}
|
||||
sd := &Sensordata{
|
||||
SourceId: SOURCE_SERIAL,
|
||||
}
|
||||
switch v := msg.(type) {
|
||||
case *ublox.NavPvt:
|
||||
//log.Println("NAV-PVT")
|
||||
sd.itow = v.ITOW_ms
|
||||
sd.Timestamp = time.Date(int(v.Year_y), time.Month(v.Month_month), int(v.Day_d), int(v.Hour_h), int(v.Min_min), int(v.Sec_s), int(v.Nano_ns), time.UTC).UnixNano()
|
||||
sd.Position[0] = float64(v.Lat_dege7)
|
||||
sd.Position[1] = float64(v.Lon_dege7)
|
||||
sd.Position[2] = float64(v.Height_mm)
|
||||
case *ublox.HnrPvt:
|
||||
//log.Println("HNR-PVT")
|
||||
sd.itow = v.ITOW_ms
|
||||
sd.Timestamp = time.Date(int(v.Year_y), time.Month(v.Month_month), int(v.Day_d), int(v.Hour_h), int(v.Min_min), int(v.Sec_s), int(v.Nano_ns), time.UTC).UnixNano()
|
||||
sd.Position[0] = float64(v.Lat_dege7)
|
||||
sd.Position[1] = float64(v.Lon_dege7)
|
||||
sd.Position[2] = float64(v.Height_mm)
|
||||
case *ublox.NavAtt:
|
||||
//log.Println("NAV-ATT")
|
||||
sd.itow = v.ITOW_ms
|
||||
sd.Orientation[0] = float64(v.Pitch_deg)
|
||||
sd.Orientation[1] = float64(v.Roll_deg)
|
||||
sd.Orientation[2] = float64(v.Heading_deg)
|
||||
case *ublox.RawMessage:
|
||||
//class := make([]byte, 2)
|
||||
//binary.LittleEndian.PutUint16(class, v.ClassID())
|
||||
//log.Printf("%#v, %#v", class[0],class[1])
|
||||
return nil, nil
|
||||
default:
|
||||
return nil, errNotImplemented
|
||||
}
|
||||
|
||||
@ -48,7 +48,7 @@ func echo(sub Subscriber) func(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
func home(w http.ResponseWriter, r *http.Request) {
|
||||
//var homeTemplate = template.Must(template.NewDispatcher("").ParseFiles("index.html"))
|
||||
tpl, err := template.ParseFiles("index.html")
|
||||
tpl, err := template.ParseFiles("static/index.html")
|
||||
if err != nil {
|
||||
log.Fatalln(err)
|
||||
}
|
||||
@ -58,10 +58,14 @@ func home(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
func NewHttpServer(sub Subscriber, httpPort string) {
|
||||
func HttpListenAndServe(sub Subscriber, httpPort string) {
|
||||
log.Println("register websocket handler")
|
||||
http.HandleFunc("/echo", echo(sub))
|
||||
log.Println("register index handler")
|
||||
http.HandleFunc("/", home)
|
||||
log.Println("register static file handler")
|
||||
http.Handle("/static/", http.FileServer(http.Dir(".")))
|
||||
|
||||
log.Println("start server")
|
||||
log.Fatal(http.ListenAndServe(httpPort, nil))
|
||||
}
|
||||
|
||||
@ -8,6 +8,8 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// TODO: adapt HNR-INS data to continue orientation stream
|
||||
|
||||
type Processor interface {
|
||||
Process(data *Sensordata) error
|
||||
}
|
||||
@ -23,11 +25,15 @@ func NewPipeline(d Publisher, publishIntervalMs int, delayUpdateIntervalMs int)
|
||||
return &pipeline{
|
||||
synchronizer{
|
||||
bufferSize: 100,
|
||||
updateTicker: time.NewTicker(time.Duration(delayUpdateIntervalMs) * time.Second),
|
||||
mutex: &sync.Mutex{},
|
||||
updateTicker: time.NewTicker(time.Duration(delayUpdateIntervalMs) * time.Millisecond),
|
||||
},
|
||||
aggregator{
|
||||
tcpMutex: &sync.Mutex{},
|
||||
serialMutex: &sync.Mutex{},
|
||||
},
|
||||
aggregator{},
|
||||
d,
|
||||
time.NewTicker(time.Duration(publishIntervalMs) * time.Second),
|
||||
time.NewTicker(time.Duration(publishIntervalMs) * time.Millisecond),
|
||||
}
|
||||
}
|
||||
|
||||
@ -37,22 +43,35 @@ func (p *pipeline) Run() {
|
||||
for {
|
||||
<-p.publishTicker.C
|
||||
err := p.Publish()
|
||||
if err != nil {
|
||||
if err != nil /*&& err.Error() != "no data available"*/ {
|
||||
log.Println(err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
log.Println("pipeline: processing service started")
|
||||
}
|
||||
|
||||
func (p *pipeline) Publish() error {
|
||||
p.tcpMutex.Lock()
|
||||
p.serialMutex.Lock()
|
||||
//log.Println(p.tcpSensorData)
|
||||
//log.Println(p.serialSensorData)
|
||||
if (p.tcpSensorData == Sensordata{} && p.serialSensorData == Sensordata{}) {
|
||||
return errors.New("No Data available")
|
||||
p.tcpMutex.Unlock()
|
||||
p.serialMutex.Unlock()
|
||||
return errors.New("no data available")
|
||||
}
|
||||
data := map[string]Sensordata{
|
||||
string(p.tcpSensorData.SourceId): p.tcpSensorData,
|
||||
string(p.serialSensorData.SourceId): p.serialSensorData,
|
||||
string(SOURCE_TCP): p.tcpSensorData,
|
||||
string(SOURCE_SERIAL): p.serialSensorData,
|
||||
}
|
||||
p.tcpSensorData = Sensordata{}
|
||||
p.serialSensorData = Sensordata{}
|
||||
p.tcpMutex.Unlock()
|
||||
p.serialMutex.Unlock()
|
||||
|
||||
jdata, err := json.Marshal(data)
|
||||
log.Println(string(jdata))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@ -86,6 +105,7 @@ type synchronizer struct {
|
||||
}
|
||||
|
||||
func (s *synchronizer) schedule() {
|
||||
log.Println("synchronizer: started")
|
||||
for {
|
||||
<-s.updateTicker.C
|
||||
err := s.refreshDelay()
|
||||
@ -101,16 +121,19 @@ func (s *synchronizer) refreshDelay() error {
|
||||
}
|
||||
|
||||
func (p *pipeline) Process(data *Sensordata) error {
|
||||
if data == nil {
|
||||
return errors.New("nil processing not allowed")
|
||||
}
|
||||
//log.Println(string(data.SourceId))
|
||||
switch data.SourceId {
|
||||
case SOURCE_TCP:
|
||||
go p.pushTcpDataToBuffer(*data)
|
||||
return nil
|
||||
case SOURCE_SERIAL:
|
||||
go p.pushSerialDataToBuffer(*data)
|
||||
return nil
|
||||
default:
|
||||
return errors.New("invalid data source")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *pipeline) pushTcpDataToBuffer(data Sensordata) {
|
||||
@ -122,6 +145,6 @@ func (p *pipeline) pushTcpDataToBuffer(data Sensordata) {
|
||||
func (p *pipeline) pushSerialDataToBuffer(data Sensordata) {
|
||||
time.Sleep(time.Duration(p.serialDelayMs))
|
||||
p.serialMutex.Lock()
|
||||
p.serialSensorData = data
|
||||
p.serialSensorData = p.serialSensorData.Consolidate(data)
|
||||
p.serialMutex.Unlock()
|
||||
}
|
||||
|
||||
@ -85,8 +85,7 @@ func (d *decoder) Decode() (msg interface{}, err error) {
|
||||
|
||||
switch d.s.Bytes()[0] {
|
||||
case '$':
|
||||
//fmt.Println("NMEA message: skipping!")
|
||||
return nil, err
|
||||
return nil, errors.New("NMEA not implemented")
|
||||
//return nmea.Decode(d.s.Bytes())
|
||||
case 0xB5:
|
||||
return decodeUbx(d.s.Bytes())
|
||||
|
||||
Loading…
Reference in New Issue
Block a user