GR25 Signal WS Endpoint

gr25/api/api.go

@@ -25,4 +25,5 @@ func SetupRouter() *gin.Engine {
 
 func InitializeRoutes(router *gin.Engine) {
 	router.GET("/gr25/ping", Ping)
+	router.GET("/gr25/live", GetLatestSignalWebSocket)
 }

gr25/api/signal.go

@@ -0,0 +1,59 @@
+package api
+
+import (
+	"gr25/service"
+	"gr25/utils"
+	"net/http"
+	"slices"
+	"strconv"
+	"strings"
+
+	"github.com/gaucho-racing/mapache-go"
+	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
+)
+
+var upgrader = websocket.Upgrader{
+	ReadBufferSize:  1024,
+	WriteBufferSize: 1024,
+	CheckOrigin: func(r *http.Request) bool {
+		return true
+	},
+}
+
+func GetLatestSignalWebSocket(c *gin.Context) {
+	vehicleID := c.Query("vehicle_id")
+	signals := strings.Split(c.Query("signals"), ",")
+
+	if vehicleID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "vehicle_id is required"})
+		return
+	}
+
+	if len(signals) == 0 {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "signals are required"})
+		return
+	}
+
+	conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
+	if err != nil {
+		return
+	}
+	defer conn.Close()
+
+	service.SubscribeSignals(func(signal mapache.Signal) {
+		if signal.VehicleID == vehicleID && slices.Contains(signals, signal.Name) {
+			conn.WriteJSON(signal)
+		}
+	})
+
+	for {
+		messageType, p, err := conn.ReadMessage()
+		if err != nil {
+			utils.SugarLogger.Errorln("[WS - gr25/live] error while reading message\n", err.Error())
+			c.AbortWithError(http.StatusInternalServerError, err)
+			break
+		}
+		utils.SugarLogger.Infoln("[WS - gr25/live] Received message ("+strconv.Itoa(messageType)+"): ", string(p))
+	}
+}

gr25/config/config.go

@@ -8,7 +8,7 @@ import (
 
 var Service rincon.Service = rincon.Service{
 	Name:    "GR25",
-	Version: "1.3.0",
+	Version: "1.4.0",
 }
 
 var Routes = []rincon.Route{

gr25/go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/bk1031/rincon-go/v2 v2.0.0
 	github.com/eclipse/paho.mqtt.golang v1.5.0
 	github.com/gin-gonic/gin v1.10.0
+	github.com/gorilla/websocket v1.5.3
 	go.uber.org/zap v1.27.0
 	gorm.io/gorm v1.25.7
 )
@@ -23,7 +24,6 @@ require (
 	github.com/go-sql-driver/mysql v1.7.0 // indirect
 	github.com/goccy/go-json v0.10.4 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
-	github.com/gorilla/websocket v1.5.3 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.5 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect

gr25/service/signal.go

@@ -8,6 +8,21 @@ import (
 	"github.com/gaucho-racing/mapache-go"
 )
 
+// signalCallbacks is a list of functions that will be called when a signal is created or updated
+var signalCallbacks = []func(signal mapache.Signal){}
+
+// signalNotify fires all the callbacks in signalCallbacks with the provided signal
+func signalNotify(signal mapache.Signal) {
+	for _, callback := range signalCallbacks {
+		callback(signal)
+	}
+}
+
+// SubscribeSignals registers a function to be called when a signal is created or updated
+func SubscribeSignals(callback func(signal mapache.Signal)) {
+	signalCallbacks = append(signalCallbacks, callback)
+}
+
 func GetSignal(timestamp int, vehicleID string, name string) mapache.Signal {
 	var signal mapache.Signal
 	database.DB.Where("timestamp = ?", timestamp).Where("vehicle_id = ?", vehicleID).Where("name = ?", name).First(&signal)
@@ -40,5 +55,6 @@ func CreateSignal(signal mapache.Signal) error {
 			"name", signal.Name,
 		)
 	}
+	go signalNotify(signal)
 	return nil
 }

query/tests/test_query.py

@@ -8,31 +8,31 @@
 
 def test_query_signals_single_signal():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=['vdm_speed'])
+    result = query_signals(vehicle_id='gr24-main', signals=['vdm_speed'])
     assert len(result) == 1
     assert isinstance(result[0], pd.DataFrame)
     assert list(result[0].columns) == ['produced_at', 'vdm_speed']
     assert len(result[0]) == 234980
 
 def test_query_signals_start_time():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=['vdm_speed'], start='2024-11-09 22:53:55.00')
+    result = query_signals(vehicle_id='gr24-main', signals=['vdm_speed'], start='2024-11-09 22:53:55.00')
     assert len(result) == 1
     assert isinstance(result[0], pd.DataFrame)
     assert list(result[0].columns) == ['produced_at', 'vdm_speed']
     assert len(result[0]) == 135540
 
 def test_query_signals_end_time():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=['vdm_speed'], end='2024-11-09 22:57:41.00')
+    result = query_signals(vehicle_id='gr24-main', signals=['vdm_speed'], end='2024-11-09 22:57:41.00')
     assert len(result) == 1
     assert isinstance(result[0], pd.DataFrame)
     assert list(result[0].columns) == ['produced_at', 'vdm_speed']
     assert len(result[0]) == 101786
 
 def test_query_signals_start_end_time():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=['vdm_speed'], start='2024-11-09 22:53:55.00', end='2024-11-09 22:57:41.00')
+    result = query_signals(vehicle_id='gr24-main', signals=['vdm_speed'], start='2024-11-09 22:53:55.00', end='2024-11-09 22:57:41.00')
     assert len(result) == 1
     assert isinstance(result[0], pd.DataFrame)
     assert list(result[0].columns) == ['produced_at', 'vdm_speed']
@@ -48,7 +48,7 @@ def test_query_signals_invalid_vehicle_id():
 
 def test_query_signals_invalid_range():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=['vdm_speed'], start='2022-11-09 22:53:55.00', end='2022-11-09 22:57:41.00')
+    result = query_signals(vehicle_id='gr24-main', signals=['vdm_speed'], start='2022-11-09 22:53:55.00', end='2022-11-09 22:57:41.00')
     assert len(result) == 1
     assert isinstance(result[0], pd.DataFrame)
     assert list(result[0].columns) == ['produced_at', 'vdm_speed']
@@ -61,7 +61,7 @@ def test_query_signals_invalid_range():
 
 def test_query_signals_two_signals():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=two_signals)
+    result = query_signals(vehicle_id='gr24-main', signals=two_signals)
     assert len(result) == 2
     for i in range(len(result)):
         assert isinstance(result[i], pd.DataFrame)
@@ -72,7 +72,7 @@ def test_query_signals_two_signals():
 
 def test_query_signals_start_time_two_signals():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=two_signals, start='2024-11-09 22:53:55.00')
+    result = query_signals(vehicle_id='gr24-main', signals=two_signals, start='2024-11-09 22:53:55.00')
     assert len(result) == 2
     for i in range(len(result)):
         assert isinstance(result[i], pd.DataFrame)
@@ -82,7 +82,7 @@ def test_query_signals_start_time_two_signals():
 
 def test_query_signals_end_time_two_signals():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=two_signals, end='2024-11-09 22:57:41.00')
+    result = query_signals(vehicle_id='gr24-main', signals=two_signals, end='2024-11-09 22:57:41.00')
     assert len(result) == 2
     for i in range(len(result)):
         assert isinstance(result[i], pd.DataFrame)
@@ -92,7 +92,7 @@ def test_query_signals_end_time_two_signals():
 
 def test_query_signals_start_end_time_two_signals():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=two_signals, start='2024-11-09 22:53:55.00', end='2024-11-09 22:57:41.00')
+    result = query_signals(vehicle_id='gr24-main', signals=two_signals, start='2024-11-09 22:53:55.00', end='2024-11-09 22:57:41.00')
     assert len(result) == 2
     for i in range(len(result)):
         assert isinstance(result[i], pd.DataFrame)
@@ -112,7 +112,7 @@ def test_query_signals_invalid_vehicle_id_two_signals():
 
 def test_query_signals_invalid_range_two_signals():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=two_signals, start='2022-11-09 22:53:55.00', end='2022-11-09 22:57:41.00')
+    result = query_signals(vehicle_id='gr24-main', signals=two_signals, start='2022-11-09 22:53:55.00', end='2022-11-09 22:57:41.00')
     assert len(result) == 2
     for i in range(len(result)):
         assert isinstance(result[i], pd.DataFrame)
@@ -127,7 +127,7 @@ def test_query_signals_invalid_range_two_signals():
 
 def test_query_signals_five_signals():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=five_signals)
+    result = query_signals(vehicle_id='gr24-main', signals=five_signals)
     assert len(result) == 5
     for i in range(len(result)):
         assert isinstance(result[i], pd.DataFrame)
@@ -141,7 +141,7 @@ def test_query_signals_five_signals():
 
 def test_query_signals_start_time_five_signals():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=five_signals, start='2024-11-09 22:53:55.00')
+    result = query_signals(vehicle_id='gr24-main', signals=five_signals, start='2024-11-09 22:53:55.00')
     assert len(result) == 5
     for i in range(len(result)):
         assert isinstance(result[i], pd.DataFrame)
@@ -154,7 +154,7 @@ def test_query_signals_start_time_five_signals():
 
 def test_query_signals_end_time_five_signals():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=five_signals, end='2024-11-09 22:57:41.00')
+    result = query_signals(vehicle_id='gr24-main', signals=five_signals, end='2024-11-09 22:57:41.00')
     assert len(result) == 5
     for i in range(len(result)):
         assert isinstance(result[i], pd.DataFrame)
@@ -167,7 +167,7 @@ def test_query_signals_end_time_five_signals():
 
 def test_query_signals_start_end_time_five_signals():
     init_test_db()
-    result = query_signals(vehicle_id='gr24-test', signals=five_signals, start='2024-11-09 22:53:55.00', end='2024-11-09 22:57:41.00')
+    result = query_signals(vehicle_id='gr24-main', signals=five_signals, start='2024-11-09 22:53:55.00', end='2024-11-09 22:57:41.00')
     assert len(result) == 5
     for i in range(len(result)):
         assert isinstance(result[i], pd.DataFrame)