mirror of
https://github.com/IS1DI/x0gp.git
synced 2026-07-16 20:47:55 +00:00
feat(server): integrate UDP video service and normalize database schema
- Implement UDPVideoService to stream MJPEG video frames from ESP32s via UDP and send control commands. - Normalize race database schema: - Remove redundant driver_ids array from races table (migration 011). - Move per-driver outcomes (total_time_ms, best_lap_ms, position) to race_drivers table (migration 012). - Add device_id column to lobby_drivers to link active physical devices (migration 013). - Update WebSocket handler to pass lobbySvc, driversSvc, and videoSvc for driver identity tracking, syncing in-memory driver metadata, and forwarding WS controls to physical cars. - Add CORS middleware to HTTP routes. - Regenerate Swagger documentation.
This commit is contained in:
x0gp
@@ -0,0 +1,381 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type deviceState struct {
|
||||
addr *net.UDPAddr
|
||||
latestFrame []byte
|
||||
currentFrame []byte
|
||||
lastSeen time.Time
|
||||
}
|
||||
|
||||
// UDPVideoService handles UDP video frame collection and command forwarding.
|
||||
type UDPVideoService struct {
|
||||
logger *slog.Logger
|
||||
listenAddr string
|
||||
espCmdPort int
|
||||
|
||||
mu sync.RWMutex
|
||||
devices map[uint8]*deviceState
|
||||
|
||||
clientsMu sync.RWMutex
|
||||
clients map[uint8]map[chan []byte]struct{}
|
||||
|
||||
conn *net.UDPConn
|
||||
}
|
||||
|
||||
// NewUDPVideoService initializes a new UDPVideoService.
|
||||
func NewUDPVideoService(logger *slog.Logger, listenAddr string, espCmdPort int) *UDPVideoService {
|
||||
return &UDPVideoService{
|
||||
logger: logger,
|
||||
listenAddr: listenAddr,
|
||||
espCmdPort: espCmdPort,
|
||||
devices: make(map[uint8]*deviceState),
|
||||
clients: make(map[uint8]map[chan []byte]struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// Start runs the UDP receiver loop (gracefully shut down by ctx).
|
||||
// Stdin command reader runs in a separate un-tracked goroutine so it doesn't block shutdown.
|
||||
func (s *UDPVideoService) Start(ctx context.Context, wg *sync.WaitGroup) {
|
||||
addr, err := net.ResolveUDPAddr("udp", s.listenAddr)
|
||||
if err != nil {
|
||||
s.logger.Error("failed to resolve UDP address", "listenAddr", s.listenAddr, "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
conn, err := net.ListenUDP("udp", addr)
|
||||
if err != nil {
|
||||
s.logger.Error("failed to listen UDP", "listenAddr", s.listenAddr, "err", err)
|
||||
return
|
||||
}
|
||||
s.conn = conn
|
||||
|
||||
s.logger.Info("UDP video server started. Listening for video frames...", "listenAddr", s.listenAddr)
|
||||
|
||||
// Goroutine for handling UDP receive loop
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer conn.Close()
|
||||
|
||||
buf := make([]byte, 65535)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
s.logger.Info("stopping UDP video receiver loop")
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
// Use a read deadline to check for context cancellation periodically
|
||||
_ = conn.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
|
||||
n, remoteAddr, err := conn.ReadFromUDP(buf)
|
||||
if err != nil {
|
||||
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
|
||||
continue
|
||||
}
|
||||
s.logger.Debug("UDP read error", "err", err)
|
||||
continue
|
||||
}
|
||||
|
||||
if n < 1 {
|
||||
continue
|
||||
}
|
||||
deviceID := buf[0] // first byte is DEVICE_ID (0..127)
|
||||
payload := buf[1:n]
|
||||
|
||||
s.mu.Lock()
|
||||
dev, ok := s.devices[deviceID]
|
||||
if !ok {
|
||||
dev = &deviceState{
|
||||
addr: remoteAddr,
|
||||
lastSeen: time.Now(),
|
||||
}
|
||||
s.devices[deviceID] = dev
|
||||
s.logger.Info("ESP32 video source discovered", "device_id", deviceID, "ip", remoteAddr.IP.String())
|
||||
} else {
|
||||
if dev.addr == nil || dev.addr.IP.String() != remoteAddr.IP.String() {
|
||||
dev.addr = remoteAddr
|
||||
s.logger.Info("ESP32 video source IP updated", "device_id", deviceID, "ip", remoteAddr.IP.String())
|
||||
}
|
||||
dev.lastSeen = time.Now()
|
||||
}
|
||||
|
||||
// Reassemble JPEG frame for this device
|
||||
if len(payload) >= 2 && payload[0] == 0xFF && payload[1] == 0xD8 {
|
||||
dev.currentFrame = make([]byte, 0, 65535)
|
||||
}
|
||||
dev.currentFrame = append(dev.currentFrame, payload...)
|
||||
|
||||
if len(payload) >= 2 && payload[len(payload)-2] == 0xFF && payload[len(payload)-1] == 0xD9 {
|
||||
// Frame fully assembled
|
||||
s.broadcastFrame(deviceID, dev.currentFrame)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
}
|
||||
}()
|
||||
|
||||
// Stdin command reader (run in background, not tracked by wg to avoid blocking shutdown)
|
||||
go func() {
|
||||
reader := bufio.NewReader(os.Stdin)
|
||||
s.logger.Info("Console command reader active. Enter commands (e.g., '1 left', '1 start', or 'left') in console:")
|
||||
|
||||
for {
|
||||
text, err := reader.ReadString('\n')
|
||||
if err != nil {
|
||||
// Stdin closed or EOF (e.g. non-interactive/background mode)
|
||||
s.logger.Debug("stdin closed, stopping console command reader", "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
text = strings.TrimSpace(text)
|
||||
if text == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
// Format: "ID command" (e.g. "1 left") or "command" (defaults to the first discovered/available device)
|
||||
parts := strings.SplitN(text, " ", 2)
|
||||
var deviceID uint8
|
||||
var cmd string
|
||||
hasDevice := false
|
||||
|
||||
if len(parts) == 2 {
|
||||
if idVal, err := strconv.Atoi(parts[0]); err == nil && idVal >= 0 && idVal <= 127 {
|
||||
deviceID = uint8(idVal)
|
||||
cmd = parts[1]
|
||||
hasDevice = true
|
||||
}
|
||||
}
|
||||
|
||||
if !hasDevice {
|
||||
cmd = text
|
||||
// Find the first discovered/active device
|
||||
s.mu.RLock()
|
||||
for id := range s.devices {
|
||||
deviceID = id
|
||||
hasDevice = true
|
||||
break
|
||||
}
|
||||
s.mu.RUnlock()
|
||||
}
|
||||
|
||||
if !hasDevice {
|
||||
s.logger.Warn("cannot send command: no ESP32 devices discovered yet")
|
||||
continue
|
||||
}
|
||||
|
||||
err = s.SendCommand(deviceID, cmd)
|
||||
if err != nil {
|
||||
s.logger.Error("failed to send command to ESP32", "device_id", deviceID, "cmd", cmd, "err", err)
|
||||
} else {
|
||||
s.logger.Info("command sent to ESP32", "device_id", deviceID, "cmd", cmd)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// SendCommand sends a command string to the ESP32 representing deviceID.
|
||||
func (s *UDPVideoService) SendCommand(deviceID uint8, cmd string) error {
|
||||
s.mu.RLock()
|
||||
dev, ok := s.devices[deviceID]
|
||||
conn := s.conn
|
||||
s.mu.RUnlock()
|
||||
|
||||
if !ok || dev.addr == nil {
|
||||
return fmt.Errorf("ESP32 device %d address unknown (no video packets received yet)", deviceID)
|
||||
}
|
||||
if conn == nil {
|
||||
return fmt.Errorf("UDP service not started")
|
||||
}
|
||||
|
||||
cmdAddr := &net.UDPAddr{
|
||||
IP: dev.addr.IP,
|
||||
Port: s.espCmdPort,
|
||||
}
|
||||
|
||||
_, err := conn.WriteToUDP([]byte(cmd), cmdAddr)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *UDPVideoService) broadcastFrame(deviceID uint8, frame []byte) {
|
||||
// dev.latestFrame is updated inside the caller's lock (Start loop) to avoid races
|
||||
if dev, ok := s.devices[deviceID]; ok {
|
||||
dev.latestFrame = make([]byte, len(frame))
|
||||
copy(dev.latestFrame, frame)
|
||||
}
|
||||
|
||||
s.clientsMu.RLock()
|
||||
defer s.clientsMu.RUnlock()
|
||||
deviceClients, ok := s.clients[deviceID]
|
||||
if ok {
|
||||
for ch := range deviceClients {
|
||||
select {
|
||||
case ch <- frame:
|
||||
default:
|
||||
// Drop frame if client is slow to read
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *UDPVideoService) registerClient(deviceID uint8, ch chan []byte) {
|
||||
s.clientsMu.Lock()
|
||||
defer s.clientsMu.Unlock()
|
||||
if s.clients[deviceID] == nil {
|
||||
s.clients[deviceID] = make(map[chan []byte]struct{})
|
||||
}
|
||||
s.clients[deviceID][ch] = struct{}{}
|
||||
}
|
||||
|
||||
func (s *UDPVideoService) unregisterClient(deviceID uint8, ch chan []byte) {
|
||||
s.clientsMu.Lock()
|
||||
defer s.clientsMu.Unlock()
|
||||
if s.clients[deviceID] != nil {
|
||||
delete(s.clients[deviceID], ch)
|
||||
if len(s.clients[deviceID]) == 0 {
|
||||
delete(s.clients, deviceID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// StreamHandler godoc
|
||||
// @Summary Get MJPEG video stream from ESP32
|
||||
// @Description Streams video frames received via UDP as multipart/x-mixed-replace. Suitable for raw img tags in browser.
|
||||
// @Tags video
|
||||
// @Produce multipart/x-mixed-replace
|
||||
// @Param id query int false "Device ID (0..127)"
|
||||
// @Success 200 {string} string "MJPEG stream"
|
||||
// @Router /api/video/stream [get]
|
||||
func (s *UDPVideoService) StreamHandler() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
idVal := r.URL.Query().Get("id")
|
||||
deviceID := uint8(1) // default device ID is 1 as per client.ino
|
||||
if idVal != "" {
|
||||
if idInt, err := strconv.Atoi(idVal); err == nil && idInt >= 0 && idInt <= 127 {
|
||||
deviceID = uint8(idInt)
|
||||
} else {
|
||||
http.Error(w, "invalid device id", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "multipart/x-mixed-replace; boundary=frame")
|
||||
w.Header().Set("Cache-Control", "no-cache, private, max-age=0, no-store, must-revalidate")
|
||||
w.Header().Set("Connection", "keep-alive")
|
||||
w.Header().Set("Pragma", "no-cache")
|
||||
|
||||
frameCh := make(chan []byte, 10)
|
||||
s.registerClient(deviceID, frameCh)
|
||||
defer s.unregisterClient(deviceID, frameCh)
|
||||
|
||||
// Send the latest frame immediately if we have one
|
||||
s.mu.RLock()
|
||||
var initFrame []byte
|
||||
if dev, ok := s.devices[deviceID]; ok {
|
||||
initFrame = dev.latestFrame
|
||||
}
|
||||
s.mu.RUnlock()
|
||||
if len(initFrame) > 0 {
|
||||
_, _ = fmt.Fprintf(w, "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: %d\r\n\r\n", len(initFrame))
|
||||
_, _ = w.Write(initFrame)
|
||||
_, _ = w.Write([]byte("\r\n"))
|
||||
if flusher, ok := w.(http.Flusher); ok {
|
||||
flusher.Flush()
|
||||
}
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-r.Context().Done():
|
||||
return
|
||||
case frame, ok := <-frameCh:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
_, err := fmt.Fprintf(w, "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: %d\r\n\r\n", len(frame))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
_, err = w.Write(frame)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
_, err = w.Write([]byte("\r\n"))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if flusher, ok := w.(http.Flusher); ok {
|
||||
flusher.Flush()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ControlHandler godoc
|
||||
// @Summary Send command to ESP32
|
||||
// @Description Sends a text command (e.g. start, stop, left, right) to the ESP32 command port (UDP 8888).
|
||||
// @Tags video
|
||||
// @Param id query int false "Device ID (0..127)"
|
||||
// @Param cmd query string true "Command text"
|
||||
// @Produce json
|
||||
// @Success 200 {object} map[string]interface{} "Command sent"
|
||||
// @Failure 400 {string} string "Missing cmd parameter"
|
||||
// @Failure 500 {string} string "Internal error"
|
||||
// @Router /api/video/control [post]
|
||||
func (s *UDPVideoService) ControlHandler() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "Only POST is allowed")
|
||||
return
|
||||
}
|
||||
cmd := r.URL.Query().Get("cmd")
|
||||
if cmd == "" {
|
||||
cmd = r.FormValue("cmd")
|
||||
}
|
||||
if cmd == "" {
|
||||
writeError(w, http.StatusBadRequest, "bad_request", "Missing 'cmd' parameter")
|
||||
return
|
||||
}
|
||||
|
||||
idVal := r.URL.Query().Get("id")
|
||||
if idVal == "" {
|
||||
idVal = r.FormValue("id")
|
||||
}
|
||||
deviceID := uint8(1)
|
||||
if idVal != "" {
|
||||
if idInt, err := strconv.Atoi(idVal); err == nil && idInt >= 0 && idInt <= 127 {
|
||||
deviceID = uint8(idInt)
|
||||
} else {
|
||||
writeError(w, http.StatusBadRequest, "bad_request", "invalid device id")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
err := s.SendCommand(deviceID, cmd)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, "internal_error", err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"status": "success",
|
||||
"device_id": deviceID,
|
||||
"command": cmd,
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user