olemd/skyview
1
0
Fork 0
Code Issues 12 Pull requests Projects Releases 11 Packages Wiki Activity Actions

Add VRS JSON format support for readsb integration

Browse source 
Added comprehensive support for VRS (Virtual Radar Server) JSON format
as a simpler alternative to Beast binary protocol, enabling integration
with readsb --net-vrs-port output.

## Key Features:
- **VRS JSON Parser**: Stream parsing of newline-delimited JSON aircraft data
- **VRS Client**: TCP client with automatic reconnection and error recovery
- **Mixed Format Support**: Use Beast and VRS sources simultaneously
- **Enhanced Aircraft Data**: Added VRS-specific fields (registration, type, operator)
- **Position Source Tracking**: Identifies ADS-B, MLAT, TIS-B, and satellite positions

## Implementation:
- `internal/vrs/parser.go`: VRS JSON message parsing and validation
- `internal/client/vrs.go`: VRS TCP client implementation
- Enhanced `MultiSourceClient` to support both Beast and VRS formats
- Extended `Aircraft` struct with validity flags and additional metadata
- Updated configuration to include `format` field ("beast" or "vrs")

## Testing:
- Successfully tested against svovel:33005 VRS JSON stream
- Verified aircraft data parsing and position tracking
- Confirmed mixed-format operation with existing Beast clients

## Documentation:
- Updated README.md with VRS format configuration examples
- Enhanced ARCHITECTURE.md with VRS parser documentation
- Added data format comparison and configuration guide

This enables simpler integration with modern readsb installations while
maintaining full backward compatibility with existing Beast deployments.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Ole-Morten Duesund 2025-08-31 11:02:27 +02:00
commit 073acb7304
10 changed files with 903 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

99
internal/client/beast.go
View file

@ -250,10 +250,10 @@ func (c *BeastClient) processMessages() {
}
}
// MultiSourceClient manages multiple Beast TCP clients for multi-receiver setups.
// MultiSourceClient manages multiple Beast and VRS clients for multi-receiver setups.
//
// This client coordinator:
// - Manages connections to multiple Beast format sources simultaneously
// - Manages connections to multiple Beast and VRS format sources simultaneously
// - Provides unified control for starting and stopping all clients
// - Runs periodic cleanup tasks for stale aircraft data
// - Aggregates statistics from all managed clients
@ -262,21 +262,23 @@ func (c *BeastClient) processMessages() {
// All clients share the same data merger, enabling automatic data fusion
// and conflict resolution across multiple receivers.
type MultiSourceClient struct {
clients []*BeastClient // Managed Beast clients
merger *merger.Merger // Shared data merger for all sources
mu sync.RWMutex // Protects clients slice
beastClients []*BeastClient // Managed Beast clients
vrsClients []*VRSClient // Managed VRS JSON clients
merger *merger.Merger // Shared data merger for all sources
mu sync.RWMutex // Protects client slices
}
// NewMultiSourceClient creates a client manager for multiple Beast format sources.
// NewMultiSourceClient creates a client manager for multiple Beast and VRS format sources.
//
// The multi-source client enables connecting to multiple dump1090 instances
// or other Beast format sources simultaneously. All sources feed into the
// same data merger, which handles automatic data fusion and conflict resolution.
// The multi-source client enables connecting to multiple dump1090/readsb instances
// using either Beast binary or VRS JSON formats simultaneously. All sources feed into
// the same data merger, which handles automatic data fusion and conflict resolution.
//
// This is essential for:
// - Improved coverage from multiple receivers
// - Redundancy in case of individual receiver failures
// - Data quality improvement through signal strength comparison
// - Support for different receiver output formats
//
// Parameters:
// - merger: Shared data merger instance for all sources
@ -284,18 +286,23 @@ type MultiSourceClient struct {
// Returns a configured multi-source client ready for source addition.
func NewMultiSourceClient(merger *merger.Merger) *MultiSourceClient {
return &MultiSourceClient{
clients: make([]*BeastClient, 0),
merger: merger,
beastClients: make([]*BeastClient, 0),
vrsClients: make([]*VRSClient, 0),
merger: merger,
}
}
// AddSource registers and configures a new Beast format data source.
// AddSource registers and configures a new data source (Beast or VRS format).
//
// This method:
// 1. Registers the source with the data merger
// 2. Creates a new BeastClient for the source
// 2. Creates appropriate client based on source format
// 3. Adds the client to the managed clients list
//
// The source format is determined by the source.Format field:
// - "beast": Creates a BeastClient for Beast binary protocol (default)
// - "vrs": Creates a VRSClient for VRS JSON protocol
//
// The source is not automatically started; call Start() to begin connections.
// Sources can be added before or after starting the multi-source client.
//
@ -305,18 +312,31 @@ func (m *MultiSourceClient) AddSource(source *merger.Source) {
m.mu.Lock()
defer m.mu.Unlock()
// Default to Beast format if not specified
if source.Format == "" {
source.Format = "beast"
}
// Register source with merger
m.merger.AddSource(source)
// Create and start client
client := NewBeastClient(source, m.merger)
m.clients = append(m.clients, client)
// Create appropriate client based on format
switch source.Format {
case "vrs":
client := NewVRSClient(source, m.merger)
m.vrsClients = append(m.vrsClients, client)
case "beast":
fallthrough
default:
client := NewBeastClient(source, m.merger)
m.beastClients = append(m.beastClients, client)
}
}
// Start begins connections to all configured Beast sources.
// Start begins connections to all configured sources (Beast and VRS).
//
// This method:
// - Starts all managed BeastClient instances in parallel
// - Starts all managed BeastClient and VRSClient instances in parallel
// - Begins the periodic cleanup routine for stale aircraft data
// - Uses the provided context for cancellation control
//
@ -330,7 +350,13 @@ func (m *MultiSourceClient) Start(ctx context.Context) {
m.mu.RLock()
defer m.mu.RUnlock()
for _, client := range m.clients {
// Start Beast clients
for _, client := range m.beastClients {
client.Start(ctx)
}
// Start VRS clients
for _, client := range m.vrsClients {
client.Start(ctx)
}
@ -338,7 +364,7 @@ func (m *MultiSourceClient) Start(ctx context.Context) {
go m.cleanupRoutine(ctx)
}
// Stop gracefully shuts down all managed Beast clients.
// Stop gracefully shuts down all managed clients (Beast and VRS).
//
// This method stops all clients in parallel and waits for their
// goroutines to complete. The shutdown is coordinated to ensure
@ -347,7 +373,13 @@ func (m *MultiSourceClient) Stop() {
m.mu.RLock()
defer m.mu.RUnlock()
for _, client := range m.clients {
// Stop Beast clients
for _, client := range m.beastClients {
client.Stop()
}
// Stop VRS clients
for _, client := range m.vrsClients {
client.Stop()
}
}
@ -382,8 +414,8 @@ func (m *MultiSourceClient) cleanupRoutine(ctx context.Context) {
//
// The statistics include:
// - All merger statistics (aircraft count, message rates, etc.)
// - Number of active client connections
// - Total number of configured clients
// - Number of active client connections (Beast and VRS)
// - Total number of configured clients by type
// - Per-source connection status and message counts
//
// This information is useful for monitoring system health, diagnosing
@ -397,15 +429,26 @@ func (m *MultiSourceClient) GetStatistics() map[string]interface{} {
stats := m.merger.GetStatistics()
// Add client-specific stats
activeClients := 0
for _, client := range m.clients {
activeBeastClients := 0
for _, client := range m.beastClients {
if client.source.Active {
activeClients++
activeBeastClients++
}
}
stats["active_clients"] = activeClients
stats["total_clients"] = len(m.clients)
activeVRSClients := 0
for _, client := range m.vrsClients {
if client.source.Active {
activeVRSClients++
}
}
stats["active_beast_clients"] = activeBeastClients
stats["active_vrs_clients"] = activeVRSClients
stats["active_clients"] = activeBeastClients + activeVRSClients
stats["total_beast_clients"] = len(m.beastClients)
stats["total_vrs_clients"] = len(m.vrsClients)
stats["total_clients"] = len(m.beastClients) + len(m.vrsClients)
return stats
}

350
internal/client/vrs.go Normal file
View file

@ -0,0 +1,350 @@
// Package client provides VRS JSON format client implementation for connecting to readsb receivers.
//
// This file implements a VRS (Virtual Radar Server) JSON format client that connects
// to readsb instances with --net-vrs-port enabled. VRS JSON is a simpler alternative
// to the Beast binary protocol, providing aircraft data as JSON over TCP.
package client
import (
"context"
"fmt"
"net"
"sync"
"time"
"skyview/internal/merger"
"skyview/internal/modes"
"skyview/internal/vrs"
)
// VRSClient handles connection to a single readsb VRS JSON TCP stream
//
// The VRS client provides robust connectivity with:
// - Automatic reconnection with exponential backoff
// - JSON message parsing and processing
// - Integration with data merger
// - Source status tracking and statistics
// - Graceful shutdown handling
//
// VRS JSON is simpler than Beast format but provides the same aircraft data
type VRSClient struct {
source *merger.Source // Source configuration and status
merger *merger.Merger // Data merger for multi-source fusion
conn net.Conn // TCP connection to VRS source
parser *vrs.Parser // VRS JSON format parser
msgChan chan *vrs.VRSMessage // Buffered channel for parsed messages
errChan chan error // Error reporting channel
stopChan chan struct{} // Shutdown signal channel
wg sync.WaitGroup // Wait group for goroutine coordination
// Reconnection parameters
reconnectDelay time.Duration // Initial reconnect delay
maxReconnect time.Duration // Maximum reconnect delay (for backoff cap)
}
// NewVRSClient creates a new VRS JSON format TCP client for a specific data source
//
// The client is configured with:
// - Buffered message channel (100 messages) for handling updates
// - Error channel for connection and parsing issues
// - Initial reconnect delay of 5 seconds
// - Maximum reconnect delay of 60 seconds (exponential backoff cap)
//
// Parameters:
// - source: Source configuration including host, port, and metadata
// - merger: Data merger instance for aircraft state management
//
// Returns a configured but not yet started VRSClient
func NewVRSClient(source *merger.Source, merger *merger.Merger) *VRSClient {
return &VRSClient{
source: source,
merger: merger,
msgChan: make(chan *vrs.VRSMessage, 100),
errChan: make(chan error, 10),
stopChan: make(chan struct{}),
reconnectDelay: 5 * time.Second,
maxReconnect: 60 * time.Second,
}
}
// Start begins the client connection and message processing in the background
//
// The client will:
// - Attempt to connect to the configured VRS source
// - Handle connection failures with exponential backoff
// - Start message reading and processing goroutines
// - Continuously reconnect if the connection is lost
//
// The method returns immediately; the client runs in background goroutines
// until Stop() is called or the context is cancelled
//
// Parameters:
// - ctx: Context for cancellation and timeout control
func (c *VRSClient) Start(ctx context.Context) {
c.wg.Add(1)
go c.run(ctx)
}
// Stop gracefully shuts down the client and all associated goroutines
//
// The shutdown process:
// 1. Signals all goroutines to stop via stopChan
// 2. Closes the TCP connection if active
// 3. Waits for all goroutines to complete
//
// This method blocks until the shutdown is complete
func (c *VRSClient) Stop() {
close(c.stopChan)
if c.conn != nil {
c.conn.Close()
}
c.wg.Wait()
}
// run implements the main client connection and reconnection loop
//
// This method handles the complete client lifecycle:
// 1. Connection establishment with timeout
// 2. Exponential backoff on connection failures
// 3. Message parsing and processing goroutine management
// 4. Connection monitoring and failure detection
// 5. Automatic reconnection on disconnection
//
// The exponential backoff starts at reconnectDelay (5s) and doubles on each
// failure up to maxReconnect (60s), then resets on successful connection
//
// Source status is updated to reflect connection state for monitoring
func (c *VRSClient) run(ctx context.Context) {
defer c.wg.Done()
reconnectDelay := c.reconnectDelay
for {
select {
case <-ctx.Done():
return
case <-c.stopChan:
return
default:
}
// Connect to VRS JSON stream
addr := fmt.Sprintf("%s:%d", c.source.Host, c.source.Port)
fmt.Printf("Connecting to VRS JSON stream at %s (%s)...\n", addr, c.source.Name)
conn, err := net.DialTimeout("tcp", addr, 30*time.Second)
if err != nil {
fmt.Printf("Failed to connect to VRS source %s: %v\n", c.source.Name, err)
c.source.Active = false
// Exponential backoff
time.Sleep(reconnectDelay)
if reconnectDelay < c.maxReconnect {
reconnectDelay *= 2
}
continue
}
c.conn = conn
c.source.Active = true
reconnectDelay = c.reconnectDelay // Reset backoff
fmt.Printf("Connected to VRS source %s at %s\n", c.source.Name, addr)
// Create parser for this connection
c.parser = vrs.NewParser(conn, c.source.ID)
// Start processing messages
c.wg.Add(2)
go c.readMessages()
go c.processMessages()
// Wait for disconnect
select {
case <-ctx.Done():
c.conn.Close()
return
case <-c.stopChan:
c.conn.Close()
return
case err := <-c.errChan:
fmt.Printf("Error from VRS source %s: %v\n", c.source.Name, err)
c.conn.Close()
c.source.Active = false
}
// Wait for goroutines to finish
time.Sleep(1 * time.Second)
}
}
// readMessages runs in a dedicated goroutine to read VRS JSON messages
//
// This method:
// - Continuously reads from the TCP connection
// - Parses VRS JSON data into Message structs
// - Queues parsed messages for processing
// - Reports parsing errors to the error channel
//
// The method blocks on the parser's ParseStream call and exits when
// the connection is closed or an unrecoverable error occurs
func (c *VRSClient) readMessages() {
defer c.wg.Done()
c.parser.ParseStream(c.msgChan, c.errChan)
}
// processMessages runs in a dedicated goroutine to process aircraft data
//
// For each received VRS message, this method:
// 1. Iterates through all aircraft in the message
// 2. Converts VRS format to internal aircraft representation
// 3. Updates the data merger with new aircraft state
// 4. Updates source statistics (message count)
//
// Invalid or unparseable aircraft data is silently discarded to maintain
// system stability. The merger handles data fusion from multiple sources
// and conflict resolution based on signal strength
func (c *VRSClient) processMessages() {
defer c.wg.Done()
for {
select {
case <-c.stopChan:
return
case msg := <-c.msgChan:
if msg == nil {
return
}
// Process each aircraft in the message
for _, vrsAircraft := range msg.AcList {
// Convert VRS aircraft to internal format
aircraft := c.convertVRSToAircraft(&vrsAircraft)
if aircraft == nil {
continue // Skip invalid aircraft
}
// Update merger with new data
// Note: VRS doesn't provide signal strength, so we use a default value
c.merger.UpdateAircraft(
c.source.ID,
aircraft,
-30.0, // Default signal strength for VRS sources
vrsAircraft.GetTimestamp(),
)
// Update source statistics
c.source.Messages++
}
}
}
}
// convertVRSToAircraft converts VRS JSON aircraft to internal aircraft format
//
// This method maps VRS JSON fields to the internal aircraft structure used
// by the merger. It handles:
// - ICAO address extraction and validation
// - Position data (lat/lon)
// - Altitude (barometric and geometric)
// - Speed and heading
// - Callsign and squawk code
// - Ground status
//
// Returns nil if the aircraft data is invalid or incomplete
func (c *VRSClient) convertVRSToAircraft(vrs *vrs.VRSAircraft) *modes.Aircraft {
// Get ICAO address
icao, err := vrs.GetICAO24()
if err != nil {
return nil // Invalid ICAO, skip this aircraft
}
// Create aircraft structure
aircraft := &modes.Aircraft{
ICAO24: icao,
}
// Set position if available
if vrs.HasPosition() {
aircraft.Latitude = vrs.Lat
aircraft.Longitude = vrs.Long
aircraft.PositionValid = true
}
// Set altitude if available
if vrs.HasAltitude() {
aircraft.Altitude = vrs.GetAltitude()
aircraft.AltitudeValid = true
// Set barometric altitude specifically
if vrs.Alt != 0 {
aircraft.BaroAltitude = vrs.Alt
}
// Set geometric altitude if different from barometric
if vrs.GAlt != 0 {
aircraft.GeomAltitude = vrs.GAlt
aircraft.GeomAltitudeValid = true
}
}
// Set speed if available
if vrs.Spd > 0 {
aircraft.GroundSpeed = int(vrs.Spd)
aircraft.GroundSpeedValid = true
}
// Set heading/track if available
if vrs.Trak > 0 {
aircraft.Track = int(vrs.Trak)
aircraft.TrackValid = true
}
// Set vertical rate if available
if vrs.Vsi != 0 {
aircraft.VerticalRate = vrs.Vsi
aircraft.VerticalRateValid = true
}
// Set callsign if available
if vrs.Call != "" {
aircraft.Callsign = vrs.Call
aircraft.CallsignValid = true
}
// Set squawk if available
if squawk, err := vrs.GetSquawk(); err == nil {
aircraft.Squawk = fmt.Sprintf("%04X", squawk) // Convert to hex string
aircraft.SquawkValid = true
}
// Set ground status
aircraft.OnGround = vrs.IsOnGround()
aircraft.OnGroundValid = true
// Set selected altitude if available
if vrs.TAlt != 0 {
aircraft.SelectedAltitude = vrs.TAlt
}
// Set position source
switch vrs.GetPositionSource() {
case "MLAT":
aircraft.PositionMLAT = true
case "TIS-B":
aircraft.PositionTISB = true
}
// Set additional metadata if available
if vrs.Reg != "" {
aircraft.Registration = vrs.Reg
}
if vrs.Type != "" {
aircraft.AircraftType = vrs.Type
}
if vrs.Op != "" {
aircraft.Operator = vrs.Op
}
return aircraft
}